lib/PublicInbox/LeiMirror.pm

   1 # Copyright (C) 2021 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # "lei add-external --mirror" support
   5 package PublicInbox::LeiMirror;
   6 use strict;
   7 use v5.10.1;
   8 use parent qw(PublicInbox::IPC);
   9 use IO::Uncompress::Gunzip qw(gunzip $GunzipError);
  10 use PublicInbox::Spawn qw(popen_rd spawn);
  11 use File::Temp ();
  12
  13 sub do_finish_mirror { # dwaitpid callback
  14         my ($arg, $pid) = @_;
  15         my ($mrr, $lei) = @$arg;
  16         my $f = "$mrr->{dst}/mirror.done";
  17         if ($?) {
  18                 $lei->child_error($?);
  19         } elsif (!unlink($f)) {
  20                 $lei->err("unlink($f): $!") unless $!{ENOENT};
  21         } else {
  22                 if ($lei->{cmd} ne 'public-inbox-clone') {
  23                         $lei->add_external_finish($mrr->{dst});
  24                 }
  25                 $lei->qerr("# mirrored $mrr->{src} => $mrr->{dst}");
  26         }
  27         $lei->dclose;
  28 }
  29
  30 sub _lei_wq_eof { # EOF callback for main daemon
  31         my ($lei) = @_;
  32         my $mrr = delete $lei->{wq1} or return $lei->fail;
  33         $mrr->wq_wait_old(\&do_finish_mirror, $lei);
  34 }
  35
  36 # for old installations without manifest.js.gz
  37 sub try_scrape {
  38         my ($self) = @_;
  39         my $uri = URI->new($self->{src});
  40         my $lei = $self->{lei};
  41         my $curl = $self->{curl} //= PublicInbox::LeiCurl->new($lei) or return;
  42         my $cmd = $curl->for_uri($lei, $uri, '--compressed');
  43         my $opt = { 0 => $lei->{0}, 2 => $lei->{2} };
  44         my $fh = popen_rd($cmd, undef, $opt);
  45         my $html = do { local $/; <$fh> } // die "read(curl $uri): $!";
  46         close($fh) or return $lei->child_error($?, "@$cmd failed");
  47
  48         # we grep with URL below, we don't want Subject/From headers
  49         # making us clone random URLs
  50         my @urls = ($html =~ m!\bgit clone --mirror ([a-z\+]+://\S+)!g);
  51         my $url = $uri->as_string;
  52         chop($url) eq '/' or die "BUG: $uri not canonicalized";
  53
  54         # since this is for old instances w/o manifest.js.gz, try v1 first
  55         return clone_v1($self) if grep(m!\A\Q$url\E/*\z!, @urls);
  56         if (my @v2_urls = grep(m!\A\Q$url\E/[0-9]+\z!, @urls)) {
  57                 my %v2_uris = map { $_ => URI->new($_) } @v2_urls; # uniq
  58                 return clone_v2($self, [ values %v2_uris ]);
  59         }
  60
  61         # filter out common URLs served by WWW (e.g /$MSGID/T/)
  62         if (@urls && $url =~ s!/+[^/]+\@[^/]+/.*\z!! &&
  63                         grep(m!\A\Q$url\E/*\z!, @urls)) {
  64                 die <<"";
  65 E: confused by scraping <$uri>, did you mean <$url>?
  66
  67         }
  68         @urls and die <<"";
  69 E: confused by scraping <$uri>, got ambiguous results:
  70 @urls
  71
  72         die "E: scraping <$uri> revealed nothing\n";
  73 }
  74
  75 sub clone_cmd {
  76         my ($lei, $opt) = @_;
  77         my @cmd = qw(git);
  78         $opt->{$_} = $lei->{$_} for (0..2);
  79         # we support "-c $key=$val" for arbitrary git config options
  80         # e.g.: git -c http.proxy=socks5h://127.0.0.1:9050
  81         push(@cmd, '-c', $_) for @{$lei->{opt}->{c} // []};
  82         push @cmd, qw(clone --mirror);
  83         push @cmd, '-q' if $lei->{opt}->{quiet};
  84         push @cmd, '-v' if $lei->{opt}->{verbose};
  85         # XXX any other options to support?
  86         # --reference is tricky with multiple epochs...
  87         @cmd;
  88 }
  89
  90 # tries the relatively new /$INBOX/_/text/config/raw endpoint
  91 sub _try_config {
  92         my ($self) = @_;
  93         my $dst = $self->{dst};
  94         if (!-d $dst || !mkdir($dst)) {
  95                 require File::Path;
  96                 File::Path::mkpath($dst);
  97                 -d $dst or die "mkpath($dst): $!\n";
  98         }
  99         my $uri = URI->new($self->{src});
 100         my $lei = $self->{lei};
 101         my $path = $uri->path;
 102         chop($path) eq '/' or die "BUG: $uri not canonicalized";
 103         $uri->path($path . '/_/text/config/raw');
 104         my $cmd = $self->{curl}->for_uri($lei, $uri, '--compressed');
 105         my $ce = "$dst/inbox.config.example";
 106         my $f = "$ce-$$.tmp";
 107         open(my $fh, '+>', $f) or return $lei->err("open $f: $! (non-fatal)");
 108         my $opt = { 0 => $lei->{0}, 1 => $fh, 2 => $lei->{2} };
 109         my $cerr = run_reap($lei, $cmd, $opt);
 110         if (($cerr >> 8) == 22) { # 404 missing
 111                 unlink($f) if -s $fh == 0;
 112                 return;
 113         }
 114         return $lei->err("# @$cmd failed (non-fatal)") if $cerr;
 115         rename($f, $ce) or return $lei->err("rename($f, $ce): $! (non-fatal)");
 116         my $cfg = PublicInbox::Config->git_config_dump($f, $lei->{2});
 117         my $ibx = $self->{ibx} = {};
 118         for my $sec (grep(/\Apublicinbox\./, @{$cfg->{-section_order}})) {
 119                 for (qw(address newsgroup nntpmirror)) {
 120                         $ibx->{$_} = $cfg->{"$sec.$_"};
 121                 }
 122         }
 123 }
 124
 125 sub index_cloned_inbox {
 126         my ($self, $iv) = @_;
 127         my $lei = $self->{lei};
 128
 129         # n.b. public-inbox-clone works w/o (SQLite || Xapian)
 130         # lei is useless without Xapian + SQLite
 131         if ($lei->{cmd} ne 'public-inbox-clone') {
 132                 my $ibx = delete($self->{ibx}) // {
 133                         address => [ 'lei@example.com' ],
 134                         version => $iv,
 135                 };
 136                 $ibx->{inboxdir} = $self->{dst};
 137                 PublicInbox::Inbox->new($ibx);
 138                 PublicInbox::InboxWritable->new($ibx);
 139                 my $opt = {};
 140                 for my $sw ($lei->index_opt) {
 141                         my ($k) = ($sw =~ /\A([\w-]+)/);
 142                         $opt->{$k} = $lei->{opt}->{$k};
 143                 }
 144                 # force synchronous dwaitpid for v2:
 145                 local $PublicInbox::DS::in_loop = 0;
 146                 my $cfg = PublicInbox::Config->new(undef, $lei->{2});
 147                 my $env = PublicInbox::Admin::index_prepare($opt, $cfg);
 148                 local %ENV = (%ENV, %$env) if $env;
 149                 PublicInbox::Admin::progress_prepare($opt, $lei->{2});
 150                 PublicInbox::Admin::index_inbox($ibx, undef, $opt);
 151         }
 152         open my $x, '>', "$self->{dst}/mirror.done"; # for do_finish_mirror
 153 }
 154
 155 sub run_reap {
 156         my ($lei, $cmd, $opt) = @_;
 157         $lei->qerr("# @$cmd");
 158         $opt->{pgid} = 0 if $lei->{sock};
 159         my $pid = spawn($cmd, undef, $opt);
 160         my $reap = PublicInbox::OnDestroy->new($lei->can('sigint_reap'), $pid);
 161         waitpid($pid, 0) == $pid or die "waitpid @$cmd: $!";
 162         @$reap = (); # cancel reap
 163         $?
 164 }
 165
 166 sub clone_v1 {
 167         my ($self) = @_;
 168         my $lei = $self->{lei};
 169         my $curl = $self->{curl} //= PublicInbox::LeiCurl->new($lei) or return;
 170         my $uri = URI->new($self->{src});
 171         my $pfx = $curl->torsocks($lei, $uri) or return;
 172         my $cmd = [ @$pfx, clone_cmd($lei, my $opt = {}),
 173                         $uri->as_string, $self->{dst} ];
 174         my $cerr = run_reap($lei, $cmd, $opt);
 175         return $lei->child_error($cerr, "@$cmd failed") if $cerr;
 176         _try_config($self);
 177         index_cloned_inbox($self, 1);
 178 }
 179
 180 sub clone_v2 {
 181         my ($self, $v2_uris) = @_;
 182         my $lei = $self->{lei};
 183         my $curl = $self->{curl} //= PublicInbox::LeiCurl->new($lei) or return;
 184         my $pfx //= $curl->torsocks($lei, $v2_uris->[0]) or return;
 185         my @epochs;
 186         my $dst = $self->{dst};
 187         my @src_edst;
 188         for my $uri (@$v2_uris) {
 189                 my $src = $uri->as_string;
 190                 my $edst = $dst;
 191                 $src =~ m!/([0-9]+)(?:\.git)?\z! or die <<"";
 192 failed to extract epoch number from $src
 193
 194                 my $nr = $1 + 0;
 195                 $edst .= "/git/$nr.git";
 196                 push @src_edst, [ $src, $edst ];
 197         }
 198         my $lk = bless { lock_path => "$dst/inbox.lock" }, 'PublicInbox::Lock';
 199         _try_config($self);
 200         my $on_destroy = $lk->lock_for_scope($$);
 201         my @cmd = clone_cmd($lei, my $opt = {});
 202         while (my $pair = shift(@src_edst)) {
 203                 my $cmd = [ @$pfx, @cmd, @$pair ];
 204                 my $cerr = run_reap($lei, $cmd, $opt);
 205                 return $lei->child_error($cerr, "@$cmd failed") if $cerr;
 206         }
 207         undef $on_destroy; # unlock
 208         index_cloned_inbox($self, 2);
 209 }
 210
 211 # PSGI mount prefixes and manifest.js.gz prefixes don't always align...
 212 sub deduce_epochs ($$) {
 213         my ($m, $path) = @_;
 214         my ($v1_bare, @v2_epochs);
 215         my $path_pfx = '';
 216         $path =~ s!/+\z!!;
 217         do {
 218                 $v1_bare = $m->{$path};
 219                 @v2_epochs = grep(m!\A\Q$path\E/git/[0-9]+\.git\z!, keys %$m);
 220         } while (!defined($v1_bare) && !@v2_epochs &&
 221                 $path =~ s!\A(/[^/]+)/!/! and $path_pfx .= $1);
 222         ($path_pfx, $v1_bare, @v2_epochs);
 223 }
 224
 225 sub decode_manifest ($$$) {
 226         my ($fh, $fn, $uri) = @_;
 227         my $js;
 228         my $gz = do { local $/; <$fh> } // die "slurp($fn): $!";
 229         gunzip(\$gz => \$js, MultiStream => 1) or
 230                 die "gunzip($uri): $GunzipError\n";
 231         my $m = eval { PublicInbox::Config->json->decode($js) };
 232         die "$uri: error decoding `$js': $@\n" if $@;
 233         ref($m) eq 'HASH' or die "$uri unknown type: ".ref($m);
 234         $m;
 235 }
 236
 237 sub try_manifest {
 238         my ($self) = @_;
 239         my $uri = URI->new($self->{src});
 240         my $lei = $self->{lei};
 241         my $curl = $self->{curl} //= PublicInbox::LeiCurl->new($lei) or return;
 242         my $path = $uri->path;
 243         chop($path) eq '/' or die "BUG: $uri not canonicalized";
 244         $uri->path($path . '/manifest.js.gz');
 245         my $pdir = $lei->rel2abs($self->{dst});
 246         $pdir =~ s!/[^/]+/?\z!!;
 247         my $ft = File::Temp->new(TEMPLATE => 'manifest-XXXX',
 248                                 UNLINK => 1, DIR => $pdir);
 249         my $fn = $ft->filename;
 250         my $cmd = $curl->for_uri($lei, $uri, '-R', '-o', $fn);
 251         my $opt = { 0 => $lei->{0}, 1 => $lei->{1}, 2 => $lei->{2} };
 252         my $cerr = run_reap($lei, $cmd, $opt);
 253         if ($cerr) {
 254                 return try_scrape($self) if ($cerr >> 8) == 22; # 404 missing
 255                 return $lei->child_error($cerr, "@$cmd failed");
 256         }
 257         my $m = decode_manifest($ft, $fn, $uri);
 258         my ($path_pfx, $v1_bare, @v2_epochs) = deduce_epochs($m, $path);
 259         if (@v2_epochs) {
 260                 # It may be possible to have v1 + v2 in parallel someday:
 261                 $lei->err(<<EOM) if defined $v1_bare;
 262 # `$v1_bare' appears to be a v1 inbox while v2 epochs exist:
 263 # @v2_epochs
 264 # ignoring $v1_bare (use --inbox-version=1 to force v1 instead)
 265 EOM
 266                 @v2_epochs = map {
 267                         $uri->path($path_pfx.$_);
 268                         $uri->clone
 269                 } @v2_epochs;
 270                 clone_v2($self, \@v2_epochs);
 271                 my $fin = "$self->{dst}/manifest.js.gz";
 272                 rename($fn, $fin) or die "E: rename($fn, $fin): $!";
 273                 $ft->unlink_on_destroy(0);
 274         } elsif (defined $v1_bare) {
 275                 clone_v1($self);
 276         } else {
 277                 die "E: confused by <$uri>, possible matches:\n\t",
 278                         join(', ', sort keys %$m), "\n";
 279         }
 280 }
 281
 282 sub start_clone_url {
 283         my ($self) = @_;
 284         return try_manifest($self) if $self->{src} =~ m!\Ahttps?://!;
 285         die "TODO: non-HTTP/HTTPS clone of $self->{src} not supported, yet";
 286 }
 287
 288 sub do_mirror { # via wq_io_do
 289         my ($self) = @_;
 290         my $lei = $self->{lei};
 291         eval {
 292                 my $iv = $lei->{opt}->{'inbox-version'};
 293                 if (defined $iv) {
 294                         return clone_v1($self) if $iv == 1;
 295                         return try_scrape($self) if $iv == 2;
 296                         die "bad --inbox-version=$iv\n";
 297                 }
 298                 return start_clone_url($self) if $self->{src} =~ m!://!;
 299                 die "TODO: cloning local directories not supported, yet";
 300         };
 301         $lei->fail($@) if $@;
 302 }
 303
 304 sub start {
 305         my ($cls, $lei, $src, $dst) = @_;
 306         my $self = bless { src => $src, dst => $dst }, $cls;
 307         if ($src =~ m!https?://!) {
 308                 require URI;
 309                 require PublicInbox::LeiCurl;
 310         }
 311         require PublicInbox::Lock;
 312         require PublicInbox::Inbox;
 313         require PublicInbox::Admin;
 314         require PublicInbox::InboxWritable;
 315         my ($op_c, $ops) = $lei->workers_start($self, 1);
 316         $lei->{wq1} = $self;
 317         $self->wq_io_do('do_mirror', []);
 318         $self->wq_close(1);
 319         $lei->wait_wq_events($op_c, $ops);
 320 }
 321
 322 sub ipc_atfork_child {
 323         my ($self) = @_;
 324         $self->{lei}->_lei_atfork_child;
 325         $SIG{TERM} = sub { exit(128 + 15) }; # trigger OnDestroy $reap
 326         $self->SUPER::ipc_atfork_child;
 327 }
 328
 329 1;