CBG-4258 allow support of a CBL V3 client (#7721)

Author: torcolvin

topologytest/couchbase_lite_mock_peer_test.go

@@ -43,6 +43,7 @@ type CouchbaseLiteMockPeer struct {
 	blipClients        map[string]*PeerBlipTesterClient
 	name               string
 	symmetricRedundant bool // there is another peer that is symmetric to this one
+	peerType           PeerType
 }
 
 func (p *CouchbaseLiteMockPeer) String() string {
@@ -98,7 +99,16 @@ func (p *CouchbaseLiteMockPeer) getSingleSGBlipClient() *PeerBlipTesterClient {
 // CreateDocument creates a document on the peer. The test will fail if the document already exists.
 func (p *CouchbaseLiteMockPeer) CreateDocument(dsName sgbucket.DataStoreName, docID string, body []byte) BodyAndVersion {
 	client := p.getSingleSGBlipClient().CollectionClient(dsName)
-	docVersion, hlv := client.AddHLVRev(docID, rest.EmptyDocVersion(), body)
+	var docVersion db.DocVersion
+	var hlv *db.HybridLogicalVector
+	switch p.Type() {
+	case PeerTypeCouchbaseLiteV3:
+		docVersion = client.AddRev(docID, rest.EmptyDocVersion(), body)
+	case PeerTypeCouchbaseLite:
+		docVersion, hlv = client.AddHLVRev(docID, rest.EmptyDocVersion(), body)
+	default:
+		require.Fail(p.TB(), fmt.Sprintf("unsupported peer type %s for creating document", p.Type()))
+	}
 	docMetadata := DocMetadataFromDocVersion(p.TB(), docID, hlv, docVersion)
 	p.TB().Logf("%s: Created document %s with %#v", p, docID, docMetadata)
 	return BodyAndVersion{
@@ -114,9 +124,18 @@ func (p *CouchbaseLiteMockPeer) WriteDocument(dsName sgbucket.DataStoreName, doc
 	_, parentMeta := p.getLatestDocVersion(dsName, docID)
 	parentVersion := rest.EmptyDocVersion()
 	if parentMeta != nil {
-		parentVersion = &db.DocVersion{CV: parentMeta.CV(p.TB())}
+		parentVersion = &db.DocVersion{CV: parentMeta.CV(p.TB()), RevTreeID: parentMeta.RevTreeID}
+	}
+	var docVersion db.DocVersion
+	var hlv *db.HybridLogicalVector
+	switch p.Type() {
+	case PeerTypeCouchbaseLiteV3:
+		docVersion = client.AddRev(docID, parentVersion, body)
+	case PeerTypeCouchbaseLite:
+		docVersion, hlv = client.AddHLVRev(docID, parentVersion, body)
+	default:
+		require.Fail(p.TB(), fmt.Sprintf("unsupported peer type %s for writing document", p.Type()))
 	}
-	docVersion, hlv := client.AddHLVRev(docID, parentVersion, body)
 	docMetadata := DocMetadataFromDocVersion(p.TB(), docID, hlv, docVersion)
 	p.TB().Logf("%s: Wrote document %s with %#+v", p, docID, docMetadata)
 	return BodyAndVersion{
@@ -132,7 +151,7 @@ func (p *CouchbaseLiteMockPeer) DeleteDocument(dsName sgbucket.DataStoreName, do
 	_, parentMeta := p.getLatestDocVersion(dsName, docID)
 	parentVersion := rest.EmptyDocVersion()
 	if parentMeta != nil {
-		parentVersion = &db.DocVersion{CV: parentMeta.CV(p.TB())}
+		parentVersion = &db.DocVersion{CV: parentMeta.CV(p.TB()), RevTreeID: parentMeta.RevTreeID}
 	}
 	docVersion, hlv := client.Delete(docID, parentVersion)
 	docMeta := DocMetadataFromDocVersion(p.TB(), docID, hlv, docVersion)
@@ -142,14 +161,19 @@ func (p *CouchbaseLiteMockPeer) DeleteDocument(dsName sgbucket.DataStoreName, do
 
 // WaitForDocVersion waits for a document to reach a specific version. The test will fail if the document does not reach the expected version in 20s.
 func (p *CouchbaseLiteMockPeer) WaitForDocVersion(dsName sgbucket.DataStoreName, docID string, expected DocMetadata, topology Topology) db.Body {
+	compareHLV := !topology.CompareRevTreeOnly()
 	var data []byte
 	require.EventuallyWithT(p.TB(), func(c *assert.CollectT) {
 		var actual *DocMetadata
 		data, actual = p.getLatestDocVersion(dsName, docID)
 		if !assert.NotNil(c, actual, "Could not find docID:%+v on %p\nVersion %#v", docID, p, expected) {
 			return
 		}
-		assertHLVEqual(c, dsName, docID, p.name, *actual, data, expected, topology)
+		if compareHLV {
+			assertHLVEqual(c, dsName, docID, p.name, *actual, data, expected, topology)
+		} else {
+			assertRevTreeIDEqual(c, dsName, docID, p.name, *actual, data, expected, topology)
+		}
 	}, totalWaitTime, pollInterval)
 	var body db.Body
 	require.NoError(p.TB(), base.JSONUnmarshal(data, &body))
@@ -176,6 +200,9 @@ func (p *CouchbaseLiteMockPeer) WaitForCV(dsName sgbucket.DataStoreName, docID s
 func (p *CouchbaseLiteMockPeer) WaitForTombstoneVersion(dsName sgbucket.DataStoreName, docID string, expected DocMetadata, topology Topology) {
 	client := p.getSingleSGBlipClient().CollectionClient(dsName)
 	expectedVersion := db.DocVersion{CV: expected.CV(p.TB())}
+	if p.Type() == PeerTypeCouchbaseLiteV3 {
+		expectedVersion = db.DocVersion{RevTreeID: expected.RevTreeID}
+	}
 	require.EventuallyWithT(p.TB(), func(c *assert.CollectT) {
 		isTombstone, err := client.IsVersionTombstone(docID, expectedVersion)
 		require.NoError(c, err)
@@ -190,9 +217,10 @@ func (p *CouchbaseLiteMockPeer) Close() {
 	}
 }
 
-// Type returns PeerTypeCouchbaseLite.
+// Type returns PeerTypeCouchbaseLite if representing a Couchbase Lite 4.x client or PeerTypeCouchbaseLiteV3
+// if representing a 3.x client.
 func (p *CouchbaseLiteMockPeer) Type() PeerType {
-	return PeerTypeCouchbaseLite
+	return p.peerType
 }
 
 // IsSymmetricRedundant returns true if there is another peer set up that is identical to this one, and this peer doesn't need to participate in unique actions.
@@ -226,8 +254,14 @@ func (p *CouchbaseLiteMockPeer) CreateReplication(peer Peer, config PeerReplicat
 	}
 	const username = "user"
 	sg.rt.CreateUser(username, []string{"*"})
-	replication.btcRunner.SetSubprotocols([]string{db.CBMobileReplicationV4.SubprotocolString()})
-	// intentionally do not use base.TestCtx to drop test name for readability
+	switch p.Type() {
+	case PeerTypeCouchbaseLite:
+		replication.btcRunner.SetSubprotocols([]string{db.CBMobileReplicationV4.SubprotocolString()})
+	case PeerTypeCouchbaseLiteV3:
+		replication.btcRunner.SetSubprotocols([]string{db.CBMobileReplicationV3.SubprotocolString()})
+	default:
+		require.Fail(p.TB(), fmt.Sprintf("unsupported peer type %v for pull replication", p.Type()))
+	}
 	ctx := base.CorrelationIDLogCtx(sg.rt.TB().Context(), p.name)
 	replication.btc = replication.btcRunner.NewBlipTesterClientOptsWithRTAndContext(ctx, sg.rt, &rest.BlipTesterClientOpts{
 		Username: "user",

topologytest/couchbase_server_peer_test.go

@@ -216,7 +216,11 @@ func (p *CouchbaseServerPeer) waitForDocVersion(dsName sgbucket.DataStoreName, d
 			return
 		}
 		version = getDocVersion(docID, p, cas, xattrs)
-		assertHLVEqual(c, dsName, docID, p.name, version, docBytes, expected, topology)
+		if expected.HasHLV() {
+			assertHLVEqual(c, dsName, docID, p.name, version, docBytes, expected, topology)
+		} else {
+			assertRevTreeIDEqual(c, dsName, docID, p.name, version, docBytes, expected, topology)
+		}
 	}, totalWaitTime, pollInterval)
 	return docBytes
 }

topologytest/hlv_test.go

@@ -172,7 +172,11 @@ func waitForConvergingTombstones(t *testing.T, dsName base.ScopeAndCollectionNam
 				nonCBLVersion = &version
 				continue
 			}
-			assertHLVEqual(c, dsName, docID, peer, version, nil, *nonCBLVersion, topology)
+			if topology.CompareRevTreeOnly() {
+				assertRevTreeIDEqual(c, dsName, docID, peer, version, nil, *nonCBLVersion, topology)
+			} else {
+				assertHLVEqual(c, dsName, docID, peer, version, nil, *nonCBLVersion, topology)
+			}
 		}
 	}, totalWaitTime, pollInterval)
 }

topologytest/multi_actor_no_conflict_test.go

@@ -24,8 +24,8 @@ func TestMultiActorUpdate(t *testing.T) {
 		t.Run(topologySpec.description, func(t *testing.T) {
 			collectionName, topology := setupTests(t, topologySpec)
 			topology.StartReplications()
-			for createPeerName, createPeer := range topology.peers.ActivePeers() {
-				for updatePeerName, updatePeer := range topology.SortedPeers() {
+			for createPeerName, createPeer := range topology.ActivePeers() {
+				for updatePeerName, updatePeer := range topology.ActivePeers() {
 					topology.Run(t, "create="+createPeerName+",update="+updatePeerName, func(t *testing.T) {
 						docID := getDocID(t) + "_create=" + createPeerName + ",update=" + updatePeerName
 						body1 := []byte(fmt.Sprintf(`{"activePeer": "%s", "createPeer": "%s", "updatePeer": "%s", "topology": "%s", "action": "create"}`, createPeerName, createPeerName, updatePeer, topology.specDescription))
@@ -55,7 +55,7 @@ func TestMultiActorDelete(t *testing.T) {
 			collectionName, topology := setupTests(t, topologySpec)
 			topology.StartReplications()
 			for createPeerName, createPeer := range topology.ActivePeers() {
-				for deletePeerName, deletePeer := range topology.SortedPeers() {
+				for deletePeerName, deletePeer := range topology.ActivePeers() {
 					topology.Run(t, "create="+createPeerName+",delete="+deletePeerName, func(t *testing.T) {
 						docID := getDocID(t) + "_create=" + createPeerName + ",update=" + deletePeerName
 						body1 := []byte(fmt.Sprintf(`{"activePeer": "%s", "createPeer": "%s", "deletePeer": "%s", "topology": "%s", "action": "create"}`, createPeerName, createPeerName, deletePeer, topology.specDescription))
@@ -85,8 +85,8 @@ func TestMultiActorResurrect(t *testing.T) {
 			collectionName, topology := setupTests(t, topologySpec)
 			topology.StartReplications()
 			for createPeerName, createPeer := range topology.ActivePeers() {
-				for deletePeerName, deletePeer := range topology.SortedPeers() {
-					for resurrectPeerName, resurrectPeer := range topology.SortedPeers() {
+				for deletePeerName, deletePeer := range topology.ActivePeers() {
+					for resurrectPeerName, resurrectPeer := range topology.ActivePeers() {
 						topology.Run(t, fmt.Sprintf("create=%s,delete=%s,resurrect=%s", createPeerName, deletePeerName, resurrectPeerName), func(t *testing.T) {
 							docID := getDocID(t) + "_create=" + createPeerName + ",delete=" + deletePeerName + ",resurrect=" + resurrectPeerName
 							body1 := []byte(fmt.Sprintf(`{"activePeer": "%s", "createPeer": "%s", "deletePeer": "%s", "resurrectPeer": "%s", "topology": "%s", "action": "create"}`, createPeerName, createPeerName, deletePeer, resurrectPeer, topologySpec.description))

topologytest/peer_test.go

@@ -233,6 +233,9 @@ func (to Topology) GetDocState(t assert.TestingT, dsName sgbucket.DataStoreName,
 
 // ActivePeers returns a list of unique peers in the topology. If there is an identical symmetric peer, do not return it.
 func (to Topology) ActivePeers() iter.Seq2[string, Peer] {
+	if to.CompareRevTreeOnly() {
+		return to.peers.NonImportSortedPeers()
+	}
 	return to.peers.ActivePeers()
 }
 
@@ -241,6 +244,16 @@ func (to Topology) SortedPeers() iter.Seq2[string, Peer] {
 	return to.peers.SortedPeers()
 }
 
+// CompareRevTreeOnly is true for a given topology when comparing only revtree is correct.
+func (to Topology) CompareRevTreeOnly() bool {
+	for _, peer := range to.peers.ActivePeers() {
+		if peer.Type() == PeerTypeCouchbaseLiteV3 {
+			return true
+		}
+	}
+	return false
+}
+
 // Run is equivalent to testing.T.Run() but updates the underlying TB to the new testing.T
 // so that checks are made against the right instance (otherwise the outer test complains
 // "subtest may have called FailNow on a parent test")
@@ -308,10 +321,27 @@ const (
 	PeerTypeCouchbaseServer PeerType = iota
 	// PeerTypeCouchbaseLite represents a Couchbase Lite peer. This is currently backed in memory but will be backed by in memory structure that will send and receive blip messages. Future expansion to real Couchbase Lite peer in CBG-4260.
 	PeerTypeCouchbaseLite
+	// PeerTypeCouchbaseLiteV3 represents a Couchbase Lite peer. This is currently backed in memory but will be backed by in memory structure that will send and receive blip messages. Future expansion to real Couchbase Lite peer in CBG-4260.
+	PeerTypeCouchbaseLiteV3
 	// PeerTypeSyncGateway represents a Sync Gateway peer backed by a RestTester.
 	PeerTypeSyncGateway
 )
 
+func (pt PeerType) String() string {
+	switch pt {
+	case PeerTypeCouchbaseServer:
+		return "Couchbase Server Peer"
+	case PeerTypeCouchbaseLite:
+		return "Couchbase Lite Peer"
+	case PeerTypeCouchbaseLiteV3:
+		return "Couchbase Lite V3 Peer"
+	case PeerTypeSyncGateway:
+		return "Sync Gateway Peer"
+	default:
+		return fmt.Sprintf("Unknown Peer Type %d", pt)
+	}
+}
+
 // PeerBucketID represents a specific bucket for a test. This allows multiple Sync Gateway instances to point to the same bucket, or a different buckets. There is no significance to the numbering of the buckets. We can use as many buckets as the MainTestBucketPool allows.
 type PeerBucketID int
 
@@ -349,14 +379,15 @@ func NewPeer(t *testing.T, name string, buckets map[PeerBucketID]*base.TestBucke
 		}
 		p.UpdateTB(t)
 		return p
-	case PeerTypeCouchbaseLite:
+	case PeerTypeCouchbaseLite, PeerTypeCouchbaseLiteV3:
 		require.Equal(t, PeerBucketNoBackingBucket, opts.BucketID, "bucket should not be specified for Couchbase Lite peer %+v", opts)
 		_, ok := buckets[opts.BucketID]
 		require.False(t, ok, "bucket should not be specified for Couchbase Lite peer")
 		p := &CouchbaseLiteMockPeer{
 			name:               name,
 			blipClients:        make(map[string]*PeerBlipTesterClient),
 			symmetricRedundant: opts.Symmetric,
+			peerType:           opts.Type,
 		}
 		p.UpdateTB(t)
 		return p
@@ -365,7 +396,7 @@ func NewPeer(t *testing.T, name string, buckets map[PeerBucketID]*base.TestBucke
 		require.True(t, ok, "bucket not found for bucket ID %d", opts.BucketID)
 		return newSyncGatewayPeer(t, name, bucket, opts.Symmetric)
 	default:
-		require.Fail(t, fmt.Sprintf("unsupported peer type %T", opts.Type))
+		require.Fail(t, fmt.Sprintf("unsupported peer type %s", opts.Type))
 	}
 	return nil
 }

topologytest/sync_gateway_peer_test.go

@@ -161,7 +161,11 @@ func (p *SyncGatewayPeer) WaitForDocVersion(dsName sgbucket.DataStoreName, docID
 		// Only assert on CV since RevTreeID might not be present if this was a Couchbase Server write
 		bodyBytes, err := doc.BodyBytes(ctx)
 		assert.NoError(c, err)
-		assertHLVEqual(c, dsName, docID, p.name, version, bodyBytes, expected, topology)
+		if expected.HasHLV() {
+			assertHLVEqual(c, dsName, docID, p.name, version, bodyBytes, expected, topology)
+		} else {
+			assertRevTreeIDEqual(c, dsName, docID, p.name, version, bodyBytes, expected, topology)
+		}
 	}, totalWaitTime, pollInterval)
 	return doc.Body(ctx)
 }

topologytest/topologies_test.go

@@ -58,6 +58,48 @@ var TopologySpecifications = []TopologySpecification{
 			},
 		},
 	},
+	{
+		/*
+			+ - - - - - - +
+			' +---------+ '
+			' |  cbs1   | '
+			' +---------+ '
+			' +---------+ '
+			' |   sg1   | '
+			' +---------+ '
+			+ - - - - - - +
+			      ^
+			      |
+			      |
+			      v
+			  +---------+
+			  |cbl1 (v3)|
+			  +---------+
+		*/
+		description: "CBLV3<->SG<->CBS 1.1",
+		peers: map[string]PeerOptions{
+			"cbs1": {Type: PeerTypeCouchbaseServer, BucketID: PeerBucketID1},
+			"sg1":  {Type: PeerTypeSyncGateway, BucketID: PeerBucketID1},
+			"cbl1": {Type: PeerTypeCouchbaseLiteV3},
+		},
+		replications: []PeerReplicationDefinition{
+			{
+				activePeer:  "cbl1",
+				passivePeer: "sg1",
+				config: PeerReplicationConfig{
+					direction: PeerReplicationDirectionPull,
+				},
+			},
+			{
+				activePeer:  "cbl1",
+				passivePeer: "sg1",
+				config: PeerReplicationConfig{
+					direction: PeerReplicationDirectionPush,
+				},
+			},
+		},
+	},
+
 	{
 		/*
 			Test topology 1.2

topologytest/version_test.go

@@ -29,6 +29,15 @@ type DocMetadata struct {
 	ImplicitHLV *db.HybridLogicalVector // ImplicitHLV is the version of the document, if there was no HLV
 }
 
+// HasHLV returns true if the version has an HLV. This will always return true on Couchbase Server or Sync Gateway
+// peers since they can construct an implicit HLV from the CAS.
+func (v DocMetadata) HasHLV() bool {
+	if v.HLV != nil && v.HLV.GetCurrentVersionString() != "" {
+		return true
+	}
+	return v.ImplicitHLV != nil && v.ImplicitHLV.GetCurrentVersionString() != ""
+}
+
 // CV returns the current version of the document.
 func (v DocMetadata) CV(t assert.TestingT) db.Version {
 	if v.ImplicitHLV != nil {
@@ -98,3 +107,8 @@ func assertHLVEqual(t assert.TestingT, dsName sgbucket.DataStoreName, docID stri
 func assertCVEqual(t assert.TestingT, dsName sgbucket.DataStoreName, docID string, p string, version DocMetadata, body []byte, expected DocMetadata, topology Topology) {
 	assert.Equal(t, expected.CV(t), version.CV(t), "Actual HLV's CV does not match expected on %s for peer %s.  Expected: %#v, Actual: %#v\nActual Body: %s\n%s", docID, p, expected, version, body, topology.GetDocState(t, dsName, docID))
 }
+
+// assertRevTreeIDEqual asserts that revtree ID of the version is equal to the expected CV.
+func assertRevTreeIDEqual(t assert.TestingT, dsName sgbucket.DataStoreName, docID string, p string, version DocMetadata, body []byte, expected DocMetadata, topology Topology) {
+	assert.Equal(t, expected.RevTreeID, version.RevTreeID, "Actual revtree id does not match expected on %s for peer %s.  Expected: %#v, Actual: %#v\nActual Body: %s\n%s", docID, p, expected, version, body, topology.GetDocState(t, dsName, docID))
+}