X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FLeiMirror.pm;h=fca11ccff037cae4d5b1f4b37e609c36733629fa;hb=de39bfcb19571f5c723c03d35a26a7af2e03d993;hp=5ba69287fdd50c5039a10e56b8796de1320f91c3;hpb=757652fd1ad6843c984610263a2a0b336c974111;p=public-inbox.git diff --git a/lib/PublicInbox/LeiMirror.pm b/lib/PublicInbox/LeiMirror.pm index 5ba69287..fca11ccf 100644 --- a/lib/PublicInbox/LeiMirror.pm +++ b/lib/PublicInbox/LeiMirror.pm @@ -8,20 +8,25 @@ use v5.10.1; use parent qw(PublicInbox::IPC); use IO::Uncompress::Gunzip qw(gunzip $GunzipError); use PublicInbox::Spawn qw(popen_rd spawn); -use PublicInbox::PktOp; sub do_finish_mirror { # dwaitpid callback my ($arg, $pid) = @_; my ($mrr, $lei) = @$arg; - if ($? == 0 && unlink("$mrr->{dst}/mirror.done")) { + my $f = "$mrr->{dst}/mirror.done"; + if ($?) { + $lei->child_error($?); + } elsif (!unlink($f)) { + $lei->err("unlink($f): $!"); + } else { $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); } @@ -31,9 +36,9 @@ 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); + my $cmd = $curl->for_uri($lei, $uri, '--compressed'); my $opt = { 0 => $lei->{0}, 2 => $lei->{2} }; - my $fh = popen_rd($cmd, $lei->{env}, $opt); + my $fh = popen_rd($cmd, undef, $opt); my $html = do { local $/; <$fh> } // die "read(curl $uri): $!"; close($fh) or return $lei->child_error($?, "@$cmd failed"); @@ -93,8 +98,7 @@ sub _try_config { my $path = $uri->path; chop($path) eq '/' or die "BUG: $uri not canonicalized"; $uri->path($path . '/_/text/config/raw'); - my $cmd = $self->{curl}->for_uri($lei, $uri); - push @$cmd, '--compressed'; # curl decompresses for us + my $cmd = $self->{curl}->for_uri($lei, $uri, '--compressed'); my $ce = "$dst/inbox.config.example"; my $f = "$ce-$$.tmp"; open(my $fh, '+>', $f) or return $lei->err("open $f: $! (non-fatal)"); @@ -106,7 +110,7 @@ sub _try_config { } 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); + my $cfg = PublicInbox::Config->git_config_dump($f); my $ibx = $self->{ibx} = {}; for my $sec (grep(/\Apublicinbox\./, @{$cfg->{-section_order}})) { for (qw(address newsgroup nntpmirror)) { @@ -144,7 +148,7 @@ sub run_reap { my ($lei, $cmd, $opt) = @_; $lei->qerr("# @$cmd"); $opt->{pgid} = 0; - my $pid = spawn($cmd, $lei->{env}, $opt); + 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: $!"; @$reap = (); # cancel reap @@ -196,6 +200,19 @@ failed to extract epoch number from $src 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 = ''; + 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 try_manifest { my ($self) = @_; my $uri = URI->new($self->{src}); @@ -207,7 +224,7 @@ sub try_manifest { my $cmd = $curl->for_uri($lei, $uri); $lei->qerr("# @$cmd"); my $opt = { 0 => $lei->{0}, 2 => $lei->{2} }; - my ($fh, $pid) = popen_rd($cmd, $lei->{env}, $opt); + 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; @@ -225,8 +242,7 @@ sub try_manifest { 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 ($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) { + } 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"; } } @@ -264,14 +282,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; - $lei->{mrr} = $self; + my $self = bless { src => $src, dst => $dst }, $cls; if ($src =~ m!https?://!) { require URI; require PublicInbox::LeiCurl; @@ -280,27 +296,16 @@ sub start { require PublicInbox::Inbox; require PublicInbox::Admin; require PublicInbox::InboxWritable; - my $ops = { - '!' => [ $lei->can('fail_handler'), $lei ], - 'x_it' => [ $lei->can('x_it'), $lei ], - 'child_error' => [ $lei->can('child_error'), $lei ], - '' => [ \&mirror_done, $lei ], - }; - ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops); - $self->wq_workers_start('lei_mirror', 1, $lei->oldset, {lei => $lei}); - my $op = delete $lei->{pkt_op_c}; - delete $lei->{pkt_op_p}; + my ($op_c, $ops) = $lei->workers_start($self, 1); + $lei->{wq1} = $self; $self->wq_io_do('do_mirror', []); $self->wq_close(1); - $lei->event_step_init; # wait for shutdowns - if ($lei->{oneshot}) { - while ($op->{sock}) { $op->event_step } - } + $lei->wait_wq_events($op_c, $ops); } sub ipc_atfork_child { my ($self) = @_; - $self->{lei}->lei_atfork_child; + $self->{lei}->_lei_atfork_child; $SIG{TERM} = sub { exit(128 + 15) }; # trigger OnDestroy $reap $self->SUPER::ipc_atfork_child; }