]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/ExtSearchIdx.pm
extindex: guard against buggy unrefs
[public-inbox.git] / lib / PublicInbox / ExtSearchIdx.pm
index c0fd282358f97f6c8368acf9536f1d43c9282371..3d7a6e7d76223a47904b62cb7bef1d9b42734d03 100644 (file)
@@ -18,6 +18,7 @@ use strict;
 use v5.10.1;
 use parent qw(PublicInbox::ExtSearch PublicInbox::Lock);
 use Carp qw(croak carp);
+use Scalar::Util qw(blessed);
 use Sys::Hostname qw(hostname);
 use POSIX qw(strftime);
 use File::Glob qw(bsd_glob GLOB_NOSORT);
@@ -129,25 +130,59 @@ sub apply_boost ($$) {
        $req->{self}->{oidx}->add_overview($req->{eml}, $new_smsg);
 }
 
+sub remove_doc ($$) {
+       my ($self, $docid) = @_;
+       $self->{oidx}->delete_by_num($docid);
+       $self->{oidx}->eidxq_del($docid);
+       $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);
+}
+
 sub _unref_doc ($$$$$;$) {
        my ($sync, $docid, $ibx, $xnum, $oidbin, $eml) = @_;
-       my $s = 'DELETE FROM xref3 WHERE ibx_id = ? AND oidbin = ?';
+       my $smsg;
+       if (ref($docid)) {
+               $smsg = $docid;
+               $docid = $smsg->{num};
+       }
+       if (defined($oidbin) && defined($xnum) && blessed($ibx) && $ibx->over) {
+               my $smsg = $ibx->over->get_art($xnum);
+               if ($smsg && pack('H*', $smsg->{blob}) eq $oidbin) {
+                       carp("BUG: (non-fatal) ".$ibx->eidx_key.
+                               " #$xnum $smsg->{blob} still valid");
+                       return;
+               }
+       }
+       my $s = 'DELETE FROM xref3 WHERE oidbin = ?';
+       $s .= ' AND ibx_id = ?' if defined($ibx);
        $s .= ' AND xnum = ?' if defined($xnum);
        my $del = $sync->{self}->{oidx}->dbh->prepare_cached($s);
-       $del->bind_param(1, $ibx->{-ibx_id});
-       $del->bind_param(2, $oidbin, SQL_BLOB);
-       $del->bind_param(3, $xnum) if defined($xnum);
+       my $col = 0;
+       $del->bind_param(++$col, $oidbin, SQL_BLOB);
+       $del->bind_param(++$col, $ibx->{-ibx_id}) if $ibx;
+       $del->bind_param(++$col, $xnum) if defined($xnum);
        $del->execute;
-       my $xr3 = $sync->{self}->{oidx}->get_xref3($docid, 1);
-       my $idx = $sync->{self}->idx_shard($docid);
+       my $xr3 = $sync->{self}->{oidx}->get_xref3($docid);
        if (scalar(@$xr3) == 0) { # all gone
-               $sync->{self}->{oidx}->delete_by_num($docid);
-               $sync->{self}->{oidx}->eidxq_del($docid);
-               $idx->ipc_do('xdb_remove', $docid);
+               remove_doc($sync->{self}, $docid);
        } else { # enqueue for reindex of remaining messages
-               my $ekey = $ibx->{-gc_eidx_key} // $ibx->eidx_key;
-               $idx->ipc_do('remove_eidx_info', $docid, $ekey, $eml);
-               $sync->{self}->{oidx}->eidxq_add($docid); # yes, add
+               if ($ibx) {
+                       my $ekey = $ibx->{-gc_eidx_key} // $ibx->eidx_key;
+                       my $idx = $sync->{self}->idx_shard($docid);
+                       $idx->ipc_do('remove_eidx_info', $docid, $ekey, $eml);
+               } # else: we can't remove_eidx_info in reindex-only path
+
+               # replace invalidated blob ASAP with something which should be
+               # readable since we may commit the transaction on checkpoint.
+               # eidxq processing will re-apply boost
+               $smsg //= $sync->{self}->{oidx}->get_art($docid);
+               my $hex = unpack('H*', $oidbin);
+               if ($smsg && $smsg->{blob} eq $hex) {
+                       $xr3->[0] =~ /:([a-f0-9]{40,}+)\z/ or
+                               die "BUG: xref $xr3->[0] has no OID";
+                       $sync->{self}->{oidx}->update_blob($smsg, $1);
+               }
+               # yes, add, we'll need to re-apply boost
+               $sync->{self}->{oidx}->eidxq_add($docid);
        }
        @$xr3
 }
@@ -167,6 +202,7 @@ sub do_xpost ($$) {
                $idx->ipc_do('add_eidx_info', $docid, $eidx_key, $eml);
                apply_boost($req, $smsg) if $req->{boost_in_use};
        } else { # 'd' no {xnum}
+               $self->git->async_wait_all;
                $oid = pack('H*', $oid);
                _unref_doc($req, $docid, $xibx, undef, $oid, $eml);
        }
@@ -230,24 +266,13 @@ sub do_step ($) { # main iterator for adding messages to the index
        do_finalize($req);
 }
 
