CBG-4416 Use waitForTombstoneVersion to validate tombstone versions (#…

…7252)

* CBG-4416 Use waitForTombstoneVersion to validate tombstone versions

Validates CV for deletes.

* cleanup

* Logging cleanup

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/couchbase/sg-bucket v0.0.0-20241018143914-45ef51a0c1be
 	github.com/couchbaselabs/go-fleecedelta v0.0.0-20220909152808-6d09efa7a338
 	github.com/couchbaselabs/gocbconnstr v1.0.5
-	github.com/couchbaselabs/rosmar v0.0.0-20240924211003-933f0fd5bba0
+	github.com/couchbaselabs/rosmar v0.0.0-20241219222419-f9921fccab90
 	github.com/elastic/gosigar v0.14.3
 	github.com/felixge/fgprof v0.9.5
 	github.com/go-jose/go-jose/v4 v4.0.4
@@ -68,7 +68,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
-	github.com/mattn/go-sqlite3 v1.14.23 // indirect
+	github.com/mattn/go-sqlite3 v1.14.24 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect

go.sum

@@ -76,8 +76,8 @@ github.com/couchbaselabs/gocbconnstr v1.0.5 h1:e0JokB5qbcz7rfnxEhNRTKz8q1svoRvDo
 github.com/couchbaselabs/gocbconnstr v1.0.5/go.mod h1:KV3fnIKMi8/AzX0O9zOrO9rofEqrRF1d2rG7qqjxC7o=
 github.com/couchbaselabs/gocbconnstr/v2 v2.0.0-20240607131231-fb385523de28 h1:lhGOw8rNG6RAadmmaJAF3PJ7MNt7rFuWG7BHCYMgnGE=
 github.com/couchbaselabs/gocbconnstr/v2 v2.0.0-20240607131231-fb385523de28/go.mod h1:o7T431UOfFVHDNvMBUmUxpHnhivwv7BziUao/nMl81E=
-github.com/couchbaselabs/rosmar v0.0.0-20240924211003-933f0fd5bba0 h1:CQil6oxiHYhJBITdKTlxEUOetPdcgN6bk8wOZd4maDM=
-github.com/couchbaselabs/rosmar v0.0.0-20240924211003-933f0fd5bba0/go.mod h1:Abf5EPwi/7j5caDy2OPmo+L36I02H7sp9dkgek5t4bM=
+github.com/couchbaselabs/rosmar v0.0.0-20241219222419-f9921fccab90 h1:rQfOVEJvF8uGdRsWqNC4WZo5I6gOysFdE/1r0RyxEoE=
+github.com/couchbaselabs/rosmar v0.0.0-20241219222419-f9921fccab90/go.mod h1:suZBurj14d2YtLOW8pBc8mjQN8MhPFHHgPbqX1fDDlE=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -155,8 +155,8 @@ github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80/go.mod h1:imJHygn/1
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
-github.com/mattn/go-sqlite3 v1.14.23 h1:gbShiuAP1W5j9UOksQ06aiiqPMxYecovVGwmTxWtuw0=
-github.com/mattn/go-sqlite3 v1.14.23/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
+github.com/mattn/go-sqlite3 v1.14.24 h1:tpSp2G2KyMnnQu99ngJ47EIkWVmliIizyZBfPrBWDRM=
+github.com/mattn/go-sqlite3 v1.14.24/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=

topologytest/couchbase_server_peer_test.go

@@ -174,7 +174,7 @@ func (p *CouchbaseServerPeer) WaitForDeletion(dsName sgbucket.DataStoreName, doc
 // WaitForTombstoneVersion waits for a document to reach a specific version, this must be a tombstone. The test will fail if the document does not reach the expected version in 20s.
 func (p *CouchbaseServerPeer) WaitForTombstoneVersion(dsName sgbucket.DataStoreName, docID string, expected DocMetadata) {
 	docBytes := p.waitForDocVersion(dsName, docID, expected)
-	require.Nil(p.tb, docBytes, "expected tombstone for docID %s, got %s", docID, docBytes)
+	require.Empty(p.tb, docBytes, "expected tombstone for docID %s, got %s", docID, docBytes)
 }
 
 // waitForDocVersion waits for a document to reach a specific version and returns the body in bytes. The bytes will be nil if the document is a tombstone. The test will fail if the document does not reach the expected version in 20s.

topologytest/hlv_test.go

@@ -58,14 +58,14 @@ func waitForVersionAndBody(t *testing.T, dsName base.ScopeAndCollectionName, pee
 	}
 }
 
