X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FExtSearchIdx.pm;h=7c44a1a406308604c1319b91eaa7130a0be645d2;hb=23af251dd607c4e75ab1e68063f2c885c48cc035;hp=67d720368922d6013253fc4be27c7464c02f3265;hpb=84897536c91023566556d3564da6f3d4316f557e;p=public-inbox.git

diff --git a/lib/PublicInbox/ExtSearchIdx.pm b/lib/PublicInbox/ExtSearchIdx.pm
index 67d72036..7c44a1a4 100644
--- a/lib/PublicInbox/ExtSearchIdx.pm
+++ b/lib/PublicInbox/ExtSearchIdx.pm
@@ -1,4 +1,4 @@
-# Copyright (C) 2020-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
 # Detached/external index cross inbox search indexing support
@@ -59,6 +59,7 @@ sub new {
 		nproc_shards({ nproc => $opt->{jobs} });
 	my $oidx = PublicInbox::OverIdx->new("$self->{xpfx}/over.sqlite3");
 	$self->{-no_fsync} = $oidx->{-no_fsync} = 1 if !$opt->{fsync};
+	$self->{-dangerous} = 1 if $opt->{dangerous};
 	$self->{oidx} = $oidx;
 	$self
 }
@@ -292,8 +293,8 @@ sub ck_existing { # git->cat_async callback
 
 # is the messages visible in the inbox currently being indexed?
 # return the number if so
-sub cur_ibx_xnum ($$) {
-	my ($req, $bref) = @_;
+sub cur_ibx_xnum ($$;$) {
+	my ($req, $bref, $mismatch) = @_;
 	my $ibx = $req->{ibx} or die 'BUG: current {ibx} missing';
 
 	$req->{eml} = PublicInbox::Eml->new($bref);
@@ -303,6 +304,7 @@ sub cur_ibx_xnum ($$) {
 		my ($id, $prev);
 		while (my $x = $ibx->over->next_by_mid($mid, \$id, \$prev)) {
 			return $x->{num} if $x->{blob} eq $req->{oid};
+			push @$mismatch, $x if $mismatch;
 		}
 	}
 	undef;
@@ -317,8 +319,15 @@ sub index_oid { # git->cat_async callback for 'm'
 		blob => $oid,
 	}, 'PublicInbox::Smsg';
 	$new_smsg->set_bytes($$bref, $size);
-	defined($req->{xnum} = cur_ibx_xnum($req, $bref)) or return;
 	++${$req->{nr}};
+	my $mismatch = [];
+	$req->{xnum} = cur_ibx_xnum($req, $bref, $mismatch) // do {
+		warn "# deleted\n";
+		warn "# mismatch $_->{blob}\n" for @$mismatch;
+		${$req->{latest_cmt}} = $req->{cur_cmt} //
+			die "BUG: {cur_cmt} unset ($oid)\n";
+		return;
+	};
 	do_step($req);
 }
 
@@ -859,14 +868,20 @@ sub _reindex_check_ibx ($$$) {
 	my $slice = 10000;
 	my $opt = { limit => $slice };
 	my ($beg, $end) = (1, $slice);
-	my $err = sync_inbox($self, $sync, $ibx) and return;
-	my $max = $ibx->mm->num_highwater;
+	my $ekey = $ibx->eidx_key;
+	my ($max, $max0);
+	do {
+		$max0 = $ibx->mm->num_highwater;
+		sync_inbox($self, $sync, $ibx) and return; # warned
+		$max = $ibx->mm->num_highwater;
+		return if $sync->{quit};
+	} while ($max > $max0 &&
+		warn("# $ekey moved $max0..$max, resyncing..\n"));
 	$end = $max if $end > $max;
 
 	# first, check if we missed any messages in target $ibx
 	my $msgs;
 	my $pr = $sync->{-opt}->{-progress};
-	my $ekey = $ibx->eidx_key;
 	local $sync->{-regen_fmt} = "$ekey checking %u/$max\n";
 	${$sync->{nr}} = 0;
 	my $fast = $sync->{-opt}->{fast};
@@ -902,10 +917,9 @@ ibx_id = ? AND xnum >= ? AND xnum <= ?
 				for my $num (@$docids) {
 					$self->{oidx}->eidxq_add($num);
 				}
-				return if $sync->{quit};
 			}
+			return if $sync->{quit};
 		}
-		return if $sync->{quit};
 		next unless scalar keys %x3m;
 		$self->git->async_wait_all; # wait for reindex_unseen
 
@@ -915,11 +929,22 @@ ibx_id = ? AND xnum >= ? AND xnum <= ?
 			my ($xnum, $hex) = unpack('JH*', $k);
 			my $bin = pack('H*', $hex);
 			my $exp = $mismatch{$xnum};
+			if (defined $exp) {
+				my $smsg = $ibx->over->get_art($xnum) // next;
+				# $xnum may be expired by another process
+				if ($smsg->{blob} eq $hex) {
+					warn <<"";
+BUG: (non-fatal) $ekey #$xnum $smsg->{blob} still matches (old exp: $exp)
+
+					next;
+				} # else: continue to unref
+			}
 			my $m = defined($exp) ? "mismatch (!= $exp)" : 'stale';
 			warn("# $xnum:$hex (#@$docids): $m\n");
 			for my $i (@$docids) {
 				_unref_doc($sync, $i, $ibx, $xnum, $bin);
 			}
+			return if $sync->{quit};
 		}
 	}
 	defined($hi) and ($hi < $max) and