X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FLeiMirror.pm;h=c128d13d09855be78eb43e54f647458ca0107eea;hb=b45a1dffa647f6427d0c900fcc55753db7a1994c;hp=6e62625d73502d6eb881aae49ec28e86a3505193;hpb=97caa1fe259cd6904573f36e0ac078a269b6ec5f;p=public-inbox.git diff --git a/lib/PublicInbox/LeiMirror.pm b/lib/PublicInbox/LeiMirror.pm index 6e62625d..c128d13d 100644 --- a/lib/PublicInbox/LeiMirror.pm +++ b/lib/PublicInbox/LeiMirror.pm @@ -8,19 +8,28 @@ use v5.10.1; use parent qw(PublicInbox::IPC); use IO::Uncompress::Gunzip qw(gunzip $GunzipError); use PublicInbox::Spawn qw(popen_rd spawn); +use File::Temp (); sub do_finish_mirror { # dwaitpid callback my ($arg, $pid) = @_; my ($mrr, $lei) = @$arg; - if ($? == 0 && unlink("$mrr->{dst}/mirror.done")) { - $lei->add_external_finish($mrr->{dst}); + my $f = "$mrr->{dst}/mirror.done"; + if ($?) { + $lei->child_error($?); + } elsif (!unlink($f)) { + $lei->err("unlink($f): $!") unless $!{ENOENT}; + } else { + if ($lei->{cmd} ne 'public-inbox-clone') { + $lei->add_external_finish($mrr->{dst}); + } + $lei->qerr("# mirrored $mrr->{src} => $mrr->{dst}"); } $lei->dclose; } -sub mirror_done { # EOF callback for main daemon +sub _lei_wq_eof { # EOF callback for main daemon my ($lei) = @_; - my $mrr = delete $lei->{mrr} or return; + my $mrr = delete $lei->{wq1} or return $lei->fail; $mrr->wq_wait_old(\&do_finish_mirror, $lei); } @@ -97,14 +106,14 @@ sub _try_config { my $f = "$ce-$$.tmp"; open(my $fh, '+>', $f) or return $lei->err("open $f: $! (non-fatal)"); my $opt = { 0 => $lei->{0}, 1 => $fh, 2 => $lei->{2} }; - my $cerr = run_reap($lei, $cmd, $opt) // return; + my $cerr = run_reap($lei, $cmd, $opt); if (($cerr >> 8) == 22) { # 404 missing unlink($f) if -s $fh == 0; return; } return $lei->err("# @$cmd failed (non-fatal)") if $cerr; - rename($f, $ce) or return $lei->err("link($f, $ce): $! (non-fatal)"); - my $cfg = PublicInbox::Config::git_config_dump($f); + rename($f, $ce) or return $lei->err("rename($f, $ce): $! (non-fatal)"); + my $cfg = PublicInbox::Config->git_config_dump($f, $lei->{2}); my $ibx = $self->{ibx} = {}; for my $sec (grep(/\Apublicinbox\./, @{$cfg->{-section_order}})) { for (qw(address newsgroup nntpmirror)) { @@ -115,38 +124,43 @@ sub _try_config { sub index_cloned_inbox { my ($self, $iv) = @_; - my $ibx = delete($self->{ibx}) // { - address => [ 'lei@example.com' ], - version => $iv, - }; - $ibx->{inboxdir} = $self->{dst}; - PublicInbox::Inbox->new($ibx); - PublicInbox::InboxWritable->new($ibx); - my $opt = {}; my $lei = $self->{lei}; - for my $sw ($lei->index_opt) { - my ($k) = ($sw =~ /\A([\w-]+)/); - $opt->{$k} = $lei->{opt}->{$k}; + + # n.b. public-inbox-clone works w/o (SQLite || Xapian) + # lei is useless without Xapian + SQLite + if ($lei->{cmd} ne 'public-inbox-clone') { + my $ibx = delete($self->{ibx}) // { + address => [ 'lei@example.com' ], + version => $iv, + }; + $ibx->{inboxdir} = $self->{dst}; + PublicInbox::Inbox->new($ibx); + PublicInbox::InboxWritable->new($ibx); + my $opt = {}; + for my $sw ($lei->index_opt) { + my ($k) = ($sw =~ /\A([\w-]+)/); + $opt->{$k} = $lei->{opt}->{$k}; + } + # force synchronous dwaitpid for v2: + local $PublicInbox::DS::in_loop = 0; + my $cfg = PublicInbox::Config->new(undef, $lei->{2}); + my $env = PublicInbox::Admin::index_prepare($opt, $cfg); + local %ENV = (%ENV, %$env) if $env; + PublicInbox::Admin::progress_prepare($opt, $lei->{2}); + PublicInbox::Admin::index_inbox($ibx, undef, $opt); } - # force synchronous dwaitpid for v2: - local $PublicInbox::DS::in_loop = 0; - my $cfg = PublicInbox::Config->new; - my $env = PublicInbox::Admin::index_prepare($opt, $cfg); - local %ENV = (%ENV, %$env) if $env; - PublicInbox::Admin::progress_prepare($opt, $lei->{2}); - PublicInbox::Admin::index_inbox($ibx, undef, $opt); open my $x, '>', "$self->{dst}/mirror.done"; # for do_finish_mirror } sub run_reap { my ($lei, $cmd, $opt) = @_; $lei->qerr("# @$cmd"); - $opt->{pgid} = 0; + $opt->{pgid} = 0 if $lei->{sock}; my $pid = spawn($cmd, undef, $opt); my $reap = PublicInbox::OnDestroy->new($lei->can('sigint_reap'), $pid); - my $err = waitpid($pid, 0) == $pid ? undef : "waitpid @$cmd: $!"; + waitpid($pid, 0) == $pid or die "waitpid @$cmd: $!"; @$reap = (); # cancel reap - $err ? $lei->err($err) : $? + $? } sub clone_v1 { @@ -157,7 +171,7 @@ sub clone_v1 { my $pfx = $curl->torsocks($lei, $uri) or return; my $cmd = [ @$pfx, clone_cmd($lei, my $opt = {}), $uri->as_string, $self->{dst} ]; - my $cerr = run_reap($lei, $cmd, $opt) // return; + my $cerr = run_reap($lei, $cmd, $opt); return $lei->child_error($cerr, "@$cmd failed") if $cerr; _try_config($self); index_cloned_inbox($self, 1); @@ -187,13 +201,39 @@ failed to extract epoch number from $src my @cmd = clone_cmd($lei, my $opt = {}); while (my $pair = shift(@src_edst)) { my $cmd = [ @$pfx, @cmd, @$pair ]; - my $cerr = run_reap($lei, $cmd, $opt) // return; + my $cerr = run_reap($lei, $cmd, $opt); return $lei->child_error($cerr, "@$cmd failed") if $cerr; } undef $on_destroy; # unlock index_cloned_inbox($self, 2); } +# PSGI mount prefixes and manifest.js.gz prefixes don't always align... +sub deduce_epochs ($$) { + my ($m, $path) = @_; + my ($v1_bare, @v2_epochs); + my $path_pfx = ''; + $path =~ s!/+\z!!; + do { + $v1_bare = $m->{$path}; + @v2_epochs = grep(m!\A\Q$path\E/git/[0-9]+\.git\z!, keys %$m); + } while (!defined($v1_bare) && !@v2_epochs && + $path =~ s!\A(/[^/]+)/!/! and $path_pfx .= $1); + ($path_pfx, $v1_bare, @v2_epochs); +} + +sub decode_manifest ($$$) { + my ($fh, $fn, $uri) = @_; + my $js; + my $gz = do { local $/; <$fh> } // die "slurp($fn): $!"; + gunzip(\$gz => \$js, MultiStream => 1) or + die "gunzip($uri): $GunzipError\n"; + my $m = eval { PublicInbox::Config->json->decode($js) }; + die "$uri: error decoding `$js': $@\n" if $@; + ref($m) eq 'HASH' or die "$uri unknown type: ".ref($m); + $m; +} + sub try_manifest { my ($self) = @_; my $uri = URI->new($self->{src}); @@ -202,29 +242,20 @@ sub try_manifest { my $path = $uri->path; chop($path) eq '/' or die "BUG: $uri not canonicalized"; $uri->path($path . '/manifest.js.gz'); - my $cmd = $curl->for_uri($lei, $uri); - $lei->qerr("# @$cmd"); - my $opt = { 0 => $lei->{0}, 2 => $lei->{2} }; - my ($fh, $pid) = popen_rd($cmd, undef, $opt); - my $reap = PublicInbox::OnDestroy->new($lei->can('sigint_reap'), $pid); - my $gz = do { local $/; <$fh> } // die "read(curl $uri): $!"; - close $fh; - my $err = waitpid($pid, 0) == $pid ? undef : "waitpid @$cmd: $!"; - @$reap = (); - return $lei->err($err) if $err; - if ($?) { - return try_scrape($self) if ($? >> 8) == 22; # 404 missing - return $lei->child_error($?, "@$cmd failed"); + my $pdir = $lei->rel2abs($self->{dst}); + $pdir =~ s!/[^/]+/?\z!!; + my $ft = File::Temp->new(TEMPLATE => 'manifest-XXXX', + UNLINK => 1, DIR => $pdir); + my $fn = $ft->filename; + my $cmd = $curl->for_uri($lei, $uri, '-R', '-o', $fn); + my $opt = { 0 => $lei->{0}, 1 => $lei->{1}, 2 => $lei->{2} }; + my $cerr = run_reap($lei, $cmd, $opt); + if ($cerr) { + return try_scrape($self) if ($cerr >> 8) == 22; # 404 missing + return $lei->child_error($cerr, "@$cmd failed"); } - my $js; - gunzip(\$gz => \$js, MultiStream => 1) or - die "gunzip($uri): $GunzipError"; - my $m = eval { PublicInbox::Config->json->decode($js) }; - die "$uri: error decoding `$js': $@" if $@; - ref($m) eq 'HASH' or die "$uri unknown type: ".ref($m); - - my $v1_bare = $m->{$path}; - my @v2_epochs = grep(m!\A\Q$path\E/git/[0-9]+\.git\z!, keys %$m); + my $m = decode_manifest($ft, $fn, $uri); + my ($path_pfx, $v1_bare, @v2_epochs) = deduce_epochs($m, $path); if (@v2_epochs) { # It may be possible to have v1 + v2 in parallel someday: $lei->err(<path($_); $uri->clone } @v2_epochs; + @v2_epochs = map { + $uri->path($path_pfx.$_); + $uri->clone + } @v2_epochs; clone_v2($self, \@v2_epochs); - } elsif ($v1_bare) { + my $fin = "$self->{dst}/manifest.js.gz"; + rename($fn, $fin) or die "E: rename($fn, $fin): $!"; + $ft->unlink_on_destroy(0); + } elsif (defined $v1_bare) { clone_v1($self); - } elsif (my @maybe = grep(m!\Q$path\E!, keys %$m)) { - die "E: confused by <$uri>, possible matches:\n@maybe"; } else { - die "E: confused by <$uri>"; + die "E: confused by <$uri>, possible matches:\n\t", + join(', ', sort keys %$m), "\n"; } } @@ -262,13 +298,12 @@ sub do_mirror { # via wq_io_do return start_clone_url($self) if $self->{src} =~ m!://!; die "TODO: cloning local directories not supported, yet"; }; - return $lei->fail($@) if $@; - $lei->qerr("# mirrored $self->{src} => $self->{dst}"); + $lei->fail($@) if $@; } sub start { my ($cls, $lei, $src, $dst) = @_; - my $self = bless { lei => $lei, src => $src, dst => $dst }, $cls; + my $self = bless { src => $src, dst => $dst }, $cls; if ($src =~ m!https?://!) { require URI; require PublicInbox::LeiCurl; @@ -277,13 +312,11 @@ sub start { require PublicInbox::Inbox; require PublicInbox::Admin; require PublicInbox::InboxWritable; - my $op = $lei->workers_start($self, 'lei_mirror', 1, { - '' => [ \&mirror_done, $lei ] - }); - $lei->{mrr} = $self; + my ($op_c, $ops) = $lei->workers_start($self, 1); + $lei->{wq1} = $self; $self->wq_io_do('do_mirror', []); $self->wq_close(1); - while ($op && $op->{sock}) { $op->event_step } + $lei->wait_wq_events($op_c, $ops); } sub ipc_atfork_child {