extsearchidx: handle edits

We can now handle cases where messages are edited in one inbox
but not another, bifurcating the message.

V2Writable::log_range handles some edge-cases which could happen
in v2-only code paths, as well, but weren't usually triggered
due to default git-gc knobs not pruning immediately

3 files changed, 105 insertions, 31 deletions

diff --git a/lib/PublicInbox/ExtSearchIdx.pm b/lib/PublicInbox/ExtSearchIdx.pm
index 026e1377..bfe39891 100644
--- a/lib/PublicInbox/ExtSearchIdx.pm
+++ b/lib/PublicInbox/ExtSearchIdx.pm
@@ -19,7 +19,7 @@ use v5.10.1;
 use parent qw(PublicInbox::ExtSearch PublicInbox::Lock);
 use Carp qw(croak carp);
 use PublicInbox::Search;
-use PublicInbox::SearchIdx qw(crlf_adjust prepare_stack);
+use PublicInbox::SearchIdx qw(crlf_adjust prepare_stack is_ancestor);
 use PublicInbox::OverIdx;
 use PublicInbox::MID qw(mids);
 use PublicInbox::V2Writable;
@@ -119,6 +119,7 @@ sub do_xpost ($$) {
                 $self->{oidx}->add_xref3($docid, $xnum, $oid, $xibx->eidx_key);
                 $idx->shard_add_eidx_info($docid, $oid, $xibx, $eml);
         } else { # 'd'
+                $self->{oidx}->remove_xref3($docid, $oid, $xibx->eidx_key);
                 $idx->shard_remove_eidx_info($docid, $oid, $xibx, $eml);
         }
 }
@@ -176,14 +177,35 @@ sub do_step ($) { # main iterator for adding messages to the index
         do_finalize($req);
 }
 
+sub _blob_missing ($) { # called when req->{cur_smsg}->{blob} is bad
+        my ($req) = @_;
+        my $smsg = $req->{cur_smsg} or die 'BUG: {cur_smsg} missing';
+        my $self = $req->{self};
+        my $xref3 = $self->{oidx}->get_xref3($smsg->{num});
+        my @keep = grep(!/:$smsg->{blob}\z/, @$xref3);
+        if (@keep) {
+                $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});
+                my $upd = $self->{oidx}->update_blob($smsg, $oidhex);
+                my $saved = $self->{oidx}->get_art($smsg->{num});
+        } else {
+                $self->{oidx}->delete_by_num($smsg->{num});
+        }
+}
+
 sub ck_existing { # git->cat_async callback
         my ($bref, $oid, $type, $size, $req) = @_;
         my $smsg = $req->{cur_smsg} or die 'BUG: {cur_smsg} missing';
-        return if is_bad_blob($oid, $type, $size, $smsg->{blob});
-        my $cur = PublicInbox::Eml->new($bref);
-        if (content_hash($cur) eq $req->{chash}) {
-                push @{$req->{indexed}}, $smsg; # for do_xpost
-        } # else { index_unseen later }
+        if ($type eq 'missing') {
+                _blob_missing($req);
+        } elsif (!is_bad_blob($oid, $type, $size, $smsg->{blob})) {
+                my $cur = PublicInbox::Eml->new($bref);
+                if (content_hash($cur) eq $req->{chash}) {
+                        push @{$req->{indexed}}, $smsg; # for do_xpost
+                } # else { index_unseen later }
+        }
         do_step($req);
 }
 
@@ -281,15 +303,33 @@ sub eidx_sync { # main entry point
         PublicInbox::V2Writable::done($self);
 }
 