-func waitForDeletion(t *testing.T, dsName base.ScopeAndCollectionName, peers Peers, docID string, deleteActor string) {
-	for peerName, peer := range peers {
+func waitForTombstoneVersion(t *testing.T, dsName base.ScopeAndCollectionName, peers Peers, docID string, expectedVersion BodyAndVersion) {
+	for peerName, peer := range peers.SortedPeers() {
 		if peer.Type() == PeerTypeCouchbaseLite {
 			t.Logf("skipping deletion check for Couchbase Lite peer %s, CBG-4432", peerName)
 			continue
 		}
-		t.Logf("waiting for doc to be deleted on %s, written from %s", peer, deleteActor)
-		peer.WaitForDeletion(dsName, docID)
+		t.Logf("waiting for tombstone version %#v on %s, written from %s", expectedVersion, peer, expectedVersion.updatePeer)
+		peer.WaitForTombstoneVersion(dsName, docID, expectedVersion.docMeta)
 	}
 }
 
@@ -138,7 +138,11 @@ func deleteConflictDocs(t *testing.T, dsName base.ScopeAndCollectionName, peers
 			continue
 		}
 		deleteVersion := peer.DeleteDocument(dsName, docID)
-		t.Logf("deleteVersion: %+v", deleteVersion)
+		if peer.Type() == PeerTypeCouchbaseLite {
+			t.Logf("Don't include deleteVersion from Couchbase Lite peers when determining lastWrite %s, CBG-4432", peerName)
+			continue
+		}
+		t.Logf("deleteVersion on peer %s: %+v", peerName, deleteVersion)
 		documentVersion = append(documentVersion, BodyAndVersion{docMeta: deleteVersion, updatePeer: peerName})
 	}
 	index := len(documentVersion) - 1

topologytest/multi_actor_conflict_test.go

@@ -101,7 +101,7 @@ func TestMultiActorConflictDelete(t *testing.T) {
 			lastWrite := deleteConflictDocs(t, collectionName, peers, docID)
 
 			replications.Start()
-			waitForDeletion(t, collectionName, peers, docID, lastWrite.updatePeer)
+			waitForTombstoneVersion(t, collectionName, peers, docID, lastWrite)
 		})
 	}
 }
@@ -141,7 +141,7 @@ func TestMultiActorConflictResurrect(t *testing.T) {
 
 			replications.Start()
 
-			waitForDeletion(t, collectionName, peers, docID, lastWrite.updatePeer)
+			waitForTombstoneVersion(t, collectionName, peers, docID, lastWrite)
 			replications.Stop()
 
 			lastWriteVersion := updateConflictingDocs(t, collectionName, peers, docID, topology.description)

topologytest/multi_actor_no_conflict_test.go

@@ -67,8 +67,8 @@ func TestMultiActorDelete(t *testing.T) {
 					waitForVersionAndBody(t, collectionName, peers, docID, createVersion)
 
 					deleteVersion := deletePeer.DeleteDocument(collectionName, docID)
-					t.Logf("deleteVersion: %+v\n", deleteVersion) // FIXME: verify hlv in CBG-4416
-					waitForDeletion(t, collectionName, peers, docID, deletePeerName)
+					t.Logf("deleteVersion: %+v\n", deleteVersion)
+					waitForTombstoneVersion(t, collectionName, peers, docID, BodyAndVersion{docMeta: deleteVersion, updatePeer: deletePeerName})
 				}
 			}
 		})