-sub _blob_missing ($$) { # called when $smsg->{blob} is bad
+sub _blob_missing ($$) { # called when a known $smsg->{blob} is gone
        my ($req, $smsg) = @_;
-       my $self = $req->{self};
-       my $xref3 = $self->{oidx}->get_xref3($smsg->{num});
-       my @keep = grep(!/:$smsg->{blob}\z/, @$xref3);
-       if (@keep) {
-               warn "E: $smsg->{blob} gone, removing #$smsg->{num}\n";
-               $keep[0] =~ /:([a-f0-9]{40,}+)\z/ or
-                       die "BUG: xref $keep[0] has no OID";
-               my $oidhex = $1;
-               $self->{oidx}->remove_xref3($smsg->{num}, $smsg->{blob});
-               $self->{oidx}->update_blob($smsg, $oidhex) or warn <<EOM;
-E: #$smsg->{num} gone ($smsg->{blob} => $oidhex)
-EOM
-       } else {
-               warn "E: $smsg->{blob} gone, removing #$smsg->{num}\n";
-               $self->{oidx}->delete_by_num($smsg->{num});
-       }
+       # xnum and ibx are unknown, we only call this when an entry from
+       # /ei*/over.sqlite3 is bad, not on entries from xap*/over.sqlite3
+       my $oidbin = pack('H*', $smsg->{blob});
+       $req->{self}->git->async_wait_all;
+       _unref_doc($req, $smsg, undef, undef, $oidbin);
 }
 
 sub ck_existing { # git->cat_async callback
@@ -424,6 +449,12 @@ 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);
 
+       $nr = $self->{oidx}->dbh->do(<<'');
+DELETE FROM eidxq WHERE docid NOT IN (SELECT num FROM over)
+
+       warn "I: eliminated $nr stale reindex queue 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
@@ -532,12 +563,14 @@ sub _reindex_finalize ($$$) {
        }
        return if $nr == 1; # likely, all good
 
+       $self->git->async_wait_all;
        warn "W: #$docid split into $nr due to deduplication change\n";
        my @todo;
        for my $ary (values %$by_chash) {
                for my $x (reverse @$ary) {
                        warn "removing #$docid xref3 $x->{blob}\n";
-                       my $n = $self->{oidx}->remove_xref3($docid, $x->{blob});
+                       my $bin = pack('H*', $x->{blob});
+                       my $n = _unref_doc($sync, $docid, undef, undef, $bin);
                        die "BUG: $x->{blob} invalidated #$docid" if $n == 0;
                }
                my $x = pop(@$ary) // die "BUG: #$docid {by_chash} empty";
@@ -568,15 +601,14 @@ sub _reindex_oid { # git->cat_async callback
        my $expect_oid = $req->{xr3r}->[$req->{ix}]->[2];
        my $docid = $orig_smsg->{num};
        if (is_bad_blob($oid, $type, $size, $expect_oid)) {
-               my $remain = $self->{oidx}->remove_xref3($docid, $expect_oid);
+               my $oidbin = pack('H*', $expect_oid);
+               my $remain = _unref_doc($sync, $docid, undef, undef, $oidbin);
                if ($remain == 0) {
-                       warn "W: #$docid gone or corrupted\n";
-                       $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);
+                       warn "W: #$docid ($oid) gone or corrupt\n";
                } elsif (my $next_oid = $req->{xr3r}->[++$req->{ix}]->[2]) {
                        $self->git->cat_async($next_oid, \&_reindex_oid, $req);
                } else {
-                       warn "BUG: #$docid gone (UNEXPECTED)\n";
-                       $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);
+                       warn "BUG: #$docid ($oid) gone (UNEXPECTED)\n";
                }
                return;
        }
@@ -609,8 +641,7 @@ sub _reindex_smsg ($$$) {
                warn <<"";
 BUG? #$docid $smsg->{blob} is not referenced by inboxes during reindex
 
-               $self->{oidx}->delete_by_num($docid);
-               $self->idx_shard($docid)->ipc_do('xdb_remove', $docid);
+               remove_doc($self, $docid);
                return;
        }
 
@@ -732,9 +763,8 @@ sub prep_id2pos ($) {
 
 sub eidxq_process ($$) { # for reindexing
        my ($self, $sync) = @_;
-       return unless $self->{cfg};
-
-       return unless eidxq_lock_acquire($self);
+       local $self->{current_info} = 'eidxq process';
+       return unless ($self->{cfg} && eidxq_lock_acquire($self));
        my $dbh = $self->{oidx}->dbh;
        my $tot = $dbh->selectrow_array('SELECT COUNT(*) FROM eidxq') or return;
        ${$sync->{nr}} = 0;
@@ -877,6 +907,7 @@ ibx_id = ? AND xnum >= ? AND xnum <= ?
                }
                return if $sync->{quit};
                next unless scalar keys %x3m;
+               $self->git->async_wait_all; # wait for reindex_unseen
 
                # eliminate stale/mismatched entries
                my %mismatch = map { $_->{num} => $_->{blob} } @$msgs;
@@ -957,8 +988,7 @@ sub dd_smsg { # git->cat_async callback
                for my $smsg (@$ary) {
                        my $gone = $smsg->{num};
                        $oidx->merge_xref3($keep->{num}, $gone, $smsg->{blob});
-                       $self->idx_shard($gone)->ipc_do('xdb_remove', $gone);
-                       $oidx->delete_by_num($gone);
+                       remove_doc($self, $gone);
                }
        }
 }