CBG-5547 handle topologytest conflicts for matching timestamps (#8424)

Author: torcolvin

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)
+}

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

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=

rest/replicatortest/replicator_conflict_test.go

@@ -16,6 +16,7 @@ import (
 	"testing"
 	"time"
 
+	sgbucket "github.com/couchbase/sg-bucket"
 	"github.com/couchbase/sync_gateway/base"
 	"github.com/couchbase/sync_gateway/db"
 	"github.com/couchbase/sync_gateway/rest"
@@ -2212,6 +2213,78 @@ func TestActiveReplicatorV4DefaultResolverWithTombstoneRemote(t *testing.T) {
 	})
 }
 
+// TestDefaultConflictResolverMatchingTimestamps verifies that default (HLV-aware) conflict resolution for different
+// source IDs but matching versions is local wins. See CBG-5570 for changing this.
+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)
+	})
+}
+
 // getRevTreeID create a revtree ID for a new revision that is a child of the parentRevID for a given body.
 func getRevTreeID(t *testing.T, parentRevID string, body []byte) string {
 	prevGeneration, _ := db.ParseRevID(t.Context(), parentRevID)

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 {

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
+}