Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions common/dynamicconfig/constants.go
Original file line number Diff line number Diff line change
Expand Up @@ -1552,6 +1552,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,
Expand Down
2 changes: 2 additions & 0 deletions service/matching/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,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
Expand Down Expand Up @@ -289,6 +290,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),
Expand Down
22 changes: 21 additions & 1 deletion service/matching/matching_engine.go
Original file line number Diff line number Diff line change
Expand Up @@ -2511,9 +2511,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.
Expand Down
1 change: 1 addition & 0 deletions service/matching/task_queue_user_data_merge.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading
Loading