Fix Pusher bugs that omit sequences from checkpoint

I saw a failure of "Continuous Super-Fast Push" from a Windows build,
where the ending checkpoint was incorrect. Near the end of that test
it logged the active replicator checkpoint as:
    `{"local":42,"localCompleted":[0,43,48,154]}`
which is clearly wrong as it shows some sequences unable to be sent.

The test log showed the Pusher holding off on a bunch of consecutive
sequences ("...Found 0 changes up to #43" etc) because it was already
sending a revision of that doc. This was logged for sequences 43 thru
48, which is very close to the missing sequences in the checkpoint
(43-47).

From inspection, I deduced the problem to be that the code block
with comment "This doc already has a revision being sent; wait till
that one is done" stores the current rev into `iDoc->second->nextRev`
to be processed later; but if that field already holds a revision
that one gets dropped on the floor: it's obsoleted by the current
one and doesn't need to be sent, but the checkpointer is never
notified.

Adding a call to `_checkpointer.completedSequence` in that case should
fix the problem. (Also, the existing call to
`_checkpointer.addPendingSequence` is unnecessary since the Changes-
Feed already marks sequences as pending.)

Unfortunately I'm unable to reproduce the test failure locally; it
probably only occurs because the test machine is slow.

PS: Later I found another bug with the same effect, this one involving
`_conflictsIMightRetry`. Added it to the fix.

Author: snej

Replicator/Pusher+Revs.cc

@@ -289,7 +289,7 @@ namespace litecore::repl {
         }
     }
 
