Activate Drained/Inactive version if a workflow is "moved" to it

common/dynamicconfig/constants.go

@@ -2794,6 +2794,27 @@ instead of the previous HSM backed implementation.`,
 		`Maximum number of entries in the version membership cache.`,
 	)
 
+	VersionReactivationSignalCacheTTL = NewGlobalDurationSetting(
+		"history.versionReactivationSignalCacheTTL",
+		10*time.Second,
+		`TTL for caching drainage reactivation signals to version workflows. These signals are sent from the history service to update the version workflow's 
+		draining status to DRAINING from DRAINED/INACTIVE states.`,
+	)
+
+	VersionReactivationSignalCacheMaxSize = NewGlobalIntSetting(
+		"history.versionReactivationSignalCacheMaxSize",
+		10000,
+		`Maximum number of entries in the version reactivation signal cache.`,
+	)
+
+	EnableVersionReactivationSignals = NewGlobalBoolSetting(
+		"history.enableVersionReactivationSignals",
+		true,
+		`EnableVersionReactivationSignals controls whether reactivation signals are sent to version workflows
+		when workflows are pinned to a potentially DRAINED/INACTIVE version. Set to false to disable signals
+		globally if load becomes problematic.`,
+	)
+
 	RoutingInfoCacheTTL = NewGlobalDurationSetting(
 		"history.routingInfoCacheTTL",
 		1*time.Second,

common/metrics/metric_defs.go

@@ -44,6 +44,7 @@ const (
 	MutableStateCacheTypeTagValue                     = "mutablestate"
 	EventsCacheTypeTagValue                           = "events"
 	VersionMembershipCacheTypeTagValue                = "version_membership"
+	VersionReactivationSignalCacheTypeTagValue        = "version_reactivation_signal"
 	RoutingInfoCacheTypeTagValue                      = "routing_info"
 	NexusEndpointRegistryReadThroughCacheTypeTagValue = "nexus_endpoint_registry_readthrough"
 
@@ -458,6 +459,8 @@ const (
 	VersionMembershipCacheGetScope = "VersionMembershipCacheGet"
 	// VersionMembershipCachePutScope is the scope used by version membership cache
 	VersionMembershipCachePutScope = "VersionMembershipCachePut"
+	// VersionReactivationSignalCacheShouldSendScope is the scope used by version reactivation signal cache
+	VersionReactivationSignalCacheShouldSendScope = "VersionReactivationSignalCacheShouldSend"
 	// RoutingInfoCacheGetScope is the scope used by routing info cache
 	RoutingInfoCacheGetScope = "RoutingInfoCacheGet"
 	// RoutingInfoCachePutScope is the scope used by routing info cache

common/worker_versioning/version_reactivation_signal_cache.go

@@ -0,0 +1,65 @@
+//nolint:staticcheck
+package worker_versioning
+
+import (
+	"go.temporal.io/server/common/cache"
+	"go.temporal.io/server/common/metrics"
+)
+
+// ReactivationSignalCache deduplicates reactivation signals to version workflows.
+//
+// Implementations are expected to be safe for concurrent use.
+type (
+	ReactivationSignalCache interface {
+		// ShouldSendSignal returns true if signal should be sent (not recently sent)
+		// and atomically marks it as sent. Returns false if recently sent.
+		ShouldSendSignal(namespaceID, deploymentName, buildID string) bool
+	}
+
+	reactivationSignalCacheKey struct {
+		namespaceID    string
+		deploymentName string
+		buildID        string
+	}
+
+	ReactivationSignalCacheImpl struct {
+		cache.Cache
+		metricsHandler metrics.Handler
+	}
+)
+
+// NewReactivationSignalCache wraps the provided cache with a typed API and metrics.
+func NewReactivationSignalCache(c cache.Cache, metricsHandler metrics.Handler) ReactivationSignalCache {
+	h := metricsHandler.WithTags(metrics.CacheTypeTag(metrics.VersionReactivationSignalCacheTypeTagValue))
+	return &ReactivationSignalCacheImpl{
+		Cache:          c,
+		metricsHandler: h,
+	}
+}
+
+func (c *ReactivationSignalCacheImpl) ShouldSendSignal(
+	namespaceID, deploymentName, buildID string,
+) bool {
+	handler := c.metricsHandler.WithTags(
+		metrics.OperationTag(metrics.VersionReactivationSignalCacheShouldSendScope),
+		metrics.NamespaceIDTag(namespaceID),
+	)
+	metrics.CacheRequests.With(handler).Record(1)
+
+	key := reactivationSignalCacheKey{
+		namespaceID:    namespaceID,
+		deploymentName: deploymentName,
+		buildID:        buildID,
+	}
+
+	// Check if we recently sent a signal for this version
+	if c.Cache.Get(key) != nil {
+		// Entry exists, signal was recently sent - deduplicate
+		return false
+	}
+
+	// No recent signal, mark as sent and return true
+	metrics.CacheMissCounter.With(handler).Record(1)
+	c.Cache.Put(key, true)
+	return true
+}

service/history/api/multioperation/api.go

@@ -61,6 +61,8 @@ func Invoke(
 	tokenSerializer *tasktoken.Serializer,
 	matchingClient matchingservice.MatchingServiceClient,
 	versionMembershipCache worker_versioning.VersionMembershipCache,
+	reactivationSignalCache worker_versioning.ReactivationSignalCache,
+	reactivationSignaler api.VersionReactivationSignalerFn,
 	testHooks testhooks.TestHooks,
 ) (*historyservice.ExecuteMultiOperationResponse, error) {
 	namespaceEntry, err := api.GetActiveNamespace(shardContext, namespace.ID(req.GetNamespaceId()), req.WorkflowId)
@@ -101,6 +103,8 @@ func Invoke(
 			startReq,
 			matchingClient,
 			versionMembershipCache,
+			reactivationSignalCache,
+			reactivationSignaler,
 			uws.workflowLeaseCallback(ctx),
 		)
 		if err != nil {

service/history/api/resetworkflow/api.go

@@ -29,6 +29,8 @@ func Invoke(
 	workflowConsistencyChecker api.WorkflowConsistencyChecker,
 	matchingClient matchingservice.MatchingServiceClient,
 	versionMembershipCache worker_versioning.VersionMembershipCache,
+	reactivationSignalCache worker_versioning.ReactivationSignalCache,
+	reactivationSignaler api.VersionReactivationSignalerFn,
 ) (_ *historyservice.ResetWorkflowExecutionResponse, retError error) {
 	namespaceID := namespace.ID(resetRequest.GetNamespaceId())
 	err := api.ValidateNamespaceUUID(namespaceID)
@@ -177,6 +179,15 @@ func Invoke(
 	); err != nil {
 		return nil, err
 	}
+
+	// Notify version workflow if we're pinning to a potentially drained version via post-reset operations
+	for _, operation := range request.GetPostResetOperations() {
+		if updateOpts, ok := operation.GetVariant().(*workflowpb.PostResetOperation_UpdateWorkflowOptions_); ok {
+			api.ReactivateVersionWorkflowIfPinned(ctx, namespaceEntry,
+				updateOpts.UpdateWorkflowOptions.GetWorkflowExecutionOptions().GetVersioningOverride(), reactivationSignalCache, reactivationSignaler, shardContext.GetConfig().EnableVersionReactivationSignals())
+		}
+	}
+
 	return &historyservice.ResetWorkflowExecutionResponse{
 		RunId: resetRunID,
 	}, nil

service/history/api/signalwithstartworkflow/api.go

@@ -24,6 +24,8 @@ func Invoke(
 	workflowConsistencyChecker api.WorkflowConsistencyChecker,
 	matchingClient matchingservice.MatchingServiceClient,
 	versionMembershipCache worker_versioning.VersionMembershipCache,
+	reactivationSignalCache worker_versioning.ReactivationSignalCache,
+	reactivationSignaler api.VersionReactivationSignalerFn,
 ) (_ *historyservice.SignalWithStartWorkflowExecutionResponse, retError error) {
 	namespaceEntry, err := api.GetActiveNamespace(shard, namespace.ID(signalWithStartRequest.GetNamespaceId()), signalWithStartRequest.SignalWithStartRequest.WorkflowId)
 	if err != nil {
@@ -91,6 +93,12 @@ func Invoke(
 	if err != nil {
 		return nil, err
 	}
+
+	// Notify version workflow if we're starting a new workflow pinned to a potentially drained version
+	if started {
+		api.ReactivateVersionWorkflowIfPinned(ctx, namespaceEntry, request.GetVersioningOverride(), reactivationSignalCache, reactivationSignaler, shard.GetConfig().EnableVersionReactivationSignals())
+	}
+
 	return &historyservice.SignalWithStartWorkflowExecutionResponse{
 		RunId:   runID,
 		Started: started,

service/history/api/startworkflow/api.go

@@ -60,6 +60,8 @@ type Starter struct {
 	namespace                  *namespace.Namespace
 	createOrUpdateLeaseFn      api.CreateOrUpdateLeaseFunc
 	versionMembershipCache     worker_versioning.VersionMembershipCache
+	reactivationSignalCache    worker_versioning.ReactivationSignalCache
+	reactivationSignaler       api.VersionReactivationSignalerFn
 }
 
 // creationParams is a container for all information obtained from creating the uncommitted execution.
@@ -89,6 +91,8 @@ func NewStarter(
 	request *historyservice.StartWorkflowExecutionRequest,
 	matchingClient matchingservice.MatchingServiceClient,
 	versionMembershipCache worker_versioning.VersionMembershipCache,
+	reactivationSignalCache worker_versioning.ReactivationSignalCache,
+	reactivationSignaler api.VersionReactivationSignalerFn,
 	createLeaseFn api.CreateOrUpdateLeaseFunc,
 ) (*Starter, error) {
 	namespaceEntry, err := api.GetActiveNamespace(shardContext, namespace.ID(request.GetNamespaceId()), request.StartRequest.WorkflowId)
@@ -107,6 +111,8 @@ func NewStarter(
 		namespace:                  namespaceEntry,
 		createOrUpdateLeaseFn:      createLeaseFn,
 		versionMembershipCache:     versionMembershipCache,
+		reactivationSignalCache:    reactivationSignalCache,
+		reactivationSignaler:       reactivationSignaler,
 	}, nil
 }
 
@@ -208,12 +214,21 @@ func (s *Starter) Invoke(
 		var currentWorkflowConditionFailedError *persistence.CurrentWorkflowConditionFailedError
 		if errors.As(err, &currentWorkflowConditionFailedError) && len(currentWorkflowConditionFailedError.RunID) > 0 {
 			// The history and mutable state generated above will be deleted by a background process.
-			return s.handleConflict(ctx, creationParams, currentWorkflowConditionFailedError)
+			resp, outcome, conflictErr := s.handleConflict(ctx, creationParams, currentWorkflowConditionFailedError)
+			if conflictErr == nil && outcome == StartNew {
+				// Notify version workflow if we are starting a workflow execution on a potentially drained version.
+				// Only signal when a new workflow was actually created (StartNew), not for deduped retries
+				// (StartDeduped) or reused existing workflows (StartReused) where the pinned override is not applied.
+				api.ReactivateVersionWorkflowIfPinned(ctx, s.namespace, s.request.StartRequest.GetVersioningOverride(), s.reactivationSignalCache, s.reactivationSignaler, s.shardContext.GetConfig().EnableVersionReactivationSignals())
+			}
+			return resp, outcome, conflictErr
 		}
-
 		return nil, StartErr, err
 	}
 
+	// Notify version workflow if we're pinning to a potentially drained version
+	api.ReactivateVersionWorkflowIfPinned(ctx, s.namespace, s.request.StartRequest.GetVersioningOverride(), s.reactivationSignalCache, s.reactivationSignaler, s.shardContext.GetConfig().EnableVersionReactivationSignals())
+
 	resp, err = s.generateResponse(
 		creationParams.runID,
 		creationParams.workflowTaskInfo,

service/history/api/updateworkflowoptions/api.go

@@ -28,6 +28,8 @@ func Invoke(
 	workflowConsistencyChecker api.WorkflowConsistencyChecker,
 	matchingClient matchingservice.MatchingServiceClient,
 	versionMembershipCache worker_versioning.VersionMembershipCache,
+	reactivationSignalCache worker_versioning.ReactivationSignalCache,
+	reactivationSignaler api.VersionReactivationSignalerFn,
 ) (*historyservice.UpdateWorkflowExecutionOptionsResponse, error) {
 	req := request.GetUpdateRequest()
 	ns, err := api.GetActiveNamespace(shardCtx, namespace.ID(request.GetNamespaceId()), req.GetWorkflowExecution().GetWorkflowId())
@@ -36,6 +38,9 @@ func Invoke(
 	}
 	ret := &historyservice.UpdateWorkflowExecutionOptionsResponse{}
 
+	// Store versioning override to send reactivation signal after successful persistence
+	var versioningOverrideForReactivation *workflowpb.VersioningOverride
+
 	err = api.GetAndUpdateWorkflowWithNew(
 		ctx,
 		nil,
@@ -94,6 +99,9 @@ func Invoke(
 				}, nil
 			}
 
+			// Store versioning override to send reactivation signal after successful persistence
+			versioningOverrideForReactivation = mergedOpts.GetVersioningOverride()
+
 			// TODO (carly) part 2: handle safe deployment change --> CreateWorkflowTask=true
 			return &api.UpdateWorkflowAction{
 				Noop:               false,
@@ -107,6 +115,11 @@ func Invoke(
 	if err != nil {
 		return nil, err
 	}
+
+	// Notify version workflow if we're pinning to a potentially drained version.
+	// This is done after successful persistence to avoid signaling if the update fails.
+	api.ReactivateVersionWorkflowIfPinned(ctx, ns, versioningOverrideForReactivation, reactivationSignalCache, reactivationSignaler, shardCtx.GetConfig().EnableVersionReactivationSignals())
+
 	return ret, nil
 }
 

service/history/api/updateworkflowoptions/api_test.go

@@ -53,6 +53,17 @@ func (noopVersionMembershipCache) Put(
 ) {
 }
 
+type noopReactivationSignalCache struct{}
+
+func (noopReactivationSignalCache) ShouldSendSignal(_, _, _ string) bool {
+	return false // Always return false to skip sending signals in tests
+}
+
+// noopReactivationSignaler is a no-op signaler function for tests
+func noopReactivationSignaler(_ context.Context, _ *namespace.Namespace, _, _ string) error {
+	return nil
+}
+
 var (
 	emptyOptions            = &workflowpb.WorkflowExecutionOptions{}
 	unpinnedOverrideOptions = &workflowpb.WorkflowExecutionOptions{
@@ -184,6 +195,7 @@ func (s *updateWorkflowOptionsSuite) SetupTest() {
 	s.shardContext = historyi.NewMockShardContext(s.controller)
 	s.shardContext.EXPECT().GetNamespaceRegistry().Return(s.namespaceRegistry)
 	s.shardContext.EXPECT().GetClusterMetadata().Return(clustertest.NewMetadataForTest(cluster.NewTestClusterMetadataConfig(true, true)))
+	s.shardContext.EXPECT().GetConfig().Return(tests.NewDynamicConfig()).AnyTimes()
 
 	// mock a mutable state with an existing versioning override
 	s.currentMutableState = historyi.NewMockMutableState(s.controller)
@@ -263,7 +275,9 @@ func (s *updateWorkflowOptionsSuite) TestInvoke_Success() {
 		s.shardContext,
 		s.workflowConsistencyChecker,
 		s.mockMatchingClient,
-		noopVersionMembershipCache{}, // cache not meant to be used in this test
+		noopVersionMembershipCache{},  // cache not meant to be used in this test
+		noopReactivationSignalCache{}, // cache not meant to be used in this test
+		noopReactivationSignaler,      // signaler not meant to be used in this test
 	)
 	s.NoError(err)
 	s.NotNil(resp)

service/history/api/worker_versioning_util.go

@@ -0,0 +1,65 @@
+package api
+
+import (
+	"context"
+
+	workflowpb "go.temporal.io/api/workflow/v1"
+	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/worker_versioning"
+)
+
+// VersionReactivationSignalerFn is a function type for sending reactivation signals to version workflows.
+// This abstraction allows the history API layer to use the deployment client without importing it directly,
+// avoiding import cycles between history/api and worker/workerdeployment packages.
+type VersionReactivationSignalerFn func(
+	ctx context.Context,
+	namespaceEntry *namespace.Namespace,
+	deploymentName, buildID string,
+) error
+
+// ReactivateVersionWorkflowIfPinned sends a reactivation signal to the version workflow
+// when workflows are pinned to a potentially DRAINED/INACTIVE version. It also deduplicates
+// signals within the cache TTL window.
+// This is a fire-and-forget operation - the signal is sent asynchronously and errors are
+// logged by the signaler implementation.
+//
+//nolint:revive,errcheck
+func ReactivateVersionWorkflowIfPinned(
+	ctx context.Context,
+	namespaceEntry *namespace.Namespace,
+	override *workflowpb.VersioningOverride,
+	signalCache worker_versioning.ReactivationSignalCache,
+	signaler VersionReactivationSignalerFn,
+	enabled bool,
+) {
+	// Check if signals are enabled globally
+	if !enabled {
+		return
+	}
+
+	// Only process if we're pinning to a specific version
+	if !worker_versioning.OverrideIsPinned(override) {
+		return
+	}
+
+	pinnedVersion := worker_versioning.GetOverridePinnedVersion(override)
+	if pinnedVersion == nil {
+		return
+	}
+
+	// Check cache - skip if signal was recently sent
+	if signalCache != nil && !signalCache.ShouldSendSignal(
+		namespaceEntry.ID().String(),
+		pinnedVersion.GetDeploymentName(),
+		pinnedVersion.GetBuildId(),
+	) {
+		return
+	}
+
+	// Send the signal asynchronously to avoid adding latency to the caller's request.
+	// Errors are logged by the signaler implementation (e.g. via convertAndRecordError). However,
+	// errors are not propagated to the caller as this is a fire-and-forget operation.
+	go func() {
+		signaler(context.Background(), namespaceEntry, pinnedVersion.GetDeploymentName(), pinnedVersion.GetBuildId()) //nolint:errcheck
+	}()
+}