Allow config.ReplicationLowPriorityTaskParallelism to take effect. (#10051)

## What changed?

Use `WorkflowKey` for the low and high priority replication scheduler's queue ID
rather than a string. Deterministically map RunID to a slot number for low priority using
`config.ReplicationLowPriorityTaskParallelism` slots.

## Why?

Using strings for the queue ID was preventing concurrent tx map from
optimising queue operations.

While we had code to allow parallelising tasks for different executions
of the same workflow ID, the code was not reachable. Switch the
parallelisation to the queue ID level which is subjectively more
intuitive.

## Potential risks
The `config.ReplicationLowPriorityTaskParallelism` could be set to a
non-default number on clusters but would not currently be taking effect.
After this change the parallelism would kick in, which may cause
surprising behaviour.

Author: robholland

service/history/replication/fx.go

@@ -2,7 +2,6 @@ package replication
 
 import (
 	"context"
-	"math/rand"
 	"strconv"
 
 	"github.com/dgryski/go-farm"
@@ -181,22 +180,26 @@ func replicationStreamLowPrioritySchedulerProvider(
 	metricsHandler metrics.Handler,
 	lc fx.Lifecycle,
 ) ctasks.Scheduler[TrackableExecutableTask] {
+	// P-way parallelism for executions of the same workflow (per ReplicationLowPriorityTaskParallelism)
+	// is modeled as P distinct per-namespace-workflow queue IDs. We bucket by execution (RunID) so all
+	// low-priority tasks for one execution share a queue; the third field stores the slot index, not
+	// the run UUID.
 	queueFactory := func(task TrackableExecutableTask) ctasks.SequentialTaskQueue[TrackableExecutableTask] {
 		item := task.QueueID()
 		workflowKey, ok := item.(definition.WorkflowKey)
 		if !ok {
 			return NewSequentialTaskQueueWithID(item)
 		}
-		return NewSequentialTaskQueueWithID(workflowKey.NamespaceID + "_" + workflowKey.WorkflowID)
-	}
-	taskQueueHashFunc := func(item any) uint32 {
-		workflowKey, ok := item.(definition.WorkflowKey)
-		if !ok {
-			return 0
+		parallelism := config.ReplicationLowPriorityTaskParallelism()
+		if parallelism < 1 {
+			parallelism = 1
 		}
-
-		idBytes := []byte(workflowKey.NamespaceID + "_" + workflowKey.WorkflowID + "_" + strconv.Itoa(rand.Intn(config.ReplicationLowPriorityTaskParallelism())))
-		return farm.Fingerprint32(idBytes)
+		// 0..parallelism-1, stable for a given RunID. Different runs of the same workflow can share
+		// a slot, so up to P sequential queues (and workers) can progress them concurrently.
+		slot := int(farm.Fingerprint32([]byte(workflowKey.RunID)) % uint32(parallelism))
+		return NewSequentialTaskQueueWithID(
+			definition.NewWorkflowKey(workflowKey.NamespaceID, workflowKey.WorkflowID, strconv.Itoa(slot)),
+		)
 	}
 	// SequentialScheduler has panic wrapper when executing task,
 	// if changing the executor, please make sure other executor has panic wrapper
@@ -205,7 +208,7 @@ func replicationStreamLowPrioritySchedulerProvider(
 			QueueSize:   config.ReplicationProcessorSchedulerQueueSize(),
 			WorkerCount: config.ReplicationLowPriorityProcessorSchedulerWorkerCount,
 		},
-		taskQueueHashFunc,
+		WorkflowKeyHashFn,
 		queueFactory,
 		logger,
 	)
@@ -303,7 +306,9 @@ func sequentialTaskQueueFactoryProvider(
 		if !ok {
 			return NewSequentialTaskQueueWithID(item)
 		}
-		return NewSequentialTaskQueueWithID(workflowKey.NamespaceID + "_" + workflowKey.WorkflowID)
+		return NewSequentialTaskQueueWithID(
+			definition.NewWorkflowKey(workflowKey.NamespaceID, workflowKey.WorkflowID, ""),
+		)
 	}
 }