From 04d686e233a1a2cd9ce5dfbb50c4f5d368af8a90 Mon Sep 17 00:00:00 2001 From: Kannan Date: Thu, 5 Nov 2026 19:39:41 -0800 Subject: [PATCH 1/2] Fix deleted deployment version accumulation on passive clusters (#12966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problem: Deleted deployment versions accumulate on passive clusters. They can block new builds after failover and eventually make replication updates too large to store or transmit. 1. **Normal deletion — the case this PR addresses** **Setup:** Active and passive start with `{A, B}`, clock 10. Active deletes A and sends `{B}`, clock 20. **Before:** The passive unions both snapshots and retains `{A, B}`. **After:** The passive replaces its deployment data with `{B}`. An equal incoming clock also replaces, allowing force replication to clear existing accumulation. 2. **Delayed replication — fixed by [#12351](https://github.com/temporalio/temporal/pull/12351), which introduced the normal-deletion issue this PR addresses** **Setup:** After failover, the target has `{B}`, clock 20. A delayed snapshot from the old active contains `{A}`, clock 10. **Before:** The union fix merged to `{A, B}`, preserving valid A despite its older parent clock. **After:** Still merges to `{A, B}`. This change retains the original fix's protection for delayed replication. 3. **Accepted edge case** **Setup:** The target has valid `{B}`, clock 20, but a delayed incoming snapshot contains only `{A}`, clock 30, for example because of clock skew. **Before:** Union preserves `{A, B}`. **After:** Replacement produces `{A}`, losing valid B. This remains an accepted limitation outside normal single-active clock ordering; this fix covers routine deletions and the older-incoming race in example 2. **Rollback protection:** `matching.enableTaskQueueUserDataRevisionMerge` defaults to `true`. If this fix causes issues, turn it off to restore the parent-clock snapshot selection used before [#12351](https://github.com/temporalio/temporal/pull/12351). This also gives up the delayed-replication protection in example 2, so we would need to revisit a different solution. Missing clocks retain compatibility merging while enabled. ### Testing Unit tests; replication tests also run with Go's race detector. (cherry picked from commit e4160f8e41999b55534650d91df4b1417bcff636) --- common/dynamicconfig/constants.go | 6 + service/matching/config.go | 2 + service/matching/matching_engine.go | 22 +- .../matching/task_queue_user_data_merge.go | 1 + .../user_data_replication_merge_test.go | 501 +++++++++++++++--- 5 files changed, 453 insertions(+), 79 deletions(-) diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 82c739f2f45..72a4f2eed43 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -1581,6 +1581,12 @@ these log lines can be noisy, we want to be able to turn on and sample selective 100, `MatchingMaxVersionsInDeployment represents the maximum number of versions that can be registered in a single worker deployment`, ) + MatchingEnableTaskQueueUserDataRevisionMerge = NewNamespaceBoolSetting( + "matching.enableTaskQueueUserDataRevisionMerge", + true, + `When enabled, merges incoming and current worker deployment data by revision when needed to preserve delayed updates. +When disabled, always uses the deployment snapshot selected by the parent clock without merging.`, + ) MatchingMaxVersionsInTaskQueue = NewNamespaceIntSetting( "matching.maxVersionsInTaskQueue", 200, diff --git a/service/matching/config.go b/service/matching/config.go index 81ed1d7c673..c4d65b2194f 100644 --- a/service/matching/config.go +++ b/service/matching/config.go @@ -34,6 +34,7 @@ type ( EnableDeployments dynamicconfig.BoolPropertyFnWithNamespaceFilter // [cleanup-wv-pre-release] EnableDeploymentVersions dynamicconfig.BoolPropertyFnWithNamespaceFilter UseRevisionNumberForWorkerVersioning dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableTaskQueueUserDataRevisionMerge dynamicconfig.BoolPropertyFnWithNamespaceFilter MaxTaskQueuesInDeployment dynamicconfig.IntPropertyFnWithNamespaceFilter MaxVersionsInTaskQueue dynamicconfig.IntPropertyFnWithNamespaceFilter MaxIDLengthLimit dynamicconfig.IntPropertyFn @@ -276,6 +277,7 @@ func NewConfig( EnableDeployments: dynamicconfig.EnableDeployments.Get(dc), // [cleanup-wv-pre-release] EnableDeploymentVersions: dynamicconfig.EnableDeploymentVersions.Get(dc), UseRevisionNumberForWorkerVersioning: dynamicconfig.UseRevisionNumberForWorkerVersioning.Get(dc), + EnableTaskQueueUserDataRevisionMerge: dynamicconfig.MatchingEnableTaskQueueUserDataRevisionMerge.Get(dc), MaxTaskQueuesInDeployment: dynamicconfig.MatchingMaxTaskQueuesInDeployment.Get(dc), MaxVersionsInTaskQueue: dynamicconfig.MatchingMaxVersionsInTaskQueue.Get(dc), RPS: dynamicconfig.MatchingRPS.Get(dc), diff --git a/service/matching/matching_engine.go b/service/matching/matching_engine.go index d7ec0a0bd12..c6fd081163e 100644 --- a/service/matching/matching_engine.go +++ b/service/matching/matching_engine.go @@ -2573,9 +2573,29 @@ func (e *matchingEngineImpl) ApplyTaskQueueUserDataReplicationEvent( // Merge logic: Task queue user data has an overall clock. Deployment routing configs and // individual versions within it have their own revision numbers. To merge the two user data values: + // // 1. Select the base user data using the overall clock. - // 2. Merge the routing configs and versions from the other user data into the base. + // + // 2. When revision-based merging is disabled, use the selected base without merging. + // + // 3. When revision-based merging is enabled, there are three cases: + // + // 3a. Both clocks are present and incoming is equal or newer: use incoming without merging. + // This lets missing versions propagate as deletions, including during same-clock force replication. + // + // 3b. Both clocks are present and incoming is older: merge its routing configs and versions into the base. + // This preserves valid versions or higher revisions arriving in delayed replication after failover. + // + // 3c. Either clock is missing: retain compatibility merging because snapshot ordering is unknown. baseUserData, otherUserData, selectionReason := selectTaskQueueUserData(current, incoming) + revisionMergeEnabled := e.config.EnableTaskQueueUserDataRevisionMerge(ns.Name().String()) + currentClock, incomingClock := current.GetClock(), incoming.GetClock() + hasMissingClock := currentClock == nil || incomingClock == nil + incomingClockIsOlder := !hasMissingClock && hlc.Greater(currentClock, incomingClock) + shouldMergeDeploymentData := revisionMergeEnabled && (incomingClockIsOlder || hasMissingClock) + if !shouldMergeDeploymentData { + otherUserData = nil + } mergedUserData, mergeConflicts := mergeTaskQueueUserDataDeployments(baseUserData, otherUserData) // Merge V1 sets. diff --git a/service/matching/task_queue_user_data_merge.go b/service/matching/task_queue_user_data_merge.go index 39ce78a7e12..d487728fc80 100644 --- a/service/matching/task_queue_user_data_merge.go +++ b/service/matching/task_queue_user_data_merge.go @@ -70,6 +70,7 @@ func (reason taskQueueUserDataSelectionReason) selectedUserDataSource() string { // worker versions found only in otherUserData are added. When both contain the // same routing configuration or worker version, the value with the higher revision is retained. // Equal revisions retain the value from baseUserData. +// A nil otherUserData skips merging; a nil baseUserData starts with empty user data. func mergeTaskQueueUserDataDeployments( baseUserData *persistencespb.TaskQueueUserData, otherUserData *persistencespb.TaskQueueUserData, diff --git a/service/matching/user_data_replication_merge_test.go b/service/matching/user_data_replication_merge_test.go index 52fff2d79b1..5fcc1c5536a 100644 --- a/service/matching/user_data_replication_merge_test.go +++ b/service/matching/user_data_replication_merge_test.go @@ -1,7 +1,11 @@ package matching import ( + "fmt" + "testing" + "github.com/google/uuid" + "github.com/stretchr/testify/require" deploymentpb "go.temporal.io/api/deployment/v1" enumspb "go.temporal.io/api/enums/v1" taskqueuepb "go.temporal.io/api/taskqueue/v1" @@ -10,9 +14,14 @@ import ( enumsspb "go.temporal.io/server/api/enums/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/common" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/persistence/visibility/manager" "go.temporal.io/server/common/testing/protorequire" "go.temporal.io/server/common/testing/testlogger" + "go.temporal.io/server/common/testing/testvars" + "go.uber.org/mock/gomock" ) func taskQueueUserDataWithDeployment( @@ -36,7 +45,7 @@ func taskQueueUserDataWithDeployment( } // Replicating identical user data leaves persisted data unchanged and emits no conflict diagnostics. -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventIdenticalUserData() { +func (s *matchingEngineSuite) TestReplicateIdenticalData() { captureHandler := s.captureDroppedOnEngine() current := taskQueueUserDataWithDeployment( &clockspb.HybridLogicalClock{WallClock: 20, Version: 1, ClusterId: 1}, @@ -82,7 +91,9 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventIdentica } } -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventUnionsDeploymentDataByRevision() { +// After failover, delayed replication can contribute valid versions and higher revisions despite its older parent clock. +func (s *matchingEngineSuite) TestMergeDelayedSnapshot() { + // The target has accepted newer writes after failover; the source event was delayed. currentClock := &clockspb.HybridLogicalClock{WallClock: 200, ClusterId: 1} incomingClock := &clockspb.HybridLogicalClock{WallClock: 100, ClusterId: 1} currentRouting := &deploymentpb.RoutingConfig{ @@ -125,6 +136,7 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventUnionsDe CurrentDeploymentVersion: &deploymentpb.WorkerDeploymentVersion{DeploymentName: "zero-routing", BuildId: "zero"}, RevisionNumber: 0, } + // The delayed snapshot has newer routing but older data for A, plus versions missing from the target. incoming := taskQueueUserDataWithDeployment( incomingClock, enumspb.TASK_QUEUE_TYPE_WORKFLOW, @@ -147,6 +159,7 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventUnionsDe taskQueue := uuid.NewString() s.seedTaskQueueUserData(taskQueue, current) + // Receive the delayed event and resolve routing and version revisions independently. got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) protorequire.ProtoEqual(s.T(), currentClock, got.GetClock()) @@ -163,23 +176,23 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventUnionsDe protorequire.ProtoEqual(s.T(), incomingZeroRouting, gotDeployments["zero-routing"].GetRoutingConfig()) } -// Reordered and repeated user data updates retain the highest revisions, all build IDs, and the newest parent clock and config. -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventReorderedAndRepeatedUserData() { - older := taskQueueUserDataWithDeployment( - &clockspb.HybridLogicalClock{WallClock: 10, ClusterId: 1}, +// Redelivering an older snapshot must preserve higher version and routing revisions already merged locally. +// The target's newer parent clock and task queue config must also survive each delivery. +func (s *matchingEngineSuite) TestRedeliveryPreservesHigherRevisions() { + target := taskQueueUserDataWithDeployment( + &clockspb.HybridLogicalClock{WallClock: 30, ClusterId: 1}, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "deployment", &persistencespb.WorkerDeploymentData{ - RoutingConfig: newTestRoutingConfig(1, "A"), Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ - "A": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING), + "B": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE), }, }, ) - older.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ - FairnessWeightOverrides: map[string]float32{"older": 1}, + target.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ + FairnessWeightOverrides: map[string]float32{"target": 1}, } - newer := taskQueueUserDataWithDeployment( + higherRevision := taskQueueUserDataWithDeployment( &clockspb.HybridLogicalClock{WallClock: 20, ClusterId: 1}, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "deployment", @@ -190,72 +203,153 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventReordere }, }, ) - newer.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ - FairnessWeightOverrides: map[string]float32{"newer": 1}, + higherRevision.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ + FairnessWeightOverrides: map[string]float32{"source": 1}, } - disjoint := taskQueueUserDataWithDeployment( - &clockspb.HybridLogicalClock{WallClock: 30, ClusterId: 1}, + lowerRevision := common.CloneProto(higherRevision) + lowerRevision.Clock.WallClock = 10 + lowerDeployment := lowerRevision.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"] + lowerDeployment.RoutingConfig = newTestRoutingConfig(1, "A") + lowerDeployment.Versions["A"] = newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING) + + expected := common.CloneProto(target) + expectedDeployment := expected.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"] + expectedDeployment.RoutingConfig = newTestRoutingConfig(2, "A") + expectedDeployment.Versions["A"] = newTestWorkerDeploymentVersionData(2, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT) + taskQueue := testvars.New(s.T()).TaskQueue().Name + s.seedTaskQueueUserData(taskQueue, target) + + // First receive A at revision 2. Its older parent clock still allows it to contribute missing data. + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, higherRevision) + protorequire.ProtoEqual(s.T(), expected, got) + + // Then receive A at revision 1. Keep revision 2, B, and the target's snapshot fields. + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, lowerRevision) + protorequire.ProtoEqual(s.T(), expected, got) + + // Redeliver the same revision-1 snapshot. Receiving it again must leave that result unchanged. + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, lowerRevision) + protorequire.ProtoEqual(s.T(), expected, got) +} + +// Older replication snapshots can restore a version removed by a newer snapshot because deletion is encoded as absence. +// This accepted limitation applies to redelivery of either the original version or its later revision. +func (s *matchingEngineSuite) TestRedeliveryCanRestoreDeletedVersions() { + firstSnapshot := taskQueueUserDataWithDeployment( + &clockspb.HybridLogicalClock{WallClock: 10, ClusterId: 1}, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "deployment", &persistencespb.WorkerDeploymentData{ + RoutingConfig: newTestRoutingConfig(1, "A"), Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ - "B": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE), + "A": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING), }, }, ) - disjoint.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ - FairnessWeightOverrides: map[string]float32{"disjoint": 1}, + firstSnapshot.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ + FairnessWeightOverrides: map[string]float32{"first": 1}, } - want := taskQueueUserDataWithDeployment( + updatedSnapshot := common.CloneProto(firstSnapshot) + updatedSnapshot.Clock.WallClock = 20 + updatedDeployment := updatedSnapshot.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"] + updatedDeployment.RoutingConfig = newTestRoutingConfig(2, "A") + updatedDeployment.Versions["A"] = newTestWorkerDeploymentVersionData(2, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT) + deletionSnapshot := taskQueueUserDataWithDeployment( &clockspb.HybridLogicalClock{WallClock: 30, ClusterId: 1}, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "deployment", &persistencespb.WorkerDeploymentData{ - RoutingConfig: newTestRoutingConfig(2, "A"), Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ - "A": newTestWorkerDeploymentVersionData(2, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT), "B": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE), }, }, ) - want.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ - FairnessWeightOverrides: map[string]float32{"disjoint": 1}, + deletionSnapshot.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ + FairnessWeightOverrides: map[string]float32{"latest": 1}, } - tests := []struct { - name string - userDataUpdates []*persistencespb.TaskQueueUserData - }{ - { - name: "ordered updates then stale user data", - userDataUpdates: []*persistencespb.TaskQueueUserData{older, newer, disjoint, older}, - }, - { - name: "newest parent first then stale user data", - userDataUpdates: []*persistencespb.TaskQueueUserData{disjoint, newer, older, older}, - }, - { - name: "newer version before stale then repeated subset", - userDataUpdates: []*persistencespb.TaskQueueUserData{newer, older, disjoint, newer}, - }, + taskQueue := testvars.New(s.T()).TaskQueue().Name + s.seedTaskQueueUserData(taskQueue, firstSnapshot) + + // Receive A's revision-2 update, then the newest snapshot that removes A and contains only B. + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, updatedSnapshot) + protorequire.ProtoEqual(s.T(), updatedSnapshot, got) + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, deletionSnapshot) + protorequire.ProtoEqual(s.T(), deletionSnapshot, got) + + // Redeliver the original snapshot. A returns at revision 1 because its revision-2 entry was removed. + restoredOriginal := common.CloneProto(deletionSnapshot) + restoredOriginalDeployment := restoredOriginal.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"] + restoredOriginalDeployment.RoutingConfig = newTestRoutingConfig(1, "A") + restoredOriginalDeployment.Versions["A"] = newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING) + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, firstSnapshot) + protorequire.ProtoEqual(s.T(), restoredOriginal, got) + + // Redeliver the authoritative deletion snapshot at the same clock to remove A again. + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, deletionSnapshot) + protorequire.ProtoEqual(s.T(), deletionSnapshot, got) + + // Redelivering A's revision-2 update can also restore it; the clock and config still come from the latest snapshot. + restoredUpdate := common.CloneProto(restoredOriginal) + restoredUpdateDeployment := restoredUpdate.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"] + restoredUpdateDeployment.RoutingConfig = newTestRoutingConfig(2, "A") + restoredUpdateDeployment.Versions["A"] = newTestWorkerDeploymentVersionData(2, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT) + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, updatedSnapshot) + protorequire.ProtoEqual(s.T(), restoredUpdate, got) +} + +// Each deployment forgets one old version and registers one new version; the passive must retain the same bounded set. +func (s *matchingEngineSuite) TestRolloutsDoNotAccumulateVersions() { + const retainedVersions = 12 + const rollouts = 50 + initialCurrentBuildID := fmt.Sprintf("build-%03d", retainedVersions-1) + s.matchingEngine.config.EnableTaskQueueUserDataRevisionMerge = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true) + active := &persistencespb.TaskQueueUserData{ + Clock: &clockspb.HybridLogicalClock{WallClock: 100, ClusterId: 1}, + PerType: make(map[int32]*persistencespb.TaskQueueTypeUserData), } - for _, test := range tests { - s.Run(test.name, func() { - taskQueue := uuid.NewString() - s.seedTaskQueueUserData(taskQueue, test.userDataUpdates[0]) - var got *persistencespb.TaskQueueUserData - for _, incoming := range test.userDataUpdates[1 : len(test.userDataUpdates)-1] { - got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) - } - protorequire.ProtoEqual(s.T(), want, got) + // Both task types begin with the same 12 versions: one current version and 11 drained versions. + for _, taskType := range []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW, enumspb.TASK_QUEUE_TYPE_ACTIVITY} { + versions := make(map[string]*deploymentspb.WorkerDeploymentVersionData) + for i := range retainedVersions { + versions[fmt.Sprintf("build-%03d", i)] = newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED) + } + versions[initialCurrentBuildID].Status = enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT + active.PerType[int32(taskType)] = taskQueueUserDataWithDeployment(nil, taskType, "deployment", &persistencespb.WorkerDeploymentData{ + RoutingConfig: newTestRoutingConfig(1, initialCurrentBuildID), + Versions: versions, + }).PerType[int32(taskType)] + } + taskQueue := testvars.New(s.T()).TaskQueue().Name + s.seedTaskQueueUserData(taskQueue, common.CloneProto(active)) - got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, test.userDataUpdates[len(test.userDataUpdates)-1]) - protorequire.ProtoEqual(s.T(), want, got) - }) + // Every rollout forgets one version and adds one, so the active's retained set never grows. + for nextBuild := retainedVersions; nextBuild < retainedVersions+rollouts; nextBuild++ { + active = common.CloneProto(active) + active.Clock.WallClock++ + for _, perType := range active.PerType { + deployment := perType.DeploymentData.DeploymentsData["deployment"] + // Drain the previous current version and forget the oldest retained version. + previous := deployment.Versions[fmt.Sprintf("build-%03d", nextBuild-1)] + previous.Status = enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED + previous.RevisionNumber++ + delete(deployment.Versions, fmt.Sprintf("build-%03d", nextBuild-retainedVersions)) + // Register the next build and make it the routing target. + buildID := fmt.Sprintf("build-%03d", nextBuild) + deployment.Versions[buildID] = newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT) + deployment.RoutingConfig = newTestRoutingConfig(int64(nextBuild+1), buildID) + } + + // Check the whole passive snapshot after each delivery, including its 12 versions and routing. + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, active) + + protorequire.ProtoEqual(s.T(), active, got) } } +// An older snapshot can add deployment data for a missing task type without importing its snapshot-level settings. +// //nolint:staticcheck // SA1019 verifies that deprecated deployment fields are selected by the parent clock. -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventImportsOnlyDeploymentDataForMissingType() { +func (s *matchingEngineSuite) TestImportDeploymentDataForMissingType() { current := taskQueueUserDataWithDeployment( &clockspb.HybridLogicalClock{WallClock: 20, ClusterId: 1}, enumspb.TASK_QUEUE_TYPE_WORKFLOW, @@ -299,41 +393,38 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventImportsO s.NotContains(got.GetPerType(), int32(enumspb.TASK_QUEUE_TYPE_NEXUS)) } -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventRetainsEntryMissingFromNewerUserData() { - currentVersion := &deploymentspb.WorkerDeploymentVersionData{ - RevisionNumber: 5, - Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, - } - current := taskQueueUserDataWithDeployment( - &clockspb.HybridLogicalClock{WallClock: 100, ClusterId: 1}, +// Normal deletion must remove a forgotten version from the passive rather than retaining it in a union. +func (s *matchingEngineSuite) TestReplicateDeletion() { + passive := taskQueueUserDataWithDeployment( + &clockspb.HybridLogicalClock{WallClock: 10, ClusterId: 1}, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "deployment", &persistencespb.WorkerDeploymentData{ - Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{"A": currentVersion}, + RoutingConfig: newTestRoutingConfig(5, "B"), + Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ + "A": newTestWorkerDeploymentVersionData(5, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED), + "B": newTestWorkerDeploymentVersionData(5, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT), + }, }, ) - incomingClock := &clockspb.HybridLogicalClock{WallClock: 200, ClusterId: 1} - incomingConfig := &taskqueuepb.TaskQueueConfig{FairnessWeightOverrides: map[string]float32{"newer": 1}} - incoming := &persistencespb.TaskQueueUserData{ - Clock: incomingClock, - PerType: map[int32]*persistencespb.TaskQueueTypeUserData{ - int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW): {Config: incomingConfig}, - }, - } + // Both replicas start with {A, B}; the active then forgets A and advances its clock. + active := common.CloneProto(passive) + active.Clock = &clockspb.HybridLogicalClock{WallClock: 20, ClusterId: 1} + delete(active.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"].Versions, "A") taskQueue := uuid.NewString() - s.seedTaskQueueUserData(taskQueue, current) - got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) + s.seedTaskQueueUserData(taskQueue, passive) + // The newer active snapshot contains only B; the passive must remove its old A entry. + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, active) - protorequire.ProtoEqual(s.T(), incomingClock, got.GetClock()) - protorequire.ProtoEqual(s.T(), incomingConfig, got.GetPerType()[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].GetConfig()) - gotDeployment := got.GetPerType()[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].GetDeploymentData().GetDeploymentsData()["deployment"] - s.Require().NotNil(gotDeployment) - protorequire.ProtoEqual(s.T(), currentVersion, gotDeployment.GetVersions()["A"]) + protorequire.ProtoEqual(s.T(), active, got) } +// Snapshot-level config, fairness, legacy deployment data, and V2 rules follow the parent-clock winner. +// Revision-based deployment merging must not change how these other fields are selected. +// //nolint:staticcheck // SA1019 verifies that deprecated deployment fields are selected by the parent clock. -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventUsesParentClockForNonDeploymentFields() { +func (s *matchingEngineSuite) TestSelectSnapshotFieldsByParentClock() { newUserData := func( clock *clockspb.HybridLogicalClock, label string, @@ -427,7 +518,9 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventUsesPare } } -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventResolvesEqualRevisionConflictByParentClock() { +// Equal-revision differences emit diagnostics only when the snapshots are merged. +// Replacing the whole snapshot with incoming data must not report a merge conflict. +func (s *matchingEngineSuite) TestReportEqualRevisionConflicts() { const conflictLogMessage = "found conflicting user data with same revision number during replication" captureHandler := s.captureDroppedOnEngine() tests := []struct { @@ -437,6 +530,7 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventResolves wantStatus enumspb.WorkerDeploymentVersionStatus wantSide string wantResolution string + wantNoConflict bool }{ { name: "newer current clock", @@ -453,6 +547,7 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventResolves wantStatus: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING, wantSide: "incoming", wantResolution: "incoming_parent_clock", + wantNoConflict: true, }, { name: "only current is clocked", @@ -481,6 +576,7 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventResolves wantStatus: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING, wantSide: "incoming", wantResolution: "incoming_clock_fallback", + wantNoConflict: true, }, } @@ -518,6 +614,10 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventResolves gotVersion := got.GetPerType()[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].GetDeploymentData().GetDeploymentsData()["deployment"].GetVersions()["build"] s.Equal(test.wantStatus, gotVersion.GetStatus()) recordings := metricCapture.SnapshotMetric(metrics.TaskQueueUserDataReplicationEqualRevisionConflicts.Name()) + if test.wantNoConflict { + s.Require().Empty(recordings) + return + } s.Require().Len(recordings, 1) s.Equal(int64(1), recordings[0].Value) s.Len(recordings[0].Tags, 4) @@ -635,7 +735,8 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventResolves }) } -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventClonesMergedDeploymentData() { +// Replication must copy imported deployment data so later mutation of the incoming message cannot alter cached data. +func (s *matchingEngineSuite) TestCloneMergedDeploymentData() { incomingRouting := &deploymentpb.RoutingConfig{ CurrentDeploymentVersion: &deploymentpb.WorkerDeploymentVersion{DeploymentName: "shared", BuildId: "incoming"}, RevisionNumber: 2, @@ -693,3 +794,247 @@ func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventClonesMe protorequire.ProtoEqual(s.T(), wantVersion, gotShared.GetVersions()["A"]) protorequire.ProtoEqual(s.T(), wantIncomingOnly, gotDeployments["incoming-only"]) } + +// Revision merging defaults on; a namespace-specific override must not disable it for other namespaces. +func TestRevisionMergeConfig(t *testing.T) { + t.Parallel() + + client := dynamicconfig.NewMemoryClient() + config := NewConfig(dynamicconfig.NewCollection(client, log.NewNoopLogger())) + require.True(t, config.EnableTaskQueueUserDataRevisionMerge("disabled")) + t.Cleanup(client.OverrideSetting(dynamicconfig.MatchingEnableTaskQueueUserDataRevisionMerge, []dynamicconfig.ConstrainedValue{ + {Constraints: dynamicconfig.Constraints{Namespace: "disabled"}, Value: false}, + })) + require.False(t, config.EnableTaskQueueUserDataRevisionMerge("disabled")) + require.True(t, config.EnableTaskQueueUserDataRevisionMerge("other")) +} + +// Clockless compatibility, HLC tie-breakers, and disabling revision merging follow their explicit policies. +// Each row varies the flag or clocks while sharing deployment data with conflicting revisions and disjoint versions. +func (s *matchingEngineSuite) TestReplicationMergePolicy() { + current := taskQueueUserDataWithDeployment(nil, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "deployment", &persistencespb.WorkerDeploymentData{ + RoutingConfig: newTestRoutingConfig(5, "shared"), + Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ + "shared": newTestWorkerDeploymentVersionData(8, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT), + "current-only": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED), + }, + }) + current.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ + FairnessWeightOverrides: map[string]float32{"current": 1}, + } + incoming := taskQueueUserDataWithDeployment(nil, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "deployment", &persistencespb.WorkerDeploymentData{ + RoutingConfig: newTestRoutingConfig(6, "shared"), + Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ + "shared": newTestWorkerDeploymentVersionData(7, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING), + "incoming-only": newTestWorkerDeploymentVersionData(0, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE), + }, + }) + incoming.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = &taskqueuepb.TaskQueueConfig{ + FairnessWeightOverrides: map[string]float32{"incoming": 1}, + } + // A merge keeps the higher version revision from current and the higher routing revision from incoming. + // Snapshot fields such as config still come from whichever parent clock wins. + mergedCurrent := common.CloneProto(current) + mergedCurrentDeployment := mergedCurrent.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"] + mergedCurrentDeployment.RoutingConfig = newTestRoutingConfig(6, "shared") + mergedCurrentDeployment.Versions["incoming-only"] = newTestWorkerDeploymentVersionData(0, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE) + mergedIncoming := common.CloneProto(mergedCurrent) + mergedIncoming.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config = common.CloneProto(incoming.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].Config) + + older := &clockspb.HybridLogicalClock{WallClock: 10, ClusterId: 1} + newer := &clockspb.HybridLogicalClock{WallClock: 20, ClusterId: 1} + laterCounter := &clockspb.HybridLogicalClock{WallClock: 10, Version: 1, ClusterId: 1} + laterCluster := &clockspb.HybridLogicalClock{WallClock: 10, ClusterId: 2} + tests := []struct { + name string + enabled bool + currentClock *clockspb.HybridLogicalClock + incomingClock *clockspb.HybridLogicalClock + want *persistencespb.TaskQueueUserData + wantClock *clockspb.HybridLogicalClock + }{ + {"merge_disabled_newer_incoming_replaces", false, older, newer, incoming, newer}, + {"merge_disabled_equal_clocks_replace", false, older, older, incoming, older}, + {"merge_disabled_older_incoming_is_discarded", false, newer, older, current, newer}, + {"merge_disabled_missing_current_clock", false, nil, newer, incoming, newer}, + {"merge_disabled_missing_incoming_clock", false, newer, nil, current, newer}, + {"merge_disabled_both_clocks_missing", false, nil, nil, incoming, nil}, + {"equal_clocks_replace", true, older, older, incoming, older}, + {"missing_current_clock_merges", true, nil, newer, mergedIncoming, newer}, + {"missing_incoming_clock_merges", true, newer, nil, mergedCurrent, newer}, + {"both_clocks_missing_merge", true, nil, nil, mergedIncoming, nil}, + {"newer_incoming_counter_replaces", true, older, laterCounter, incoming, laterCounter}, + {"older_incoming_counter_merges", true, laterCounter, older, mergedCurrent, laterCounter}, + {"newer_incoming_cluster_replaces", true, older, laterCluster, incoming, laterCluster}, + } + for _, test := range tests { + s.Run(test.name, func() { + s.matchingEngine.config.EnableTaskQueueUserDataRevisionMerge = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(test.enabled) + currentData := common.CloneProto(current) + currentData.Clock = common.CloneProto(test.currentClock) + incomingData := common.CloneProto(incoming) + incomingData.Clock = common.CloneProto(test.incomingClock) + want := common.CloneProto(test.want) + want.Clock = common.CloneProto(test.wantClock) + currentBefore := common.CloneProto(currentData) + incomingBefore := common.CloneProto(incomingData) + taskQueue := testvars.New(s.T()).TaskQueue().Name + s.seedTaskQueueUserData(taskQueue, currentData) + + // Apply this row's policy, then verify the complete result and that neither input was changed. + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, incomingData) + + protorequire.ProtoEqual(s.T(), want, got) + protorequire.ProtoEqual(s.T(), currentBefore, currentData) + protorequire.ProtoEqual(s.T(), incomingBefore, incomingData) + // Mutating the caller's message afterward must not change the cached result either. + incomingData.PerType[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)].DeploymentData.DeploymentsData["deployment"].Versions["incoming-only"].RevisionNumber++ + protorequire.ProtoEqual(s.T(), want, got) + }) + } +} + +// A fresh authoritative snapshot removes accumulated versions, whole deployments, and missing task queue types. +func (s *matchingEngineSuite) TestForceReplicationClearsOrphans() { + s.matchingEngine.config.EnableTaskQueueUserDataRevisionMerge = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true) + current := &persistencespb.TaskQueueUserData{ + Clock: &clockspb.HybridLogicalClock{WallClock: 100, ClusterId: 1}, + PerType: make(map[int32]*persistencespb.TaskQueueTypeUserData), + } + // Both task types have extra drained versions, plus a whole forgotten deployment. + for _, taskType := range []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW, enumspb.TASK_QUEUE_TYPE_ACTIVITY} { + versions := map[string]*deploymentspb.WorkerDeploymentVersionData{ + "current": newTestWorkerDeploymentVersionData(10, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT), + } + for i := range 3 { + versions[fmt.Sprintf("drained-%03d", i)] = newTestWorkerDeploymentVersionData(5, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED) + } + current.PerType[int32(taskType)] = taskQueueUserDataWithDeployment(nil, taskType, "deployment", &persistencespb.WorkerDeploymentData{ + RoutingConfig: newTestRoutingConfig(20, "current"), + Versions: versions, + }).PerType[int32(taskType)] + current.PerType[int32(taskType)].DeploymentData.DeploymentsData["forgotten-deployment"] = &persistencespb.WorkerDeploymentData{ + Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ + "forgotten": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED), + }, + } + } + current.PerType[int32(enumspb.TASK_QUEUE_TYPE_NEXUS)] = &persistencespb.TaskQueueTypeUserData{} + // The active retains one drained version and the current version, but has forgotten the other entries. + incoming := common.CloneProto(current) + incoming.Clock = &clockspb.HybridLogicalClock{WallClock: 200, ClusterId: 1} + delete(incoming.PerType, int32(enumspb.TASK_QUEUE_TYPE_NEXUS)) + for _, perType := range incoming.PerType { + delete(perType.DeploymentData.DeploymentsData, "forgotten-deployment") + versions := perType.DeploymentData.DeploymentsData["deployment"].Versions + for i := 1; i < 3; i++ { + delete(versions, fmt.Sprintf("drained-%03d", i)) + } + } + taskQueue := testvars.New(s.T()).TaskQueue().Name + s.seedTaskQueueUserData(taskQueue, current) + + // A fresh snapshot clears the accumulated entries. Redelivering it must keep that result unchanged. + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) + protorequire.ProtoEqual(s.T(), incoming, got) + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) + protorequire.ProtoEqual(s.T(), incoming, got) + + // A delayed snapshot can restore absent entries without advancing the parent clock. + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, current) + wantRestored := common.CloneProto(current) + wantRestored.Clock = common.CloneProto(incoming.Clock) + delete(wantRestored.PerType, int32(enumspb.TASK_QUEUE_TYPE_NEXUS)) + protorequire.ProtoEqual(s.T(), wantRestored, got) + + // Force replication of the same-clock authoritative snapshot clears those entries again. + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) + protorequire.ProtoEqual(s.T(), incoming, got) + + // Subsequent newer snapshots keep propagating version removals. + incoming.Clock = &clockspb.HybridLogicalClock{WallClock: 300, ClusterId: 1} + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) + protorequire.ProtoEqual(s.T(), incoming, got) + + // Removing the remaining per-type data must also propagate an empty snapshot. + empty := &persistencespb.TaskQueueUserData{Clock: &clockspb.HybridLogicalClock{WallClock: 400, ClusterId: 1}} + got = s.applyTaskQueueUserDataReplicationEvent(taskQueue, empty) + protorequire.ProtoEqual(s.T(), empty, got) +} + +// Accepted limitation: a newer incoming parent clock wins even if its snapshot omits a valid target version. +func (s *matchingEngineSuite) TestNewerSnapshotDropsTargetOnlyVersions() { + target := taskQueueUserDataWithDeployment( + &clockspb.HybridLogicalClock{WallClock: 20, ClusterId: 1}, + enumspb.TASK_QUEUE_TYPE_WORKFLOW, + "deployment", + &persistencespb.WorkerDeploymentData{ + RoutingConfig: newTestRoutingConfig(5, "B"), + Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ + "B": newTestWorkerDeploymentVersionData(5, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT), + }, + }, + ) + // A delayed source snapshot can have a higher clock because of clock skew while still missing B. + incoming := taskQueueUserDataWithDeployment( + &clockspb.HybridLogicalClock{WallClock: 30, ClusterId: 1}, + enumspb.TASK_QUEUE_TYPE_WORKFLOW, + "deployment", + &persistencespb.WorkerDeploymentData{ + RoutingConfig: newTestRoutingConfig(4, "A"), + Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ + "A": newTestWorkerDeploymentVersionData(4, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT), + }, + }, + ) + taskQueue := testvars.New(s.T()).TaskQueue().Name + s.seedTaskQueueUserData(taskQueue, target) + + // Document the accepted loss: the newer parent clock replaces B even though B has a higher revision. + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) + + protorequire.ProtoEqual(s.T(), incoming, got) +} + +// Replacing deployment membership must not bypass legacy build ID merging or parent-clock selection of V2 rules. +func (s *matchingEngineSuite) TestPreserveLegacyVersioning() { + s.matchingEngine.config.EnableTaskQueueUserDataRevisionMerge = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true) + currentClock := &clockspb.HybridLogicalClock{WallClock: 10, ClusterId: 1} + incomingClock := &clockspb.HybridLogicalClock{WallClock: 20, ClusterId: 1} + current := taskQueueUserDataWithDeployment(currentClock, enumspb.TASK_QUEUE_TYPE_WORKFLOW, "forgotten", &persistencespb.WorkerDeploymentData{ + Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{ + "forgotten": newTestWorkerDeploymentVersionData(1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED), + }, + }) + current.VersioningData = &persistencespb.VersioningData{ + VersionSets: []*persistencespb.CompatibleVersionSet{mkNewSet("old-v1-build", currentClock)}, + AssignmentRules: []*persistencespb.AssignmentRule{ + mkAssignmentRulePersistence(mkAssignmentRuleWithoutRamp("old-v2-target"), currentClock, nil), + }, + } + incoming := &persistencespb.TaskQueueUserData{ + Clock: incomingClock, + VersioningData: &persistencespb.VersioningData{ + VersionSets: []*persistencespb.CompatibleVersionSet{mkNewSet("new-v1-build", incomingClock)}, + AssignmentRules: []*persistencespb.AssignmentRule{ + mkAssignmentRulePersistence(mkAssignmentRuleWithoutRamp("new-v2-target"), incomingClock, nil), + }, + RedirectRules: []*persistencespb.RedirectRule{ + mkRedirectRulePersistence(mkRedirectRule("source", "new-v2-target"), incomingClock, nil), + }, + }, + } + // Deployment membership is replaced, but old V1 builds survive and V2 rules follow the incoming parent. + want := common.CloneProto(incoming) + want.VersioningData.VersionSets = []*persistencespb.CompatibleVersionSet{ + mkNewSet("old-v1-build", currentClock), + mkNewSet("new-v1-build", incomingClock), + } + s.mockVisibilityManager.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{}, nil) + taskQueue := testvars.New(s.T()).TaskQueue().Name + s.seedTaskQueueUserData(taskQueue, current) + + got := s.applyTaskQueueUserDataReplicationEvent(taskQueue, incoming) + + protorequire.ProtoEqual(s.T(), want, got) +} From f356e7f9e564ad39ac8ac5bfed0d2252bfe03771 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 5 Nov 2026 20:11:16 -0800 Subject: [PATCH 2/2] Update release-only replication test for conditional merging --- service/matching/user_data_source_replay_test.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/service/matching/user_data_source_replay_test.go b/service/matching/user_data_source_replay_test.go index 724c97090ca..1c7606d9098 100644 --- a/service/matching/user_data_source_replay_test.go +++ b/service/matching/user_data_source_replay_test.go @@ -7,23 +7,24 @@ import ( deploymentspb "go.temporal.io/server/api/deployment/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/common" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/testing/protorequire" "go.temporal.io/server/common/worker_versioning" "google.golang.org/protobuf/types/known/timestamppb" ) -func (s *matchingEngineSuite) TestApplyTaskQueueUserDataReplicationEventSourceReplayPreservesDestination() { +// Redelivering an older or clockless source snapshot preserves higher destination revisions. +// Equal and newer clock replacement is covered by TestReplicationMergePolicy and TestForceReplicationClearsOrphans. +func (s *matchingEngineSuite) TestRedeliveryPreservesDestination() { + s.matchingEngine.config.EnableTaskQueueUserDataRevisionMerge = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true) tests := []struct { name string currentClock *clockspb.HybridLogicalClock incomingClock *clockspb.HybridLogicalClock }{ {name: "destination_clock_newer", currentClock: &clockspb.HybridLogicalClock{WallClock: 20}, incomingClock: &clockspb.HybridLogicalClock{WallClock: 10}}, - {name: "source_clock_newer", currentClock: &clockspb.HybridLogicalClock{WallClock: 10}, incomingClock: &clockspb.HybridLogicalClock{WallClock: 20}}, {name: "source_clock_missing", currentClock: &clockspb.HybridLogicalClock{WallClock: 20}}, - {name: "destination_clock_missing", incomingClock: &clockspb.HybridLogicalClock{WallClock: 20}}, {name: "both_clocks_missing"}, - {name: "equal_clocks", currentClock: &clockspb.HybridLogicalClock{WallClock: 20}, incomingClock: &clockspb.HybridLogicalClock{WallClock: 20}}, } for _, test := range tests { s.Run(test.name, func() {