-sub update_last_commit {
+sub update_last_commit { # overrides V2Writable
         my ($self, $sync, $unit, $latest_cmt) = @_;
+        return unless defined $latest_cmt;
 
-        my $ALL = $self->git;
-        # while (scalar(@{$ALL->{inflight_c}}) || scalar(@{$ALL->{inflight}})) {
-                # $ALL->check_async_wait;
-                # $ALL->cat_async_wait;
-        # }
-        # TODO
+        $self->git->async_wait_all;
+        my $ibx = $sync->{ibx} or die 'BUG: {ibx} missing';
+        my $ekey = $ibx->eidx_key;
+        my $uv = $ibx->uidvalidity;
+        my $epoch = $unit->{epoch};
+        my $meta_key;
+        my $v = $ibx->version;
+        if ($v == 2) {
+                die 'No {epoch} for v2 unit' unless defined $epoch;
+                $meta_key = "lc-v2:$ekey//$uv;$epoch";
+        } elsif ($v == 1) {
+                die 'Unexpected {epoch} for v1 unit' if defined $epoch;
+                $meta_key = "lc-v1:$ekey//$uv";
+        } else {
+                die "Unsupported inbox version: $v";
+        }
+        my $last = $self->{oidx}->eidx_meta($meta_key);
+        if (defined $last && is_ancestor($unit->{git}, $last, $latest_cmt)) {
+                my @cmd = (qw(rev-list --count), "$last..$latest_cmt");
+                chomp(my $n = $unit->{git}->qx(@cmd));
+                return if $n ne '' && $n == 0;
+        }
+        $self->{oidx}->eidx_meta($meta_key, $latest_cmt);
 }
 
 sub idx_init { # similar to V2Writable
diff --git a/lib/PublicInbox/OverIdx.pm b/lib/PublicInbox/OverIdx.pm
index dff2780d..173e3220 100644
--- a/lib/PublicInbox/OverIdx.pm
+++ b/lib/PublicInbox/OverIdx.pm
@@ -261,6 +261,13 @@ sub subject_path ($) {
         lc($subj);
 }
 
+sub ddd_for ($) {
+        my ($smsg) = @_;
+        my $dd = $smsg->to_doc_data;
+        utf8::encode($dd);
+        compress($dd);
+}
+
 sub add_overview {
         my ($self, $eml, $smsg) = @_;
         $smsg->{lines} = $eml->body_raw =~ tr!\n!\n!;
@@ -272,10 +279,7 @@ sub add_overview {
                 $xpath = subject_path($subj);
                 $xpath = id_compress($xpath);
         }
-        my $dd = $smsg->to_doc_data;
-        utf8::encode($dd);
-        $dd = compress($dd);
-        add_over($self, $smsg, $mids, $refs, $xpath, $dd);
+        add_over($self, $smsg, $mids, $refs, $xpath, ddd_for($smsg));
 }
 
 sub _add_over {
@@ -589,14 +593,36 @@ INSERT OR IGNORE INTO xref3 (docid, ibx_id, xnum, oidbin) VALUES (?, ?, ?, ?)
 sub remove_xref3 {
         my ($self, $docid, $oidhex, $eidx_key) = @_;
         begin_lazy($self);
-        my $ibx_id = id_for($self, 'inboxes', 'ibx_id', eidx_key => $eidx_key);
         my $oidbin = pack('H*', $oidhex);
-        my $sth = $self->{dbh}->prepare_cached(<<'');
+        my $sth;
+        if (defined $eidx_key) {
+                my $ibx_id = id_for($self, 'inboxes', 'ibx_id',
+                                        eidx_key => $eidx_key);
+                $sth = $self->{dbh}->prepare_cached(<<'');
 DELETE FROM xref3 WHERE docid = ? AND ibx_id = ? AND oidbin = ?
 
-        $sth->bind_param(1, $docid);
-        $sth->bind_param(2, $ibx_id);
-        $sth->bind_param(3, $oidbin, SQL_BLOB);
+                $sth->bind_param(1, $docid);
+                $sth->bind_param(2, $ibx_id);
+                $sth->bind_param(3, $oidbin, SQL_BLOB);
+        } else {
+                $sth = $self->{dbh}->prepare_cached(<<'');
+DELETE FROM xref3 WHERE docid = ? AND oidbin = ?
+
+                $sth->bind_param(1, $docid);
+                $sth->bind_param(2, $oidbin, SQL_BLOB);
+        }
+        $sth->execute;
+}
+
+# for when an xref3 goes missing, this does NOT update {ts}
+sub update_blob {
+        my ($self, $smsg, $oidhex) = @_;
+        my $sth = $self->{dbh}->prepare(<<'');
+UPDATE over SET ddd = ? WHERE num = ?
+
+        $smsg->{blob} = $oidhex;
+        $sth->bind_param(1, ddd_for($smsg), SQL_BLOB);
+        $sth->bind_param(2, $smsg->{num});
         $sth->execute;
 }
 
diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm
index efda7907..0364857f 100644
--- a/lib/PublicInbox/V2Writable.pm
+++ b/lib/PublicInbox/V2Writable.pm
@@ -567,7 +567,7 @@ sub last_epoch_commit ($$;$) {
         $self->{mm}->last_commit_xap($v, $i, $cmt);
 }
 
-sub set_last_commits ($) {
+sub set_last_commits ($) { # this is NOT for ExtSearchIdx
         my ($self) = @_;
         defined(my $epoch_max = $self->{epoch_max}) or return;
         my $last_commit = $self->{last_commit};
@@ -992,9 +992,10 @@ sub log_range ($$$) {
 
         my $range = "$cur..$tip";
         $pr->("$i.git checking contiguity... ") if $pr;
-        if (is_ancestor($unit->{git}, $cur, $tip)) { # common case
+        my $git = $unit->{git};
+        if (is_ancestor($git, $cur, $tip)) { # common case
                 $pr->("OK\n") if $pr;
-                my $n = $unit->{git}->qx(qw(rev-list --count), $range);
+                my $n = $git->qx(qw(rev-list --count), $range);
                 chomp($n);
                 if ($n == 0) {
                         $sync->{ranges}->[$i] = undef;
@@ -1006,9 +1007,9 @@ sub log_range ($$$) {
                 $pr->("FAIL\n") if $pr;
                 warn <<"";
 discontiguous range: $range
-Rewritten history? (in $unit->{git}->{git_dir})
+Rewritten history? (in $git->{git_dir})
 
-                chomp(my $base = $unit->{git}->qx('merge-base', $tip, $cur));
+                chomp(my $base = $git->qx('merge-base', $tip, $cur));
                 if ($base) {
                         $range = "$base..$tip";
                         warn "found merge-base: $base\n"
@@ -1017,10 +1018,17 @@ Rewritten history? (in $unit->{git}->{git_dir})
                         warn "discarding history at $cur\n";
                 }
                 warn <<"";
-reindexing $unit->{git}->{git_dir} starting at
-$range
-
-                $unit->{unindex_range} = "$base..$cur";
+reindexing $git->{git_dir}
+starting at $range
+
+                # $cur^0 may no longer exist if pruned by git
+                if ($git->qx(qw(rev-parse -q --verify), "$cur^0")) {
+                        $unit->{unindex_range} = "$base..$cur";
+                } elsif ($base && $git->qx(qw(rev-parse -q --verify), $base)) {
+                        $unit->{unindex_range} = "$base..";
+                } else {
+                        warn "W: unable to unindex before $range\n";
+                }
         }
         $range;
 }