extindex: --gc checkpoints

We need to ensure -extindex --gc runs don't prevent other
work from happening in the meantime.  I actually caused
my -extindex to OOM due to the lack of checkpoints :x

We'll also hoist out the shard scanning into its own sub
in preparation for lei/store usage.

diff --git a/lib/PublicInbox/ExtSearchIdx.pm b/lib/PublicInbox/ExtSearchIdx.pm
index 3a1856c847096f71df18ad3d6291a96422185e1a..7cc8dd952559d17b3c5c600a7ac8155547ad34ce 100644 (file)
--- a/lib/PublicInbox/ExtSearchIdx.pm
+++ b/lib/PublicInbox/ExtSearchIdx.pm
@@ -369,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};
@@ -390,44 +387,84 @@ 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;
 
        # fixup from old bugs:
-       $nr = $dbh->do(<<'');
+       $nr = $self->{oidx}->dbh->do(<<'');
 DELETE FROM over WHERE num NOT IN (SELECT docid FROM xref3)
 
        warn "I: eliminated $nr stale over entries\n" if $nr != 0;
+
+       my ($cur) = $self->{oidx}->dbh->selectrow_array(<<EOM);
+SELECT MIN(num) FROM over
+EOM
+       my ($max) = $self->{oidx}->dbh->selectrow_array(<<EOM);
+SELECT MAX(num) FROM over
+EOM
+       my $exists;
+restart:
+       $exists = $self->{oidx}->dbh->prepare(<<EOM);
+SELECT COUNT(num) FROM over WHERE num = ?
+EOM
+       for (; $cur <= $max; $cur++) {
+               $exists->execute($cur);
+               next if $exists->fetchrow_array != 0;
+               $self->idx_shard($cur)->ipc_do('xdb_remove_quiet', $cur);
+               if (checkpoint_due($sync)) {
+                       $exists = undef;
+                       reindex_checkpoint($self, $sync);
+                       goto restart;
+               }
+       }
+}
+
+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);
 }
 

diff --git a/lib/PublicInbox/SearchIdx.pm b/lib/PublicInbox/SearchIdx.pm
index e5c872d52fd3fd477f42301f7407acbd8d0061fb..78db329d9da9145ff97091004265ce9264231aca 100644 (file)
--- a/lib/PublicInbox/SearchIdx.pm
+++ b/lib/PublicInbox/SearchIdx.pm
@@ -637,14 +637,21 @@ sub update_vmd {
 
 sub xdb_remove {
        my ($self, @docids) = @_;
-       $self->begin_txn_lazy;
-       my $xdb = $self->{xdb} or return;
+       begin_txn_lazy($self);
+       my $xdb = $self->{xdb} // die 'BUG: missing {xdb}';
        for my $docid (@docids) {
                eval { $xdb->delete_document($docid) };
                warn "E: #$docid not in in Xapian? $@\n" if $@;
        }
 }
 
+sub xdb_remove_quiet {
+       my ($self, $docid) = @_;
+       begin_txn_lazy($self);
+       my $xdb = $self->{xdb} // die 'BUG: missing {xdb}';
+       eval { $xdb->delete_document($docid) };
+}
+
 sub index_git_blob_id {
        my ($doc, $pfx, $objid) = @_;
 
@@ -1098,25 +1105,4 @@ sub eidx_shard_new {
        $self;
 }
 
-# ensure there's no stale Xapian docs by treating $over as canonical
-sub over_check {
-       my ($self, $over) = @_;
-       begin_txn_lazy($self);
-       my $sth = $over->dbh->prepare(<<'');
-SELECT COUNT(*) FROM over WHERE num = ?
-
-       my $xdb = $self->{xdb};
-       my $cur = $xdb->postlist_begin('');
-       my $end = $xdb->postlist_end('');
-       my $xdir = $self->xdir;
-       for (; $cur != $end; $cur++) {
-               my $docid = $cur->get_docid;
-               $sth->execute($docid);
-               my $x = $sth->fetchrow_array;
-               next if $x > 0;
-               warn "I: removing $xdir #$docid, not in `over'\n";
-               $xdb->delete_document($docid);
-       }
-}
-
 1;