mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Chetan GowdaandGitHub
2312f6752f
## What changed? - Prevents workflow task generation when paused (ex: signals, activity completions, child workflow completions etc) - Prevents activity task generation when a workflow is paused and an inflight workflow task comes back with start activity commands. - Prevents eager activity starts when the workflow is paused. ## Why? When a workflow is paused, we should not generate any more new activity and workflow tasks. ## How did you test it? - [x] built - [ ] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [x] added new functional test(s) --- > [!NOTE] > Prevents workflow/activity task creation and certain API operations when a workflow is paused, adds safeguards and tests across history, NDC, and task refresh paths. > > - **Behavior/State**: > - Add `MutableState.IsWorkflowExecutionStatusPaused()` and honor paused status across scheduling paths. > - `workflow/util.ScheduleWorkflowTask` returns early if paused. > - **API Handlers**: > - `RespondWorkflowTaskCompleted`: reject `ForceCreateNewWorkflowTask` when paused; block eager activity and bypass activity task gen when paused. > - `SignalWithStartWorkflow`: do not schedule a new workflow task if paused. > - `UpdateWorkflow`: reject updates while paused. > - Activity failure/respond paths: avoid new WFT when paused. > - **NDC/Replication**: > - Reapply events and buffered-event flusher: skip scheduling WFT if paused. > - `ndc.Workflow.FlushBufferedEvents`: don't schedule WFT when paused. > - **Task Refresher**: > - Skip refreshing WFT/activity tasks when paused; allow timers/search-attr updates; skip close tasks for running/paused. > - **Interfaces/Mocks**: > - Add `IsWorkflowExecutionStatusPaused` to `MutableState` interface and update mocks. > - **Tests**: > - New unit tests for paused behavior in handlers, NDC, task refresher, and transfer queue; functional tests for pause/unpause workflow behavior. > > <sup>Written by [Cursor Bugbot]() for commit 17c11c687a0d620cc5a77e6ce9369622feb9214b. This will update automatically on new commits. Configure [here]().</sup>
126 lines
3.3 KiB
Go
126 lines
3.3 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
|
|
clockspb "go.temporal.io/server/api/clock/v1"
|
|
enumsspb "go.temporal.io/server/api/enums/v1"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/locks"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/workflow/update"
|
|
)
|
|
|
|
func GetAndUpdateWorkflowWithNew(
|
|
ctx context.Context,
|
|
reqClock *clockspb.VectorClock,
|
|
workflowKey definition.WorkflowKey,
|
|
action UpdateWorkflowActionFunc,
|
|
newWorkflowFn func() (historyi.WorkflowContext, historyi.MutableState, error),
|
|
shard historyi.ShardContext,
|
|
workflowConsistencyChecker WorkflowConsistencyChecker,
|
|
) (retError error) {
|
|
workflowLease, err := workflowConsistencyChecker.GetWorkflowLease(
|
|
ctx,
|
|
reqClock,
|
|
workflowKey,
|
|
locks.PriorityHigh,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { workflowLease.GetReleaseFn()(retError) }()
|
|
|
|
return UpdateWorkflowWithNew(shard, ctx, workflowLease, action, newWorkflowFn)
|
|
}
|
|
|
|
func GetAndUpdateWorkflowWithConsistencyCheck(
|
|
ctx context.Context,
|
|
reqClock *clockspb.VectorClock,
|
|
consistencyCheckFn MutableStateConsistencyPredicate,
|
|
workflowKey definition.WorkflowKey,
|
|
action UpdateWorkflowActionFunc,
|
|
newWorkflowFn func() (historyi.WorkflowContext, historyi.MutableState, error),
|
|
shardContext historyi.ShardContext,
|
|
workflowConsistencyChecker WorkflowConsistencyChecker,
|
|
) (retError error) {
|
|
workflowLease, err := workflowConsistencyChecker.GetWorkflowLeaseWithConsistencyCheck(
|
|
ctx,
|
|
reqClock,
|
|
consistencyCheckFn,
|
|
workflowKey,
|
|
locks.PriorityHigh,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { workflowLease.GetReleaseFn()(retError) }()
|
|
|
|
return UpdateWorkflowWithNew(shardContext, ctx, workflowLease, action, newWorkflowFn)
|
|
}
|
|
|
|
func UpdateWorkflowWithNew(
|
|
shardContext historyi.ShardContext,
|
|
ctx context.Context,
|
|
workflowLease WorkflowLease,
|
|
action UpdateWorkflowActionFunc,
|
|
newWorkflowFn func() (historyi.WorkflowContext, historyi.MutableState, error),
|
|
) (retError error) {
|
|
|
|
// conduct caller action
|
|
postActions, err := action(workflowLease)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if postActions.Noop {
|
|
return nil
|
|
}
|
|
|
|
mutableState := workflowLease.GetMutableState()
|
|
if postActions.CreateWorkflowTask {
|
|
// Create a transfer task to schedule a workflow task only if the workflow is not paused and there is no pending workflow task.
|
|
if !mutableState.HasPendingWorkflowTask() && !mutableState.IsWorkflowExecutionStatusPaused() {
|
|
if _, err := mutableState.AddWorkflowTaskScheduledEvent(
|
|
false,
|
|
enumsspb.WORKFLOW_TASK_TYPE_NORMAL,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
var updateErr error
|
|
if newWorkflowFn != nil {
|
|
newContext, newMutableState, err := newWorkflowFn()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
lastWriteVersion, err := mutableState.GetLastWriteVersion()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = NewWorkflowVersionCheck(shardContext, lastWriteVersion, newMutableState); err != nil {
|
|
return err
|
|
}
|
|
|
|
updateErr = workflowLease.GetContext().UpdateWorkflowExecutionWithNewAsActive(
|
|
ctx,
|
|
shardContext,
|
|
newContext,
|
|
newMutableState,
|
|
)
|
|
} else {
|
|
updateErr = workflowLease.GetContext().UpdateWorkflowExecutionAsActive(ctx, shardContext)
|
|
}
|
|
|
|
if updateErr != nil {
|
|
return updateErr
|
|
}
|
|
|
|
if postActions.AbortUpdates {
|
|
workflowLease.GetContext().UpdateRegistry(ctx).Abort(update.AbortReasonWorkflowCompleted)
|
|
}
|
|
|
|
return nil
|
|
}
|