[CHASM] timer queue pure task processing logic

service/history/timer_queue_active_task_executor.go

@@ -12,6 +12,7 @@ import (
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/backoff"
 	"go.temporal.io/server/common/definition"
@@ -114,6 +115,8 @@ func (t *timerQueueActiveTaskExecutor) Execute(
 		err = t.executeDeleteHistoryEventTask(ctx, task)
 	case *tasks.StateMachineTimerTask:
 		err = t.executeStateMachineTimerTask(ctx, task)
+	case *tasks.ChasmTaskPure:
+		err = t.executeChasmPureTimerTask(ctx, task)
 	default:
 		err = queues.NewUnprocessableTaskError("unknown task type")
 	}
@@ -911,3 +914,53 @@ func (t *timerQueueActiveTaskExecutor) processActivityWorkflowRules(
 
 	return nil
 }
+
+func (t *timerQueueActiveTaskExecutor) executeChasmPureTimerTask(
+	ctx context.Context,
+	task *tasks.ChasmTaskPure,
+) (err error) {
+	ctx, cancel := context.WithTimeout(ctx, taskTimeout)
+	defer cancel()
+
+	wfCtx, release, err := getWorkflowExecutionContextForTask(ctx, t.shardContext, t.cache, task)
+	if err != nil {
+		return err
+	}
+	defer func() { release(err) }()
+
+	ms, err := loadMutableStateForTimerTask(ctx, t.shardContext, wfCtx, task, t.metricsHandler, t.logger)
+	if err != nil {
+		return err
+	}
+	if ms == nil {
+		return nil
+	}
+
+	// Execute all fired pure tasks for a component while holding the workflow lock.
+	processedTimers, err := t.executeChasmPureTimers(
+		ctx,
+		wfCtx,
+		ms,
+		task,
+		func(node *chasm.Node, task any) error {
+			return node.ExecutePureTask(task)
+		},
+	)
+
+	// Commit changes only if we processed any timers.
+	if processedTimers == 0 {
+		return nil
+	}
+
+	if t.config.EnableUpdateWorkflowModeIgnoreCurrent() {
+		return wfCtx.UpdateWorkflowExecutionAsActive(ctx, t.shardContext)
+	}
+
+	// TODO: remove following code once EnableUpdateWorkflowModeIgnoreCurrent config is deprecated.
+	if ms.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
+		// Can't use UpdateWorkflowExecutionAsActive since it updates the current run, and we are operating on a
+		// closed workflow.
+		return wfCtx.SubmitClosedWorkflowSnapshot(ctx, t.shardContext, historyi.TransactionPolicyActive)
+	}
+	return wfCtx.UpdateWorkflowExecutionAsActive(ctx, t.shardContext)
+}

service/history/timer_queue_standby_task_executor.go

@@ -12,6 +12,7 @@ import (
 	taskqueuepb "go.temporal.io/api/taskqueue/v1"
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/client"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/log"
@@ -100,6 +101,8 @@ func (t *timerQueueStandbyTaskExecutor) Execute(
 		err = t.executeDeleteHistoryEventTask(ctx, task)
 	case *tasks.StateMachineTimerTask:
 		err = t.executeStateMachineTimerTask(ctx, task)
+	case *tasks.ChasmTaskPure:
+		err = t.executeChasmPureTimerTask(ctx, task)
 	default:
 		err = queues.NewUnprocessableTaskError("unknown task type")
 	}
@@ -111,6 +114,70 @@ func (t *timerQueueStandbyTaskExecutor) Execute(
 	}
 }
 