-    // If sending a rev that's been obsoleted by a newer one, mark the sequence as complete and send
+    // This RevToSend has been obsoleted by a newer one; mark the sequence as complete and set `*c4Err` to
     // a 410 Gone error. (Common subroutine of sendRevision & shouldRetryConflictWithNewerAncestor.)
     void Pusher::revToSendIsObsolete(const RevToSend& request, C4Error* c4err) {
         logInfo("Revision '%.*s' #%.*s is obsolete; not sending it", SPLAT(request.docID), SPLAT(request.revID));
@@ -375,7 +375,6 @@ namespace litecore::repl {
     // `synced` - whether the revision was successfully stored on the peer
     void Pusher::doneWithRev(RevToSend* rev, bool completed, bool synced) {
         if ( !passive() ) {
-            logDebug("** doneWithRev %.*s #%.*s", SPLAT(rev->docID), SPLAT(rev->revID));  //TEMP
             addProgress({rev->bodySize, 0});
             if ( completed ) {
                 _checkpointer.completedSequence(rev->sequence);

Replicator/Pusher.cc

@@ -125,10 +125,11 @@ namespace litecore::repl {
                 ++iChange;
             } else {
                 // This doc already has a revision being sent; wait till that one is done
-                logVerbose("Holding off on change '%.*s' %.*s till earlier rev %.*s is done", SPLAT(rev->docID),
-                           SPLAT(rev->revID), SPLAT(iDoc->second->revID));
-                iDoc->second->nextRev = rev;
-                if ( !passive() ) _checkpointer.addPendingSequence(rev->sequence);
+                logInfo("Holding off on change '%.*s' %.*s till earlier rev %.*s is done", SPLAT(rev->docID),
+                        SPLAT(rev->revID), SPLAT(iDoc->second->revID));
+                auto& nextRev = iDoc->second->nextRev;
+                if ( nextRev && !passive() ) revToSendIsObsolete(*nextRev, nullptr);
+                nextRev = std::move(rev);
                 iChange = changes.revs.erase(iChange);  // remove from `changes`
             }
         }
@@ -377,15 +378,18 @@ namespace litecore::repl {
             }
 
             if ( shouldRetryConflictWithNewerAncestor(change, serverRevID) ) {
-                // I have a newer revision to send in its place:
+                // Retry this revision with its updated remoteAncestorRevID:
                 RevToSendList changes = {change};
                 sendChanges(changes);
-                return true;
+                return true;  // return early, w/o caling doneWithRev()
             } else if ( _options->pull(collectionIndex()) <= kC4Passive ) {
+                // I'm not pulling, so there's no way to get the server revision. Give up.
                 C4Error error = C4Error::make(WebSocketDomain, 409, "conflicts with newer server revision"_sl);
                 finishedDocumentWithError(change, error, false);
             } else {
-                completed = false;
+                // The current server rev will be pulled soon, either updating the doc or creating
+                // a conflict that the app will merge (creating a new rev.)
+                // Either way, this rev is done, so leave `completed` true.
             }
         } else {
             // Other error:
@@ -447,9 +451,17 @@ namespace litecore::repl {
                         return true;
                     }
                 } else {
-                    // No change to remote ancestor, but try again later if it changes:
+                    // No change to remote ancestor, but try again later if it changes.
+                    // (if an earlier rev of the doc was in this state, overwrite it & mark its seq complete.)
                     logInfo("Will try again if remote rev of '%.*s' is updated", SPLAT(rev->docID));
-                    _conflictsIMightRetry.emplace(rev->docID, rev);
+                    if ( auto i = _conflictsIMightRetry.find(rev->docID); i != _conflictsIMightRetry.end() ) {
+                        if ( i->second->sequence < rev->sequence ) {
+                            revToSendIsObsolete(*i->second);
+                            i->second = rev;
+                        }
+                    } else {
+                        _conflictsIMightRetry.emplace(rev->docID, rev);
+                    }
                 }
             } else {
                 // Doc has changed, so this rev is obsolete
@@ -478,6 +490,7 @@ namespace litecore::repl {
                            "but local doc has changed",
                            SPLAT(docID), SPLAT(collectionSpec().scope), SPLAT(collectionSpec().name),
                            SPLAT(foreignAncestor));
+                revToSendIsObsolete(*rev, nullptr);
             } else if ( doc->selectRevision(foreignAncestor, false) && !(doc->selectedRev().flags & kRevIsConflict) ) {
                 // The remote rev is an ancestor of my revision, so retry it:
                 doc->selectCurrentRevision();

Replicator/tests/ReplicatorLoopbackTest.hh

@@ -437,7 +437,7 @@ class ReplicatorLoopbackTest
         auto& conflictHandlerRunning = _conflictHandlerRunning;
         _conflictHandler             = [resolvDB, &conflictHandlerRunning](ReplicatedRev* rev) {
             // Note: Can't use Catch (CHECK, REQUIRE) on a background thread
-            Log("Resolving conflict in '%.*s' ...", SPLAT(rev->docID));
+            Log("Resolving conflict in '%.*s' %.*s ...", SPLAT(rev->docID), SPLAT(rev->revID));
 
             conflictHandlerRunning = true;
             TransactionHelper t(resolvDB);
@@ -449,6 +449,12 @@ class ReplicatorLoopbackTest
                 WarnError("conflictHandler: Couldn't read doc '%.*s'", SPLAT(rev->docID));
                 Require(doc);
             }
+            if ( !(doc->flags & kDocConflicted) ) {
+                Log("conflictHandler: Doc '%.*s' not conflicted anymore (at %.*s)", SPLAT(rev->docID),
+                                SPLAT(doc->revID));
+                conflictHandlerRunning = false;
+                return;
+            }
             alloc_slice     localRevID = doc->selectedRev.revID;
             C4RevisionFlags localFlags = doc->selectedRev.flags;
             FLDict          localBody  = c4doc_getProperties(doc);
@@ -529,7 +535,9 @@ class ReplicatorLoopbackTest
                 string json = stringprintf(R"({"db":"%p","i":%d})", db, revNo);
                 revID       = createFleeceRev(collection, docID, nullslice, slice(json));
             }
-            Log("-------- %s %d: Created rev '%.*s' #%s --------", logName, revNo, SPLAT(docID), revID.c_str());
+            unsigned long long sequence = collection->getLastSequence();
+            Log("-------- %s %d: Created rev '%.*s' %s (seq #%llu) --------", logName, revNo, SPLAT(docID),
+                revID.c_str(), sequence);
         }
         Log("-------- %s: Done creating revs --------", logName);
     }