From 9ee8d7bc15ae2ec9ac42aaa6574a3d2c0ba9a300 Mon Sep 17 00:00:00 2001 From: Tor Colvin Date: Sat, 4 Jul 2026 10:18:44 -0400 Subject: [PATCH] CBG-5547 handle topologytest conflicts when version is matching If sourceIDs are different but version element is exaclty the same, the topologytest would fail with unexpected revisions. Implement tests with existing behavior: - XDCR: local wins - CBL: remote wins - ISGR: local wins Add some hooks to force the ability to set the clock for these tests so that this behavior doesn't regress. --- db/utilities_hlv_testing.go | 5 + go.mod | 2 +- go.sum | 4 +- rest/replicatortest/replicator_test.go | 73 +++++++++ rest/utilities_testing_blip_client.go | 10 +- .../multi_actor_conflict_clock_tie_test.go | 91 ++++++++++ topologytest/multi_actor_conflict_test.go | 155 +++++++++++++----- xdcr/xdcr_test.go | 54 ++++++ 8 files changed, 346 insertions(+), 48 deletions(-) create mode 100644 topologytest/multi_actor_conflict_clock_tie_test.go diff --git a/db/utilities_hlv_testing.go b/db/utilities_hlv_testing.go index c03d30a5e4..1fa2ad8665 100644 --- a/db/utilities_hlv_testing.go +++ b/db/utilities_hlv_testing.go @@ -421,3 +421,8 @@ func AlterHLVForTest(t *testing.T, ctx context.Context, dataStore base.DataStore func (db *DatabaseContext) GetHLCValueForTest(floorValue uint64) uint64 { return db.hlc.Now(floorValue) } + +// SetHLCClockForTest overrides the database's HLC clock function, for deterministic HLV version generation in tests. +func (db *DatabaseContext) SetHLCClockForTest(clockFn func() uint64) { + db.hlc.SetClockForTest(clockFn) +} diff --git a/go.mod b/go.mod index ab804445ac..310032b613 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,7 @@ require ( github.com/couchbasedeps/fast-skiplist v0.0.0-20250722125747-e0dd031fe2ac github.com/couchbaselabs/go-fleecedelta v0.0.0-20220909152808-6d09efa7a338 github.com/couchbaselabs/gocbconnstr v1.0.5 - github.com/couchbaselabs/rosmar v0.0.0-20260706131717-11d9680fd14d + github.com/couchbaselabs/rosmar v0.0.0-20260706134933-2f0dabde975c github.com/elastic/gosigar v0.14.4 github.com/felixge/fgprof v0.9.5 github.com/go-jose/go-jose/v4 v4.1.4 diff --git a/go.sum b/go.sum index 4b02f19dd1..5ba9d561fe 100644 --- a/go.sum +++ b/go.sum @@ -68,8 +68,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 h1:HU9DlAYYWR69jQnLN6cpg0fh0hxW/8d5hnglCXXjW78= github.com/couchbaselabs/gocbconnstr/v2 v2.0.0/go.mod h1:o7T431UOfFVHDNvMBUmUxpHnhivwv7BziUao/nMl81E= -github.com/couchbaselabs/rosmar v0.0.0-20260706131717-11d9680fd14d h1:aO+wy8QnEj2XsLI5jk01zXvxSjVbOMbpWeLzTtWdkms= -github.com/couchbaselabs/rosmar v0.0.0-20260706131717-11d9680fd14d/go.mod h1:9tmSd+tF1TDFlrlbUB8Q1XKxOoZyJu97wIQ5pTN6B20= +github.com/couchbaselabs/rosmar v0.0.0-20260706134933-2f0dabde975c h1:rTggaJZ5WvEizK3YV6B+wFvqlC3pUxHRCqH4oYzff9M= +github.com/couchbaselabs/rosmar v0.0.0-20260706134933-2f0dabde975c/go.mod h1:9tmSd+tF1TDFlrlbUB8Q1XKxOoZyJu97wIQ5pTN6B20= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= diff --git a/rest/replicatortest/replicator_test.go b/rest/replicatortest/replicator_test.go index cfb40bff03..13103a3da0 100644 --- a/rest/replicatortest/replicator_test.go +++ b/rest/replicatortest/replicator_test.go @@ -27,6 +27,7 @@ import ( "time" "github.com/couchbase/gocb/v2" + sgbucket "github.com/couchbase/sg-bucket" "github.com/couchbase/sync_gateway/testing/assert" "github.com/couchbase/sync_gateway/testing/require" "github.com/google/uuid" @@ -6141,6 +6142,78 @@ func TestDefaultConflictResolverWithTombstoneLocal(t *testing.T) { } } +// TestDefaultConflictResolverMatchingTimestamps verifies that default (HLV-aware) conflict resolution for different +// source IDs but matching versions is local wins. +func TestDefaultConflictResolverMatchingTimestamps(t *testing.T) { + base.LongRunningTest(t) + + base.RequireNumTestBuckets(t, 2) + + // Matching-timestamps ties are an HLV/V4-only scenario (V3/revTree conflicts are resolved by generation and + // digest, not by timestamp), so only the V4 subtest is run. + sgrRunner := rest.NewSGRTestRunner(t) + sgrRunner.RunSubprotocolV4(func(t *testing.T) { + activeRT, passiveRT, remoteDBURLString := sgrRunner.SetupSGRPeers(t) + remoteURL, err := url.Parse(remoteDBURLString) + require.NoError(t, err) + activeRTCtx := activeRT.Context() + + fixedTime := sgbucket.HLCWallClock() + tieClock := func() uint64 { return fixedTime } + activeRT.GetDatabase().SetHLCClockForTest(tieClock) + passiveRT.GetDatabase().SetHLCClockForTest(tieClock) + + docID := "doc" + activeRTVersion := createDoc(activeRT, docID, "local") + passiveRTVersion := createDoc(passiveRT, docID, "remote") + require.Equal(t, activeRTVersion.CV.Value, passiveRTVersion.CV.Value, "test requires activeRT and passiveRT to generate matching HLC timestamps") + require.NotEqual(t, activeRTVersion.CV.SourceID, passiveRTVersion.CV.SourceID) + + replicationStats := dbReplicatorStats(t) + config := db.ActiveReplicatorConfig{ + ID: t.Name(), + Direction: db.ActiveReplicatorTypePushAndPull, + RemoteDBURL: remoteURL, + ActiveDB: &db.Database{ + DatabaseContext: activeRT.GetDatabase(), + }, + Continuous: true, + ConflictResolverFuncForHLV: db.DefaultLWWConflictResolutionType, + ReplicationStatsMap: replicationStats, + CollectionsEnabled: !activeRT.GetDatabase().OnlyDefaultCollection(), + SupportedBLIPProtocols: sgrRunner.SupportedSubprotocols, + } + + // Create active replicator and start replication. + ar, err := db.NewActiveReplicator(activeRTCtx, &config) + require.NoError(t, err) + require.NoError(t, ar.Start(activeRTCtx), "Error starting replication") + defer func() { require.NoError(t, ar.Stop(), "Error stopping replication") }() + + requireConflictResolvedLocally := func(t *testing.T, rt *rest.RestTester) { + t.Helper() + collection, ctx := rt.GetSingleTestDatabaseCollectionWithUser() + require.EventuallyWithT(t, func(c *assert.CollectT) { + doc, err := collection.GetDocument(ctx, docID, db.DocUnmarshalAll) + if !assert.NoError(c, err) { + return + } + if !assert.NotNil(c, doc.HLV) { + return + } + assert.Equal(c, activeRTVersion.CV.SourceID, doc.HLV.SourceID, "expected local (activeRT) source to win the tie") + assert.Equal(c, "local", doc.Body(base.TestCtx(t))["key"], "expected local (activeRT) body to win the tie") + }, 10*time.Second, 100*time.Millisecond) + } + requireConflictResolvedLocally(t, activeRT) + requireConflictResolvedLocally(t, passiveRT) + + base.RequireWaitForStat(t, replicationStats.ConflictResolvedLocalCount.Value, 1) + base.RequireWaitForStat(t, replicationStats.ConflictResolvedMergedCount.Value, 0) + base.RequireWaitForStat(t, replicationStats.ConflictResolvedRemoteCount.Value, 0) + }) +} + // This test ensures that the remote tombstone revision wins over non-tombstone revision // whilst applying default conflict resolution policy through pushAndPull replication. func TestDefaultConflictResolverWithTombstoneRemote(t *testing.T) { diff --git a/rest/utilities_testing_blip_client.go b/rest/utilities_testing_blip_client.go index 4815389077..22ab8360cd 100644 --- a/rest/utilities_testing_blip_client.go +++ b/rest/utilities_testing_blip_client.go @@ -358,8 +358,9 @@ func (btcc *BlipTesterCollectionClient) _resolveConflict(incomingHLV *db.HybridL func (btcc *BlipTesterCollectionClient) _resolveConflictLWW(incomingHLV *db.HybridLogicalVector, incomingBody []byte, latestLocalRev *clientDocRev) (body []byte, hlv db.HybridLogicalVector) { latestLocalHLV := latestLocalRev.HLV updatedHLV := latestLocalRev.HLV.Copy() - // resolve conflict in favor of remote document - if incomingHLV.Version > latestLocalHLV.Version { + // Resolve conflict in favor of the remote document, including on a tie: this mirrors real Couchbase Lite's + // LWW conflict resolution, which favors the incoming (remote) revision rather than the existing local one. + if incomingHLV.Version >= latestLocalHLV.Version { updatedHLV.UpdateWithIncomingHLV(incomingHLV) return incomingBody, *updatedHLV } @@ -1053,6 +1054,11 @@ func (btc *BlipTesterClient) TB() testing.TB { return btc.rt.TB() } +// SetHLCClockForTest overrides the client's HLC clock function, for deterministic HLV version generation in tests. +func (btc *BlipTesterClient) SetHLCClockForTest(clockFn func() uint64) { + btc.hlc.SetClockForTest(clockFn) +} + // Close shuts down all the clients and clears all messages stored. func (btc *BlipTesterClient) Close() { for _, collectionClient := range btc.collectionClients { diff --git a/topologytest/multi_actor_conflict_clock_tie_test.go b/topologytest/multi_actor_conflict_clock_tie_test.go new file mode 100644 index 0000000000..958e346dff --- /dev/null +++ b/topologytest/multi_actor_conflict_clock_tie_test.go @@ -0,0 +1,91 @@ +// 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 topologytest + +import ( + "testing" + + sgbucket "github.com/couchbase/sg-bucket" + "github.com/couchbase/sync_gateway/base" + "github.com/couchbase/sync_gateway/testing/require" + "github.com/couchbaselabs/rosmar" +) + +// This file supports the forceClockTie dimension of the TestMultiActorConflict* tests in +// multi_actor_conflict_test.go, to allow conflicting documents to be generated with matching timestamps. +// +// There are several independent clocks: +// +// - Sync Gateway, used for REST writes +// - Couchbase Lite mock client +// - Couchbase Server +// - Couchbase Server has independent clocks per bucket +// - Rosmar shares one clock across all buckets (implementation detail) +// +// controllableClockPeerCount returns the number of peers in the spec whose HLC clock forceHLCClockTieForTest can +// override. Used to skip the ClockTie subtest before setupTests, since skipping a topology mid-test (after +// peers and replications are configured but before they're started) trips replication teardown, which assumes +// StartReplications was called. +func controllableClockPeerCount(topologySpec TopologySpecification) (count int) { + for _, opts := range topologySpec.peers { + switch opts.Type { + case PeerTypeSyncGateway, PeerTypeCouchbaseLite, PeerTypeCouchbaseLiteV3: + count++ + case PeerTypeCouchbaseServer: + // A Couchbase Server peer's clock is only controllable when it's backed by rosmar - see + // forceHLCClockTieForTest. + if base.UnitTestUrlIsWalrus() { + count++ + } + } // exhaustive:enforce + } + return count +} + +// forceHLCClockTieForTest pins the HLC clocks of every controllable peer in the topology to the same fixed +// timestamp, forcing each of their next brand-new-document writes to generate an identical HLV current-version +// Value. Returns the number of peers whose clock was overridden. +// +// Unlike a Sync Gateway/Couchbase Lite peer's clock (a fresh instance per peer, discarded with the test), rosmar's +// clock is a single instance shared by every rosmar bucket in the whole test process. Freezing it here without +// restoring it would leak into every later test in the process (CAS values frozen in the past, or colliding with +// already-written docs) - so t is required purely to register a t.Cleanup that restores the real clock once this +// specific (sub)test finishes, before any sibling or later test runs. +func forceHLCClockTieForTest(t testing.TB, topology Topology) (forced int) { + ctx := base.TestCtx(t) + fixedTime := sgbucket.HLCWallClock() + tieClock := func() uint64 { return fixedTime } + rosmarClockSet := false + for name, peer := range topology.peers.NonImportSortedPeers() { + switch peer.Type() { + case PeerTypeSyncGateway: + p, ok := peer.(*SyncGatewayPeer) + require.True(t, ok, "peer %s (%T) is a PeerTypeSyncGateway but not a *SyncGatewayPeer", name, peer) + p.rt.GetDatabase().SetHLCClockForTest(tieClock) + forced++ + case PeerTypeCouchbaseLite, PeerTypeCouchbaseLiteV3: + p, ok := peer.(*CouchbaseLiteMockPeer) + require.True(t, ok, "peer %s (%T) is a %s but not a *CouchbaseLiteMockPeer", name, peer, peer.Type()) + p.getSingleSGBlipClient().btc.SetHLCClockForTest(tieClock) + forced++ + case PeerTypeCouchbaseServer: + if !base.UnitTestUrlIsWalrus() { + base.InfofCtx(ctx, base.KeySGTest, "forceHLCClockTieForTest: not overriding clock for peer %s (%T), Couchbase Server-backed peers can only have their clock forced when backed by rosmar", name, peer) + continue + } + if !rosmarClockSet { + rosmar.SetClockForTest(tieClock) + t.Cleanup(func() { rosmar.SetClockForTest(sgbucket.HLCWallClock) }) + rosmarClockSet = true + } + forced++ + } // exhaustive:enforce + } + return forced +} diff --git a/topologytest/multi_actor_conflict_test.go b/topologytest/multi_actor_conflict_test.go index 2d0190f3d4..613eef7485 100644 --- a/topologytest/multi_actor_conflict_test.go +++ b/topologytest/multi_actor_conflict_test.go @@ -14,21 +14,45 @@ import ( "github.com/couchbase/sync_gateway/base" ) +// clockTieSubtestName names the forceClockTie dimension run by each TestMultiActorConflict* test below. +func clockTieSubtestName(forceClockTie bool) string { + if forceClockTie { + return "ClockTie" + } + return "default" +} + // TestMultiActorConflictCreate // 1. create document on each peer with different contents // 2. start replications // 3. wait for documents to exist with a matching CV for Couchbase Lite peers, and a full HLV match for non Couchbase // Lite peers. The body should match. +// +// The forceClockTie dimension forces every controllable peer's HLC clock to the same fixed timestamp +// immediately before step 1, so the creates in step 1 tie on HLV Value (see forceHLCClockTieForTest) instead of +// relying on real clock jitter to hit that case. func TestMultiActorConflictCreate(t *testing.T) { base.LongRunningTest(t) for _, topologySpec := range append(simpleTopologySpecifications, TopologySpecifications...) { t.Run(topologySpec.description, func(t *testing.T) { - collectionName, topology := setupTests(t, topologySpec) - docID := getDocID(t) - docVersion := createConflictingDocs(collectionName, docID, topology) - topology.StartReplications() - waitForCVAndBody(t, collectionName, docID, docVersion, topology) + for _, forceClockTie := range []bool{false, true} { + t.Run(clockTieSubtestName(forceClockTie), func(t *testing.T) { + if forceClockTie && controllableClockPeerCount(topologySpec) < 2 { + t.Skipf("only %d controllable (Sync Gateway/Couchbase Lite) peer(s) in this topology, need at least 2 to force an HLC clock tie", controllableClockPeerCount(topologySpec)) + } + + collectionName, topology := setupTests(t, topologySpec) + if forceClockTie { + forceHLCClockTieForTest(t, topology) + } + + docID := getDocID(t) + docVersion := createConflictingDocs(collectionName, docID, topology) + topology.StartReplications() + waitForCVAndBody(t, collectionName, docID, docVersion, topology) + }) + } }) } } @@ -43,24 +67,39 @@ func TestMultiActorConflictCreate(t *testing.T) { // 6. start replications // 7. wait for documents to exist with a matching CV for Couchbase Lite peers, and a full HLV match for non Couchbase // Lite peers. The body should match. +// +// The forceClockTie dimension forces every controllable peer's HLC clock to the same fixed timestamp +// immediately before step 5, so the updates in step 5 tie on HLV Value instead of relying on real clock jitter +// to hit that case. func TestMultiActorConflictUpdate(t *testing.T) { base.LongRunningTest(t) for _, topologySpec := range append(simpleTopologySpecifications, TopologySpecifications...) { t.Run(topologySpec.description, func(t *testing.T) { - collectionName, topology := setupTests(t, topologySpec) - - docID := getDocID(t) - docVersion := createConflictingDocs(collectionName, docID, topology) - - topology.StartReplications() - waitForCVAndBody(t, collectionName, docID, docVersion, topology) - - topology.StopReplications() - - docVersion = updateConflictingDocs(collectionName, docID, topology) - topology.StartReplications() - waitForCVAndBody(t, collectionName, docID, docVersion, topology) + for _, forceClockTie := range []bool{false, true} { + t.Run(clockTieSubtestName(forceClockTie), func(t *testing.T) { + if forceClockTie && controllableClockPeerCount(topologySpec) < 2 { + t.Skipf("only %d controllable (Sync Gateway/Couchbase Lite) peer(s) in this topology, need at least 2 to force an HLC clock tie", controllableClockPeerCount(topologySpec)) + } + + collectionName, topology := setupTests(t, topologySpec) + + docID := getDocID(t) + docVersion := createConflictingDocs(collectionName, docID, topology) + + topology.StartReplications() + waitForCVAndBody(t, collectionName, docID, docVersion, topology) + + topology.StopReplications() + + if forceClockTie { + forceHLCClockTieForTest(t, topology) + } + docVersion = updateConflictingDocs(collectionName, docID, topology) + topology.StartReplications() + waitForCVAndBody(t, collectionName, docID, docVersion, topology) + }) + } }) } } @@ -73,23 +112,38 @@ func TestMultiActorConflictUpdate(t *testing.T) { // 5. delete documents on all peers // 6. start replications // 7. assert that the documents are deleted on all peers and have hlv sources equal to the number of active peers +// +// The forceClockTie dimension forces every controllable peer's HLC clock to the same fixed timestamp +// immediately before step 5, so the concurrent deletes in step 5 tie on HLV Value instead of relying on real +// clock jitter to hit that case. func TestMultiActorConflictDelete(t *testing.T) { base.LongRunningTest(t) for _, topologySpec := range append(simpleTopologySpecifications, TopologySpecifications...) { t.Run(topologySpec.description, func(t *testing.T) { - collectionName, topology := setupTests(t, topologySpec) - docID := getDocID(t) - docVersion := createConflictingDocs(collectionName, docID, topology) - - topology.StartReplications() - waitForCVAndBody(t, collectionName, docID, docVersion, topology) - - topology.StopReplications() - deleteConflictDocs(collectionName, docID, topology) - - topology.StartReplications() - waitForConvergingTombstones(t, collectionName, docID, topology) + for _, forceClockTie := range []bool{false, true} { + t.Run(clockTieSubtestName(forceClockTie), func(t *testing.T) { + if forceClockTie && controllableClockPeerCount(topologySpec) < 2 { + t.Skipf("only %d controllable (Sync Gateway/Couchbase Lite) peer(s) in this topology, need at least 2 to force an HLC clock tie", controllableClockPeerCount(topologySpec)) + } + + collectionName, topology := setupTests(t, topologySpec) + docID := getDocID(t) + docVersion := createConflictingDocs(collectionName, docID, topology) + + topology.StartReplications() + waitForCVAndBody(t, collectionName, docID, docVersion, topology) + + topology.StopReplications() + if forceClockTie { + forceHLCClockTieForTest(t, topology) + } + deleteConflictDocs(collectionName, docID, topology) + + topology.StartReplications() + waitForConvergingTombstones(t, collectionName, docID, topology) + }) + } }) } } @@ -110,31 +164,46 @@ func TestMultiActorConflictDelete(t *testing.T) { // 10. start replications // 11. assert that the documents are resurrected on all peers and have matching hlvs for non Couchbase Lite peers and // matching CV for Couchbase Lite peers. +// +// The forceClockTie dimension forces every controllable peer's HLC clock to the same fixed timestamp +// immediately before step 9, so the concurrent resurrections in step 9 tie on HLV Value instead of relying on +// real clock jitter to hit that case. func TestMultiActorConflictResurrect(t *testing.T) { base.LongRunningTest(t) for _, topologySpec := range append(simpleTopologySpecifications, TopologySpecifications...) { t.Run(topologySpec.description, func(t *testing.T) { - collectionName, topology := setupTests(t, topologySpec) + for _, forceClockTie := range []bool{false, true} { + t.Run(clockTieSubtestName(forceClockTie), func(t *testing.T) { + if forceClockTie && controllableClockPeerCount(topologySpec) < 2 { + t.Skipf("only %d controllable (Sync Gateway/Couchbase Lite) peer(s) in this topology, need at least 2 to force an HLC clock tie", controllableClockPeerCount(topologySpec)) + } + + collectionName, topology := setupTests(t, topologySpec) - docID := getDocID(t) - docVersion := createConflictingDocs(collectionName, docID, topology) + docID := getDocID(t) + docVersion := createConflictingDocs(collectionName, docID, topology) - topology.StartReplications() - waitForCVAndBody(t, collectionName, docID, docVersion, topology) + topology.StartReplications() + waitForCVAndBody(t, collectionName, docID, docVersion, topology) - topology.StopReplications() - deleteConflictDocs(collectionName, docID, topology) + topology.StopReplications() + deleteConflictDocs(collectionName, docID, topology) - topology.StartReplications() + topology.StartReplications() - waitForConvergingTombstones(t, collectionName, docID, topology) - topology.StopReplications() + waitForConvergingTombstones(t, collectionName, docID, topology) + topology.StopReplications() - resurrectVersion := updateConflictingDocs(collectionName, docID, topology) - topology.StartReplications() + if forceClockTie { + forceHLCClockTieForTest(t, topology) + } + resurrectVersion := updateConflictingDocs(collectionName, docID, topology) + topology.StartReplications() - waitForCVAndBody(t, collectionName, docID, resurrectVersion, topology) + waitForCVAndBody(t, collectionName, docID, resurrectVersion, topology) + }) + } }) } } diff --git a/xdcr/xdcr_test.go b/xdcr/xdcr_test.go index c487e7d431..54c2114144 100644 --- a/xdcr/xdcr_test.go +++ b/xdcr/xdcr_test.go @@ -538,6 +538,60 @@ func TestLWWAfterInitialReplication(t *testing.T) { requireCV(t, xattrs[base.VvXattrName], fromBucketSourceID, fromCAS) } +// TestXDCRMatchingTimestamps verifies XDCR's local wins conflict resolution when the source and target's +// CAS match but the sourceIDs are different. +func TestXDCRMatchingTimestamps(t *testing.T) { + fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t) + ctx := base.TestCtx(t) + + docID := "doc" + // Fixed value shared by both sides to force a tie. CAS is roughly nanoseconds since the Unix epoch, so this + // must stay comfortably in the past (this value is ~2020) - real Couchbase Server rejects an HLV whose + // cvCAS is newer than the document's actual CAS. + const matchingCAS = 0x1600000000000000 + const fromSourceID = "fromSource" + const toSourceID = "toSource" + + writeTiedDoc := func(ds sgbucket.DataStore, sourceID string, body string) { + hlv := &db.HybridLogicalVector{ + SourceID: db.EncodeSource(sourceID), + Version: matchingCAS, + CurrentVersionCAS: matchingCAS, + } + xattrs := map[string][]byte{ + base.VvXattrName: base.MustJSONMarshal(t, hlv), + base.MouXattrName: base.MustJSONMarshal(t, &db.MetadataOnlyUpdate{HexCAS: "expand"}), + } + opts := &sgbucket.MutateInOptions{ + MacroExpansion: []sgbucket.MacroExpansionSpec{ + sgbucket.NewMacroExpansionSpec(db.XattrMouCasPath(), sgbucket.MacroCas), + }, + } + _, err := ds.WriteWithXattrs(ctx, docID, 0, 0, []byte(body), xattrs, nil, opts) + require.NoError(t, err) + } + + writeTiedDoc(fromDs, fromSourceID, `{"source":"fromDs"}`) + writeTiedDoc(toDs, toSourceID, `{"source":"toDs"}`) + + xdcr := startXDCR(t, fromBucket, toBucket, XDCROptions{Mobile: MobileOn}) + defer func() { + assert.NoError(t, xdcr.Stop(ctx)) + }() + requireWaitForXDCRDocsProcessed(t, xdcr, 1) + + stats, err := xdcr.Stats(ctx) + require.NoError(t, err) + assert.Equal(t, uint64(0), stats.DocsWritten) + assert.Equal(t, uint64(1), stats.TargetNewerDocs) + + // The target (toDs) keeps its own document: the tied source write does not win. + body, xattrs, _, err := toDs.GetWithXattrs(ctx, docID, []string{base.VvXattrName}) + require.NoError(t, err) + require.JSONEq(t, `{"source":"toDs"}`, string(body)) + requireCV(t, xattrs[base.VvXattrName], db.EncodeSource(toSourceID), matchingCAS) +} + func TestReplicateXattrs(t *testing.T) { base.LongRunningTest(t)