fix: trim orphaned workflow timer tasks on workflow close and deletion

Fixes #7568

Workflows with long timeouts that complete early leave orphaned timer
task rows in Cassandra that persist until the timer fires. Cron
workflows are the most severe case: because they reuse the same workflow
ID, all their orphaned timer rows accumulate in the same Cassandra
partition, causing large-partition problems. High-frequency regular
workflows also accumulate orphaned timers, though spread across
partitions.

This change tracks workflow-level timer tasks at creation time and
deletes them in two places:

1. At workflow close (best-effort, async goroutine) — prevents
   accumulation during the retention window
2. At retention-based deletion — safety net that catches any timers
   missed at close time

Implementation:
- New WorkflowTimerTaskInfo type persisted as a blob on the execution
  row (workflow_timer_tasks / workflow_timer_tasks_encoding columns,
  Cassandra schema v0.47)
- syncExecutionInfoWithTasks() captures timer task IDs from TasksByCategory
  immediately after task ID assignment during CreateWorkflowExecution
- deleteWorkflowTimerTasksBestEffortAsync() in execution/context.go
  fires a goroutine on workflow completion to clean up tracked timers
- deleteWorkflowTimerTasksBestEffort() in timer_task_executor_base.go
  performs the same cleanup at retention time with retry
- TaskCleanupTimeoutThreshold (default 24h) skips timers that will fire
  soon anyway, avoiding unnecessary DB writes
- Everything gated behind system.enableExecutionInfoTracking (default
  false); enabled in development.yaml for local testing
- SQL DeleteTimerTask is a stub (returns nil); Cassandra is fully
  implemented; the Cassandra large-partition problem does not affect SQL

Not in this PR (deferred):
- User-timer cleanup (requires adding timer_task_id to TimerInfo in IDL)
- Tracking timers created during workflow updates (only creation-time
  timers are tracked; documented in CloseTransactionAsMutation)

Author: fimanishi

common/dynamicconfig/dynamicproperties/constants.go

