]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/V2Writable.pm
Merge remote-tracking branch 'origin/danga-bundle'
[public-inbox.git] / lib / PublicInbox / V2Writable.pm
index 412eb6a9a4cc0ff3aa9b8ec98c348a042bab6c18..87e8f3eb581a97076b5fd22ea5af917bb623306e 100644 (file)
@@ -2,6 +2,7 @@
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
 # This interface wraps and mimics PublicInbox::Import
+# Used to write to V2 inboxes (see L<public-inbox-v2-format(5)>).
 package PublicInbox::V2Writable;
 use strict;
 use warnings;
@@ -71,6 +72,7 @@ sub new {
                im => undef, #  PublicInbox::Import
                parallel => 1,
                transact_bytes => 0,
+               current_info => '',
                xpfx => $xpfx,
                over => PublicInbox::OverIdx->new("$xpfx/over.sqlite3", 1),
                lock_path => "$dir/inbox.lock",
@@ -83,11 +85,14 @@ sub new {
 }
 
 sub init_inbox {
-       my ($self, $parallel) = @_;
+       my ($self, $parallel, $skip) = @_;
        $self->{parallel} = $parallel;
        $self->idx_init;
        my $epoch_max = -1;
        git_dir_latest($self, \$epoch_max);
+       if (defined $skip && $epoch_max == -1) {
+               $epoch_max = $skip;
+       }
        $self->git_init($epoch_max >= 0 ? $epoch_max : 0);
        $self->done;
 }
@@ -160,6 +165,19 @@ sub num_for {
                        return if $existing;
                }
 
+               # AltId may pre-populate article numbers (e.g. X-Mail-Count
+               # or NNTP article number), use that article number if it's
+               # not in Over.
+               my $altid = $self->{-inbox}->{altid};
+               if ($altid && grep(/:file=msgmap\.sqlite3\z/, @$altid)) {
+                       my $num = $self->{mm}->num_for($mid);
+
+                       if (defined $num && !$self->{over}->get_art($num)) {
+                               $$mid0 = $mid;
+                               return $num;
+                       }
+               }
+
                # very unlikely:
                warn "<$mid> reused for mismatched content\n";
 
@@ -220,11 +238,19 @@ sub idx_init {
        # frequently activated.
        delete $ibx->{$_} foreach (qw(git mm search));
 
-       if ($self->{parallel}) {
-               pipe(my ($r, $w)) or die "pipe failed: $!";
-               $self->{bnote} = [ $r, $w ];
-               $w->autoflush(1);
-       }
+       my $indexlevel = $ibx->{indexlevel};
+       if ($indexlevel && $indexlevel eq 'basic') {
+               $self->{parallel} = 0;
+       }
+
+       if ($self->{parallel}) {
+               pipe(my ($r, $w)) or die "pipe failed: $!";
+               # pipe for barrier notifications doesn't need to be big,
+               # 1031: F_SETPIPE_SZ
+               fcntl($w, 1031, 4096) if $^O eq 'linux';
+               $self->{bnote} = [ $r, $w ];
+               $w->autoflush(1);
+       }
 
        my $over = $self->{over};
        $ibx->umask_prepare;
@@ -260,10 +286,19 @@ sub purge_oids {
        $self->done;
        my $pfx = "$self->{-inbox}->{mainrepo}/git";
        my $purges = [];
-       foreach my $i (0..$self->{epoch_max}) {
-               my $git = PublicInbox::Git->new("$pfx/$i.git");
+       my $max = $self->{epoch_max};
+
+       unless (defined($max)) {
+               defined(my $latest = git_dir_latest($self, \$max)) or return;
+               $self->{epoch_max} = $max;
+       }
+       foreach my $i (0..$max) {
+               my $git_dir = "$pfx/$i.git";
+               -d $git_dir or next;
+               my $git = PublicInbox::Git->new($git_dir);
                my $im = $self->import_init($git, 0, 1);
                $purges->[$i] = $im->purge_oids($purge);
+               $im->done;
        }
        $purges;
 }
@@ -365,7 +400,7 @@ sub purge {
        my ($self, $mime) = @_;
        my $purges = $self->{-inbox}->with_umask(sub {
                remove_internal($self, $mime, undef, {});
-       });
+       }) or return;
        $self->idx_init if @$purges; # ->done is called on purges
        for my $i (0..$#$purges) {
                defined(my $cmt = $purges->[$i]) or next;
@@ -472,16 +507,15 @@ sub done {
        delete $self->{bnote};
        $self->{transact_bytes} = 0;
        $self->lock_release if $parts;
+       $self->{-inbox}->git->cleanup;
 }
 
-sub git_init {
+sub fill_alternates ($$) {
        my ($self, $epoch) = @_;
-       my $pfx = "$self->{-inbox}->{mainrepo}/git";
-       my $git_dir = "$pfx/$epoch.git";
-       my @cmd = (qw(git init --bare -q), $git_dir);
-       PublicInbox::Import::run_die(\@cmd);
 
+       my $pfx = "$self->{-inbox}->{mainrepo}/git";
        my $all = "$self->{-inbox}->{mainrepo}/all.git";
+       my @cmd;
        unless (-d $all) {
                @cmd = (qw(git init --bare -q), $all);
                PublicInbox::Import::run_die(\@cmd);
@@ -489,8 +523,7 @@ sub git_init {
                                'repack.writeBitmaps', 'true');
                PublicInbox::Import::run_die(\@cmd);
        }
-
-       @cmd = (qw/git config/, "--file=$git_dir/config",
+       @cmd = (qw/git config/, "--file=$pfx/$epoch.git/config",
                        'include.path', '../../all.git/config');
        PublicInbox::Import::run_die(\@cmd);
 
@@ -505,12 +538,20 @@ sub git_init {
                my $dir = "../../git/$i.git/objects";
                push @add, $dir if !$alts{$dir} && -d "$pfx/$i.git";
        }
-       return $git_dir unless @add;
+       return unless @add;
        open my $fh, '>>', $alt or die "open >> $alt: $!\n";
        foreach my $dir (@add) {
                print $fh "$dir\n" or die "print >> $alt: $!\n";
        }
        close $fh or die "close $alt: $!\n";
+}
+
+sub git_init {
+       my ($self, $epoch) = @_;
+       my $git_dir = "$self->{-inbox}->{mainrepo}/git/$epoch.git";
+       my @cmd = (qw(git init --bare -q), $git_dir);
+       PublicInbox::Import::run_die(\@cmd);
+       fill_alternates($self, $epoch);
        $git_dir
 }
 
@@ -553,7 +594,9 @@ sub importer {
        if (defined $latest) {
                my $git = PublicInbox::Git->new($latest);
                my $packed_bytes = $git->packed_bytes;
-               if ($packed_bytes >= $self->{rotate_bytes}) {
+               my $unpacked_bytes = $packed_bytes / $PACKING_FACTOR;
+
+               if ($unpacked_bytes >= $self->{rotate_bytes}) {
                        $epoch = $max + 1;
                } else {
                        $self->{epoch_max} = $max;
@@ -653,12 +696,12 @@ sub mark_deleted {
        my $mids = mids($mime->header_obj);
        my $cid = content_id($mime);
        foreach my $mid (@$mids) {
-               $D->{"$mid\0$cid"} = 1;
+               $D->{"$mid\0$cid"} = $oid;
        }
 }
 
 sub reindex_oid {
-       my ($self, $mm_tmp, $D, $git, $oid, $regen) = @_;
+       my ($self, $mm_tmp, $D, $git, $oid, $regen, $reindex) = @_;
        my $len;
        my $msgref = $git->cat_file($oid, \$len);
        my $mime = PublicInbox::MIME->new($$msgref);
@@ -671,11 +714,12 @@ sub reindex_oid {
        my $num = -1;
        my $del = 0;
        foreach my $mid (@$mids) {
-               $del += (delete $D->{"$mid\0$cid"} || 0);
+               $del += delete($D->{"$mid\0$cid"}) ? 1 : 0;
                my $n = $mm_tmp->num_for($mid);
                if (defined $n && $n > $num) {
                        $mid0 = $mid;
                        $num = $n;
+                       $self->{mm}->mid_set($num, $mid0);
                }
        }
        if (!defined($mid0) && $regen && !$del) {
@@ -700,7 +744,8 @@ sub reindex_oid {
 
        if (!defined($mid0) || $del) {
                if (!defined($mid0) && $del) { # expected for deletes
-                       $$regen--;
+                       $num = $$regen--;
+                       $self->{mm}->num_highwater($num) unless $reindex;
                        return
                }
 
@@ -767,7 +812,8 @@ sub index_prepare {
                my $git_dir = git_dir_n($self, $i);
                -d $git_dir or next; # missing parts are fine
                my $git = PublicInbox::Git->new($git_dir);
-               chomp(my $tip = $git->qx('rev-parse', $head));
+               chomp(my $tip = $git->qx(qw(rev-parse -q --verify), $head));
+               next if $?; # new repo
                my $range;
                if (defined(my $cur = $ranges->[$i])) {
                        $range = "$cur..$tip";
@@ -840,7 +886,10 @@ sub unindex_oid {
                        warn "BUG: multiple articles linked to $oid\n",
                                join(',',sort keys %gone), "\n";
                }
-               $self->{unindexed}->{$_}++ foreach keys %gone;
+               foreach my $num (keys %gone) {
+                       $self->{unindexed}->{$_}++;
+                       $self->{mm}->num_delete($num);
+               }
                $self->unindex_oid_remote($oid, $mid);
        }
 }
@@ -877,12 +926,13 @@ sub index_sync {
        return unless defined $latest;
        $self->idx_init; # acquire lock
        my $mm_tmp = $self->{mm}->tmp_clone;
-       my $ranges = $opts->{reindex} ? [] : $self->last_commits($epoch_max);
+       my $reindex = $opts->{reindex};
+       my $ranges = $reindex ? [] : $self->last_commits($epoch_max);
 
-       my ($min, $max) = $mm_tmp->minmax;
+       my $high = $self->{mm}->num_highwater();
        my $regen = $self->index_prepare($opts, $epoch_max, $ranges);
-       $$regen += $max if $max;
-       my $D = {};
+       $$regen += $high if $high;
+       my $D = {}; # "$mid\0$cid" => $oid
        my @cmd = qw(log --raw -r --pretty=tformat:%H
                        --no-notes --no-color --no-abbrev --no-renames);
 
@@ -892,6 +942,7 @@ sub index_sync {
                my $git_dir = git_dir_n($self, $i);
                die "already reindexing!\n" if delete $self->{reindex_pipe};
                -d $git_dir or next; # missing parts are fine
+               fill_alternates($self, $i);
                my $git = PublicInbox::Git->new($git_dir);
                my $unindex = delete $self->{"unindex-range.$i"};
                $self->unindex($opts, $git, $unindex) if $unindex;
@@ -899,11 +950,13 @@ sub index_sync {
                my $fh = $self->{reindex_pipe} = $git->popen(@cmd, $range);
                my $cmt;
                while (<$fh>) {
+                       chomp;
+                       $self->{current_info} = "$i.git $_";
                        if (/\A$x40$/o && !defined($cmt)) {
-                               chomp($cmt = $_);
+                               $cmt = $_;
                        } elsif (/\A:\d{6} 100644 $x40 ($x40) [AM]\tm$/o) {
                                $self->reindex_oid($mm_tmp, $D, $git, $1,
-                                               $regen);
+                                               $regen, $reindex);
                        } elsif (/\A:\d{6} 100644 $x40 ($x40) [AM]\td$/o) {
                                $self->mark_deleted($D, $git, $1);
                        }
@@ -912,13 +965,13 @@ sub index_sync {
                delete $self->{reindex_pipe};
                $self->update_last_commit($git, $i, $cmt) if defined $cmt;
        }
-       my @d = sort keys %$D;
-       if (@d) {
-               warn "BUG: ", scalar(@d)," unseen deleted messages marked\n";
-               foreach (@d) {
-                       my ($mid, undef) = split(/\0/, $_, 2);
-                       warn "<$mid>\n";
-               }
+
+       # unindex is required for leftovers if "deletes" affect messages
+       # in a previous fetch+index window:
+       if (scalar keys %$D) {
+               my $git = $self->{-inbox}->git;
+               $self->unindex_oid($git, $_) for values %$D;
+               $git->cleanup;
        }
        $self->done;
 }