+func (t *timerQueueStandbyTaskExecutor) executeChasmPureTimerTask(
+	ctx context.Context,
+	task *tasks.ChasmTaskPure,
+) error {
+	actionFn := func(
+		ctx context.Context,
+		wfContext historyi.WorkflowContext,
+		mutableState historyi.MutableState,
+	) (any, error) {
+		processedTimers, err := t.executeChasmPureTimers(
+			ctx,
+			wfContext,
+			mutableState,
+			task,
+			func(node *chasm.Node, task any) error {
+				// If this line of code is reached, the task's Validate() function returned no error, which indicates
+				// that it is still expected to run. Return ErrTaskRetry to wait the machine to transition on the active
+				// cluster.
+				return consts.ErrTaskRetry
+			},
+		)
+		if err != nil {
+			if errors.Is(err, consts.ErrTaskRetry) {
+				// This handles the ErrTaskRetry error returned by executeChasmPureTimers.
+				return &struct{}{}, nil
+			}
+			return nil, err
+		}
+
+		// We haven't done any work, return without committing.
+		if processedTimers == 0 {
+			return nil, nil
+		}
+
+		if t.config.EnableUpdateWorkflowModeIgnoreCurrent() {
+			return nil, wfContext.UpdateWorkflowExecutionAsPassive(ctx, t.shardContext)
+		}
+
+		// TODO: remove following code once EnableUpdateWorkflowModeIgnoreCurrent config is deprecated.
+		if mutableState.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
+			// Can't use UpdateWorkflowExecutionAsPassive since it updates the current run,
+			// and we are operating on a closed workflow.
+			return nil, wfContext.SubmitClosedWorkflowSnapshot(
+				ctx,
+				t.shardContext,
+				historyi.TransactionPolicyPassive,
+			)
+		}
+		return nil, wfContext.UpdateWorkflowExecutionAsPassive(ctx, t.shardContext)
+	}
+
+	return t.processTimer(
+		ctx,
+		task,
+		actionFn,
+		getStandbyPostActionFn(
+			task,
+			t.getCurrentTime,
+			t.config.StandbyTaskMissingEventsDiscardDelay(task.GetType()),
+			t.checkWorkflowStillExistOnSourceBeforeDiscard,
+		),
+	)
+}
+
 func (t *timerQueueStandbyTaskExecutor) executeUserTimerTimeoutTask(
 	ctx context.Context,
 	timerTask *tasks.UserTimerTask,

service/history/timer_queue_task_executor_base.go

@@ -10,6 +10,7 @@ import (
 	"go.temporal.io/api/serviceerror"
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common/locks"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
@@ -240,8 +241,51 @@ func (t *timerQueueTaskExecutorBase) executeSingleStateMachineTimer(
 	return nil
 }
 
-// executeStateMachineTimers gets the state machine timers, processed the expired timers,
-// and return a slice of unprocessed timers.
+// executeChasmPureTimers walks a CHASM tree for expired pure task timers,
+// executes them, and returns a count of timers processed.
+func (t *timerQueueTaskExecutorBase) executeChasmPureTimers(
+	ctx context.Context,
+	workflowContext historyi.WorkflowContext,
+	ms historyi.MutableState,
+	task *tasks.ChasmTaskPure,
+	execute func(node *chasm.Node, task any) error,
+) (processedCount int, err error) {
+	// Because CHASM timers can target closed workflows, we need to specifically
+	// exclude zombie workflows, instead of merely checking that the workflow is
+	// running.
+	if ms.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {
+		return 0, consts.ErrWorkflowZombie
+	}
+
+	tree := ms.ChasmTree()
+	if tree == nil {
+		return 0, consts.ErrStaleReference
+	}
+
+	rootNode, ok := tree.(*chasm.Node)
+	if !ok {
+		return 0, fmt.Errorf("failed type assertion for ChasmTree")
+	}
+
+	pureTasks, err := tree.GetPureTasks(t.Now())
+	if err != nil {
+		return
+	}
+
+	for _, pureTask := range pureTasks {
+		err = execute(rootNode, pureTask)
+		if err != nil {
+			return
+		}
+
+		processedCount += 1
+	}
+
+	return
+}
+
+// executeStateMachineTimers gets the state machine timers, processes the expired timers,
+// and returns a count of timers processed.
 func (t *timerQueueTaskExecutorBase) executeStateMachineTimers(
 	ctx context.Context,
 	workflowContext historyi.WorkflowContext,