remove fastForwardNotifier from context provider and simiplify the hash

function

Author: feiyang3cat

common/testing/mocksdk/client_mock.go

@@ -221,7 +221,7 @@ func TestInvoke(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		// not running, but NewExecutionRunId set (retry/cron/CaN) => not "closed"; polls and times out.
 		checker := mockConsistencyChecker{lease: mockLease{ms: mutableState(ctrl, fastForwardTSI(testFastForwardID, false), false, "next-run")}}
-		ffNotifier := notification.NewTimeSkippingFastForwardNotifier(func(namespace.ID, string) int32 { return 1 })
+		ffNotifier := notification.NewTimeSkippingFastForwardNotifier()
 		resp, err := Invoke(context.Background(), pollReq(uuid.NewString(), testWorkflowID, testFastForwardID),
 			shardContext(ctrl, 20*time.Millisecond), checker, ffNotifier)
 		require.NoError(t, err)
@@ -232,7 +232,7 @@ func TestInvoke(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		checker := mockConsistencyChecker{lease: mockLease{ms: mutableState(ctrl, fastForwardTSI(testFastForwardID, false), true, "")}}
 		// Real notifier: never notified, so the wait blocks until the soft timeout.
-		ffNotifier := notification.NewTimeSkippingFastForwardNotifier(func(namespace.ID, string) int32 { return 1 })
+		ffNotifier := notification.NewTimeSkippingFastForwardNotifier()
 		resp, err := Invoke(context.Background(), pollReq(uuid.NewString(), testWorkflowID, testFastForwardID),
 			shardContext(ctrl, 20*time.Millisecond), checker, ffNotifier)
 		require.NoError(t, err)

service/history/api/polltimeskipping/api_test.go

@@ -47,7 +47,6 @@ import (
 	"go.temporal.io/server/service/history/consts"
 	"go.temporal.io/server/service/history/events"
 	"go.temporal.io/server/service/history/hsm"
-	"go.temporal.io/server/service/history/notification"
 	"go.temporal.io/server/service/history/replication"
 	"go.temporal.io/server/service/history/shard"
 	"go.temporal.io/server/service/history/workflow"
@@ -93,7 +92,6 @@ var Module = fx.Options(
 	service.PersistenceLazyLoadedServiceResolverModule,
 	fx.Provide(ServiceResolverProvider),
 	fx.Provide(EventNotifierProvider),
-	fx.Provide(TimeSkippingFastForwardNotifierProvider),
 	fx.Provide(HistoryEngineFactoryProvider),
 	fx.Provide(HandlerProvider),
 	fx.Provide(HistoryServiceServerProvider),
@@ -484,12 +482,6 @@ func EventNotifierProvider(
 	)
 }
 
-func TimeSkippingFastForwardNotifierProvider(
-	serviceConfig *configs.Config,
-) notification.TimeSkippingFastForwardNotifier {
-	return notification.NewTimeSkippingFastForwardNotifier(serviceConfig.GetShardID)
-}
-
 func ServiceLifetimeHooks(lc fx.Lifecycle, svc *Service) {
 	lc.Append(fx.StartStopHook(svc.Start, svc.Stop))
 }

service/history/fx.go

@@ -161,7 +161,6 @@ func NewEngineWithShardContext(
 	matchingClient matchingservice.MatchingServiceClient,
 	sdkClientFactory sdk.ClientFactory,
 	eventNotifier events.Notifier,
-	fastForwardNotifier notification.TimeSkippingFastForwardNotifier,
 	config *configs.Config,
 	versionCache worker_versioning.VersionMembershipAndReactivationStatusCache,
 	workerDeploymentClient workerdeployment.Client,
@@ -218,7 +217,7 @@ func NewEngineWithShardContext(
 		throttledLogger:            log.With(shard.GetThrottledLogger(), tag.ComponentHistoryEngine),
 		metricsHandler:             shard.GetMetricsHandler(),
 		eventNotifier:              eventNotifier,
-		fastForwardNotifier:        fastForwardNotifier,
+		fastForwardNotifier:        notification.NewTimeSkippingFastForwardNotifier(),
 		config:                     config,
 		sdkClientFactory:           sdkClientFactory,
 		matchingClient:             matchingClient,
@@ -895,7 +894,7 @@ func (e *historyEngineImpl) NotifyFastForwardUpdate(
 	fastforwardNotification *notification.TimeSkippingFastForwardNotification,
 ) {
 	if e.fastForwardNotifier == nil {
-		// Always injected in production via fx; a nil here means a misconfigured engine.
+		// Always set by NewEngineWithShardContext; a nil here means a hand-built engine.
 		// Fast-forward notification is best-effort (waiters re-poll on timeout), so log and
 		// skip rather than panic.
 		e.logger.Warn("fastForwardNotifier is not configured; skipping fast-forward notification",

service/history/history_engine.go

@@ -17,7 +17,6 @@ import (
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/events"
 	historyi "go.temporal.io/server/service/history/interfaces"
-	"go.temporal.io/server/service/history/notification"
 	"go.temporal.io/server/service/history/replication"
 	"go.temporal.io/server/service/history/tasks"
 	"go.temporal.io/server/service/history/workflow"
@@ -34,7 +33,6 @@ type (
 		MatchingClient                  resource.MatchingClient
 		SdkClientFactory                sdk.ClientFactory
 		EventNotifier                   events.Notifier
-		TimeSkippingFastForwardNotifier notification.TimeSkippingFastForwardNotifier
 		Config                          *configs.Config
 		RawMatchingClient               resource.MatchingRawClient
 		WorkflowCache                   wcache.Cache
@@ -73,7 +71,6 @@ func (f *historyEngineFactory) CreateEngine(
 		f.MatchingClient,
 		f.SdkClientFactory,
 		f.EventNotifier,
-		f.TimeSkippingFastForwardNotifier,
 		f.Config,
 		f.VersionMembershipCache,
 		f.WorkerDeploymentClient,

service/history/history_engine_factory.go

@@ -56,9 +56,8 @@ func (noopNotifier[K, T]) Unwatch(K, string) error { return nil }
 // concurrent waiters per key; Watch beyond that returns a ResourceExhausted error.
 //
 // hashKey only stripes the internal map's locks for concurrency — any uniform hash of the
-// key works and it has no history-shard meaning here (callers just pass the workflow->shard
-// hash as a convenient, well-distributed one). It need not cover every field of K: keys are
-// matched by equality, so a hash over a subset only affects lock distribution.
+// key works and it carries no history-shard meaning here. It need not cover every field of K:
+// keys are matched by equality, so a hash over a subset only affects lock distribution.
 func NewPubSubNotifier[K comparable, T any](hashKey func(K) uint32, maxSubscribersPerKey int) PubSubNotifier[K, T] {
 	if hashKey == nil {
 		// A caller bug, but not a fatal one, and better caught here than as a nil-func panic on

service/history/notification/notifier.go

@@ -1,9 +1,11 @@
 package notification
 
 import (
+	"strconv"
+
+	"github.com/dgryski/go-farm"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
-	"go.temporal.io/server/common/namespace"
 )
 
 // TimeSkippingFastForwardNotification is the payload delivered to subscribers waiting for
@@ -78,13 +80,17 @@ func NewTimeSkippingNotificationKey(
 
 const maxFastForwardWaitersPerExecution = 5
 
-func NewTimeSkippingFastForwardNotifier(hashKey func(namespace.ID, string) int32) TimeSkippingFastForwardNotifier {
+func NewTimeSkippingFastForwardNotifier() TimeSkippingFastForwardNotifier {
 	return NewPubSubNotifier[TimeSkippingNotificationKey, *TimeSkippingFastForwardNotification](
-		func(key TimeSkippingNotificationKey) uint32 {
-			return uint32(hashKey(namespace.ID(key.NamespaceID), key.WorkflowID))
-		},
+		hashTimeSkippingNotificationKey,
 		maxFastForwardWaitersPerExecution,
 	)
 }
 
+func hashTimeSkippingNotificationKey(key TimeSkippingNotificationKey) uint32 {
+	return farm.Fingerprint32([]byte(
+		key.NamespaceID + "_" + key.WorkflowID + "_" + strconv.FormatUint(uint64(key.ArchetypeID), 10),
+	))
+}
+
 var NoopTimeSkippingFastForwardNotifier = NewNoopNotifier[TimeSkippingNotificationKey, *TimeSkippingFastForwardNotification]()

service/history/notification/timeskipping_notifier.go

@@ -5,16 +5,11 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"go.temporal.io/server/chasm"
-	"go.temporal.io/server/common/namespace"
 )
 
 func TestTimeSkippingFastForwardNotifier(t *testing.T) {
-	// The notifier's hash covers only (namespace, workflowID), so keys differing solely by
-	// archetype land on the same lock shard and stay distinct by struct equality alone.
-	oneShardByWorkflow := func(namespace.ID, string) int32 { return 1 }
-
 	t.Run("archetype isolates executions sharing a business id", func(t *testing.T) {
-		n := NewTimeSkippingFastForwardNotifier(oneShardByWorkflow)
+		n := NewTimeSkippingFastForwardNotifier()
 		workflowKey := NewTimeSkippingNotificationKey("ns", "wf", chasm.WorkflowArchetypeID)
 		schedulerKey := NewTimeSkippingNotificationKey("ns", "wf", chasm.SchedulerArchetypeID)
 
@@ -38,7 +33,7 @@ func TestTimeSkippingFastForwardNotifier(t *testing.T) {
 	// The waiter always names the workflow archetype while the publisher forwards whatever the
 	// execution carries, so a pre-archetype record (unspecified) must reach the workflow waiter.
 	t.Run("unspecified archetype is normalized to workflow", func(t *testing.T) {
-		n := NewTimeSkippingFastForwardNotifier(oneShardByWorkflow)
+		n := NewTimeSkippingFastForwardNotifier()
 		require.Equal(t,
 			NewTimeSkippingNotificationKey("ns", "wf", chasm.WorkflowArchetypeID),
 			NewTimeSkippingNotificationKey("ns", "wf", chasm.UnspecifiedArchetypeID))
@@ -71,7 +66,7 @@ func TestTimeSkippingFastForwardNotifier(t *testing.T) {
 	})
 
 	t.Run("keys match by value, not identity", func(t *testing.T) {
-		n := NewTimeSkippingFastForwardNotifier(oneShardByWorkflow)
+		n := NewTimeSkippingFastForwardNotifier()
 		_, ch, err := n.Watch(NewTimeSkippingNotificationKey("ns", "wf", chasm.WorkflowArchetypeID))
 		require.NoError(t, err)