From ed425ec59a59fc93fa475b261bfa05b738cbe0d4 Mon Sep 17 00:00:00 2001 From: Gregory Newman-Smith Date: Mon, 28 Sep 2026 14:00:01 +0100 Subject: [PATCH 1/5] CBG-5829: have better isgr test assertions --- rest/blip_api_crud_test.go | 2 +- .../replicator_conflict_test.go | 15 +- rest/replicatortest/replicator_test.go | 138 ++++++------ .../replicator_test_legacy_rev_test.go | 59 ++--- rest/revocation_test.go | 6 +- rest/utilities_testing_isgr.go | 202 ++++++++++++++++-- rest/utilities_testing_isgr_test.go | 80 +++++++ 7 files changed, 377 insertions(+), 125 deletions(-) create mode 100644 rest/utilities_testing_isgr_test.go diff --git a/rest/blip_api_crud_test.go b/rest/blip_api_crud_test.go index 5298157a78..b19aeab152 100644 --- a/rest/blip_api_crud_test.go +++ b/rest/blip_api_crud_test.go @@ -2773,7 +2773,7 @@ func TestProcessRevIncrementsStat(t *testing.T) { defer func() { require.NoError(t, ar.Stop()) }() activeRT.WaitForPendingChanges() - sgrRunner.WaitForVersion(docID, activeRT, version) + sgrRunner.RequireDocReplicated(docID, remoteRT, activeRT, version) base.RequireWaitForStat(t, pullStats.HandleRevCount.Value, 1) assert.NotEqualValues(t, 0, pullStats.HandleRevBytes.Value()) diff --git a/rest/replicatortest/replicator_conflict_test.go b/rest/replicatortest/replicator_conflict_test.go index 3c4333a174..dbf666fc64 100644 --- a/rest/replicatortest/replicator_conflict_test.go +++ b/rest/replicatortest/replicator_conflict_test.go @@ -2117,7 +2117,9 @@ func TestActiveReplicatorConflictRemoveCVFromCache(t *testing.T) { newRev := db.CreateRevIDWithBytes(2, rt2Version.RevTreeID, docBodyBytes) // generate what the new revID will be after conflict resolution conflictResVersion := rt1Version conflictResVersion.RevTreeID = newRev - rt2.WaitForVersion(docID, conflictResVersion) + // rt1 has resolved once the result reaches rt2. Don't wait on rt1 over REST: that would populate the rev + // cache this test asserts on. + sgrRunner.RequireDocReplicated(docID, rt1, rt2, conflictResVersion) // assert we cannot keep old cache entry for original wrote (before conflict resolution) on active cluster rest testers collectionActive1, ctxActive1 := active.GetSingleTestDatabaseCollectionWithUser() @@ -2172,8 +2174,10 @@ func TestActiveReplicatorV4DefaultResolverWithTombstoneLocal(t *testing.T) { conflictResVersion := rt1Version conflictResVersion.RevTreeID = newRev - // expect local doc to win and still be tombstone with remote docs revision history - sgrRunner.WaitForTombstone(docID, activeRT, conflictResVersion) + // expect local doc to win and still be tombstone with remote docs revision history. The local CV is new to + // passiveRT, so the resolved tombstone is pushed back. + activeRT.WaitForTombstone(docID, conflictResVersion) + sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, conflictResVersion) rt1Doc := activeRT.GetDocument(docID) // assert that remote cv is in pv now @@ -2202,6 +2206,7 @@ func TestActiveReplicatorV4DefaultResolverWithTombstoneRemote(t *testing.T) { rt2Version := passiveRT.PutDoc(docID, `{"source":"passiveRT","channels":["alice"]}`) rt2Version = passiveRT.DeleteDoc(docID, rt2Version) passiveRT.WaitForPendingChanges() + passiveDocBeforeReplication := rest.ExpectedISGRDocFromPeer(t, passiveRT, docID) // create conflicting update on activeRT rt1Version := activeRT.PutDoc(docID, `{"source":"activeRT","channels":["alice"]}`) // delete local version @@ -2223,7 +2228,9 @@ func TestActiveReplicatorV4DefaultResolverWithTombstoneRemote(t *testing.T) { newRev := getRevTreeID(t, rt1Version.RevTreeID, []byte(db.DeletedDocument)) conflictResVersion := rt2Version conflictResVersion.RevTreeID = newRev - sgrRunner.WaitForTombstone(docID, activeRT, conflictResVersion) + sgrRunner.RequireDoc(docID, activeRT, rest.ExpectedISGRDoc{Version: conflictResVersion, Deleted: true}) + // activeRT adopted passiveRT's CV, so there is nothing new to push back and passiveRT must be unchanged + sgrRunner.RequireDoc(docID, passiveRT, passiveDocBeforeReplication) rt1Doc := activeRT.GetDocument(docID) // assert that remote cv is in pv now diff --git a/rest/replicatortest/replicator_test.go b/rest/replicatortest/replicator_test.go index 41f549420b..41840b0106 100644 --- a/rest/replicatortest/replicator_test.go +++ b/rest/replicatortest/replicator_test.go @@ -2077,7 +2077,7 @@ func TestActiveReplicatorPullBasic(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.WaitForVersion(docID, rt1, version) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -2170,7 +2170,7 @@ func TestActiveReplicatorPullSkippedSequence(t *testing.T) { pullCheckpointer := ar.Pull.GetSingleCollection(t).Checkpointer // wait for the documents originally written to rt2 to arrive at rt1 - sgrRunner.WaitForVersion(docID1, rt1, doc1Version) + sgrRunner.RequireDocReplicated(docID1, rt2, rt1, doc1Version) base.RequireWaitForStat(t, func() int64 { return pullCheckpointer.Stats().ExpectedSequenceCount }, 1) base.RequireWaitForStat(t, func() int64 { return pullCheckpointer.Stats().ProcessedSequenceCount }, 1) @@ -2594,7 +2594,7 @@ func TestActiveReplicatorPullAttachments(t *testing.T) { assert.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.WaitForVersion(docID, rt1, version) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -2610,7 +2610,7 @@ func TestActiveReplicatorPullAttachments(t *testing.T) { version = rt2.PutDoc(docID, `{"source":"rt2","doc_num":2,`+attachment+`,"channels":["alice"]}`) // wait for the new document written to rt2 to arrive at rt1 - sgrRunner.WaitForVersion(docID, rt1, version) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) doc2, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) assert.NoError(t, err) @@ -3211,20 +3211,8 @@ func TestActiveReplicatorPushBasic(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt1 to arrive at rt2 - sgrRunner.WaitForVersion(docID, rt2, version) - - rt2collection, rt2ctx := rt2.GetSingleTestDatabaseCollection() - doc, err := rt2collection.GetDocument(rt2ctx, docID, db.DocUnmarshalAll) - assert.NoError(t, err) - - body, err := doc.GetDeepMutableBody() - require.NoError(t, err) - assert.Equal(t, "rt1", body["source"]) - - // the receiving peer must run its own sync function and persist the resulting channels - ChannelMap values - // are nil for channels the doc is currently in, so check for key presence - _, inChannel := doc.Channels[username] - assert.True(t, inChannel, "channels not persisted on pushed doc, channels=%v", doc.Channels) + // the receiving peer runs its own sync function, so this also requires the channel set to be persisted + sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) assert.Equal(t, strconv.FormatUint(localDoc.Sequence, 10), ar.GetStatus(ctx1).LastSeqPush) }) @@ -3279,16 +3267,7 @@ func TestActiveReplicatorPushAttachments(t *testing.T) { assert.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt1 to arrive at rt2 - sgrRunner.WaitForVersion(docID, rt2, version) - - rt2collection, rt2ctx := rt2.GetSingleTestDatabaseCollection() - doc, err := rt2collection.GetDocument(rt2ctx, docID, db.DocUnmarshalAll) - assert.NoError(t, err) - - body, err := doc.GetDeepMutableBody() - require.NoError(t, err) - assert.Equal(t, "rt1", body["source"]) - assert.Equal[any](t, json.Number("1"), body["doc_num"]) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) assert.Equal(t, int64(1), ar.Push.GetStats().HandleGetAttachment.Value()) @@ -3296,15 +3275,7 @@ func TestActiveReplicatorPushAttachments(t *testing.T) { version = rt1.PutDoc(docID, `{"source":"rt1","doc_num":2,`+attachment+`,"channels":["alice"]}`) // wait for the new document written to rt1 to arrive at rt2 - sgrRunner.WaitForVersion(docID, rt2, version) - - doc2, err := rt2collection.GetDocument(rt2ctx, docID, db.DocUnmarshalAll) - assert.NoError(t, err) - - body, err = doc2.GetDeepMutableBody() - require.NoError(t, err) - assert.Equal(t, "rt1", body["source"]) - assert.Equal[any](t, json.Number("2"), body["doc_num"]) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) // When targeting a Hydrogen node that supports proveAttachments, we typically end up sending // the attachment only once. However, targeting a Lithium node sends the attachment twice like @@ -3679,7 +3650,7 @@ func TestActiveReplicatorPushOneshot(t *testing.T) { doc, err := rt2collection.GetDocument(rt2ctx, docID, db.DocUnmarshalAll) require.NoError(t, err) - sgrRunner.WaitForVersion(docID, rt2, version) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) body, err := doc.GetDeepMutableBody() require.NoError(t, err) @@ -3731,7 +3702,7 @@ func TestActiveReplicatorPullTombstone(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.WaitForVersion(docID, rt1, version) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -3745,7 +3716,7 @@ func TestActiveReplicatorPullTombstone(t *testing.T) { deletedVersion := rt2.DeleteDoc(docID, version) // wait for the tombstone written to rt2 to arrive at rt1 - sgrRunner.WaitForTombstone(docID, rt1, deletedVersion) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, deletedVersion) doc, err = rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) assert.NoError(t, err) assert.True(t, doc.IsDeleted()) @@ -3797,7 +3768,7 @@ func TestActiveReplicatorPullPurgeOnRemoval(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.WaitForVersion(docID, rt1, version) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -4017,14 +3988,13 @@ func TestActiveReplicatorPullConflict(t *testing.T) { test.expectedLocalVersion.CV = rt2Version.CV } } + // A pull-only replication resolves on rt1 without pushing back, so rt2 isn't a valid source to compare + // against. Tombstones have no body to assert against. + expectedDoc := rest.ExpectedISGRDoc{Version: test.expectedLocalVersion, Deleted: test.skipBodyAssertion} if !test.skipBodyAssertion { - sgrRunner.WaitForVersion(docID, rt1, test.expectedLocalVersion) - // This is skipped for tombstone tests running with xattr as xattr tombstones don't have a body to assert - // against - requireBodyEqual(t, test.expectedLocalBody, doc) - } else { - sgrRunner.WaitForTombstone(docID, rt1, test.expectedLocalVersion) + expectedDoc.Body = test.expectedLocalBody } + sgrRunner.RequireDoc(docID, rt1, expectedDoc) t.Logf("Doc %s is %+v", docID, doc) for revID, revInfo := range doc.SyncData.History { t.Logf("doc revision [%s]: %+v", revID, revInfo) @@ -4165,6 +4135,7 @@ func TestActiveReplicatorPushAndPullConflict(t *testing.T) { rt2collection, rt2ctx := rt2.GetSingleTestDatabaseCollection() remoteDoc, err := rt2collection.GetDocument(rt2ctx, docID, db.DocUnmarshalSync) require.NoError(t, err) + remoteDocBeforeReplication := rest.ExpectedISGRDocFromPeer(t, rt2, docID) ctx1 := rt1.Context() // Create revision on rt1 (local) @@ -4269,9 +4240,15 @@ func TestActiveReplicatorPushAndPullConflict(t *testing.T) { rest.RequireChangeRev(t, test.expectedVersion, changesResults.Results[0].Changes[0], db.ChangesVersionTypeRevTreeID) t.Logf("Changes response is %+v", changesResults) - doc, err = rt2collection.GetDocument(rt2ctx, docID, db.DocUnmarshalAll) - require.NoError(t, err) - sgrRunner.WaitForVersion(docID, rt2, test.expectedVersion) + if sgrRunner.IsV4Protocol() && test.winner == remote { + // rt1 adopted rt2's CV, so there is nothing new to push back: rt2 must be exactly as it was. rt1's + // HLV also carries its losing version in pv, which rt2 never needs to learn about. + doc = sgrRunner.RequireDoc(docID, rt2, remoteDocBeforeReplication) + } else { + // rt1's resolution produced a version rt2 doesn't have, which is pushed back, so rt2 must now + // hold rt1's document + doc = sgrRunner.RequireDocReplicated(docID, rt1, rt2, test.expectedVersion) + } requireBodyEqual(t, test.expectedBody, doc) t.Logf("Remote Doc %s is %+v", docID, doc) t.Logf("Remote Doc %s attachments are %+v", docID, doc.Attachments()) @@ -4354,7 +4331,7 @@ func TestActiveReplicatorPushBasicWithInsecureSkipVerifyEnabled(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt1 to arrive at rt2 - sgrRunner.WaitForVersion(docID, rt2, version) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -5022,7 +4999,7 @@ func TestActiveReplicatorIgnoreNoConflicts(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt1 to arrive at rt2 - sgrRunner.WaitForVersion(rt1docID, rt2, rt1Version) + sgrRunner.RequireDocReplicated(rt1docID, rt1, rt2, rt1Version) rt2collection, rt2ctx := rt2.GetSingleTestDatabaseCollection() doc, err := rt2collection.GetDocument(rt2ctx, rt1docID, db.DocUnmarshalAll) @@ -5036,7 +5013,7 @@ func TestActiveReplicatorIgnoreNoConflicts(t *testing.T) { rt2Version := rt2.PutDoc(rt2docID, `{"source":"rt2","channels":["alice"]}`) // ... and wait to arrive at rt1 - sgrRunner.WaitForVersion(rt2docID, rt1, rt2Version) + sgrRunner.RequireDocReplicated(rt2docID, rt2, rt1, rt2Version) changesResults := rt1.WaitForChanges(2, "/{{.keyspace}}/_changes?since=0", "", true) assert.Equal(t, rt1docID, changesResults.Results[0].ID) assert.Equal(t, rt2docID, changesResults.Results[1].ID) @@ -5854,8 +5831,7 @@ func TestSGR2TombstoneConflictHandling(t *testing.T) { const doc2ID = "docid2" doc2Version := rest.DocVersion{RevTreeID: "3-abc"} - sgrRunner.WaitForVersion(doc2ID, localActiveRT, doc2Version) - sgrRunner.WaitForVersion(doc2ID, remotePassiveRT, doc2Version) + sgrRunner.RequireDocReplicated(doc2ID, localActiveRT, remotePassiveRT, doc2Version) // Stop the replication rest.RequireStatus(t, localActiveRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/replication?action=stop", ""), http.StatusOK) @@ -5896,8 +5872,12 @@ func TestSGR2TombstoneConflictHandling(t *testing.T) { localActiveRT.WaitForReplicationStatus("replication", db.ReplicationStateRunning) // Wait for the recently longest branch to show up on both sides - sgrRunner.WaitForTombstone(doc2ID, localActiveRT, deleteVersion) - sgrRunner.WaitForTombstone(doc2ID, remotePassiveRT, deleteVersion) + source, dest := localActiveRT, remotePassiveRT + if !test.longestBranchLocal { + source, dest = remotePassiveRT, localActiveRT + } + source.WaitForTombstoneRevIDOnly(doc2ID, deleteVersion) + sgrRunner.RequireDocReplicated(doc2ID, source, dest, deleteVersion) // Stop the replication rest.RequireStatus(t, localActiveRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/replication?action=stop", ""), http.StatusOK) @@ -5953,9 +5933,9 @@ func TestSGR2TombstoneConflictHandling(t *testing.T) { // Wait for doc to replicate from side resurrection was done on to the other side if test.resurrectLocal { - remotePassiveRT.WaitForVersion(doc2ID, expectedVersion) + sgrRunner.RequireDocReplicated(doc2ID, localActiveRT, remotePassiveRT, expectedVersion) } else { - localActiveRT.WaitForVersion(doc2ID, expectedVersion) + sgrRunner.RequireDocReplicated(doc2ID, remotePassiveRT, localActiveRT, expectedVersion) } }) } @@ -6337,7 +6317,7 @@ func TestLocalWinsConflictResolution(t *testing.T) { activeRT.CreateReplication(replicationID, remoteURLString, db.ActiveReplicatorTypePushAndPull, nil, true, db.ConflictResolverLocalWins, "") activeRT.WaitForReplicationStatus(replicationID, db.ReplicationStateRunning) - sgrRunner.WaitForVersion(docID, remoteRT, newVersion) + sgrRunner.RequireDocReplicated(docID, activeRT, remoteRT, newVersion) // Stop the replication response := activeRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/"+replicationID+"?action=stop", "") @@ -6534,7 +6514,7 @@ func TestReplicatorConflictAttachment(t *testing.T) { defer activeRT.DeleteReplication(replicationID) activeRT.WaitForReplicationStatus(replicationID, db.ReplicationStateRunning) - passiveRT.WaitForVersion(docID, newVersion) + sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, newVersion) response := activeRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/"+replicationID+"?action=stop", "") rest.RequireStatus(t, response, http.StatusOK) @@ -6561,6 +6541,7 @@ func TestReplicatorConflictAttachment(t *testing.T) { remoteGen++ newRemoteVersion = rest.NewDocVersionFromFakeRev(fmt.Sprintf("%d-remote", remoteGen)) remoteWinsVersion := passiveRT.PutNewEditsFalse(docID, newRemoteVersion, &remoteParentVersion, fmt.Sprintf(`{"_attachments": {"attach": {"stub": true, "revpos": %d, "digest":"sha1-gwwPApfQR9bzBKpqoEYwFmKp98A="}}}`, remoteGen-1)) + passiveDocBeforeReplication := rest.ExpectedISGRDocFromPeer(t, passiveRT, docID) response = activeRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/"+replicationID+"?action=start", "") rest.RequireStatus(t, response, http.StatusOK) @@ -6579,8 +6560,13 @@ func TestReplicatorConflictAttachment(t *testing.T) { } } } - activeRT.WaitForVersion(docID, expVersion) - passiveRT.WaitForVersion(docID, expVersion) + if sgrRunner.IsV4Protocol() && test.conflictResolution == db.ConflictResolverRemoteWins { + // activeRT adopted passiveRT's CV, so there is nothing new to push back and passiveRT must be unchanged + sgrRunner.RequireDoc(docID, activeRT, rest.ExpectedISGRDoc{Version: expVersion}) + sgrRunner.RequireDoc(docID, passiveRT, passiveDocBeforeReplication) + } else { + sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, expVersion) + } localDoc := activeRT.GetDocBody(docID) localVersion := localDoc.ExtractRev() @@ -6695,20 +6681,20 @@ func TestReplicatorDoNotSendDeltaWhenSrcIsTombstone(t *testing.T) { require.NoError(t, ar.Start(activeCtx)) // Wait for active to replicate to passive - sgrRunner.WaitForVersion("test", passiveRT, version) + sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, version) // Delete active document deletedVersion := activeRT.DeleteDoc("test", version) // Assert that the tombstone is replicated to passive // Get revision 2 on passive peer to assert it has been (a) replicated and (b) deleted - sgrRunner.WaitForTombstone("test", passiveRT, deletedVersion) + sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, deletedVersion) // Resurrect tombstoned document resurrectedVersion := activeRT.UpdateDoc("test", deletedVersion, `{"field2":"f2_2"}`) // Replicate resurrection to passive - sgrRunner.WaitForVersion("test", passiveRT, resurrectedVersion) + sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, resurrectedVersion) base.RequireWaitForStat(t, func() int64 { // should be 1 given delta is sent for delete, but not for resurrection @@ -6771,7 +6757,7 @@ func TestUnprocessableDeltas(t *testing.T) { require.NoError(t, ar.Start(activeCtx)) - sgrRunner.WaitForVersion("test", passiveRT, version) + sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, version) require.NoError(t, ar.Stop()) @@ -6792,7 +6778,7 @@ func TestUnprocessableDeltas(t *testing.T) { base.AssertLogContains(t, "Unable to unmarshal mutable body for doc test", func() { require.NoError(t, ar.Start(activeCtx)) // Check if it replicated - sgrRunner.WaitForVersion("test", passiveRT, version2) + sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, version2) require.NoError(t, ar.Stop()) }) }) @@ -6819,15 +6805,15 @@ func TestReplicatorIgnoreRemovalBodies(t *testing.T) { // Create the docs // // Doc rev 1 version1 := activeRT.PutDoc(docID, `{"key":"12","channels": ["rev1chan"]}`) - sgrRunner.WaitForVersion(docID, activeRT, version1) + activeRT.WaitForVersion(docID, version1) // doc rev 2 version2 := activeRT.UpdateDoc(docID, version1, `{"key":"12","channels":["rev2+3chan"]}`) - sgrRunner.WaitForVersion(docID, activeRT, version2) + activeRT.WaitForVersion(docID, version2) // Doc rev 3 version3 := activeRT.UpdateDoc(docID, version2, `{"key":"3","channels":["rev2+3chan"]}`) - sgrRunner.WaitForVersion(docID, activeRT, version3) + activeRT.WaitForVersion(docID, version3) activeRT.GetDatabase().FlushRevisionCacheForTest() err := collection.PurgeOldRevisionJSON(activeCtx, docID, version2.RevTreeID) @@ -7125,7 +7111,7 @@ func TestReplicatorCheckpointOnStop(t *testing.T) { activeRT.CreateReplication(replicationID, remoteURL, db.ActiveReplicatorTypePush, nil, true, db.ConflictResolverDefault, "") activeRT.WaitForReplicationStatus(replicationID, db.ReplicationStateRunning) - sgrRunner.WaitForVersion("test", passiveRT, rest.DocVersion{RevTreeID: revID, CV: *doc.HLV.ExtractCurrentVersionFromHLV()}) + sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, rest.DocVersion{RevTreeID: revID, CV: *doc.HLV.ExtractCurrentVersionFromHLV()}) // assert on the processed seq list being updated before stopping the active replicator ar, ok := activeRT.GetDatabase().SGReplicateMgr.GetLocalActiveReplicatorForTest(t, replicationID) @@ -8138,7 +8124,7 @@ func TestActiveReplicatorPullNewDocChannels(t *testing.T) { docID := rest.SafeDocumentName(t, t.Name()) + "rt2doc1" version := passiveRT.PutDoc(docID, `{"source":"rt2","channels":["`+chanName+`"]}`) - sgrRunner.WaitForVersion(docID, activeRT, version) + sgrRunner.RequireDocReplicated(docID, passiveRT, activeRT, version) rt1collection, rt1ctx := activeRT.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -8193,7 +8179,7 @@ func TestActiveReplicatorPushNewDocChannels(t *testing.T) { docID := rest.SafeDocumentName(t, t.Name()) + "rt1doc1" version := activeRT.PutDoc(docID, `{"source":"rt1","channels":["`+chanName+`"]}`) - sgrRunner.WaitForVersion(docID, passiveRT, version) + sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, version) passiveRTCollection, passiveRTCtx := passiveRT.GetSingleTestDatabaseCollection() doc, err := passiveRTCollection.GetDocument(passiveRTCtx, docID, db.DocUnmarshalAll) @@ -8255,11 +8241,11 @@ func TestActiveReplicatorPullUpdatedDocChannels(t *testing.T) { _, hasA := localDoc.Channels[chanA] require.True(t, hasA, "precondition: local write should set chanA, channels=%v", localDoc.Channels) - sgrRunner.WaitForVersion(docID, passiveRT, v1) + sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, v1) // now update the doc on rt2, moving it from chanA to chanB, and let it replicate back to rt1 v2 := passiveRT.UpdateDoc(docID, v1, `{"channels":["`+chanB+`"]}`) - sgrRunner.WaitForVersion(docID, activeRT, v2) + sgrRunner.RequireDocReplicated(docID, passiveRT, activeRT, v2) doc, err := activeRTCollection.GetDocument(activeRTCtx, docID, db.DocUnmarshalAll) require.NoError(t, err) diff --git a/rest/replicatortest/replicator_test_legacy_rev_test.go b/rest/replicatortest/replicator_test_legacy_rev_test.go index 829316a369..e2e8c25638 100644 --- a/rest/replicatortest/replicator_test_legacy_rev_test.go +++ b/rest/replicatortest/replicator_test_legacy_rev_test.go @@ -65,8 +65,8 @@ func TestActiveReplicatorPushPullLegacyRev(t *testing.T) { // Start the replicator require.NoError(t, ar.Start(ctx1)) - rt1.WaitForLegacyRev(docIDRT2, legacyRevRt2, []byte(`{"source":"rt2","channels":["alice"]}`)) - rt2.WaitForLegacyRev(docIDRT1, legacyRevRt1, []byte(`{"source":"rt1","channels":["alice"]}`)) + sgrRunner.RequireDocReplicated(docIDRT2, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRt2}) + sgrRunner.RequireDocReplicated(docIDRT1, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) }) } @@ -174,14 +174,14 @@ func TestActiveReplicatorBiDirectionalPreUpgradedDocOnPeer(t *testing.T) { require.NoError(t, ar.Start(ctx1)) if tc.newRevOnActivePeer { - rt2.WaitForLegacyRev(docID, legacyRevRT1, []byte(`{"source":"rt1","channels":["alice"]}`)) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRT1}) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{legacyRevRT1, initLegacyRevRT1}) // add new doc for some replication activity on pull side to allow us to assert that legacy rev 2-abc added // isn't pulled back to rt1 now it has a legacy revID encoded CV newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - rt1.WaitForVersion("newdoc", newDocVersion) // wait for it to arrive at rt2 + sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt2 // assert that the document isn't replicated back to rt1 after legacy rev CV is written on rt2 rt1Doc := rt1.GetDocument(docID) @@ -189,14 +189,14 @@ func TestActiveReplicatorBiDirectionalPreUpgradedDocOnPeer(t *testing.T) { assert.Nil(t, rt1Doc.HLV) rest.RequireHistoryContains(t, rt1Doc.History, []string{legacyRevRT1, initLegacyRevRT1}) } else { - rt1.WaitForLegacyRev(docID, legacyRevRT2, []byte(`{"source":"rt2","channels":["alice"]}`)) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) rt1Doc := rt1.GetDocument(docID) rest.RequireHistoryContains(t, rt1Doc.History, []string{initLegacyRevRT2, legacyRevRT2}) // add new doc for some replication activity on push side to allow us to assert that legacy rev 2-abc added // isn't pushed back to rt2 now it has a legacy revID encoded CV newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - rt2.WaitForVersion("newdoc", newDocVersion) // wait for it to arrive at rt2 + sgrRunner.RequireDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 // assert that the document isn't replicated back to rt2 after legacy rev CV is written on rt1 rt2Doc := rt2.GetDocument(docID) @@ -424,7 +424,7 @@ func TestActiveReplicatorBiDirectionalPreUpgradedRevInHistory(t *testing.T) { require.NoError(t, ar.Start(ctx1)) if !tc.activePeerHasUpgradedRev { - rt1.WaitForVersion(docID, upgradedDocVersion) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, upgradedDocVersion) rt1Doc := rt1.GetDocument(docID) rest.RequireHistoryContains(t, rt1Doc.History, []string{legacyRevRT1, initLegacyRevRT1, upgradedRevID}) @@ -442,7 +442,7 @@ func TestActiveReplicatorBiDirectionalPreUpgradedRevInHistory(t *testing.T) { require.JSONEq(t, expectedBody, string(actualBodyRT2)) require.JSONEq(t, expectedBody, string(actualBodyRT1)) } else { - rt2.WaitForVersion(docID, upgradedDocVersion) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, upgradedDocVersion) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{legacyRevRT2, initLegacyRevRT2, upgradedRevID}) @@ -530,8 +530,8 @@ func TestActiveReplicatorPushPullNewDocLegacyRevAndAllowUpdateAfter(t *testing.T // Start the replicator require.NoError(t, ar.Start(ctx1)) - rt1.WaitForLegacyRev(docIDToPull, legacyRevRt2, []byte(`{"source":"rt2","channels":["alice"]}`)) - rt2.WaitForLegacyRev(docIDToPush, legacyRevRt1, []byte(`{"source":"rt1","channels":["alice"]}`)) + sgrRunner.RequireDocReplicated(docIDToPull, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRt2}) + sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) // now update both docs to create a new revision on each side giving them each hlv based off their nodes source cvVersion, err := db.LegacyRevToRevTreeEncodedVersion(legacyRevRt1) @@ -540,7 +540,7 @@ func TestActiveReplicatorPushPullNewDocLegacyRevAndAllowUpdateAfter(t *testing.T RevTreeID: legacyRevRt1, } updateVer := rt1.UpdateDoc(docIDToPush, rt1DocVersion, `{"channels": ["alice"], "source": "rt1-updated"}`) - rt2.WaitForVersion(docIDToPush, updateVer) + sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, updateVer) // check rev tree encoded version is in pv finalRT2Doc := rt2.GetDocument(docIDToPush) assert.Equal(t, cvVersion.Value, finalRT2Doc.HLV.PreviousVersions[cvVersion.SourceID]) @@ -551,7 +551,7 @@ func TestActiveReplicatorPushPullNewDocLegacyRevAndAllowUpdateAfter(t *testing.T RevTreeID: legacyRevRt2, } updateVer = rt2.UpdateDoc(docIDToPull, rt2DocVersion, `{"channels": ["alice"], "source": "rt2-updated"}`) - rt1.WaitForVersion(docIDToPull, updateVer) + sgrRunner.RequireDocReplicated(docIDToPull, rt2, rt1, updateVer) finalRT1Doc := rt1.GetDocument(docIDToPull) assert.Equal(t, cvVersion.Value, finalRT1Doc.HLV.PreviousVersions[cvVersion.SourceID]) @@ -779,13 +779,13 @@ func TestActiveReplicatorConflictPreUpgradedVersionEachSide(t *testing.T) { verPostConflictRes, _ = rt1.GetDoc(docID) assert.NotEqual(c, legacyRevRT1, verPostConflictRes.RevTreeID) }, 10*time.Second, 50*time.Millisecond) - rt2.WaitForLegacyRev(docID, verPostConflictRes.RevTreeID, []byte(`{"channels":["alice"],"source":"rt1"}`)) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, rest.DocVersion{RevTreeID: verPostConflictRes.RevTreeID}) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, legacyRevRT2, verPostConflictRes.RevTreeID}) // add doc for some replication activity on pull to assert that local doesn't change after remote is written with legacy CV newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - rt1.WaitForVersion("newdoc", newDocVersion) // wait for it to arrive at rt1 + sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 // assert active side doc hasn't changed rt1Doc := rt1.GetDocument(docID) @@ -802,14 +802,14 @@ func TestActiveReplicatorConflictPreUpgradedVersionEachSide(t *testing.T) { return replicationStats.ConflictResolvedRemoteCount.Value() }, 1) - rt1.WaitForLegacyRev(docID, legacyRevRT2, []byte(`{"channels":["alice"],"source":"rt2"}`)) + sgrRunner.RequireDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) rt1Doc := rt1.GetDocument(docID) tombstonedID := db.CreateRevIDWithBytes(3, legacyRevRT1, []byte(db.DeletedDocument)) // create what would be the tombstone rev id for local branch rest.RequireHistoryContains(t, rt1Doc.History, []string{initLegacyRevRT1, legacyRevRT1, legacyRevRT2, tombstonedID}) // add doc for some replication activity on push to assert that local doesn't change after remote is written with legacy CV newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - rt2.WaitForVersion("newdoc", newDocVersion) // wait for it to arrive at rt2 + sgrRunner.RequireDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 // assert passive side doc hasn't changed rt2Doc := rt2.GetDocument(docID) @@ -944,6 +944,8 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { require.NoError(t, ar.Stop()) }() + passiveDocBeforeReplication := rest.ExpectedISGRDocFromPeer(t, rt2, docID) + // Start the replicator require.NoError(t, ar.Start(ctx1)) @@ -961,7 +963,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { assert.NotEqual(c, legacyRevRT1, verPostConflictRes.RevTreeID) }, 10*time.Second, 50*time.Millisecond) // wait for this resolution to be pushed back to passive peer - rt2.WaitForVersion(docID, verPostConflictRes) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, verPostConflictRes) // assert original upgraded version in PV history assert.Equal(t, upgradedDocVersion.CV.Value, replicatedDoc.HLV.PreviousVersions[upgradedDocVersion.CV.SourceID]) @@ -976,7 +978,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { // add doc for some replication activity on pull to assert that local doesn't change after remote is written with legacy CV newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - rt1.WaitForVersion("newdoc", newDocVersion) // wait for it to arrive at rt1 + sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 // assert active side doc hasn't changed rt1Doc := rt1.GetDocument(docID) @@ -993,7 +995,12 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { base.RequireWaitForStat(t, func() int64 { return replicationStats.ConflictResolvedRemoteCount.Value() }, 1) - rt1.WaitForVersion(docID, upgradedDocVersion) + // rt1 adopts rt2's CV and keeps its losing legacy branch in pv, so its HLV legitimately differs from rt2's + sgrRunner.RequireDoc(docID, rt1, rest.ExpectedISGRDoc{ + Version: upgradedDocVersion, + Body: expectedBody, + Channels: []string{username}, + }) rt1Doc := rt1.GetDocument(docID) tombstonedID := db.CreateRevIDWithBytes(3, legacyRevRT1, []byte(db.DeletedDocument)) // create what would be the tombstone rev id for local branch @@ -1004,10 +1011,10 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { // add doc for some replication activity on push to assert that remote doesn't change after remote wins conflict res is done newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - rt2.WaitForVersion("newdoc", newDocVersion) // wait for it to arrive at rt2 + sgrRunner.RequireDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 - // assert passive side doc hasn't changed - rt2Doc := rt2.GetDocument(docID) + // assert passive side doc hasn't changed - rt1 adopted its CV, so there was nothing to push back + rt2Doc := sgrRunner.RequireDoc(docID, rt2, passiveDocBeforeReplication) rest.RequireDocVersionEqual(t, upgradedDocVersion, rt2Doc.ExtractDocVersion()) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, upgradedDocVersion.RevTreeID}) @@ -1019,7 +1026,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { // update doc on active side to ensure we can push with no conflict updateVer := rt1.UpdateDoc(docID, upgradedDocVersion, `{"channels": ["alice"], "source": "rt1-updated"}`) - rt2.WaitForVersion(docID, updateVer) + sgrRunner.RequireDocReplicated(docID, rt1, rt2, updateVer) } }) } @@ -1088,7 +1095,7 @@ func TestActiveReplicatorDeltaSyncWhenBothSidesLegacy(t *testing.T) { // Start the replicator require.NoError(t, ar.Start(ctx1)) - rt2.WaitForLegacyRev(docIDToPush, legacyRevRt1, []byte(`{"source":"rt1","channels":["alice"]}`)) + sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) base.RequireWaitForStat(t, func() int64 { return replicationStats.PushDeltaSentCount.Value() @@ -1144,7 +1151,7 @@ func TestDeltaSyncWhenOneSideHasEncodedCV(t *testing.T) { // Start the replicator require.NoError(t, ar.Start(ctx1)) - rt2.WaitForLegacyRev(docIDToPush, legacyInitRevRt1, []byte(`{"source":"rt1","channels":["alice"]}`)) + sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyInitRevRt1}) // flush revision cache to remove old reference to rev 1 in rev cache rt1.GetDatabase().FlushRevisionCacheForTest() @@ -1154,7 +1161,7 @@ func TestDeltaSyncWhenOneSideHasEncodedCV(t *testing.T) { // 1. update doc on rt1 to give HLV based of rt1 sourceID // 2. push doc to rt2 with delta from rev1 to rev2 upgradeVersion := rt1.UpdateDoc(docIDToPush, db.DocVersion{RevTreeID: legacyInitRevRt1}, `{"channels": ["alice"], "source": "rt1-updated"}`) - rt1.WaitForVersion(docIDToPush, upgradeVersion) + sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, upgradeVersion) base.RequireWaitForStat(t, func() int64 { return replicationStats.PushDeltaSentCount.Value() diff --git a/rest/revocation_test.go b/rest/revocation_test.go index e2ed2866a9..b0379e623a 100644 --- a/rest/revocation_test.go +++ b/rest/revocation_test.go @@ -2121,9 +2121,9 @@ func TestReplicatorRevocationsFromZero(t *testing.T) { rt1.WaitForReplicationStatus(ar.ID, db.ReplicationStateStopped) // Be sure docs have arrived - sgrRunner.WaitForVersion("docA", rt1, docAVersion) - sgrRunner.WaitForVersion("docA1", rt1, docA1Version) - sgrRunner.WaitForVersion("docA2", rt1, docA2Version) + sgrRunner.RequireDocReplicated("docA", rt2, rt1, docAVersion) + sgrRunner.RequireDocReplicated("docA1", rt2, rt1, docA1Version) + sgrRunner.RequireDocReplicated("docA2", rt2, rt1, docA2Version) // Reset checkpoint (since 0) require.NoError(t, ar.Reset()) diff --git a/rest/utilities_testing_isgr.go b/rest/utilities_testing_isgr.go index ce011970eb..ba38c15324 100644 --- a/rest/utilities_testing_isgr.go +++ b/rest/utilities_testing_isgr.go @@ -10,6 +10,7 @@ package rest import ( "context" + "maps" "net/http/httptest" "net/url" "slices" @@ -255,25 +256,196 @@ func (runner *SGRTestRunner) IsV4Protocol() bool { return slices.Contains(runner.SupportedSubprotocols, db.CBMobileReplicationV4.SubprotocolString()) } -// WaitForVersion will wait for revtree if v3 protocol or full version otherwise. -func (runner *SGRTestRunner) WaitForVersion(docID string, rt *RestTester, version DocVersion) { - rt.TB().Helper() - if !runner.IsV4Protocol() { - // only assert on rev tree IDs when we're not replicating using v4 protocol - rt.WaitForVersionRevIDOnly(docID, version) - return +// ExpectedISGRDoc describes the state a document must be in on a peer after replication. Version is always checked; +// the remaining fields are checked when set, except in RequireDocReplicated where every field comes from the source. +type ExpectedISGRDoc struct { + // Version is waited for before anything else is asserted - revTreeID only in v3, revTreeID and CV in v4. + Version DocVersion + // Deleted is true if the current revision must be a tombstone. + Deleted bool + // HLV is compared in full apart from cvCAS, which each peer sets from its own write. Only checked in v4, since in + // v3 the receiving peer mints a new HLV of its own. + HLV *db.HybridLogicalVector + // RevChain is the ancestry of the current revision, current revision first. + RevChain []string + // Channels are the channels the document is currently in, ignoring channels it has been removed from. + Channels []string + // Body is the expected document body as JSON. Not checked for tombstones. + Body string + // Attachments maps attachment name to digest. + Attachments map[string]string +} + +// RequireDocReplicated waits for version to arrive on dest, then requires dest to hold the same document as source: +// HLV (v4), rev tree ancestry, channels, body and attachments. Both peers are waited on to reach version, so it can't +// be used where the peers legitimately end up different - use RequireDoc with an explicit expectation there. Only the +// parts of version that are set are waited on, so a revTreeID alone is enough. +func (runner *SGRTestRunner) RequireDocReplicated(docID string, source, dest *RestTester, version DocVersion) *db.Document { + t := dest.TB() + t.Helper() + // Wait on both peers: source's state can itself be the product of the replication (a conflict resolution, say), + // and dest can already be at version before the replication has done anything. + runner.waitForDocVersion(source, docID, version) + runner.waitForDocVersion(dest, docID, version) + return runner.RequireDoc(docID, dest, ExpectedISGRDocFromPeer(t, source, docID)) +} + +// RequireDoc waits for exp.Version to arrive on rt, then requires the document to match every field set in exp. +func (runner *SGRTestRunner) RequireDoc(docID string, rt *RestTester, exp ExpectedISGRDoc) *db.Document { + t := rt.TB() + t.Helper() + runner.waitForDocVersion(rt, docID, exp.Version) + + doc := rt.GetDocument(docID) + actual := expectedISGRDocFromDoc(t, rt.Context(), doc) + peer := rt.GetDatabase().Name + + require.NotNil(t, doc.HLV, "doc %q has no HLV on %s", docID, peer) + if runner.IsV4Protocol() { + if exp.HLV != nil { + assert.True(t, hlvEqualAllowingEncodedRevs(t, exp.HLV, doc.HLV, exp.RevChain), + "HLV mismatch for doc %q on %s. Expected: %s, Actual: %s", docID, peer, exp.HLV.HLVDebugString(), doc.HLV.HLVDebugString()) + } + } else { + assert.Equal(t, rt.GetDatabase().EncodedSourceID, doc.HLV.SourceID, + "doc %q on %s should have a CV minted locally in v3, HLV: %s", docID, peer, doc.HLV.HLVDebugString()) + } + + assert.Equal(t, exp.Deleted, actual.Deleted, "deleted mismatch for doc %q on %s", docID, peer) + if exp.RevChain != nil { + assert.Equal(t, exp.RevChain, actual.RevChain, "rev tree ancestry mismatch for doc %q on %s", docID, peer) } - rt.WaitForVersion(docID, version) + // Any branch other than the current revision's must have been resolved to a tombstone. + for _, leaf := range doc.History.GetLeaves() { + if leaf != doc.GetRevTreeID() { + assert.True(t, doc.History[leaf].Deleted, "doc %q on %s has live conflicting leaf %q alongside current revision %q", + docID, peer, leaf, doc.GetRevTreeID()) + } + } + if exp.Channels != nil { + assert.ElementsMatch(t, exp.Channels, actual.Channels, "channel mismatch for doc %q on %s", docID, peer) + } + if exp.Body != "" && !exp.Deleted { + assert.JSONEq(t, exp.Body, actual.Body, "body mismatch for doc %q on %s", docID, peer) + } + if exp.Attachments != nil { + assert.Equal(t, exp.Attachments, actual.Attachments, "attachment mismatch for doc %q on %s", docID, peer) + } + return doc } -func (runner *SGRTestRunner) WaitForTombstone(docID string, rt *RestTester, version DocVersion) { - rt.TB().Helper() - if !slices.Contains(runner.SupportedSubprotocols, db.CBMobileReplicationV4.SubprotocolString()) { - // only assert on rev tree IDs when we're not replicating using v4 protocol - rt.WaitForTombstoneRevIDOnly(docID, version) - return +// waitForDocVersion waits for docID on rt to be at version - revTreeID only in v3, where the receiving peer mints its own +// CV. It reads the document directly rather than over REST so that waiting doesn't populate the revision cache, and so +// that tombstones and live documents are waited for the same way. +func (runner *SGRTestRunner) waitForDocVersion(rt *RestTester, docID string, version DocVersion) { + t := rt.TB() + t.Helper() + checkCV := runner.IsV4Protocol() && !version.CV.IsEmpty() + require.True(t, version.RevTreeID != "" || checkCV, "nothing to wait for in version %#v", version) + collection, ctx := rt.GetSingleTestDatabaseCollection() + require.EventuallyWithT(t, func(c *assert.CollectT) { + doc, err := collection.GetDocument(ctx, docID, db.DocUnmarshalSync) + if !assert.NoError(c, err, "doc %q not found on %s", docID, rt.GetDatabase().Name) { + return + } + if version.RevTreeID != "" { + assert.Equal(c, version.RevTreeID, doc.GetRevTreeID(), "doc %q on %s", docID, rt.GetDatabase().Name) + } + if checkCV { + assert.Equal(c, version.CV.String(), doc.HLV.GetCurrentVersionString(), "doc %q on %s", docID, rt.GetDatabase().Name) + } + }, 10*time.Second, 50*time.Millisecond) +} + +// hlvEqualAllowingEncodedRevs is HybridLogicalVector.Equal, except that actual may also carry pv entries for the +// revTreeID-encoded version of revisions in revChain. A peer that received a document while it had no HLV stores its +// revTreeID as an encoded CV, and keeps that in pv once the document is updated - but the peer that wrote the update +// never stored the encoded version, since it only ever knew the revision through its rev tree. Both peers know the +// same versions, so the HLVs are equivalent. +func hlvEqualAllowingEncodedRevs(t testing.TB, expected, actual *db.HybridLogicalVector, revChain []string) bool { + t.Helper() + if expected.SourceID != actual.SourceID || expected.Version != actual.Version || + !maps.Equal(expected.MergeVersions, actual.MergeVersions) { + return false + } + // Every encoded version shares one source ID, so collect them as versions rather than keyed by source. + encodedRevs := make(map[db.Version]struct{}, len(revChain)) + for _, revID := range revChain { + encoded, err := db.LegacyRevToRevTreeEncodedVersion(revID) + if err != nil { + // A revTreeID without a hex digest (a fake one written by a test) can't be encoded, so no peer can hold an + // encoded version of it. + continue + } + encodedRevs[encoded] = struct{}{} + } + for source, value := range expected.PreviousVersions { + if actualValue, ok := actual.PreviousVersions[source]; !ok || actualValue != value { + return false + } + } + for source, value := range actual.PreviousVersions { + if _, ok := expected.PreviousVersions[source]; ok { + continue + } + if _, ok := encodedRevs[db.Version{SourceID: source, Value: value}]; !ok { + return false + } + } + return true +} + +// ExpectedISGRDocFromPeer returns the current state of docID on rt as an ExpectedISGRDoc. Taken before a replication +// starts, it lets RequireDoc assert that a peer the replication shouldn't touch is left unchanged. +func ExpectedISGRDocFromPeer(t testing.TB, rt *RestTester, docID string) ExpectedISGRDoc { + t.Helper() + return expectedISGRDocFromDoc(t, rt.Context(), rt.GetDocument(docID)) +} + +// expectedISGRDocFromDoc returns the replication-relevant state of doc, with every field populated. +func expectedISGRDocFromDoc(t testing.TB, ctx context.Context, doc *db.Document) ExpectedISGRDoc { + t.Helper() + hlv := doc.HLV.Copy() + if hlv == nil { + // A document written before HLVs existed replicates with a CV encoded from its revTreeID, so that's the HLV a + // peer receiving it must end up with. + encodedCV, err := db.LegacyRevToRevTreeEncodedVersion(doc.GetRevTreeID()) + require.NoError(t, err) + hlv = db.NewHybridLogicalVector() + require.NoError(t, hlv.AddVersion(encodedCV)) + } + exp := ExpectedISGRDoc{ + Version: DocVersion{RevTreeID: doc.GetRevTreeID(), CV: *hlv.ExtractCurrentVersionFromHLV()}, + Deleted: doc.IsDeleted(), + HLV: hlv, + RevChain: []string{}, + Channels: []string{}, + Attachments: map[string]string{}, + } + for revID := doc.GetRevTreeID(); revID != ""; { + revInfo, ok := doc.History[revID] + require.True(t, ok, "rev %q missing from rev tree of doc %q", revID, doc.ID) + exp.RevChain = append(exp.RevChain, revID) + revID = revInfo.Parent + } + for channel, removal := range doc.Channels { + if removal == nil { + exp.Channels = append(exp.Channels, channel) + } + } + slices.Sort(exp.Channels) + for name, meta := range doc.Attachments() { + metaMap, ok := meta.(map[string]any) + require.True(t, ok, "attachment %q of doc %q has unexpected metadata %T", name, doc.ID, meta) + digest, _ := metaMap["digest"].(string) + exp.Attachments[name] = digest + } + if !exp.Deleted { + body, err := doc.BodyBytes(ctx) + require.NoError(t, err) + exp.Body = string(body) } - rt.WaitForTombstone(docID, version) + return exp } // Run is equivalent to testing.T.Run() but updates underlying the RestTesters' TB to the new testing.T. diff --git a/rest/utilities_testing_isgr_test.go b/rest/utilities_testing_isgr_test.go new file mode 100644 index 0000000000..b2e1b47ece --- /dev/null +++ b/rest/utilities_testing_isgr_test.go @@ -0,0 +1,80 @@ +// Copyright 2026-Present Couchbase, Inc. +// +// Use of this software is governed by the Business Source License included +// in the file licenses/BSL-Couchbase.txt. As of the Change Date specified +// in that file, in accordance with the Business Source License, use of this +// software will be governed by the Apache License, Version 2.0, included in +// the file licenses/APL2.txt. + +package rest + +import ( + "testing" + + "github.com/couchbase/sync_gateway/db" + "github.com/couchbase/sync_gateway/testing/assert" + "github.com/couchbase/sync_gateway/testing/require" +) + +func TestHLVEqualAllowingEncodedRevs(t *testing.T) { + const chainRev, otherRev = "2-abcdef1234", "1-9876543210" + // "1-initial" is a fake revTreeID with no hex digest, which has no encoded version and must not stop the comparison + revChain := []string{chainRev, "1-initial"} + + encoded := func(revID string) db.Version { + v, err := db.LegacyRevToRevTreeEncodedVersion(revID) + require.NoError(t, err) + return v + } + newHLV := func(pv ...db.Version) *db.HybridLogicalVector { + hlv := db.NewHybridLogicalVector() + hlv.SourceID, hlv.Version = "cv", 100 + for _, v := range pv { + hlv.PreviousVersions[v.SourceID] = v.Value + } + return hlv + } + + testCases := []struct { + name string + expected *db.HybridLogicalVector + actual *db.HybridLogicalVector + equal bool + }{ + { + name: "identical", + expected: newHLV(db.Version{SourceID: "a", Value: 1}), + actual: newHLV(db.Version{SourceID: "a", Value: 1}), + equal: true, + }, + { + name: "extra encoded version of a revision in the chain", + expected: newHLV(), + actual: newHLV(encoded(chainRev)), + equal: true, + }, + { + name: "extra encoded version of a revision outside the chain", + expected: newHLV(), + actual: newHLV(encoded(otherRev)), + equal: false, + }, + { + name: "extra version that isn't revTreeID encoded", + expected: newHLV(), + actual: newHLV(db.Version{SourceID: "a", Value: 1}), + equal: false, + }, + { + name: "missing expected pv entry", + expected: newHLV(encoded(chainRev)), + actual: newHLV(), + equal: false, + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.equal, hlvEqualAllowingEncodedRevs(t, tc.expected, tc.actual, revChain)) + }) + } +} From 25dad9bac172dcc0751ef42c2c0c69d9c3dd0a9c Mon Sep 17 00:00:00 2001 From: Gregory Newman-Smith Date: Tue, 29 Sep 2026 10:47:27 +0100 Subject: [PATCH 2/5] update comment --- rest/replicatortest/replicator_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rest/replicatortest/replicator_test.go b/rest/replicatortest/replicator_test.go index 41840b0106..4305f9019e 100644 --- a/rest/replicatortest/replicator_test.go +++ b/rest/replicatortest/replicator_test.go @@ -5871,7 +5871,8 @@ func TestSGR2TombstoneConflictHandling(t *testing.T) { rest.RequireStatus(t, localActiveRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/replication?action=start", ""), http.StatusOK) localActiveRT.WaitForReplicationStatus("replication", db.ReplicationStateRunning) - // Wait for the recently longest branch to show up on both sides + // Wait for the recently longest branch to show up on both sides. The side that wrote the winning tombstone + // is the source, so a mismatch is reported against the peer that received it. source, dest := localActiveRT, remotePassiveRT if !test.longestBranchLocal { source, dest = remotePassiveRT, localActiveRT From ff553db066293ad2e00c4271ed0217fb6cbe6310 Mon Sep 17 00:00:00 2001 From: Gregory Newman-Smith Date: Tue, 29 Sep 2026 11:17:18 +0100 Subject: [PATCH 3/5] address review --- .../replicator_test_legacy_rev_test.go | 2 +- rest/utilities_testing_isgr.go | 47 +++++--- rest/utilities_testing_isgr_test.go | 109 ++++++++++++++++++ 3 files changed, 141 insertions(+), 17 deletions(-) diff --git a/rest/replicatortest/replicator_test_legacy_rev_test.go b/rest/replicatortest/replicator_test_legacy_rev_test.go index e2e8c25638..183cfdcdad 100644 --- a/rest/replicatortest/replicator_test_legacy_rev_test.go +++ b/rest/replicatortest/replicator_test_legacy_rev_test.go @@ -181,7 +181,7 @@ func TestActiveReplicatorBiDirectionalPreUpgradedDocOnPeer(t *testing.T) { // add new doc for some replication activity on pull side to allow us to assert that legacy rev 2-abc added // isn't pulled back to rt1 now it has a legacy revID encoded CV newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt2 + sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 // assert that the document isn't replicated back to rt1 after legacy rev CV is written on rt2 rt1Doc := rt1.GetDocument(docID) diff --git a/rest/utilities_testing_isgr.go b/rest/utilities_testing_isgr.go index ba38c15324..b2c87463fd 100644 --- a/rest/utilities_testing_isgr.go +++ b/rest/utilities_testing_isgr.go @@ -266,6 +266,9 @@ type ExpectedISGRDoc struct { // HLV is compared in full apart from cvCAS, which each peer sets from its own write. Only checked in v4, since in // v3 the receiving peer mints a new HLV of its own. HLV *db.HybridLogicalVector + // NoHLV is true if the document must have no HLV at all - a document written before HLVs existed, that nothing has + // written to since. Needed as well as HLV because a nil HLV means the HLV isn't checked. + NoHLV bool // RevChain is the ancestry of the current revision, current revision first. RevChain []string // Channels are the channels the document is currently in, ignoring channels it has been removed from. @@ -287,7 +290,18 @@ func (runner *SGRTestRunner) RequireDocReplicated(docID string, source, dest *Re // and dest can already be at version before the replication has done anything. runner.waitForDocVersion(source, docID, version) runner.waitForDocVersion(dest, docID, version) - return runner.RequireDoc(docID, dest, ExpectedISGRDocFromPeer(t, source, docID)) + expected := ExpectedISGRDocFromPeer(t, source, docID) + if expected.NoHLV { + // A document written before HLVs existed replicates with a CV encoded from its revTreeID, so that's the HLV + // dest must end up with. + encodedCV, err := db.LegacyRevToRevTreeEncodedVersion(expected.Version.RevTreeID) + require.NoError(t, err) + expected.HLV = db.NewHybridLogicalVector() + require.NoError(t, expected.HLV.AddVersion(encodedCV)) + expected.Version.CV = encodedCV + expected.NoHLV = false + } + return runner.RequireDoc(docID, dest, expected) } // RequireDoc waits for exp.Version to arrive on rt, then requires the document to match every field set in exp. @@ -300,13 +314,18 @@ func (runner *SGRTestRunner) RequireDoc(docID string, rt *RestTester, exp Expect actual := expectedISGRDocFromDoc(t, rt.Context(), doc) peer := rt.GetDatabase().Name - require.NotNil(t, doc.HLV, "doc %q has no HLV on %s", docID, peer) - if runner.IsV4Protocol() { + switch { + case exp.NoHLV: + // HLVDebugString isn't nil-safe, and message arguments are evaluated even when the assertion passes + assert.Nil(t, doc.HLV, "doc %q on %s should have no HLV, HLV: %+v", docID, peer, doc.HLV) + case doc.HLV == nil: + require.FailNow(t, "doc has no HLV", "doc %q on %s", docID, peer) + case runner.IsV4Protocol(): if exp.HLV != nil { assert.True(t, hlvEqualAllowingEncodedRevs(t, exp.HLV, doc.HLV, exp.RevChain), "HLV mismatch for doc %q on %s. Expected: %s, Actual: %s", docID, peer, exp.HLV.HLVDebugString(), doc.HLV.HLVDebugString()) } - } else { + default: assert.Equal(t, rt.GetDatabase().EncodedSourceID, doc.HLV.SourceID, "doc %q on %s should have a CV minted locally in v3, HLV: %s", docID, peer, doc.HLV.HLVDebugString()) } @@ -402,26 +421,22 @@ func ExpectedISGRDocFromPeer(t testing.TB, rt *RestTester, docID string) Expecte return expectedISGRDocFromDoc(t, rt.Context(), rt.GetDocument(docID)) } -// expectedISGRDocFromDoc returns the replication-relevant state of doc, with every field populated. +// expectedISGRDocFromDoc returns the replication-relevant state of doc as it is, with every field populated. A document +// with no HLV is described as one, with NoHLV set and a revTreeID-only Version. func expectedISGRDocFromDoc(t testing.TB, ctx context.Context, doc *db.Document) ExpectedISGRDoc { t.Helper() - hlv := doc.HLV.Copy() - if hlv == nil { - // A document written before HLVs existed replicates with a CV encoded from its revTreeID, so that's the HLV a - // peer receiving it must end up with. - encodedCV, err := db.LegacyRevToRevTreeEncodedVersion(doc.GetRevTreeID()) - require.NoError(t, err) - hlv = db.NewHybridLogicalVector() - require.NoError(t, hlv.AddVersion(encodedCV)) - } exp := ExpectedISGRDoc{ - Version: DocVersion{RevTreeID: doc.GetRevTreeID(), CV: *hlv.ExtractCurrentVersionFromHLV()}, + Version: DocVersion{RevTreeID: doc.GetRevTreeID()}, Deleted: doc.IsDeleted(), - HLV: hlv, + HLV: doc.HLV.Copy(), + NoHLV: doc.HLV == nil, RevChain: []string{}, Channels: []string{}, Attachments: map[string]string{}, } + if doc.HLV != nil { + exp.Version.CV = *doc.HLV.ExtractCurrentVersionFromHLV() + } for revID := doc.GetRevTreeID(); revID != ""; { revInfo, ok := doc.History[revID] require.True(t, ok, "rev %q missing from rev tree of doc %q", revID, doc.ID) diff --git a/rest/utilities_testing_isgr_test.go b/rest/utilities_testing_isgr_test.go index b2e1b47ece..1bff370f01 100644 --- a/rest/utilities_testing_isgr_test.go +++ b/rest/utilities_testing_isgr_test.go @@ -9,8 +9,10 @@ package rest import ( + "net/url" "testing" + "github.com/couchbase/sync_gateway/base" "github.com/couchbase/sync_gateway/db" "github.com/couchbase/sync_gateway/testing/assert" "github.com/couchbase/sync_gateway/testing/require" @@ -34,6 +36,20 @@ func TestHLVEqualAllowingEncodedRevs(t *testing.T) { } return hlv } + withCV := func(hlv *db.HybridLogicalVector, sourceID string, value uint64) *db.HybridLogicalVector { + hlv.SourceID, hlv.Version = sourceID, value + return hlv + } + withMV := func(hlv *db.HybridLogicalVector, mv ...db.Version) *db.HybridLogicalVector { + for _, v := range mv { + hlv.MergeVersions[v.SourceID] = v.Value + } + return hlv + } + withCVCAS := func(hlv *db.HybridLogicalVector, cvCAS uint64) *db.HybridLogicalVector { + hlv.CurrentVersionCAS = cvCAS + return hlv + } testCases := []struct { name string @@ -47,6 +63,50 @@ func TestHLVEqualAllowingEncodedRevs(t *testing.T) { actual: newHLV(db.Version{SourceID: "a", Value: 1}), equal: true, }, + { + name: "identical merge versions", + expected: withMV(newHLV(), db.Version{SourceID: "m", Value: 5}), + actual: withMV(newHLV(), db.Version{SourceID: "m", Value: 5}), + equal: true, + }, + { + // cvCAS is set by each peer from its own write, so it is never expected to match + name: "different cvCAS", + expected: withCVCAS(newHLV(), 1), + actual: withCVCAS(newHLV(), 2), + equal: true, + }, + { + name: "different current version source", + expected: newHLV(), + actual: withCV(newHLV(), "other", 100), + equal: false, + }, + { + name: "different current version value", + expected: newHLV(), + actual: withCV(newHLV(), "cv", 101), + equal: false, + }, + { + name: "different merge version value", + expected: withMV(newHLV(), db.Version{SourceID: "m", Value: 5}), + actual: withMV(newHLV(), db.Version{SourceID: "m", Value: 6}), + equal: false, + }, + { + name: "missing merge versions", + expected: withMV(newHLV(), db.Version{SourceID: "m", Value: 5}), + actual: newHLV(), + equal: false, + }, + { + // the encoded-revision allowance only applies to pv, never to mv + name: "extra encoded version of a revision in the chain in merge versions", + expected: newHLV(), + actual: withMV(newHLV(), encoded(chainRev)), + equal: false, + }, { name: "extra encoded version of a revision in the chain", expected: newHLV(), @@ -78,3 +138,52 @@ func TestHLVEqualAllowingEncodedRevs(t *testing.T) { }) } } + +// TestExpectedISGRDocFromPeerLegacyDoc checks that a snapshot of a document with no HLV describes it as it is, so +// RequireDoc can assert a peer holding a legacy document was left unchanged, while RequireDocReplicated still expects +// the receiving peer to have the revTreeID-encoded CV. +func TestExpectedISGRDocFromPeerLegacyDoc(t *testing.T) { + base.RequireNumTestBuckets(t, 2) + runner := NewSGRTestRunner(t) + runner.Run(func(t *testing.T) { + peers := runner.SetupSGRPeers(t) + const docID = "legacyDoc" + legacyDoc := peers.PassiveRT.CreateDocNoHLV(docID, db.Body{"channels": []string{"alice"}}) + + snapshot := ExpectedISGRDocFromPeer(t, peers.PassiveRT, docID) + assert.True(t, snapshot.NoHLV) + assert.Nil(t, snapshot.HLV) + assert.Equal(t, DocVersion{RevTreeID: legacyDoc.GetRevTreeID()}, snapshot.Version) + runner.RequireDoc(docID, peers.PassiveRT, snapshot) + + ar, err := db.NewActiveReplicator(peers.ActiveRT.Context(), &db.ActiveReplicatorConfig{ + ID: t.Name(), + Direction: db.ActiveReplicatorTypePull, + RemoteDBURL: passiveDBURL(t, peers), + ActiveDB: &db.Database{DatabaseContext: peers.ActiveRT.GetDatabase()}, + ChangesBatchSize: 200, + ReplicationStatsMap: DbReplicatorStats(t, peers.ActiveRT.GetDatabase(), t.Name()), + CollectionsEnabled: !peers.ActiveRT.GetDatabase().OnlyDefaultCollection(), + SupportedBLIPProtocols: runner.SupportedSubprotocols, + }) + require.NoError(t, err) + require.NoError(t, ar.Start(peers.ActiveRT.Context())) + defer func() { require.NoError(t, ar.Stop()) }() + + doc := runner.RequireDocReplicated(docID, peers.PassiveRT, peers.ActiveRT, snapshot.Version) + require.NotNil(t, doc.HLV) + if runner.IsV4Protocol() { + encodedCV, err := db.LegacyRevToRevTreeEncodedVersion(legacyDoc.GetRevTreeID()) + require.NoError(t, err) + assert.Equal(t, encodedCV, *doc.HLV.ExtractCurrentVersionFromHLV()) + } + // pulling must not have given the passive peer's document an HLV + runner.RequireDoc(docID, peers.PassiveRT, snapshot) + }) +} + +func passiveDBURL(t *testing.T, peers *TestISGRPeers) *url.URL { + u, err := url.Parse(peers.PassiveDBURL) + require.NoError(t, err) + return u +} From 80d4beab593c2f461da5a615613b8e309aac3ca0 Mon Sep 17 00:00:00 2001 From: Gregory Newman-Smith Date: Fri, 2 Oct 2026 12:56:22 +0100 Subject: [PATCH 4/5] address review --- rest/blip_api_crud_test.go | 2 +- .../replicator_conflict_test.go | 7 +- rest/replicatortest/replicator_test.go | 93 +++++++++---------- .../replicator_test_legacy_rev_test.go | 50 +++++----- rest/revocation_test.go | 6 +- rest/utilities_testing_isgr.go | 27 ++++-- rest/utilities_testing_isgr_test.go | 6 +- 7 files changed, 96 insertions(+), 95 deletions(-) diff --git a/rest/blip_api_crud_test.go b/rest/blip_api_crud_test.go index b19aeab152..ebc162be2c 100644 --- a/rest/blip_api_crud_test.go +++ b/rest/blip_api_crud_test.go @@ -2773,7 +2773,7 @@ func TestProcessRevIncrementsStat(t *testing.T) { defer func() { require.NoError(t, ar.Stop()) }() activeRT.WaitForPendingChanges() - sgrRunner.RequireDocReplicated(docID, remoteRT, activeRT, version) + sgrRunner.WaitForDocReplicated(docID, remoteRT, activeRT, version) base.RequireWaitForStat(t, pullStats.HandleRevCount.Value, 1) assert.NotEqualValues(t, 0, pullStats.HandleRevBytes.Value()) diff --git a/rest/replicatortest/replicator_conflict_test.go b/rest/replicatortest/replicator_conflict_test.go index dbf666fc64..aab8692d21 100644 --- a/rest/replicatortest/replicator_conflict_test.go +++ b/rest/replicatortest/replicator_conflict_test.go @@ -2119,7 +2119,7 @@ func TestActiveReplicatorConflictRemoveCVFromCache(t *testing.T) { conflictResVersion.RevTreeID = newRev // rt1 has resolved once the result reaches rt2. Don't wait on rt1 over REST: that would populate the rev // cache this test asserts on. - sgrRunner.RequireDocReplicated(docID, rt1, rt2, conflictResVersion) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, conflictResVersion) // assert we cannot keep old cache entry for original wrote (before conflict resolution) on active cluster rest testers collectionActive1, ctxActive1 := active.GetSingleTestDatabaseCollectionWithUser() @@ -2176,8 +2176,7 @@ func TestActiveReplicatorV4DefaultResolverWithTombstoneLocal(t *testing.T) { // expect local doc to win and still be tombstone with remote docs revision history. The local CV is new to // passiveRT, so the resolved tombstone is pushed back. - activeRT.WaitForTombstone(docID, conflictResVersion) - sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, conflictResVersion) + sgrRunner.WaitForDocReplicated(docID, activeRT, passiveRT, conflictResVersion) rt1Doc := activeRT.GetDocument(docID) // assert that remote cv is in pv now @@ -2230,7 +2229,7 @@ func TestActiveReplicatorV4DefaultResolverWithTombstoneRemote(t *testing.T) { conflictResVersion.RevTreeID = newRev sgrRunner.RequireDoc(docID, activeRT, rest.ExpectedISGRDoc{Version: conflictResVersion, Deleted: true}) // activeRT adopted passiveRT's CV, so there is nothing new to push back and passiveRT must be unchanged - sgrRunner.RequireDoc(docID, passiveRT, passiveDocBeforeReplication) + sgrRunner.RequireDocUnchanged(docID, activeRT, passiveRT, passiveDocBeforeReplication) rt1Doc := activeRT.GetDocument(docID) // assert that remote cv is in pv now diff --git a/rest/replicatortest/replicator_test.go b/rest/replicatortest/replicator_test.go index 4305f9019e..7f21d2afb0 100644 --- a/rest/replicatortest/replicator_test.go +++ b/rest/replicatortest/replicator_test.go @@ -2077,7 +2077,7 @@ func TestActiveReplicatorPullBasic(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -2170,7 +2170,7 @@ func TestActiveReplicatorPullSkippedSequence(t *testing.T) { pullCheckpointer := ar.Pull.GetSingleCollection(t).Checkpointer // wait for the documents originally written to rt2 to arrive at rt1 - sgrRunner.RequireDocReplicated(docID1, rt2, rt1, doc1Version) + sgrRunner.WaitForDocReplicated(docID1, rt2, rt1, doc1Version) base.RequireWaitForStat(t, func() int64 { return pullCheckpointer.Stats().ExpectedSequenceCount }, 1) base.RequireWaitForStat(t, func() int64 { return pullCheckpointer.Stats().ProcessedSequenceCount }, 1) @@ -2594,7 +2594,7 @@ func TestActiveReplicatorPullAttachments(t *testing.T) { assert.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -2610,7 +2610,7 @@ func TestActiveReplicatorPullAttachments(t *testing.T) { version = rt2.PutDoc(docID, `{"source":"rt2","doc_num":2,`+attachment+`,"channels":["alice"]}`) // wait for the new document written to rt2 to arrive at rt1 - sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, version) doc2, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) assert.NoError(t, err) @@ -3212,7 +3212,7 @@ func TestActiveReplicatorPushBasic(t *testing.T) { // wait for the document originally written to rt1 to arrive at rt2 // the receiving peer runs its own sync function, so this also requires the channel set to be persisted - sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, version) assert.Equal(t, strconv.FormatUint(localDoc.Sequence, 10), ar.GetStatus(ctx1).LastSeqPush) }) @@ -3267,7 +3267,7 @@ func TestActiveReplicatorPushAttachments(t *testing.T) { assert.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt1 to arrive at rt2 - sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, version) assert.Equal(t, int64(1), ar.Push.GetStats().HandleGetAttachment.Value()) @@ -3275,7 +3275,7 @@ func TestActiveReplicatorPushAttachments(t *testing.T) { version = rt1.PutDoc(docID, `{"source":"rt1","doc_num":2,`+attachment+`,"channels":["alice"]}`) // wait for the new document written to rt1 to arrive at rt2 - sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, version) // When targeting a Hydrogen node that supports proveAttachments, we typically end up sending // the attachment only once. However, targeting a Lithium node sends the attachment twice like @@ -3650,7 +3650,7 @@ func TestActiveReplicatorPushOneshot(t *testing.T) { doc, err := rt2collection.GetDocument(rt2ctx, docID, db.DocUnmarshalAll) require.NoError(t, err) - sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, version) body, err := doc.GetDeepMutableBody() require.NoError(t, err) @@ -3702,7 +3702,7 @@ func TestActiveReplicatorPullTombstone(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -3716,7 +3716,7 @@ func TestActiveReplicatorPullTombstone(t *testing.T) { deletedVersion := rt2.DeleteDoc(docID, version) // wait for the tombstone written to rt2 to arrive at rt1 - sgrRunner.RequireDocReplicated(docID, rt2, rt1, deletedVersion) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, deletedVersion) doc, err = rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) assert.NoError(t, err) assert.True(t, doc.IsDeleted()) @@ -3768,7 +3768,7 @@ func TestActiveReplicatorPullPurgeOnRemoval(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt2 to arrive at rt1 - sgrRunner.RequireDocReplicated(docID, rt2, rt1, version) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -3994,7 +3994,7 @@ func TestActiveReplicatorPullConflict(t *testing.T) { if !test.skipBodyAssertion { expectedDoc.Body = test.expectedLocalBody } - sgrRunner.RequireDoc(docID, rt1, expectedDoc) + doc = sgrRunner.RequireDoc(docID, rt1, expectedDoc) t.Logf("Doc %s is %+v", docID, doc) for revID, revInfo := range doc.SyncData.History { t.Logf("doc revision [%s]: %+v", revID, revInfo) @@ -4243,11 +4243,11 @@ func TestActiveReplicatorPushAndPullConflict(t *testing.T) { if sgrRunner.IsV4Protocol() && test.winner == remote { // rt1 adopted rt2's CV, so there is nothing new to push back: rt2 must be exactly as it was. rt1's // HLV also carries its losing version in pv, which rt2 never needs to learn about. - doc = sgrRunner.RequireDoc(docID, rt2, remoteDocBeforeReplication) + doc = sgrRunner.RequireDocUnchanged(docID, rt1, rt2, remoteDocBeforeReplication) } else { // rt1's resolution produced a version rt2 doesn't have, which is pushed back, so rt2 must now // hold rt1's document - doc = sgrRunner.RequireDocReplicated(docID, rt1, rt2, test.expectedVersion) + doc = sgrRunner.WaitForDocReplicated(docID, rt1, rt2, test.expectedVersion) } requireBodyEqual(t, test.expectedBody, doc) t.Logf("Remote Doc %s is %+v", docID, doc) @@ -4331,7 +4331,7 @@ func TestActiveReplicatorPushBasicWithInsecureSkipVerifyEnabled(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt1 to arrive at rt2 - sgrRunner.RequireDocReplicated(docID, rt1, rt2, version) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, version) rt1collection, rt1ctx := rt1.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -4999,7 +4999,7 @@ func TestActiveReplicatorIgnoreNoConflicts(t *testing.T) { require.NoError(t, ar.Start(ctx1)) // wait for the document originally written to rt1 to arrive at rt2 - sgrRunner.RequireDocReplicated(rt1docID, rt1, rt2, rt1Version) + sgrRunner.WaitForDocReplicated(rt1docID, rt1, rt2, rt1Version) rt2collection, rt2ctx := rt2.GetSingleTestDatabaseCollection() doc, err := rt2collection.GetDocument(rt2ctx, rt1docID, db.DocUnmarshalAll) @@ -5013,7 +5013,7 @@ func TestActiveReplicatorIgnoreNoConflicts(t *testing.T) { rt2Version := rt2.PutDoc(rt2docID, `{"source":"rt2","channels":["alice"]}`) // ... and wait to arrive at rt1 - sgrRunner.RequireDocReplicated(rt2docID, rt2, rt1, rt2Version) + sgrRunner.WaitForDocReplicated(rt2docID, rt2, rt1, rt2Version) changesResults := rt1.WaitForChanges(2, "/{{.keyspace}}/_changes?since=0", "", true) assert.Equal(t, rt1docID, changesResults.Results[0].ID) assert.Equal(t, rt2docID, changesResults.Results[1].ID) @@ -5831,7 +5831,7 @@ func TestSGR2TombstoneConflictHandling(t *testing.T) { const doc2ID = "docid2" doc2Version := rest.DocVersion{RevTreeID: "3-abc"} - sgrRunner.RequireDocReplicated(doc2ID, localActiveRT, remotePassiveRT, doc2Version) + sgrRunner.WaitForDocReplicated(doc2ID, localActiveRT, remotePassiveRT, doc2Version) // Stop the replication rest.RequireStatus(t, localActiveRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/replication?action=stop", ""), http.StatusOK) @@ -5878,7 +5878,7 @@ func TestSGR2TombstoneConflictHandling(t *testing.T) { source, dest = remotePassiveRT, localActiveRT } source.WaitForTombstoneRevIDOnly(doc2ID, deleteVersion) - sgrRunner.RequireDocReplicated(doc2ID, source, dest, deleteVersion) + sgrRunner.WaitForDocReplicated(doc2ID, source, dest, deleteVersion) // Stop the replication rest.RequireStatus(t, localActiveRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/replication?action=stop", ""), http.StatusOK) @@ -5934,9 +5934,9 @@ func TestSGR2TombstoneConflictHandling(t *testing.T) { // Wait for doc to replicate from side resurrection was done on to the other side if test.resurrectLocal { - sgrRunner.RequireDocReplicated(doc2ID, localActiveRT, remotePassiveRT, expectedVersion) + sgrRunner.WaitForDocReplicated(doc2ID, localActiveRT, remotePassiveRT, expectedVersion) } else { - sgrRunner.RequireDocReplicated(doc2ID, remotePassiveRT, localActiveRT, expectedVersion) + sgrRunner.WaitForDocReplicated(doc2ID, remotePassiveRT, localActiveRT, expectedVersion) } }) } @@ -6318,7 +6318,7 @@ func TestLocalWinsConflictResolution(t *testing.T) { activeRT.CreateReplication(replicationID, remoteURLString, db.ActiveReplicatorTypePushAndPull, nil, true, db.ConflictResolverLocalWins, "") activeRT.WaitForReplicationStatus(replicationID, db.ReplicationStateRunning) - sgrRunner.RequireDocReplicated(docID, activeRT, remoteRT, newVersion) + sgrRunner.WaitForDocReplicated(docID, activeRT, remoteRT, newVersion) // Stop the replication response := activeRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/"+replicationID+"?action=stop", "") @@ -6515,7 +6515,7 @@ func TestReplicatorConflictAttachment(t *testing.T) { defer activeRT.DeleteReplication(replicationID) activeRT.WaitForReplicationStatus(replicationID, db.ReplicationStateRunning) - sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, newVersion) + sgrRunner.WaitForDocReplicated(docID, activeRT, passiveRT, newVersion) response := activeRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/"+replicationID+"?action=stop", "") rest.RequireStatus(t, response, http.StatusOK) @@ -6531,7 +6531,7 @@ func TestReplicatorConflictAttachment(t *testing.T) { localGen++ newLocalVersion = rest.NewDocVersionFromFakeRev(fmt.Sprintf("%d-local", localGen)) - _ = activeRT.PutNewEditsFalse(docID, newLocalVersion, &localParentVersion, fmt.Sprintf(`{"_attachments": {"attach": {"stub": true, "revpos": %d, "digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}}`, localGen-1)) + localWinsVersion := activeRT.PutNewEditsFalse(docID, newLocalVersion, &localParentVersion, fmt.Sprintf(`{"_attachments": {"attach": {"stub": true, "revpos": %d, "digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}}`, localGen-1)) remoteGen := nextGen remoteParentVersion := newVersion @@ -6547,26 +6547,17 @@ func TestReplicatorConflictAttachment(t *testing.T) { response = activeRT.SendAdminRequest("PUT", "/{{.db}}/_replicationStatus/"+replicationID+"?action=start", "") rest.RequireStatus(t, response, http.StatusOK) - expVersion := test.expectedFinalVersion - if sgrRunner.IsV4Protocol() { - if test.conflictResolution == db.ConflictResolverRemoteWins { - expVersion = *remoteWinsVersion - } else { - activeCollection, activeCtx := activeRT.GetSingleTestDatabaseCollectionWithUser() - doc, err := activeCollection.GetDocument(activeCtx, docID, db.DocUnmarshalAll) - require.NoError(t, err) - expVersion.CV = db.Version{ - SourceID: activeRT.GetDatabase().EncodedSourceID, - Value: doc.HLV.Version, - } - } - } if sgrRunner.IsV4Protocol() && test.conflictResolution == db.ConflictResolverRemoteWins { // activeRT adopted passiveRT's CV, so there is nothing new to push back and passiveRT must be unchanged - sgrRunner.RequireDoc(docID, activeRT, rest.ExpectedISGRDoc{Version: expVersion}) - sgrRunner.RequireDoc(docID, passiveRT, passiveDocBeforeReplication) + sgrRunner.RequireDoc(docID, activeRT, rest.ExpectedISGRDoc{Version: *remoteWinsVersion}) + sgrRunner.RequireDocUnchanged(docID, activeRT, passiveRT, passiveDocBeforeReplication) } else { - sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, expVersion) + expVersion := test.expectedFinalVersion + if sgrRunner.IsV4Protocol() { + // local wins moves the rev tree on but keeps activeRT's CV from its 5-local write + expVersion.CV = localWinsVersion.CV + } + sgrRunner.WaitForDocReplicated(docID, activeRT, passiveRT, expVersion) } localDoc := activeRT.GetDocBody(docID) @@ -6682,20 +6673,20 @@ func TestReplicatorDoNotSendDeltaWhenSrcIsTombstone(t *testing.T) { require.NoError(t, ar.Start(activeCtx)) // Wait for active to replicate to passive - sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, version) + sgrRunner.WaitForDocReplicated("test", activeRT, passiveRT, version) // Delete active document deletedVersion := activeRT.DeleteDoc("test", version) // Assert that the tombstone is replicated to passive // Get revision 2 on passive peer to assert it has been (a) replicated and (b) deleted - sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, deletedVersion) + sgrRunner.WaitForDocReplicated("test", activeRT, passiveRT, deletedVersion) // Resurrect tombstoned document resurrectedVersion := activeRT.UpdateDoc("test", deletedVersion, `{"field2":"f2_2"}`) // Replicate resurrection to passive - sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, resurrectedVersion) + sgrRunner.WaitForDocReplicated("test", activeRT, passiveRT, resurrectedVersion) base.RequireWaitForStat(t, func() int64 { // should be 1 given delta is sent for delete, but not for resurrection @@ -6758,7 +6749,7 @@ func TestUnprocessableDeltas(t *testing.T) { require.NoError(t, ar.Start(activeCtx)) - sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, version) + sgrRunner.WaitForDocReplicated("test", activeRT, passiveRT, version) require.NoError(t, ar.Stop()) @@ -6779,7 +6770,7 @@ func TestUnprocessableDeltas(t *testing.T) { base.AssertLogContains(t, "Unable to unmarshal mutable body for doc test", func() { require.NoError(t, ar.Start(activeCtx)) // Check if it replicated - sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, version2) + sgrRunner.WaitForDocReplicated("test", activeRT, passiveRT, version2) require.NoError(t, ar.Stop()) }) }) @@ -7112,7 +7103,7 @@ func TestReplicatorCheckpointOnStop(t *testing.T) { activeRT.CreateReplication(replicationID, remoteURL, db.ActiveReplicatorTypePush, nil, true, db.ConflictResolverDefault, "") activeRT.WaitForReplicationStatus(replicationID, db.ReplicationStateRunning) - sgrRunner.RequireDocReplicated("test", activeRT, passiveRT, rest.DocVersion{RevTreeID: revID, CV: *doc.HLV.ExtractCurrentVersionFromHLV()}) + sgrRunner.WaitForDocReplicated("test", activeRT, passiveRT, rest.DocVersion{RevTreeID: revID, CV: *doc.HLV.ExtractCurrentVersionFromHLV()}) // assert on the processed seq list being updated before stopping the active replicator ar, ok := activeRT.GetDatabase().SGReplicateMgr.GetLocalActiveReplicatorForTest(t, replicationID) @@ -8125,7 +8116,7 @@ func TestActiveReplicatorPullNewDocChannels(t *testing.T) { docID := rest.SafeDocumentName(t, t.Name()) + "rt2doc1" version := passiveRT.PutDoc(docID, `{"source":"rt2","channels":["`+chanName+`"]}`) - sgrRunner.RequireDocReplicated(docID, passiveRT, activeRT, version) + sgrRunner.WaitForDocReplicated(docID, passiveRT, activeRT, version) rt1collection, rt1ctx := activeRT.GetSingleTestDatabaseCollection() doc, err := rt1collection.GetDocument(rt1ctx, docID, db.DocUnmarshalAll) @@ -8180,7 +8171,7 @@ func TestActiveReplicatorPushNewDocChannels(t *testing.T) { docID := rest.SafeDocumentName(t, t.Name()) + "rt1doc1" version := activeRT.PutDoc(docID, `{"source":"rt1","channels":["`+chanName+`"]}`) - sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, version) + sgrRunner.WaitForDocReplicated(docID, activeRT, passiveRT, version) passiveRTCollection, passiveRTCtx := passiveRT.GetSingleTestDatabaseCollection() doc, err := passiveRTCollection.GetDocument(passiveRTCtx, docID, db.DocUnmarshalAll) @@ -8242,11 +8233,11 @@ func TestActiveReplicatorPullUpdatedDocChannels(t *testing.T) { _, hasA := localDoc.Channels[chanA] require.True(t, hasA, "precondition: local write should set chanA, channels=%v", localDoc.Channels) - sgrRunner.RequireDocReplicated(docID, activeRT, passiveRT, v1) + sgrRunner.WaitForDocReplicated(docID, activeRT, passiveRT, v1) // now update the doc on rt2, moving it from chanA to chanB, and let it replicate back to rt1 v2 := passiveRT.UpdateDoc(docID, v1, `{"channels":["`+chanB+`"]}`) - sgrRunner.RequireDocReplicated(docID, passiveRT, activeRT, v2) + sgrRunner.WaitForDocReplicated(docID, passiveRT, activeRT, v2) doc, err := activeRTCollection.GetDocument(activeRTCtx, docID, db.DocUnmarshalAll) require.NoError(t, err) diff --git a/rest/replicatortest/replicator_test_legacy_rev_test.go b/rest/replicatortest/replicator_test_legacy_rev_test.go index 183cfdcdad..c145a56fd7 100644 --- a/rest/replicatortest/replicator_test_legacy_rev_test.go +++ b/rest/replicatortest/replicator_test_legacy_rev_test.go @@ -65,8 +65,8 @@ func TestActiveReplicatorPushPullLegacyRev(t *testing.T) { // Start the replicator require.NoError(t, ar.Start(ctx1)) - sgrRunner.RequireDocReplicated(docIDRT2, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRt2}) - sgrRunner.RequireDocReplicated(docIDRT1, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) + sgrRunner.WaitForDocReplicated(docIDRT2, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRt2}) + sgrRunner.WaitForDocReplicated(docIDRT1, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) }) } @@ -174,14 +174,14 @@ func TestActiveReplicatorBiDirectionalPreUpgradedDocOnPeer(t *testing.T) { require.NoError(t, ar.Start(ctx1)) if tc.newRevOnActivePeer { - sgrRunner.RequireDocReplicated(docID, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRT1}) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRT1}) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{legacyRevRT1, initLegacyRevRT1}) // add new doc for some replication activity on pull side to allow us to assert that legacy rev 2-abc added // isn't pulled back to rt1 now it has a legacy revID encoded CV newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 + sgrRunner.WaitForDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 // assert that the document isn't replicated back to rt1 after legacy rev CV is written on rt2 rt1Doc := rt1.GetDocument(docID) @@ -189,14 +189,14 @@ func TestActiveReplicatorBiDirectionalPreUpgradedDocOnPeer(t *testing.T) { assert.Nil(t, rt1Doc.HLV) rest.RequireHistoryContains(t, rt1Doc.History, []string{legacyRevRT1, initLegacyRevRT1}) } else { - sgrRunner.RequireDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) rt1Doc := rt1.GetDocument(docID) rest.RequireHistoryContains(t, rt1Doc.History, []string{initLegacyRevRT2, legacyRevRT2}) // add new doc for some replication activity on push side to allow us to assert that legacy rev 2-abc added // isn't pushed back to rt2 now it has a legacy revID encoded CV newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.RequireDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 + sgrRunner.WaitForDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 // assert that the document isn't replicated back to rt2 after legacy rev CV is written on rt1 rt2Doc := rt2.GetDocument(docID) @@ -424,7 +424,7 @@ func TestActiveReplicatorBiDirectionalPreUpgradedRevInHistory(t *testing.T) { require.NoError(t, ar.Start(ctx1)) if !tc.activePeerHasUpgradedRev { - sgrRunner.RequireDocReplicated(docID, rt2, rt1, upgradedDocVersion) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, upgradedDocVersion) rt1Doc := rt1.GetDocument(docID) rest.RequireHistoryContains(t, rt1Doc.History, []string{legacyRevRT1, initLegacyRevRT1, upgradedRevID}) @@ -442,7 +442,7 @@ func TestActiveReplicatorBiDirectionalPreUpgradedRevInHistory(t *testing.T) { require.JSONEq(t, expectedBody, string(actualBodyRT2)) require.JSONEq(t, expectedBody, string(actualBodyRT1)) } else { - sgrRunner.RequireDocReplicated(docID, rt1, rt2, upgradedDocVersion) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, upgradedDocVersion) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{legacyRevRT2, initLegacyRevRT2, upgradedRevID}) @@ -530,8 +530,8 @@ func TestActiveReplicatorPushPullNewDocLegacyRevAndAllowUpdateAfter(t *testing.T // Start the replicator require.NoError(t, ar.Start(ctx1)) - sgrRunner.RequireDocReplicated(docIDToPull, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRt2}) - sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) + sgrRunner.WaitForDocReplicated(docIDToPull, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRt2}) + sgrRunner.WaitForDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) // now update both docs to create a new revision on each side giving them each hlv based off their nodes source cvVersion, err := db.LegacyRevToRevTreeEncodedVersion(legacyRevRt1) @@ -540,7 +540,7 @@ func TestActiveReplicatorPushPullNewDocLegacyRevAndAllowUpdateAfter(t *testing.T RevTreeID: legacyRevRt1, } updateVer := rt1.UpdateDoc(docIDToPush, rt1DocVersion, `{"channels": ["alice"], "source": "rt1-updated"}`) - sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, updateVer) + sgrRunner.WaitForDocReplicated(docIDToPush, rt1, rt2, updateVer) // check rev tree encoded version is in pv finalRT2Doc := rt2.GetDocument(docIDToPush) assert.Equal(t, cvVersion.Value, finalRT2Doc.HLV.PreviousVersions[cvVersion.SourceID]) @@ -551,7 +551,7 @@ func TestActiveReplicatorPushPullNewDocLegacyRevAndAllowUpdateAfter(t *testing.T RevTreeID: legacyRevRt2, } updateVer = rt2.UpdateDoc(docIDToPull, rt2DocVersion, `{"channels": ["alice"], "source": "rt2-updated"}`) - sgrRunner.RequireDocReplicated(docIDToPull, rt2, rt1, updateVer) + sgrRunner.WaitForDocReplicated(docIDToPull, rt2, rt1, updateVer) finalRT1Doc := rt1.GetDocument(docIDToPull) assert.Equal(t, cvVersion.Value, finalRT1Doc.HLV.PreviousVersions[cvVersion.SourceID]) @@ -779,13 +779,13 @@ func TestActiveReplicatorConflictPreUpgradedVersionEachSide(t *testing.T) { verPostConflictRes, _ = rt1.GetDoc(docID) assert.NotEqual(c, legacyRevRT1, verPostConflictRes.RevTreeID) }, 10*time.Second, 50*time.Millisecond) - sgrRunner.RequireDocReplicated(docID, rt1, rt2, rest.DocVersion{RevTreeID: verPostConflictRes.RevTreeID}) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, verPostConflictRes) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, legacyRevRT2, verPostConflictRes.RevTreeID}) // add doc for some replication activity on pull to assert that local doesn't change after remote is written with legacy CV newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 + sgrRunner.WaitForDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 // assert active side doc hasn't changed rt1Doc := rt1.GetDocument(docID) @@ -802,14 +802,14 @@ func TestActiveReplicatorConflictPreUpgradedVersionEachSide(t *testing.T) { return replicationStats.ConflictResolvedRemoteCount.Value() }, 1) - sgrRunner.RequireDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) + sgrRunner.WaitForDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) rt1Doc := rt1.GetDocument(docID) tombstonedID := db.CreateRevIDWithBytes(3, legacyRevRT1, []byte(db.DeletedDocument)) // create what would be the tombstone rev id for local branch rest.RequireHistoryContains(t, rt1Doc.History, []string{initLegacyRevRT1, legacyRevRT1, legacyRevRT2, tombstonedID}) // add doc for some replication activity on push to assert that local doesn't change after remote is written with legacy CV newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.RequireDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 + sgrRunner.WaitForDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 // assert passive side doc hasn't changed rt2Doc := rt2.GetDocument(docID) @@ -963,7 +963,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { assert.NotEqual(c, legacyRevRT1, verPostConflictRes.RevTreeID) }, 10*time.Second, 50*time.Millisecond) // wait for this resolution to be pushed back to passive peer - sgrRunner.RequireDocReplicated(docID, rt1, rt2, verPostConflictRes) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, verPostConflictRes) // assert original upgraded version in PV history assert.Equal(t, upgradedDocVersion.CV.Value, replicatedDoc.HLV.PreviousVersions[upgradedDocVersion.CV.SourceID]) @@ -978,7 +978,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { // add doc for some replication activity on pull to assert that local doesn't change after remote is written with legacy CV newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.RequireDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 + sgrRunner.WaitForDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 // assert active side doc hasn't changed rt1Doc := rt1.GetDocument(docID) @@ -1009,12 +1009,8 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { rt1BodyBytes, err := rt1Doc.BodyBytes(rt1.Context()) require.NoError(t, err) - // add doc for some replication activity on push to assert that remote doesn't change after remote wins conflict res is done - newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.RequireDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 - // assert passive side doc hasn't changed - rt1 adopted its CV, so there was nothing to push back - rt2Doc := sgrRunner.RequireDoc(docID, rt2, passiveDocBeforeReplication) + rt2Doc := sgrRunner.RequireDocUnchanged(docID, rt1, rt2, passiveDocBeforeReplication) rest.RequireDocVersionEqual(t, upgradedDocVersion, rt2Doc.ExtractDocVersion()) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, upgradedDocVersion.RevTreeID}) @@ -1026,7 +1022,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { // update doc on active side to ensure we can push with no conflict updateVer := rt1.UpdateDoc(docID, upgradedDocVersion, `{"channels": ["alice"], "source": "rt1-updated"}`) - sgrRunner.RequireDocReplicated(docID, rt1, rt2, updateVer) + sgrRunner.WaitForDocReplicated(docID, rt1, rt2, updateVer) } }) } @@ -1095,7 +1091,7 @@ func TestActiveReplicatorDeltaSyncWhenBothSidesLegacy(t *testing.T) { // Start the replicator require.NoError(t, ar.Start(ctx1)) - sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) + sgrRunner.WaitForDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRt1}) base.RequireWaitForStat(t, func() int64 { return replicationStats.PushDeltaSentCount.Value() @@ -1151,7 +1147,7 @@ func TestDeltaSyncWhenOneSideHasEncodedCV(t *testing.T) { // Start the replicator require.NoError(t, ar.Start(ctx1)) - sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyInitRevRt1}) + sgrRunner.WaitForDocReplicated(docIDToPush, rt1, rt2, rest.DocVersion{RevTreeID: legacyInitRevRt1}) // flush revision cache to remove old reference to rev 1 in rev cache rt1.GetDatabase().FlushRevisionCacheForTest() @@ -1161,7 +1157,7 @@ func TestDeltaSyncWhenOneSideHasEncodedCV(t *testing.T) { // 1. update doc on rt1 to give HLV based of rt1 sourceID // 2. push doc to rt2 with delta from rev1 to rev2 upgradeVersion := rt1.UpdateDoc(docIDToPush, db.DocVersion{RevTreeID: legacyInitRevRt1}, `{"channels": ["alice"], "source": "rt1-updated"}`) - sgrRunner.RequireDocReplicated(docIDToPush, rt1, rt2, upgradeVersion) + sgrRunner.WaitForDocReplicated(docIDToPush, rt1, rt2, upgradeVersion) base.RequireWaitForStat(t, func() int64 { return replicationStats.PushDeltaSentCount.Value() diff --git a/rest/revocation_test.go b/rest/revocation_test.go index b0379e623a..e105dfbe06 100644 --- a/rest/revocation_test.go +++ b/rest/revocation_test.go @@ -2121,9 +2121,9 @@ func TestReplicatorRevocationsFromZero(t *testing.T) { rt1.WaitForReplicationStatus(ar.ID, db.ReplicationStateStopped) // Be sure docs have arrived - sgrRunner.RequireDocReplicated("docA", rt2, rt1, docAVersion) - sgrRunner.RequireDocReplicated("docA1", rt2, rt1, docA1Version) - sgrRunner.RequireDocReplicated("docA2", rt2, rt1, docA2Version) + sgrRunner.WaitForDocReplicated("docA", rt2, rt1, docAVersion) + sgrRunner.WaitForDocReplicated("docA1", rt2, rt1, docA1Version) + sgrRunner.WaitForDocReplicated("docA2", rt2, rt1, docA2Version) // Reset checkpoint (since 0) require.NoError(t, ar.Reset()) diff --git a/rest/utilities_testing_isgr.go b/rest/utilities_testing_isgr.go index b2c87463fd..8478e3c590 100644 --- a/rest/utilities_testing_isgr.go +++ b/rest/utilities_testing_isgr.go @@ -257,7 +257,7 @@ func (runner *SGRTestRunner) IsV4Protocol() bool { } // ExpectedISGRDoc describes the state a document must be in on a peer after replication. Version is always checked; -// the remaining fields are checked when set, except in RequireDocReplicated where every field comes from the source. +// the remaining fields are checked when set, except in WaitForDocReplicated where every field comes from the source. type ExpectedISGRDoc struct { // Version is waited for before anything else is asserted - revTreeID only in v3, revTreeID and CV in v4. Version DocVersion @@ -279,11 +279,11 @@ type ExpectedISGRDoc struct { Attachments map[string]string } -// RequireDocReplicated waits for version to arrive on dest, then requires dest to hold the same document as source: +// WaitForDocReplicated waits for version to arrive on dest, then requires dest to hold the same document as source: // HLV (v4), rev tree ancestry, channels, body and attachments. Both peers are waited on to reach version, so it can't // be used where the peers legitimately end up different - use RequireDoc with an explicit expectation there. Only the // parts of version that are set are waited on, so a revTreeID alone is enough. -func (runner *SGRTestRunner) RequireDocReplicated(docID string, source, dest *RestTester, version DocVersion) *db.Document { +func (runner *SGRTestRunner) WaitForDocReplicated(docID string, source, dest *RestTester, version DocVersion) *db.Document { t := dest.TB() t.Helper() // Wait on both peers: source's state can itself be the product of the replication (a conflict resolution, say), @@ -322,7 +322,7 @@ func (runner *SGRTestRunner) RequireDoc(docID string, rt *RestTester, exp Expect require.FailNow(t, "doc has no HLV", "doc %q on %s", docID, peer) case runner.IsV4Protocol(): if exp.HLV != nil { - assert.True(t, hlvEqualAllowingEncodedRevs(t, exp.HLV, doc.HLV, exp.RevChain), + assert.True(t, hlvEqualAllowingEncodedRevs(exp.HLV, doc.HLV, exp.RevChain), "HLV mismatch for doc %q on %s. Expected: %s, Actual: %s", docID, peer, exp.HLV.HLVDebugString(), doc.HLV.HLVDebugString()) } default: @@ -353,6 +353,22 @@ func (runner *SGRTestRunner) RequireDoc(docID string, rt *RestTester, exp Expect return doc } +// RequireDocUnchanged requires dest to still match exp once source has pushed everything it held when called. dest +// already matches exp before the push does anything, so RequireDoc alone would pass before a wrong push could land: a +// barrier document is written on source and waited for on dest first, so any earlier change to docID on source has +// been pushed by then. Source must already be in the state the push is expected to leave alone - wait on it first. +func (runner *SGRTestRunner) RequireDocUnchanged(docID string, source, dest *RestTester, exp ExpectedISGRDoc) *db.Document { + t := dest.TB() + t.Helper() + barrierID := docID + "_barrier" + // Put the barrier in the same channels as docID so that a replication or user scoped to them still carries it. + barrierBody, err := base.JSONMarshal(db.Body{"channels": exp.Channels}) + require.NoError(t, err) + barrierVersion := source.PutDoc(barrierID, string(barrierBody)) + runner.WaitForDocReplicated(barrierID, source, dest, barrierVersion) + return runner.RequireDoc(docID, dest, exp) +} + // waitForDocVersion waits for docID on rt to be at version - revTreeID only in v3, where the receiving peer mints its own // CV. It reads the document directly rather than over REST so that waiting doesn't populate the revision cache, and so // that tombstones and live documents are waited for the same way. @@ -381,8 +397,7 @@ func (runner *SGRTestRunner) waitForDocVersion(rt *RestTester, docID string, ver // revTreeID as an encoded CV, and keeps that in pv once the document is updated - but the peer that wrote the update // never stored the encoded version, since it only ever knew the revision through its rev tree. Both peers know the // same versions, so the HLVs are equivalent. -func hlvEqualAllowingEncodedRevs(t testing.TB, expected, actual *db.HybridLogicalVector, revChain []string) bool { - t.Helper() +func hlvEqualAllowingEncodedRevs(expected, actual *db.HybridLogicalVector, revChain []string) bool { if expected.SourceID != actual.SourceID || expected.Version != actual.Version || !maps.Equal(expected.MergeVersions, actual.MergeVersions) { return false diff --git a/rest/utilities_testing_isgr_test.go b/rest/utilities_testing_isgr_test.go index 1bff370f01..f1299edb72 100644 --- a/rest/utilities_testing_isgr_test.go +++ b/rest/utilities_testing_isgr_test.go @@ -134,13 +134,13 @@ func TestHLVEqualAllowingEncodedRevs(t *testing.T) { } for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - assert.Equal(t, tc.equal, hlvEqualAllowingEncodedRevs(t, tc.expected, tc.actual, revChain)) + assert.Equal(t, tc.equal, hlvEqualAllowingEncodedRevs(tc.expected, tc.actual, revChain)) }) } } // TestExpectedISGRDocFromPeerLegacyDoc checks that a snapshot of a document with no HLV describes it as it is, so -// RequireDoc can assert a peer holding a legacy document was left unchanged, while RequireDocReplicated still expects +// RequireDoc can assert a peer holding a legacy document was left unchanged, while WaitForDocReplicated still expects // the receiving peer to have the revTreeID-encoded CV. func TestExpectedISGRDocFromPeerLegacyDoc(t *testing.T) { base.RequireNumTestBuckets(t, 2) @@ -170,7 +170,7 @@ func TestExpectedISGRDocFromPeerLegacyDoc(t *testing.T) { require.NoError(t, ar.Start(peers.ActiveRT.Context())) defer func() { require.NoError(t, ar.Stop()) }() - doc := runner.RequireDocReplicated(docID, peers.PassiveRT, peers.ActiveRT, snapshot.Version) + doc := runner.WaitForDocReplicated(docID, peers.PassiveRT, peers.ActiveRT, snapshot.Version) require.NotNil(t, doc.HLV) if runner.IsV4Protocol() { encodedCV, err := db.LegacyRevToRevTreeEncodedVersion(legacyDoc.GetRevTreeID()) From c9fd41d57a515afcf70e2363995a0a1f1ccd4fb3 Mon Sep 17 00:00:00 2001 From: Gregory Newman-Smith Date: Mon, 5 Oct 2026 11:35:59 +0100 Subject: [PATCH 5/5] new commnet to address --- .../replicator_conflict_test.go | 2 +- rest/replicatortest/replicator_test.go | 4 +- .../replicator_test_legacy_rev_test.go | 51 +++++++------------ rest/utilities_testing_isgr.go | 45 +++++++++++++--- 4 files changed, 59 insertions(+), 43 deletions(-) diff --git a/rest/replicatortest/replicator_conflict_test.go b/rest/replicatortest/replicator_conflict_test.go index aab8692d21..5b8d081e93 100644 --- a/rest/replicatortest/replicator_conflict_test.go +++ b/rest/replicatortest/replicator_conflict_test.go @@ -2229,7 +2229,7 @@ func TestActiveReplicatorV4DefaultResolverWithTombstoneRemote(t *testing.T) { conflictResVersion.RevTreeID = newRev sgrRunner.RequireDoc(docID, activeRT, rest.ExpectedISGRDoc{Version: conflictResVersion, Deleted: true}) // activeRT adopted passiveRT's CV, so there is nothing new to push back and passiveRT must be unchanged - sgrRunner.RequireDocUnchanged(docID, activeRT, passiveRT, passiveDocBeforeReplication) + sgrRunner.RequireDocUnchanged(docID, activeRT, passiveRT, activeRT.GetDatabase().SGReplicateMgr.GetActiveReplicator(replicationID), db.ActiveReplicatorTypePush, passiveDocBeforeReplication) rt1Doc := activeRT.GetDocument(docID) // assert that remote cv is in pv now diff --git a/rest/replicatortest/replicator_test.go b/rest/replicatortest/replicator_test.go index 7f21d2afb0..a4783d2af1 100644 --- a/rest/replicatortest/replicator_test.go +++ b/rest/replicatortest/replicator_test.go @@ -4243,7 +4243,7 @@ func TestActiveReplicatorPushAndPullConflict(t *testing.T) { if sgrRunner.IsV4Protocol() && test.winner == remote { // rt1 adopted rt2's CV, so there is nothing new to push back: rt2 must be exactly as it was. rt1's // HLV also carries its losing version in pv, which rt2 never needs to learn about. - doc = sgrRunner.RequireDocUnchanged(docID, rt1, rt2, remoteDocBeforeReplication) + doc = sgrRunner.RequireDocUnchanged(docID, rt1, rt2, ar, db.ActiveReplicatorTypePush, remoteDocBeforeReplication) } else { // rt1's resolution produced a version rt2 doesn't have, which is pushed back, so rt2 must now // hold rt1's document @@ -6550,7 +6550,7 @@ func TestReplicatorConflictAttachment(t *testing.T) { if sgrRunner.IsV4Protocol() && test.conflictResolution == db.ConflictResolverRemoteWins { // activeRT adopted passiveRT's CV, so there is nothing new to push back and passiveRT must be unchanged sgrRunner.RequireDoc(docID, activeRT, rest.ExpectedISGRDoc{Version: *remoteWinsVersion}) - sgrRunner.RequireDocUnchanged(docID, activeRT, passiveRT, passiveDocBeforeReplication) + sgrRunner.RequireDocUnchanged(docID, activeRT, passiveRT, activeRT.GetDatabase().SGReplicateMgr.GetActiveReplicator(replicationID), db.ActiveReplicatorTypePush, passiveDocBeforeReplication) } else { expVersion := test.expectedFinalVersion if sgrRunner.IsV4Protocol() { diff --git a/rest/replicatortest/replicator_test_legacy_rev_test.go b/rest/replicatortest/replicator_test_legacy_rev_test.go index c145a56fd7..bd325259ab 100644 --- a/rest/replicatortest/replicator_test_legacy_rev_test.go +++ b/rest/replicatortest/replicator_test_legacy_rev_test.go @@ -175,31 +175,23 @@ func TestActiveReplicatorBiDirectionalPreUpgradedDocOnPeer(t *testing.T) { if tc.newRevOnActivePeer { sgrRunner.WaitForDocReplicated(docID, rt1, rt2, rest.DocVersion{RevTreeID: legacyRevRT1}) + activeDocBeforePullBack := rest.ExpectedISGRDocFromPeer(t, rt1, docID) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{legacyRevRT1, initLegacyRevRT1}) - // add new doc for some replication activity on pull side to allow us to assert that legacy rev 2-abc added - // isn't pulled back to rt1 now it has a legacy revID encoded CV - newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.WaitForDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 - - // assert that the document isn't replicated back to rt1 after legacy rev CV is written on rt2 - rt1Doc := rt1.GetDocument(docID) + // assert that legacy rev 2-abc isn't pulled back to rt1 now rt2 holds it with a legacy revID encoded CV + rt1Doc := sgrRunner.RequireDocUnchanged(docID, rt2, rt1, ar, db.ActiveReplicatorTypePull, activeDocBeforePullBack) assert.Equal(t, legacyRevRT1, rt1Doc.GetRevTreeID()) assert.Nil(t, rt1Doc.HLV) rest.RequireHistoryContains(t, rt1Doc.History, []string{legacyRevRT1, initLegacyRevRT1}) } else { sgrRunner.WaitForDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) + passiveDocBeforePushBack := rest.ExpectedISGRDocFromPeer(t, rt2, docID) rt1Doc := rt1.GetDocument(docID) rest.RequireHistoryContains(t, rt1Doc.History, []string{initLegacyRevRT2, legacyRevRT2}) - // add new doc for some replication activity on push side to allow us to assert that legacy rev 2-abc added - // isn't pushed back to rt2 now it has a legacy revID encoded CV - newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.WaitForDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 - - // assert that the document isn't replicated back to rt2 after legacy rev CV is written on rt1 - rt2Doc := rt2.GetDocument(docID) + // assert that legacy rev 2-abc isn't pushed back to rt2 now rt1 holds it with a legacy revID encoded CV + rt2Doc := sgrRunner.RequireDocUnchanged(docID, rt1, rt2, ar, db.ActiveReplicatorTypePush, passiveDocBeforePushBack) assert.Equal(t, legacyRevRT2, rt2Doc.GetRevTreeID()) assert.Nil(t, rt2Doc.HLV) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, legacyRevRT2}) @@ -780,15 +772,13 @@ func TestActiveReplicatorConflictPreUpgradedVersionEachSide(t *testing.T) { assert.NotEqual(c, legacyRevRT1, verPostConflictRes.RevTreeID) }, 10*time.Second, 50*time.Millisecond) sgrRunner.WaitForDocReplicated(docID, rt1, rt2, verPostConflictRes) + activeDocAfterResolution := rest.ExpectedISGRDocFromPeer(t, rt1, docID) rt2Doc := rt2.GetDocument(docID) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, legacyRevRT2, verPostConflictRes.RevTreeID}) - // add doc for some replication activity on pull to assert that local doesn't change after remote is written with legacy CV - newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.WaitForDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 - - // assert active side doc hasn't changed - rt1Doc := rt1.GetDocument(docID) + // assert active side doc doesn't change by pulling back the resolution it pushed to rt2, which rt2 + // stored with a legacy CV + rt1Doc := sgrRunner.RequireDocUnchanged(docID, rt2, rt1, ar, db.ActiveReplicatorTypePull, activeDocAfterResolution) rest.RequireDocRevTreeEqual(t, rest.DocVersion{RevTreeID: verPostConflictRes.RevTreeID}, rest.DocVersion{RevTreeID: rt1Doc.GetRevTreeID()}) tombstonedID := db.CreateRevIDWithBytes(3, legacyRevRT1, []byte(db.DeletedDocument)) // create what would be the tombstone rev id for local branch rest.RequireHistoryContains(t, rt1Doc.History, []string{legacyRevRT1, initLegacyRevRT1, verPostConflictRes.RevTreeID, legacyRevRT2, tombstonedID}) @@ -803,16 +793,14 @@ func TestActiveReplicatorConflictPreUpgradedVersionEachSide(t *testing.T) { }, 1) sgrRunner.WaitForDocReplicated(docID, rt2, rt1, rest.DocVersion{RevTreeID: legacyRevRT2}) + passiveDocAfterResolution := rest.ExpectedISGRDocFromPeer(t, rt2, docID) rt1Doc := rt1.GetDocument(docID) tombstonedID := db.CreateRevIDWithBytes(3, legacyRevRT1, []byte(db.DeletedDocument)) // create what would be the tombstone rev id for local branch rest.RequireHistoryContains(t, rt1Doc.History, []string{initLegacyRevRT1, legacyRevRT1, legacyRevRT2, tombstonedID}) - // add doc for some replication activity on push to assert that local doesn't change after remote is written with legacy CV - newDocVersion := rt1.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.WaitForDocReplicated("newdoc", rt1, rt2, newDocVersion) // wait for it to arrive at rt2 - - // assert passive side doc hasn't changed - rt2Doc := rt2.GetDocument(docID) + // assert passive side doc doesn't change by rt1 pushing back the remote win, which rt1 stored with a + // legacy CV + rt2Doc := sgrRunner.RequireDocUnchanged(docID, rt1, rt2, ar, db.ActiveReplicatorTypePush, passiveDocAfterResolution) rest.RequireDocRevTreeEqual(t, rest.DocVersion{RevTreeID: legacyRevRT2}, rest.DocVersion{RevTreeID: rt2Doc.GetRevTreeID()}) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, legacyRevRT2}) // legacy cv written to rt1 will correspond to local rev tree ID thus no HLV should be written yet @@ -964,6 +952,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { }, 10*time.Second, 50*time.Millisecond) // wait for this resolution to be pushed back to passive peer sgrRunner.WaitForDocReplicated(docID, rt1, rt2, verPostConflictRes) + activeDocAfterResolution := rest.ExpectedISGRDocFromPeer(t, rt1, docID) // assert original upgraded version in PV history assert.Equal(t, upgradedDocVersion.CV.Value, replicatedDoc.HLV.PreviousVersions[upgradedDocVersion.CV.SourceID]) @@ -976,12 +965,8 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { rt2BodyBytes, err := rt2Doc.BodyBytes(rt2.Context()) require.NoError(t, err) - // add doc for some replication activity on pull to assert that local doesn't change after remote is written with legacy CV - newDocVersion := rt2.PutDoc("newdoc", `{"channels": ["alice"]}`) - sgrRunner.WaitForDocReplicated("newdoc", rt2, rt1, newDocVersion) // wait for it to arrive at rt1 - - // assert active side doc hasn't changed - rt1Doc := rt1.GetDocument(docID) + // assert active side doc doesn't change by pulling back the resolution it pushed to rt2 + rt1Doc := sgrRunner.RequireDocUnchanged(docID, rt2, rt1, ar, db.ActiveReplicatorTypePull, activeDocAfterResolution) rest.RequireDocVersionEqual(t, verPostConflictRes, rt1Doc.ExtractDocVersion()) tombstonedID := db.CreateRevIDWithBytes(3, legacyRevRT1, []byte(db.DeletedDocument)) // create what would be the tombstone rev id for local branch rest.RequireHistoryContains(t, rt1Doc.History, []string{legacyRevRT1, initLegacyRevRT1, verPostConflictRes.RevTreeID, legacyRevRT2, tombstonedID}) @@ -1010,7 +995,7 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) { require.NoError(t, err) // assert passive side doc hasn't changed - rt1 adopted its CV, so there was nothing to push back - rt2Doc := sgrRunner.RequireDocUnchanged(docID, rt1, rt2, passiveDocBeforeReplication) + rt2Doc := sgrRunner.RequireDocUnchanged(docID, rt1, rt2, ar, db.ActiveReplicatorTypePush, passiveDocBeforeReplication) rest.RequireDocVersionEqual(t, upgradedDocVersion, rt2Doc.ExtractDocVersion()) rest.RequireHistoryContains(t, rt2Doc.History, []string{initLegacyRevRT2, upgradedDocVersion.RevTreeID}) diff --git a/rest/utilities_testing_isgr.go b/rest/utilities_testing_isgr.go index 8478e3c590..40111c88a7 100644 --- a/rest/utilities_testing_isgr.go +++ b/rest/utilities_testing_isgr.go @@ -353,19 +353,50 @@ func (runner *SGRTestRunner) RequireDoc(docID string, rt *RestTester, exp Expect return doc } -// RequireDocUnchanged requires dest to still match exp once source has pushed everything it held when called. dest -// already matches exp before the push does anything, so RequireDoc alone would pass before a wrong push could land: a -// barrier document is written on source and waited for on dest first, so any earlier change to docID on source has -// been pushed by then. Source must already be in the state the push is expected to leave alone - wait on it first. -func (runner *SGRTestRunner) RequireDocUnchanged(docID string, source, dest *RestTester, exp ExpectedISGRDoc) *db.Document { +// RequireDocUnchanged requires dest to still match exp once ar has replicated, in direction, everything source held +// when called. ar runs on source to push (direction db.ActiveReplicatorTypePush), or on dest to pull +// (db.ActiveReplicatorTypePull). dest already matches exp before the replication does anything, so RequireDoc alone +// would pass before a wrong rev could land. Instead a barrier document is written on source, and the checkpoint for +// direction is waited on to pass the barrier's sequence: the checkpoint only passes a sequence once dest has handled +// the rev for it and every earlier one, so any earlier change to docID on source has been handled by dest by then. +// Waiting for the barrier to arrive on dest isn't enough on its own, since dest handles several revs at once and the +// barrier can overtake docID. Source must already be in the state the replication is expected to leave alone - wait +// on it first. +func (runner *SGRTestRunner) RequireDocUnchanged(docID string, source, dest *RestTester, ar *db.ActiveReplicator, direction db.ActiveReplicatorDirection, exp ExpectedISGRDoc) *db.Document { t := dest.TB() t.Helper() + require.NotNil(t, ar, "no active replicator to wait on for doc %q", docID) + // Both checkpoints are in source's sequences: a push replicates the active peer's own changes feed, and a pull the + // remote peer's. + var lastSeq func() string + switch direction { + case db.ActiveReplicatorTypePush: + require.NotNil(t, ar.Push, "replication %q doesn't push", ar.ID) + lastSeq = func() string { return ar.Push.GetStatus().LastSeqPush } + case db.ActiveReplicatorTypePull: + require.NotNil(t, ar.Pull, "replication %q doesn't pull", ar.ID) + lastSeq = func() string { return ar.Pull.GetStatus().LastSeqPull } + default: + require.FailNow(t, "direction must be push or pull", "got %q", direction) + } barrierID := docID + "_barrier" - // Put the barrier in the same channels as docID so that a replication or user scoped to them still carries it. - barrierBody, err := base.JSONMarshal(db.Body{"channels": exp.Channels}) + // Put the barrier in every channel docID has been in on source, including ones it has been removed from, so that a + // replication or user scoped to them still carries it - a tombstone is in no channels at all. + sourceDoc := source.GetDocument(docID) + barrierChannels := slices.Sorted(maps.Keys(sourceDoc.Channels)) + barrierBody, err := base.JSONMarshal(db.Body{"channels": barrierChannels}) require.NoError(t, err) barrierVersion := source.PutDoc(barrierID, string(barrierBody)) + barrierSeq := source.GetDocument(barrierID).Sequence runner.WaitForDocReplicated(barrierID, source, dest, barrierVersion) + require.EventuallyWithT(t, func(c *assert.CollectT) { + checkpointSeq := lastSeq() + seq, err := db.ParsePlainSequenceID(checkpointSeq) + if !assert.NoError(c, err, "replication %q has unparseable %s checkpoint %q", ar.ID, direction, checkpointSeq) { + return + } + assert.GreaterOrEqual(c, seq.Seq, barrierSeq, "replication %q %s checkpoint hasn't reached barrier %q", ar.ID, direction, barrierID) + }, 10*time.Second, 50*time.Millisecond) return runner.RequireDoc(docID, dest, exp) }