mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +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>
89 lines
2.4 KiB
Go
89 lines
2.4 KiB
Go
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination events_reapplier_mock.go
|
|
|
|
package ndc
|
|
|
|
import (
|
|
"context"
|
|
|
|
historypb "go.temporal.io/api/history/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
enumsspb "go.temporal.io/server/api/enums/v1"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/service/history/hsm"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/workflow/update"
|
|
)
|
|
|
|
type (
|
|
EventsReapplier interface {
|
|
ReapplyEvents(
|
|
ctx context.Context,
|
|
ms historyi.MutableState,
|
|
updateRegistry update.Registry,
|
|
historyEvents []*historypb.HistoryEvent,
|
|
runID string,
|
|
) ([]*historypb.HistoryEvent, error)
|
|
}
|
|
|
|
EventsReapplierImpl struct {
|
|
stateMachineRegistry *hsm.Registry
|
|
metricsHandler metrics.Handler
|
|
logger log.Logger
|
|
}
|
|
)
|
|
|
|
func NewEventsReapplier(
|
|
stateMachineRegistry *hsm.Registry,
|
|
metricsHandler metrics.Handler,
|
|
logger log.Logger,
|
|
) *EventsReapplierImpl {
|
|
|
|
return &EventsReapplierImpl{
|
|
stateMachineRegistry: stateMachineRegistry,
|
|
metricsHandler: metricsHandler,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
func (r *EventsReapplierImpl) ReapplyEvents(
|
|
ctx context.Context,
|
|
ms historyi.MutableState,
|
|
updateRegistry update.Registry,
|
|
historyEvents []*historypb.HistoryEvent,
|
|
runID string,
|
|
) ([]*historypb.HistoryEvent, error) {
|
|
// sanity check workflow still running
|
|
if !ms.IsWorkflowExecutionRunning() {
|
|
return nil, serviceerror.NewInternal("unable to reapply events to closed workflow.")
|
|
}
|
|
reappliedEvents, err := reapplyEvents(ctx, ms, updateRegistry, r.stateMachineRegistry, historyEvents, nil, runID, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(reappliedEvents) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
if !ms.IsWorkflowExecutionRunning() {
|
|
// workflow is closed after reapplying events, no need to schedule workflow task
|
|
return reappliedEvents, nil
|
|
}
|
|
|
|
// After reapply event, checking if we should schedule a workflow task
|
|
if ms.IsWorkflowPendingOnWorkflowTaskBackoff() {
|
|
// Do not create workflow task when the workflow has first workflow task backoff and execution is not started yet
|
|
return reappliedEvents, nil
|
|
}
|
|
|
|
if !ms.HasPendingWorkflowTask() && !ms.IsWorkflowExecutionStatusPaused() {
|
|
if _, err := ms.AddWorkflowTaskScheduledEvent(
|
|
false,
|
|
enumsspb.WORKFLOW_TASK_TYPE_NORMAL,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return reappliedEvents, nil
|
|
}
|