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)