@@ -104,8 +104,8 @@ func TestMultiActorResurrect(t *testing.T) {
 						waitForVersionAndBody(t, collectionName, peers, docID, createVersion)
 
 						deleteVersion := deletePeer.DeleteDocument(collectionName, docID)
-						t.Logf("deleteVersion: %+v\n", deleteVersion) // FIXME: verify hlv in CBG-4416
-						waitForDeletion(t, collectionName, peers, docID, deletePeerName)
+						t.Logf("deleteVersion: %+v\n", deleteVersion)
+						waitForTombstoneVersion(t, collectionName, peers, docID, BodyAndVersion{docMeta: deleteVersion, updatePeer: deletePeerName})
 
 						resBody := []byte(fmt.Sprintf(`{"activePeer": "%s", "createPeer": "%s", "deletePeer": "%s", "resurrectPeer": "%s", "topology": "%s", "action": "resurrect"}`, resurrectPeerName, createPeerName, deletePeer, resurrectPeer, topology.description))
 						resurrectVersion := resurrectPeer.WriteDocument(collectionName, docID, resBody)

topologytest/single_actor_test.go

@@ -92,7 +92,7 @@ func TestSingleActorDelete(t *testing.T) {
 					deleteVersion := activePeer.DeleteDocument(collectionName, docID)
 					t.Logf("createVersion: %+v, deleteVersion: %+v", createVersion.docMeta, deleteVersion)
 					t.Logf("waiting for document deletion on all peers")
-					waitForDeletion(t, collectionName, peers, docID, activePeerID)
+					waitForTombstoneVersion(t, collectionName, peers, docID, BodyAndVersion{docMeta: deleteVersion, updatePeer: activePeerID})
 				})
 			}
 		})
@@ -126,7 +126,7 @@ func TestSingleActorResurrect(t *testing.T) {
 					deleteVersion := activePeer.DeleteDocument(collectionName, docID)
 					t.Logf("createVersion: %+v, deleteVersion: %+v", createVersion, deleteVersion)
 					t.Logf("waiting for document deletion on all peers")
-					waitForDeletion(t, collectionName, peers, docID, activePeerID)
+					waitForTombstoneVersion(t, collectionName, peers, docID, BodyAndVersion{docMeta: deleteVersion, updatePeer: activePeerID})
 
 					body2 := []byte(fmt.Sprintf(`{"activePeer": "%s", "topology": "%s", "action": "resurrect"}`, activePeerID, topology.description))
 					resurrectVersion := activePeer.WriteDocument(collectionName, docID, body2)

topologytest/sync_gateway_peer_test.go

@@ -162,7 +162,7 @@ func (p *SyncGatewayPeer) WaitForDeletion(dsName sgbucket.DataStoreName, docID s
 // WaitForTombstoneVersion waits for a document to reach a specific version, this must be a tombstone. The test will fail if the document does not reach the expected version in 20s.
 func (p *SyncGatewayPeer) WaitForTombstoneVersion(dsName sgbucket.DataStoreName, docID string, expected DocMetadata) {
 	docBytes := p.WaitForDocVersion(dsName, docID, expected)
-	require.Nil(p.TB(), docBytes, "expected tombstone for docID %s, got %s", docID, docBytes)
+	require.Empty(p.TB(), docBytes, "expected tombstone for docID %s, got %s", docID, docBytes)
 }
 
 // RequireDocNotFound asserts that a document does not exist on the peer.

topologytest/version_test.go

@@ -34,7 +34,7 @@ func (v DocMetadata) CV(t require.TestingT) db.Version {
 	} else if v.HLV != nil {
 		return *v.HLV.ExtractCurrentVersionFromHLV()
 	}
-	require.FailNow(t, "no hlv available %#v", v)
+	require.FailNowf(t, "no hlv available", "%#v", v)
 	return db.Version{}
 }
 

xdcr/rosmar_xdcr.go

@@ -111,6 +111,11 @@ func (r *rosmarManager) processEvent(ctx context.Context, event sgbucket.FeedEve
 			base.WarnfCtx(ctx, "Skipping replicating doc %s, could not perform a kv op get doc in toBucket: %s", event.Key, err)
 			r.errorCount.Add(1)
 			return false
+		} else if base.IsDocNotFoundError(err) {
+			// Log if the target document exists as tombstone but doesn't have xattrs we care about
+			if actualTargetCas != 0 {
+				base.DebugfCtx(ctx, base.KeyVV, "Target document exists as a tombstone, but does not have _vv, _mou, _sync")
+			}
 		}
 
 		sourceHLV, sourceMou, nonMobileXattrs, body, err := processDCPEvent(&event)