]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/LeiMirror.pm
lei_mirror: break out of fgrp fetch iteration early
[public-inbox.git] / lib / PublicInbox / LeiMirror.pm
index 2812b69622deab06a09a126fd21aff80657593b4..0508c9a8c4aad99ca3cf2b8adddd92cb73a1c1cb 100644 (file)
@@ -19,8 +19,11 @@ use PublicInbox::Inbox;
 use PublicInbox::LeiCurl;
 use PublicInbox::OnDestroy;
 use Digest::SHA qw(sha256_hex sha1_hex);
+use POSIX qw(strftime);
 
 our $LIVE; # pid => callback
+our $FGRP_TODO; # objstore -> [ fgrp mirror objects ]
+our $TODO; # reference => [ non-fgrp mirror objects ]
 
 sub keep_going ($) {
        $LIVE && (!$_[0]->{lei}->{child_error} ||
@@ -49,7 +52,7 @@ sub try_scrape {
        my $uri = URI->new($self->{src});
        my $lei = $self->{lei};
        my $curl = $self->{curl} //= PublicInbox::LeiCurl->new($lei) or return;
-       my $cmd = $curl->for_uri($lei, $uri, qw(-f --compressed));
+       my $cmd = $curl->for_uri($lei, $uri, '--compressed');
        my $opt = { 0 => $lei->{0}, 2 => $lei->{2} };
        my $fh = popen_rd($cmd, undef, $opt);
        my $html = do { local $/; <$fh> } // die "read(curl $uri): $!";
@@ -149,7 +152,7 @@ sub _get_txt_start { # non-fatal
        my $f = (split(m!/!, $endpoint))[-1];
        my $ft = File::Temp->new(TEMPLATE => "$f-XXXX", TMPDIR => 1);
        my $opt = { 0 => $lei->{0}, 1 => $lei->{1}, 2 => $lei->{2} };
-       my $cmd = $self->{curl}->for_uri($lei, $uri, qw(-f --compressed -R -o),
+       my $cmd = $self->{curl}->for_uri($lei, $uri, qw(--compressed -R -o),
                                        $ft->filename);
        do_reap($self);
        $lei->qerr("# @$cmd");
@@ -178,13 +181,16 @@ sub _write_inbox_config {
        my $buf = delete($self->{'txt._/text/config/raw'}) // return;
        my $dst = $self->{cur_dst} // $self->{dst};
        my $f = "$dst/inbox.config.example";
-       open my $fh, '>', $f or die "open($f): $!";
-       print $fh $buf or die "print: $!";
-       chmod(0444 & ~umask, $fh) or die "chmod($f): $!";
        my $mtime = delete $self->{'mtime._/text/config/raw'};
-       $fh->flush or die "flush($f): $!";
-       if (defined $mtime) {
-               utime($mtime, $mtime, $fh) or die "utime($f): $!";
+       if (sysopen(my $fh, $f, O_CREAT|O_EXCL|O_WRONLY)) {
+               print $fh $buf or die "print: $!";
+               chmod(0444 & ~umask, $fh) or die "chmod($f): $!";
+               $fh->flush or die "flush($f): $!";
+               if (defined $mtime) {
+                       utime($mtime, $mtime, $fh) or die "utime($f): $!";
+               }
+       } elsif (!$!{EEXIST}) {
+               die "open($f): $!";
        }
        my $cfg = PublicInbox::Config->git_config_dump($f, $self->{lei}->{2});
        my $ibx = $self->{ibx} = {};
@@ -198,22 +204,14 @@ sub _write_inbox_config {
 sub set_description ($) {
        my ($self) = @_;
        my $dst = $self->{cur_dst} // $self->{dst};
-       my $f = "$dst/description";
-       open my $fh, '+>>', $f or die "open($f): $!";
-       seek($fh, 0, SEEK_SET) or die "seek($f): $!";
-       my $d = do { local $/; <$fh> } // die "read($f): $!";
-       chomp(my $orig = $d);
+       chomp(my $orig = PublicInbox::Git::try_cat("$dst/description"));
+       my $d = $orig;
        while (defined($d) && ($d =~ m!^\(\$INBOX_DIR/description missing\)! ||
                        $d =~ /^Unnamed repository/ || $d !~ /\S/)) {
                $d = delete($self->{'txt.description'});
        }
        $d //= 'mirror of '.($self->{cur_src} // $self->{src});
-       chomp $d;
-       return if $d eq $orig;
-       seek($fh, 0, SEEK_SET) or die "seek($f): $!";
-       truncate($fh, 0) or die "truncate($f): $!";
-       print $fh $d, "\n" or die "print($f): $!";
-       close $fh or die "close($f): $!";
+       atomic_write($dst, 'description', $d."\n") if $d ne $orig;
 }
 
 sub index_cloned_inbox {
@@ -361,7 +359,8 @@ sub fgrpv_done {
 
 sub fgrp_fetch_all {
        my ($self) = @_;
-       my $todo = delete $self->{fgrp_todo} or return;
+       my $todo = $FGRP_TODO;
+       $FGRP_TODO = \'BUG on further use';
        keys(%$todo) or return;
 
        # Rely on the fgrptmp remote groups in the config file rather
@@ -380,6 +379,7 @@ sub fgrp_fetch_all {
        push(@fetch, "-j$j") if $j;
        while (my ($osdir, $fgrpv) = each %$todo) {
                my $f = "$osdir/config";
+               return if !keep_going($self);
 
                # clobber group from previous run atomically
                my $cmd = ['git', "--git-dir=$osdir", qw(config -f),
@@ -459,43 +459,36 @@ EOM
 }
 
 sub fp_done {
-       my ($self, $go_fetch) = @_;
+       my ($self, $cb, @arg) = @_;
        return if !keep_going($self);
        my $fh = delete $self->{-show_ref} // die 'BUG: no show-ref output';
        seek($fh, SEEK_SET, 0) or die "seek(show_ref): $!";
        $self->{-ent} // die 'BUG: no -ent';
        my $A = $self->{-ent}->{fingerprint} // die 'BUG: no fingerprint';
        my $B = sha1_hex(do { local $/; <$fh> } // die("read(show_ref): $!"));
-       return if $A ne $B; # $go_fetch->DESTROY fires
-       $go_fetch->cancel;
+       return $cb->($self, @arg) if $A ne $B;
        $self->{lei}->qerr("# $self->{-key} up-to-date");
 }
 
-sub cmp_fp_fetch {
-       my ($self, $go_fetch) = @_;
-       # $go_fetch is either resume_fetch or fgrp_enqueue
-       my $new = $self->{-ent}->{fingerprint} // die 'BUG: no fingerprint';
+sub cmp_fp_do {
+       my ($self, $cb, @arg) = @_;
+       # $cb is either resume_fetch or fgrp_enqueue
+       $self->{-ent} // return $cb->($self, @arg);
+       my $new = $self->{-ent}->{fingerprint} // return $cb->($self, @arg);
        my $key = $self->{-key} // die 'BUG: no -key';
        if (my $cur_ent = $self->{-local_manifest}->{$key}) {
                # runs go_fetch->DESTROY run if eq
-               return $go_fetch->cancel if $cur_ent->{fingerprint} eq $new;
+               return if $cur_ent->{fingerprint} eq $new;
        }
        my $dst = $self->{cur_dst} // $self->{dst};
        my $cmd = ['git', "--git-dir=$dst", 'show-ref'];
        my $opt = { 2 => $self->{lei}->{2} };
        open($opt->{1}, '+>', undef) or die "open(tmp): $!";
        $self->{-show_ref} = $opt->{1};
-       my $done = PublicInbox::OnDestroy->new($$, \&fp_done, $self, $go_fetch);
+       my $done = PublicInbox::OnDestroy->new($$, \&fp_done, $self, $cb, @arg);
        start_cmd($self, $cmd, $opt, $done);
 }
 
-sub resume_fetch_maybe {
-       my ($self, $uri, $fini) = @_;
-       my $go_fetch = PublicInbox::OnDestroy->new($$, \&resume_fetch, @_);
-       cmp_fp_fetch($self, $go_fetch) if $self->{-ent} &&
-                               defined($self->{-ent}->{fingerprint});
-}
-
 sub resume_fetch {
        my ($self, $uri, $fini) = @_;
        return if !keep_going($self);
@@ -515,18 +508,10 @@ sub resume_fetch {
        start_cmd($self, $cmd, $opt, $fini);
 }
 
-sub fgrp_enqueue_maybe {
-       my ($self, $fgrp) = @_;
-       my $enq = PublicInbox::OnDestroy->new($$, \&fgrp_enqueue, $self, $fgrp);
-       cmp_fp_fetch($self, $enq) if $self->{-ent} &&
-                                       defined($self->{-ent}->{fingerprint});
-       # $enq->DESTROY calls fgrp_enqueue otherwise
-}
-
 sub fgrp_enqueue {
-       my ($self, $fgrp) = @_;
-       return if !keep_going($self);
-       my $opt = { 2 => $self->{lei}->{2} };
+       my ($fgrp, $end) = @_; # $end calls fgrp_fetch_all
+       return if !keep_going($fgrp);
+       my $opt = { 2 => $fgrp->{lei}->{2} };
        # --no-tags is required to avoid conflicts
        my $u = $fgrp->{-uri} // die 'BUG: no {-uri}';
        my $rn = $fgrp->{-remote} // die 'BUG: no {-remote}';
@@ -534,17 +519,19 @@ sub fgrp_enqueue {
        for ("url=$u", "fetch=+refs/*:refs/remotes/$rn/*", 'tagopt=--no-tags') {
                my @kv = split(/=/, $_, 2);
                $kv[0] = "remote.$rn.$kv[0]";
-               $self->{dry_run} ? $self->{lei}->qerr("# @cmd @kv") :
+               $fgrp->{dry_run} ? $fgrp->{lei}->qerr("# @cmd @kv") :
                                run_die([@cmd, @kv], undef, $opt);
        }
-       push @{$self->{fgrp_todo}->{$fgrp->{-osdir}}}, $fgrp;
+       push @{$FGRP_TODO->{$fgrp->{-osdir}}}, $fgrp;
 }
 
 sub clone_v1 {
-       my ($self, $nohang) = @_;
+       my ($self, $end) = @_;
        my $lei = $self->{lei};
        my $curl = $self->{curl} //= PublicInbox::LeiCurl->new($lei) or return;
        my $uri = URI->new($self->{cur_src} // $self->{src});
+       my $path = $uri->path;
+       $path =~ s!/*\z!! and $uri->path($path);
        defined($lei->{opt}->{epoch}) and
                die "$uri is a v1 inbox, --epoch is not supported\n";
        $self->{-torsocks} //= $curl->torsocks($lei, $uri) or return;
@@ -553,10 +540,10 @@ sub clone_v1 {
        my $resume = -d $dst;
        if (my $fgrp = forkgroup_prep($self, $uri)) {
                $fgrp->{-fini} = $fini;
-               $resume ? fgrp_enqueue_maybe($self, $fgrp) :
-                               fgrp_enqueue($self, $fgrp);
+               $resume ? cmp_fp_do($fgrp, \&fgrp_enqueue, $end)
+                       : fgrp_enqueue($fgrp, $end);
        } elsif ($resume) {
-               resume_fetch_maybe($self, $uri, $fini);
+               cmp_fp_do($self, \&resume_fetch, $uri, $fini);
        } else { # normal clone
                my $cmd = [ @{$self->{-torsocks}},
                                clone_cmd($lei, my $opt = {}), "$uri", $dst ];
@@ -576,10 +563,10 @@ sub clone_v1 {
 
        my $d = $self->{-ent} ? $self->{-ent}->{description} : undef;
        $self->{'txt.description'} = $d if defined $d;
-       (!defined($d) && !$nohang) and
+       (!defined($d) && !$end) and
                _get_txt_start($self, 'description', $fini);
 
-       $nohang or do_reap($self, 1); # for non-manifest clone
+       $end or do_reap($self, 1); # for non-manifest clone
 }
 
 sub parse_epochs ($$) {
@@ -678,6 +665,15 @@ sub up_fp_done {
        push @{$self->{chg}->{fp_mismatch}}, $self->{-key};
 }
 
+sub atomic_write ($$$) {
+       my ($dn, $bn, $raw) = @_;
+       my $ft = File::Temp->new(DIR => $dn, TEMPLATE => "$bn-XXXX");
+       print $ft $raw or die "print($ft): $!";
+       $ft->flush or die "flush($ft): $!";
+       ft_rename($ft, "$dn/$bn", 0666);
+}
+
+# modifies the to-be-written manifest entry, and sets values from it, too
 sub update_ent {
        my ($self) = @_;
        my $key = $self->{-key} // die 'BUG: no -key';
@@ -692,6 +688,49 @@ sub update_ent {
                my $done = PublicInbox::OnDestroy->new($$, \&up_fp_done, $self);
                start_cmd($self, $cmd, $opt, $done);
        }
+       $new = $self->{-ent}->{head};
+       $cur = $self->{-local_manifest}->{$key}->{head} // "\0";
+       if (defined($new) && $new ne $cur) {
+               # n.b. grokmirror writes raw contents to $dst/HEAD w/o locking
+               my $cmd = [ 'git', "--git-dir=$dst" ];
+               if ($new =~ s/\Aref: //) {
+                       push @$cmd, qw(symbolic-ref HEAD), $new;
+               } elsif ($new =~ /\A[a-f0-9]{40,}\z/) {
+                       push @$cmd, qw(update-ref --no-deref HEAD), $new;
+               } else {
+                       undef $cmd;
+                       warn "W: $key: {head} => `$new' not understood\n";
+               }
+               start_cmd($self, $cmd, { 2 => $self->{lei}->{2} }) if $cmd;
+       }
+       if (my $symlinks = $self->{-ent}->{symlinks}) {
+               my $top = File::Spec->rel2abs($self->{dst});
+               for my $p (@$symlinks) {
+                       my $ln = "$top/$p";
+                       $ln =~ tr!/!/!s;
+                       my (undef, $dn, $bn) = File::Spec->splitpath($ln);
+                       File::Path::mkpath($dn);
+                       my $tgt = "$top/$key";
+                       $tgt = File::Spec->abs2rel($tgt, $dn);
+                       if (lstat($ln)) {
+                               if (-l _) {
+                                       next if readlink($ln) eq $tgt;
+                                       unlink($ln) or die "unlink($ln): $!";
+                               } else {
+                                       push @{$self->{chg}->{badlink}}, $p;
+                               }
+                       }
+                       symlink($tgt, $ln) or die "symlink($tgt, $ln): $!";
+               }
+       }
+       if (defined(my $t = $self->{-ent}->{modified})) {
+               my ($dn, $bn) = ("$dst/info/web", 'last-modified');
+               my $orig = PublicInbox::Git::try_cat("$dn/$bn");
+               $t = strftime('%F %T', gmtime($t))." +0000\n";
+               File::Path::mkpath($dn);
+               atomic_write($dn, $bn, $t) if $orig ne $t;
+       }
+
        $new = $self->{-ent}->{owner} // return;
        $cur = $self->{-local_manifest}->{$key}->{owner} // "\0";
        return if $cur eq $new;
@@ -762,7 +801,6 @@ sub clone_v2_prep ($$;$) {
        my $dst = $self->{cur_dst} // $self->{dst};
        my $want = parse_epochs($lei->{opt}->{epoch}, $v2_epochs);
        my $task = $m ? bless { %$self }, __PACKAGE__ : $self;
-       delete $task->{todo}; # $self->{todo} still exists
        my (@skip, $desc);
        my $fini = PublicInbox::OnDestroy->new($$, \&v2_done, $task);
        for my $nr (sort { $a <=> $b } keys %$v2_epochs) {
@@ -790,7 +828,7 @@ failed to extract epoch number from $src
                        $etask->{cur_dst} = $edst;
                        $etask->{-is_epoch} = $fini;
                        my $ref = $ent->{reference} // '';
-                       push @{$self->{todo}->{$ref}}, $etask;
+                       push @{$TODO->{$ref}}, $etask;
                        $self->{any_want}->{$key} = 1;
                } else { # create a placeholder so users only need to chmod +w
                        init_placeholder($src, $edst, $ent);
@@ -894,19 +932,23 @@ sub multi_inbox ($$$) {
 
 sub clone_all {
        my ($self, $m) = @_;
-       my $todo = delete $self->{todo};
-       my $nodep = delete $todo->{''};
-
-       # do not download unwanted deps
-       my $any_want = delete $self->{any_want};
-       my @unwanted = grep { !$any_want->{$_} } keys %$todo;
-       my @nodep = delete(@$todo{@unwanted});
-       push(@$nodep, @$_) for @nodep;
-
-       # handle no-dependency repos, first
-       for (@$nodep) {
-               clone_v1($_, 1);
-               return if !keep_going($self);
+       my $todo = $TODO;
+       $TODO = \'BUG on further use';
+       my $end = PublicInbox::OnDestroy->new($$, \&fgrp_fetch_all, $self);
+       {
+               my $nodep = delete $todo->{''};
+
+               # do not download unwanted deps
+               my $any_want = delete $self->{any_want};
+               my @unwanted = grep { !$any_want->{$_} } keys %$todo;
+               my @nodep = delete(@$todo{@unwanted});
+               push(@$nodep, @$_) for @nodep;
+
+               # handle no-dependency repos, first
+               for (@$nodep) {
+                       clone_v1($_, $end);
+                       return if !keep_going($self);
+               }
        }
        # resolve references, deepest, first:
        while (scalar keys %$todo) {
@@ -922,14 +964,16 @@ EOM
                        }
                        my $y = delete $todo->{$x} // next; # already done
                        for (@$y) {
-                               clone_v1($_, 1);
+                               clone_v1($_, $end);
                                return if !keep_going($self);
                        }
                        last; # restart %$todo iteration
                }
        }
-       do_reap($self, 1); # finish all fingerprint checks
-       fgrp_fetch_all($self);
+
+       # $end->DESTROY will call fgrp_fetch_all once all references
+       # in $LIVE are gone, and do_reap will eventually drain $LIVE
+       $end = undef;
        do_reap($self, 1);
 }
 
@@ -964,7 +1008,8 @@ sub try_manifest {
                delete $opt{TMPDIR};
        }
        my $ft = File::Temp->new(TEMPLATE => '.manifest-XXXX', %opt);
-       my $cmd = $curl->for_uri($lei, $uri, qw(-f -R -o), $ft->filename);
+       my $cmd = $curl->for_uri($lei, $uri, qw(-R -o), $ft->filename);
+       push(@$cmd, '-z', $manifest) if -f $manifest;
        my $mf_url = "$uri";
        %opt = map { $_ => $lei->{$_} } (0..2);
        my $cerr = run_reap($lei, $cmd, \%opt);
@@ -972,6 +1017,10 @@ sub try_manifest {
                return try_scrape($self) if ($cerr >> 8) == 22; # 404 missing
                return $lei->child_error($cerr, "@$cmd failed");
        }
+
+       # bail out if curl -z/--timecond hit 304 Not Modified, $ft will be empty
+       return $lei->qerr("# $manifest unchanged") if -f $manifest && !-s $ft;
+
        my $m = eval { decode_manifest($ft, $ft, $uri) };
        if ($@) {
                warn $@;
@@ -982,8 +1031,6 @@ sub try_manifest {
        my ($path_pfx, $n, $multi) = multi_inbox($self, \$path, $m);
        return $lei->child_error(1, $multi) if !ref($multi);
        my $v2 = delete $multi->{v2};
-       local $self->{todo} = {};
-       local $self->{fgrp_todo} = {}; # { objstore_dir => [fgrp, ...] }
        if ($v2) {
                for my $name (sort keys %$v2) {
                        my $epochs = delete $v2->{$name};
@@ -1029,7 +1076,7 @@ E: `$task->{cur_dst}' must not contain newline
 EOM
                        $task->{cur_src} .= '/';
                        my $dep = $task->{-ent}->{reference} // '';
-                       push @{$self->{todo}->{$dep}}, $task; # for clone_all
+                       push @{$TODO->{$dep}}, $task; # for clone_all
                        $self->{any_want}->{$name} = 1;
                }
        }
@@ -1041,8 +1088,7 @@ EOM
        my $mis = delete $self->{chg}->{fp_mismatch};
        if ($mis) {
                my $t = (stat($ft))[9];
-               require POSIX;
-               $t = POSIX::strftime('%Y-%m-%d %k:%M:%S %z', localtime($t));
+               $t = strftime('%F %k:%M:%S %z', localtime($t));
                warn <<EOM;
 W: Fingerprints for the following repositories do not match
 W: $mf_url @ $t:
@@ -1054,6 +1100,10 @@ W: The above fingerprints may never match without --prune
 EOM
        }
        dump_manifest($m => $ft) if delete($self->{chg}->{manifest}) || $mis;
+       my $bad = delete $self->{chg}->{badlink};
+       warn(<<EOM, map { ("\t", $_, "\n") } @$bad) if $bad;
+W: The following exist and have not been converted to symlinks
+EOM
        ft_rename($ft, $manifest, 0666);
 }
 
@@ -1083,6 +1133,8 @@ sub do_mirror { # via wq_io_do or public-inbox-clone
                        $self->{"-$k"} = $v;
                }
                local $LIVE = {};
+               local $TODO = {};
+               local $FGRP_TODO = {};
                my $iv = $lei->{opt}->{'inbox-version'} //
                        return start_clone_url($self);
                return clone_v1($self) if $iv == 1;