]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/ExtSearchIdx.pm
extindex: sync each inbox before checking for missed messages
[public-inbox.git] / lib / PublicInbox / ExtSearchIdx.pm
index eb7c3d67e072c5435936a813868d3dcc74b8e06b..d589d2c00f1a0aa508340fcfb040031b9972beb1 100644 (file)
@@ -120,9 +120,8 @@ sub apply_boost ($$) {
                                ||
                $a->[1] <=> $b->[1] # break ties with {xnum}
        } @$xr3;
-       my $top_blob = unpack('H*', $xr3->[0]->[2]);
        my $new_smsg = $req->{new_smsg};
-       return if $top_blob ne $new_smsg->{blob}; # loser
+       return if $xr3->[0]->[2] ne pack('H*', $new_smsg->{blob}); # loser
 
        # replace the old smsg with the more boosted one
        $new_smsg->{num} = $smsg->{num};
@@ -370,19 +369,16 @@ SELECT oidbin FROM xref3 WHERE docid = ? AND ibx_id = ?
        }
 }
 
-sub eidx_gc {
-       my ($self, $opt) = @_;
-       $self->{cfg} or die "E: GC requires ->attach_config\n";
-       $opt->{-idx_gc} = 1;
-       $self->idx_init($opt); # acquire lock via V2Writable::_idx_init
-
-       my $dbh = $self->{oidx}->dbh;
-       $dbh->do('PRAGMA case_sensitive_like = ON'); # only place we use LIKE
-       my $x3_doc = $dbh->prepare('SELECT docid FROM xref3 WHERE ibx_id = ?');
-       my $ibx_ck = $dbh->prepare('SELECT ibx_id,eidx_key FROM inboxes');
-       my $lc_i = $dbh->prepare(<<'');
-SELECT key FROM eidx_meta WHERE key LIKE ? ESCAPE ?
-
+sub eidx_gc_scan_inboxes ($$) {
+       my ($self, $sync) = @_;
+       my ($x3_doc, $ibx_ck);
+restart:
+       $x3_doc = $self->{oidx}->dbh->prepare(<<EOM);
+SELECT docid FROM xref3 WHERE ibx_id = ?
+EOM
+       $ibx_ck = $self->{oidx}->dbh->prepare(<<EOM);
+SELECT ibx_id,eidx_key FROM inboxes
+EOM
        $ibx_ck->execute;
        while (my ($ibx_id, $eidx_key) = $ibx_ck->fetchrow_array) {
                next if $self->{ibx_map}->{$eidx_key};
@@ -391,44 +387,93 @@ SELECT key FROM eidx_meta WHERE key LIKE ? ESCAPE ?
                $x3_doc->execute($ibx_id);
                while (defined(my $docid = $x3_doc->fetchrow_array)) {
                        gc_unref_doc($self, $ibx_id, $eidx_key, $docid);
+                       if (checkpoint_due($sync)) {
+                               $x3_doc = $ibx_ck = undef;
+                               reindex_checkpoint($self, $sync);
+                               goto restart;
+                       }
                }
-               $dbh->prepare_cached(<<'')->execute($ibx_id);
+               $self->{oidx}->dbh->do(<<'', undef, $ibx_id);
 DELETE FROM inboxes WHERE ibx_id = ?
 
                # drop last_commit info
                my $pat = $eidx_key;
                $pat =~ s/([_%\\])/\\$1/g;
+               $self->{oidx}->dbh->do('PRAGMA case_sensitive_like = ON');
+               my $lc_i = $self->{oidx}->dbh->prepare(<<'');
+SELECT key FROM eidx_meta WHERE key LIKE ? ESCAPE ?
+
                $lc_i->execute("lc-%:$pat//%", '\\');
                while (my ($key) = $lc_i->fetchrow_array) {
                        next if $key !~ m!\Alc-v[1-9]+:\Q$eidx_key\E//!;
                        warn "I: removing $key\n";
-                       $dbh->prepare_cached(<<'')->execute($key);
+                       $self->{oidx}->dbh->do(<<'', undef, $key);
 DELETE FROM eidx_meta WHERE key = ?
 
                }
-
                warn "I: $eidx_key removed\n";
        }
+}
 
-       # it's not real unless it's in `over', we use parallelism here,
-       # shards will be reading directly from over, so commit
-       $self->{oidx}->commit_lazy;
-       $self->{oidx}->begin_lazy;
-
-       for my $idx (@{$self->{idx_shards}}) {
-               warn "I: cleaning up shard #$idx->{shard}\n";
-               $idx->shard_over_check($self->{oidx});
-       }
-       my $nr = $dbh->do(<<'');
+sub eidx_gc_scan_shards ($$) { # TODO: use for lei/store
+       my ($self, $sync) = @_;
+       my $nr = $self->{oidx}->dbh->do(<<'');
 DELETE FROM xref3 WHERE docid NOT IN (SELECT num FROM over)
 
        warn "I: eliminated $nr stale xref3 entries\n" if $nr != 0;
+       reindex_checkpoint($self, $sync) if checkpoint_due($sync);
 
        # fixup from old bugs:
-       $nr = $dbh->do(<<'');
-DELETE FROM over WHERE num NOT IN (SELECT docid FROM xref3)
+       $nr = $self->{oidx}->dbh->do(<<'');
+DELETE FROM over WHERE num > 0 AND num NOT IN (SELECT docid FROM xref3)
 
        warn "I: eliminated $nr stale over entries\n" if $nr != 0;
+       reindex_checkpoint($self, $sync) if checkpoint_due($sync);
+
+       my ($cur) = $self->{oidx}->dbh->selectrow_array(<<EOM);
+SELECT MIN(num) FROM over WHERE num > 0
+EOM
+       $cur // return; # empty
+       my ($r, $n, %active);
+       $nr = 0;
+       while (1) {
+               $r = $self->{oidx}->dbh->selectcol_arrayref(<<"", undef, $cur);
+SELECT num FROM over WHERE num >= ? ORDER BY num ASC LIMIT 10000
+
+               last unless scalar(@$r);
+               while (defined($n = shift @$r)) {
+                       for my $i ($cur..($n - 1)) {
+                               my $idx = idx_shard($self, $i);
+                               $idx->ipc_do('xdb_remove_quiet', $i);
+                               $active{$idx} = $idx;
+                       }
+                       $cur = $n + 1;
+               }
+               if (checkpoint_due($sync)) {
+                       for my $idx (values %active) {
+                               $nr += $idx->ipc_do('nr_quiet_rm')
+                       }
+                       %active = ();
+                       reindex_checkpoint($self, $sync);
+               }
+       }
+       warn "I: eliminated $nr stale Xapian documents\n" if $nr != 0;
+}
+
+sub eidx_gc {
+       my ($self, $opt) = @_;
+       $self->{cfg} or die "E: GC requires ->attach_config\n";
+       $opt->{-idx_gc} = 1;
+       my $sync = {
+               need_checkpoint => \(my $need_checkpoint = 0),
+               check_intvl => 10,
+               next_check => now() + 10,
+               checkpoint_unlocks => 1,
+               -opt => $opt,
+       };
+       $self->idx_init($opt); # acquire lock via V2Writable::_idx_init
+       eidx_gc_scan_inboxes($self, $sync);
+       eidx_gc_scan_shards($self, $sync);
        done($self);
 }
 
@@ -767,6 +812,9 @@ sub _reindex_check_unseen ($$$) {
        my $ibx_id = $ibx->{-ibx_id};
        my $slice = 1000;
        my ($beg, $end) = (1, $slice);
+       my $err = sync_inbox($self, $sync, $ibx) and return;
+       my $max = $ibx->over->max;
+       $end = $max if $end > $max;
 
        # first, check if we missed any messages in target $ibx
        my $msgs;
@@ -775,11 +823,12 @@ sub _reindex_check_unseen ($$$) {
        local $sync->{-regen_fmt} =
                        "$ekey checking unseen %u/".$ibx->over->max."\n";
        ${$sync->{nr}} = 0;
-
+       my $fast = $sync->{-opt}->{fast};
        while (scalar(@{$msgs = $ibx->over->query_xover($beg, $end)})) {
                ${$sync->{nr}} = $beg;
                $beg = $msgs->[-1]->{num} + 1;
                $end = $beg + $slice;
+               $end = $max if $end > $max;
                if (checkpoint_due($sync)) {
                        reindex_checkpoint($self, $sync); # release lock
                }
@@ -799,7 +848,7 @@ ibx_id = ? AND xnum = ? AND oidbin = ?
                        # the first time around ASAP:
                        if (scalar(@$docids) == 0) {
                                reindex_unseen($self, $sync, $ibx, $xsmsg);
-                       } else { # already seen, reindex later
+                       } elsif (!$fast) { # already seen, reindex later
                                for my $r (@$docids) {
                                        $self->{oidx}->eidxq_add($r->[0]);
                                }
@@ -817,7 +866,7 @@ sub _reindex_check_stale ($$$) {
        my $fetching;
        my $ekey = $ibx->eidx_key;
        local $sync->{-regen_fmt} =
-                       "$ekey check stale/missing %u/".$ibx->over->max."\n";
+                       "$ekey checking stale/missing %u/".$ibx->over->max."\n";
        ${$sync->{nr}} = 0;
        do {
                if (checkpoint_due($sync)) {
@@ -857,15 +906,14 @@ DELETE FROM xref3 WHERE ibx_id = ? AND xnum = ? AND oidbin = ?
                        $del->execute;
 
                        # get_xref3 over-fetches, but this is a rare path:
-                       my $xr3 = $self->{oidx}->get_xref3($docid);
+                       my $xr3 = $self->{oidx}->get_xref3($docid, 1);
                        my $idx = $self->idx_shard($docid);
                        if (scalar(@$xr3) == 0) { # all gone
                                $self->{oidx}->delete_by_num($docid);
                                $self->{oidx}->eidxq_del($docid);
                                $idx->ipc_do('xdb_remove', $docid);
                        } else { # enqueue for reindex of remaining messages
-                               $idx->ipc_do('remove_eidx_info',
-                                               $docid, $ibx->eidx_key);
+                               $idx->ipc_do('remove_eidx_info', $docid, $ekey);
                                $self->{oidx}->eidxq_add($docid); # yes, add
                        }
                }
@@ -908,6 +956,7 @@ sub sync_inbox {
        my $err = _sync_inbox($self, $sync, $ibx);
        delete @$ibx{qw(mm over)};
        warn $err, "\n" if defined($err);
+       $err;
 }
 
 sub dd_smsg { # git->cat_async callback