@@ -1718,6 +1718,15 @@ const (
 	// Default value: true
 	// Allowed filters: N/A
 	EnableGRPCOutbound
+	// EnableOrphanedTimerCleanup enables cleanup of orphaned workflow timer tasks when a workflow
+	// closes before its timers fire. When enabled, tracked timer tasks are deleted at workflow close
+	// time and again at retention-based deletion. This is a feature flag intended to be defaulted
+	// to true once validated.
+	// KeyName: system.enableOrphanedTimerCleanup
+	// Value type: Bool
+	// Default value: false
+	// Allowed filters: N/A
+	EnableOrphanedTimerCleanup
 	// EnableSQLAsyncTransaction is the key for enabling async transaction
 	// KeyName: system.enableSQLAsyncTransaction
 	// Value type: Bool
@@ -2906,6 +2915,14 @@ const (
 	// Default value: 5m (5*time.Minute)
 	// Allowed filters: N/A
 	StandbyClusterDelay
+	// TaskCleanupTimeoutThreshold Is the time, above which, it will attempt to cleanup tasks on workflow deletion
+	// but below which, it will skip cleanup attempts, based on the assumption that short-lived workflows will be mostly
+	// creating extremely short-lived timer tasks and there's no real value in explicitly going and deleting them
+	// KeyName: history.taskCleanupTimeoutThreshold
+	// Value type: Duration
+	// Default value: 1d (24 hours)
+	// Allowed filters: N/A
+	TaskCleanupTimeoutThreshold
 	// StandbyTaskMissingEventsResendDelay is the amount of time standby cluster's will wait (if events are missing)before calling remote for missing events
 	// KeyName: history.standbyTaskMissingEventsResendDelay
 	// Value type: Duration
@@ -4568,6 +4585,11 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Description:  "EnableGRPCOutbound is the key for enabling outbound GRPC traffic",
 		DefaultValue: true,
 	},
+	EnableOrphanedTimerCleanup: {
+		KeyName:      "system.enableOrphanedTimerCleanup",
+		Description:  "Enables cleanup of orphaned workflow timer tasks when a workflow closes before its timers fire. Feature flag, intended to be defaulted to true once validated.",
+		DefaultValue: false,
+	},
 	EnableSQLAsyncTransaction: {
 		KeyName:      "system.enableSQLAsyncTransaction",
 		Description:  "EnableSQLAsyncTransaction is the key for enabling async transaction",
@@ -5567,6 +5589,11 @@ var DurationKeys = map[DurationKey]DynamicDuration{
 		Description:  "StandbyClusterDelay is the artificial delay added to standby cluster's view of active cluster's time",
 		DefaultValue: time.Minute * 5,
 	},
+	TaskCleanupTimeoutThreshold: {
+		KeyName:      "history.taskCleanupTimeoutThreshold",
+		Description:  "TaskCleanupTimeoutThreshold is the time, above which, it will attempt to cleanup tasks on workflow deletion but below which, it will skip cleanup attempts, based on the assumption that short-lived workflows will be mostly creating extremely short-lived timer tasks and there's no real value in explicitly going and deleting them",
+		DefaultValue: time.Hour * 24,
+	},
 	StandbyTaskMissingEventsResendDelay: {
 		KeyName:      "history.standbyTaskMissingEventsResendDelay",
 		Description:  "StandbyTaskMissingEventsResendDelay is the amount of time standby cluster's will wait (if events are missing)before calling remote for missing events",

common/log/tag/values.go

@@ -233,6 +233,7 @@ var (
 	StoreOperationCreateFailoverMarkerTasks          = storeOperation("createFailoverMarkerTasks")
 	StoreOperationGetTimerIndexTasks                 = storeOperation("get-timer-index-tasks")
 	StoreOperationCompleteTimerTask                  = storeOperation("complete-timer-task")
+	StoreOperationDeleteTimerTask                    = storeOperation("delete-timer-task")
 	StoreOperationGetHistoryTasks                    = storeOperation("get-history-tasks")
 	StoreOperationCompleteHistoryTask                = storeOperation("complete-history-task")
 	StoreOperationRangeCompleteHistoryTask           = storeOperation("range-complete-history-task")

common/metrics/defs.go

@@ -213,6 +213,8 @@ const (
 	PersistenceCompleteHistoryTaskScope
 	// PersistenceRangeCompleteHistoryTaskScope tracks RangeCompleteHistoryTask calls made by service to persistence layer
 	PersistenceRangeCompleteHistoryTaskScope
+	// PersistenceDeleteTimerTaskScope tracks DeleteTimerTask calls made by service to persistence layer
+	PersistenceDeleteTimerTaskScope
 	// PersistenceCreateTasksScope tracks CreateTask calls made by service to persistence layer
 	PersistenceCreateTasksScope
 	// PersistenceGetTasksScope tracks GetTasks calls made by service to persistence layer
@@ -1587,6 +1589,7 @@ var ScopeDefs = map[ServiceIdx]map[ScopeIdx]scopeDefinition{
 		PersistenceGetHistoryTasksScope:                          {operation: "GetHistoryTasks"},
 		PersistenceCompleteHistoryTaskScope:                      {operation: "CompleteHistoryTask"},
 		PersistenceRangeCompleteHistoryTaskScope:                 {operation: "RangeCompleteHistoryTask"},
+		PersistenceDeleteTimerTaskScope:                          {operation: "DeleteTimerTask"},
 		PersistenceCreateTasksScope:                              {operation: "CreateTask"},
 		PersistenceGetTasksScope:                                 {operation: "GetTasks"},
 		PersistenceCompleteTaskScope:                             {operation: "CompleteTask"},

common/mocks/ExecutionManager.go

@@ -28,6 +28,7 @@ import (
 type (
 	// DynamicConfiguration represents dynamic configuration for persistence layer
 	DynamicConfiguration struct {
+		EnableOrphanedTimerCleanup              dynamicproperties.BoolPropertyFn
 		EnableSQLAsyncTransaction                dynamicproperties.BoolPropertyFn
 		EnableCassandraAllConsistencyLevelDelete dynamicproperties.BoolPropertyFn
 		EnableShardIDMetrics                     dynamicproperties.BoolPropertyFn
@@ -44,6 +45,7 @@ type (
 // NewDynamicConfiguration returns new config with default values
 func NewDynamicConfiguration(dc *dynamicconfig.Collection) *DynamicConfiguration {
 	return &DynamicConfiguration{
+		EnableOrphanedTimerCleanup:              dc.GetBoolProperty(dynamicproperties.EnableOrphanedTimerCleanup),
 		EnableSQLAsyncTransaction:                dc.GetBoolProperty(dynamicproperties.EnableSQLAsyncTransaction),
 		EnableCassandraAllConsistencyLevelDelete: dc.GetBoolProperty(dynamicproperties.EnableCassandraAllConsistencyLevelDelete),
 		EnableShardIDMetrics:                     dc.GetBoolProperty(dynamicproperties.EnableShardIDMetrics),

common/persistence/config.go

@@ -605,18 +605,19 @@ type (
 
 	// WorkflowMutableState indicates workflow related state
 	WorkflowMutableState struct {
-		ActivityInfos       map[int64]*ActivityInfo
-		TimerInfos          map[string]*TimerInfo
-		ChildExecutionInfos map[int64]*ChildExecutionInfo
-		RequestCancelInfos  map[int64]*RequestCancelInfo
-		SignalInfos         map[int64]*SignalInfo
-		SignalRequestedIDs  map[string]struct{}
-		ExecutionInfo       *WorkflowExecutionInfo
-		ExecutionStats      *ExecutionStats
-		BufferedEvents      []*types.HistoryEvent
-		VersionHistories    *VersionHistories
-		ReplicationState    *ReplicationState // TODO: remove this after all 2DC workflows complete
-		Checksum            checksum.Checksum
+		ActivityInfos          map[int64]*ActivityInfo
+		TimerInfos             map[string]*TimerInfo
+		ChildExecutionInfos    map[int64]*ChildExecutionInfo
+		RequestCancelInfos     map[int64]*RequestCancelInfo
+		SignalInfos            map[int64]*SignalInfo
+		SignalRequestedIDs     map[string]struct{}
+		WorkflowTimerTaskInfos []*WorkflowTimerTaskInfo
+		ExecutionInfo          *WorkflowExecutionInfo
+		ExecutionStats         *ExecutionStats
+		BufferedEvents         []*types.HistoryEvent
+		VersionHistories       *VersionHistories
+		ReplicationState       *ReplicationState // TODO: remove this after all 2DC workflows complete
+		Checksum               checksum.Checksum
 	}
 
 	// ActivityInfo details.
@@ -705,6 +706,15 @@ type (
 		Control               []byte
 	}
 
+	// WorkflowTimerTaskInfo contains metadata about workflow-level timer tasks.
+	// These are timer tasks that are associated with the workflow execution itself
+	// rather than user-created timers or activities (e.g., WorkflowTimeoutTask).
+	WorkflowTimerTaskInfo struct {
+		TimeoutType         int
+		TaskID              int64
+		VisibilityTimestamp time.Time
+	}
+
 	// CreateShardRequest is used to create a shard in executions table
 	CreateShardRequest struct {
 		ShardInfo *ShardInfo
@@ -892,6 +902,7 @@ type (
 		DeleteActivityInfos       []int64
 		UpsertTimerInfos          []*TimerInfo
 		DeleteTimerInfos          []string
+		WorkflowTimerTasks        []*WorkflowTimerTaskInfo
 		UpsertChildExecutionInfos []*ChildExecutionInfo
 		DeleteChildExecutionInfos []int64
 		UpsertRequestCancelInfos  []*RequestCancelInfo
@@ -919,6 +930,7 @@ type (
 
 		ActivityInfos       []*ActivityInfo
 		TimerInfos          []*TimerInfo
+		WorkflowTimerTasks  []*WorkflowTimerTaskInfo
 		ChildExecutionInfos []*ChildExecutionInfo
 		RequestCancelInfos  []*RequestCancelInfo
 		SignalInfos         []*SignalInfo
@@ -966,6 +978,15 @@ type (
 		RunID      string
 	}
 
+	// DeleteTimerTaskRequest is used to delete a timer task
+	DeleteTimerTaskRequest struct {
+		DomainID            string
+		WorkflowID          string
+		RunID               string
+		VisibilityTimestamp time.Time
+		TaskID              int64
+	}
+
 	// PutReplicationTaskToDLQRequest is used to put a replication task to dlq
 	PutReplicationTaskToDLQRequest struct {
 		ShardID           ShardID
@@ -1657,6 +1678,7 @@ type (
 		GetHistoryTasks(ctx context.Context, request *GetHistoryTasksRequest) (*GetHistoryTasksResponse, error)
 		CompleteHistoryTask(ctx context.Context, request *CompleteHistoryTaskRequest) error
 		RangeCompleteHistoryTask(ctx context.Context, request *RangeCompleteHistoryTaskRequest) (*RangeCompleteHistoryTaskResponse, error)
+		DeleteTimerTask(ctx context.Context, request *DeleteTimerTaskRequest) error
 
 		// Scan operations
 

common/persistence/data_manager_interfaces.go

@@ -130,6 +130,7 @@ type (
 		GetHistoryTasks(ctx context.Context, request *GetHistoryTasksRequest) (*GetHistoryTasksResponse, error)
 		CompleteHistoryTask(ctx context.Context, request *CompleteHistoryTaskRequest) error
 		RangeCompleteHistoryTask(ctx context.Context, request *RangeCompleteHistoryTaskRequest) (*RangeCompleteHistoryTaskResponse, error)
+		DeleteTimerTask(ctx context.Context, request *DeleteTimerTaskRequest) error
 
 		// Scan related methods
 		ListConcreteExecutions(ctx context.Context, request *ListConcreteExecutionsRequest) (*InternalListConcreteExecutionsResponse, error)
@@ -427,6 +428,7 @@ type (
 		SignalInfos         map[int64]*SignalInfo
 		SignalRequestedIDs  map[string]struct{}
 		BufferedEvents      []*DataBlob
+		WorkflowTimerTasks  *DataBlob
 
 		// Checksum field is used by Cassandra storage
 		// ChecksumData is used by All SQL storage
@@ -540,6 +542,7 @@ type (
 		DeleteActivityInfos       []int64
 		UpsertTimerInfos          []*TimerInfo
 		DeleteTimerInfos          []string
+		WorkflowTimerTasks        *DataBlob
 		UpsertChildExecutionInfos []*InternalChildExecutionInfo
 		DeleteChildExecutionInfos []int64
 		UpsertRequestCancelInfos  []*RequestCancelInfo
@@ -570,6 +573,7 @@ type (
 
 		ActivityInfos       []*InternalActivityInfo
 		TimerInfos          []*TimerInfo
+		WorkflowTimerTasks  *DataBlob
 		ChildExecutionInfos []*InternalChildExecutionInfo
 		RequestCancelInfos  []*RequestCancelInfo
 		SignalInfos         []*SignalInfo