From b68685f736cc8082b6182987b87b57b1fc7e9125 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Mon, 5 Oct 2026 13:25:05 +0100 Subject: [PATCH 1/2] task: remove the early resolution buffer A completion, failure, timer firing or child result that reached the replay before the workflow had scheduled the matching step was buffered (#125) and delivered once the step was scheduled. That made the Go SDK the only SDK that tolerated such an early resolution: Python, Java and JavaScript replay in order and ignore a resolution with no pending step, so the fix lived in one SDK while the cause, a backend that dispatches before it saves, stayed. dapr now hands the worker such a resolution after the events that lead the workflow to schedule the step, for new events and for histories already persisted in that shape (dapr/dapr#10590). The SDK goes back to ignoring an unmatched resolution, now with a warning, like the other SDKs. The applier change from #133 stays: a scheduling whose resolution is already in the state is recorded but not dispatched. That is what keeps the step from running twice, and the backend relies on it. Signed-off-by: joshvanl --- backend/runtimestate/applier.go | 7 +- backend/runtimestate/applier_test.go | 7 +- task/early_resolution_test.go | 105 ++++ task/orchestrator.go | 140 +---- task/orchestrator_buffered_test.go | 820 --------------------------- task/task.go | 14 - 6 files changed, 123 insertions(+), 970 deletions(-) create mode 100644 task/early_resolution_test.go delete mode 100644 task/orchestrator_buffered_test.go diff --git a/backend/runtimestate/applier.go b/backend/runtimestate/applier.go index 16b31085..722eee3b 100644 --- a/backend/runtimestate/applier.go +++ b/backend/runtimestate/applier.go @@ -65,9 +65,10 @@ func (a *Applier) Actions(s *protos.WorkflowRuntimeState, customStatus *wrappers s.CustomStatus = customStatus s.Stalled = nil - // An action whose resolution is already in the state's events (delivered - // from the replay buffer) still gets its scheduling event recorded so the - // history stays replayable, but its work must not be dispatched again. + // An action whose resolution is already in the state's events (the + // backend handed the worker a result that arrived before its scheduling + // was saved) still gets its scheduling event recorded so the history + // stays replayable, but its work must not be dispatched again. resolved := dedup.NewForState(s) for _, action := range actions { diff --git a/backend/runtimestate/applier_test.go b/backend/runtimestate/applier_test.go index d26b14bd..892b5734 100644 --- a/backend/runtimestate/applier_test.go +++ b/backend/runtimestate/applier_test.go @@ -26,9 +26,10 @@ import ( ) // A resolution that reached the state before its scheduling action was -// committed (delivered to the workflow from the replay buffer) must still get -// its scheduling event recorded, so the history replays and the resolution -// keeps its match, but the work must not be dispatched a second time. +// committed (the backend handed it to the worker, which scheduled the step in +// the same turn) must still get its scheduling event recorded, so the history +// replays and the resolution keeps its match, but the work must not be +// dispatched a second time. func TestActions_ResolvedScheduleIsRecordedNotDispatched(t *testing.T) { scheduleTask := func(id int32) *protos.WorkflowAction { return &protos.WorkflowAction{Id: id, WorkflowActionType: &protos.WorkflowAction_ScheduleTask{ diff --git a/task/early_resolution_test.go b/task/early_resolution_test.go new file mode 100644 index 00000000..d4f08e5e --- /dev/null +++ b/task/early_resolution_test.go @@ -0,0 +1,105 @@ +/* +Copyright 2026 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package task + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" + + "github.com/dapr/durabletask-go/api/protos" +) + +type warnLogger struct{ warns []string } + +func (l *warnLogger) Debug(...any) {} +func (l *warnLogger) Debugf(string, ...any) {} +func (l *warnLogger) Info(...any) {} +func (l *warnLogger) Infof(string, ...any) {} +func (l *warnLogger) Warn(...any) {} +func (l *warnLogger) Warnf(f string, v ...any) { l.warns = append(l.warns, fmt.Sprintf(f, v...)) } +func (l *warnLogger) Error(...any) {} +func (l *warnLogger) Errorf(string, ...any) {} + +// A resolution replayed before the workflow has scheduled its step is +// ignored, like the other SDKs do, and the replay carries on. The backend +// (daprd) is responsible for handing such a resolution to the worker after +// the events that lead the workflow to schedule the step. +func Test_UnmatchedResolutionIsIgnored(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddActivityN("a", func(ActivityContext) (any, error) { return "x", nil })) + require.NoError(t, r.AddWorkflowN("gated", func(ctx *WorkflowContext) (any, error) { + if err := ctx.WaitForSingleEvent("go", time.Hour).Await(nil); err != nil { + return nil, err + } + var out string + if err := ctx.CallActivity("a").Await(&out); err != nil { + return nil, err + } + return out, nil + })) + started := &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_ExecutionStarted{ + ExecutionStarted: &protos.ExecutionStartedEvent{Name: "gated", WorkflowInstance: &protos.WorkflowInstance{InstanceId: "id"}}, + }} + goEvent := &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_EventRaised{ + EventRaised: &protos.EventRaisedEvent{Name: "go"}, + }} + // The synthetic external event timer is id 0, so the activity is id 1. + early := &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_TaskCompleted{ + TaskCompleted: &protos.TaskCompletedEvent{TaskScheduledId: 1, Result: wrapperspb.String(`"early"`)}, + }} + run := func(newEvents ...*protos.HistoryEvent) ([]*protos.WorkflowAction, *warnLogger) { + l := new(warnLogger) + ctx := NewWorkflowContext(r, "id", []*protos.HistoryEvent{started}, newEvents) + ctx.SetLogger(l) + return ctx.start(), l + } + scheduled := func(actions []*protos.WorkflowAction) (n int) { + for _, a := range actions { + if a.GetScheduleTask() != nil { + n++ + } + } + return n + } + completed := func(actions []*protos.WorkflowAction) *protos.CompleteWorkflowAction { + for _, a := range actions { + if c := a.GetCompleteWorkflow(); c != nil { + return c + } + } + return nil + } + + t.Run("before the gating event it is ignored", func(t *testing.T) { + actions, l := run(early, goEvent) + assert.Equal(t, 1, scheduled(actions), "the activity is scheduled") + assert.Nil(t, completed(actions), "its result was not used") + require.Len(t, l.warns, 1) + assert.Contains(t, l.warns[0], "ignoring TaskCompleted for id 1") + }) + + t.Run("after the gating event it resolves the step", func(t *testing.T) { + actions, l := run(goEvent, early) + assert.Equal(t, 1, scheduled(actions), "the schedule is still emitted for the backend to record") + require.NotNil(t, completed(actions)) + assert.Equal(t, `"early"`, completed(actions).GetResult().GetValue()) + assert.Empty(t, l.warns) + }) +} diff --git a/task/orchestrator.go b/task/orchestrator.go index 45a12453..fe2ae4ca 100644 --- a/task/orchestrator.go +++ b/task/orchestrator.go @@ -17,27 +17,9 @@ import ( "github.com/dapr/durabletask-go/api/helpers" "github.com/dapr/durabletask-go/api/protos" "github.com/dapr/durabletask-go/backend" - "github.com/dapr/durabletask-go/backend/runtimestate/dedup" "github.com/dapr/kit/ptr" ) -// resolutionKey correlates a resolution event -// (TaskCompleted/TaskFailed/TimerFired/ChildWorkflowInstance{Completed,Failed}) -// with the pending entry it resolves, using the same (kind, id) semantics as -// backend/runtimestate/dedup. -type resolutionKey struct { - kind dedup.Kind - id int32 -} - -// bufferedResolution holds a resolution event that arrived before the -// workflow scheduled the matching work in this execution, so it can be -// delivered once the matching pending entry is registered. -type bufferedResolution struct { - redeliver func() error - desc string -} - // externalEventIndefiniteFireAt is the sentinel fire-at value used for the // synthetic timer backing WaitForExternalEvent calls with a negative // timeout. It is chosen far enough in the future that it effectively never @@ -75,8 +57,6 @@ type WorkflowContext struct { bufferedExternalEvents map[string]*list.List pendingExternalEventTasks map[string]*list.List saveBufferedExternalEvents bool - bufferedResolutions map[resolutionKey]bufferedResolution - resolvedResolutions map[resolutionKey]struct{} logger backend.Logger historyPatches map[string]bool appliedPatches map[string]bool @@ -207,8 +187,6 @@ func NewWorkflowContext(registry *TaskRegistry, id api.InstanceID, oldEvents []* historyPatches: make(map[string]bool), appliedPatches: make(map[string]bool), encounteredPatches: make([]string, 0), - bufferedResolutions: make(map[resolutionKey]bufferedResolution), - resolvedResolutions: make(map[resolutionKey]struct{}), logger: backend.DefaultLogger(), } } @@ -226,12 +204,6 @@ func (ctx *WorkflowContext) start() (actions []*protos.WorkflowAction) { ctx.defaultDetachedWorkflowCounter = 0 ctx.pendingActions = make(map[int32]*protos.WorkflowAction) ctx.pendingTasks = make(map[int32]*completableTask) - clear(ctx.bufferedResolutions) - clear(ctx.resolvedResolutions) - - // Registered before the recover defer so it runs on both the normal exit - // and the ErrTaskBlocked path. - defer ctx.warnUnconsumedResolutions() defer func() { result := recover() @@ -438,9 +410,7 @@ func (ctx *WorkflowContext) internalScheduleActivity(activityName, taskExecution ctx.pendingActions[scheduleTaskAction.Id] = scheduleTaskAction task := newTask(ctx) - task.kind = dedup.KindTask ctx.pendingTasks[scheduleTaskAction.Id] = task - ctx.consumeBufferedResolution(dedup.KindTask, scheduleTaskAction.Id) return task } @@ -527,9 +497,7 @@ func (ctx *WorkflowContext) internalCallChildWorkflow(workflowName string, optio ctx.pendingActions[createChildWorkflowAction.Id] = createChildWorkflowAction task := newTask(ctx) - task.kind = dedup.KindChild ctx.pendingTasks[createChildWorkflowAction.Id] = task - ctx.consumeBufferedResolution(dedup.KindChild, createChildWorkflowAction.Id) return task } @@ -621,9 +589,7 @@ func (ctx *WorkflowContext) createTimerInternal(name *string, delay time.Duratio ctx.pendingActions[timerAction.Id] = timerAction task := newTask(ctx) - task.kind = dedup.KindTimer ctx.pendingTasks[timerAction.Id] = task - ctx.consumeBufferedResolution(dedup.KindTimer, timerAction.Id) return task, createTimer } @@ -645,9 +611,7 @@ func (ctx *WorkflowContext) createExternalEventTimerInternal(eventName string, f ctx.pendingActions[timerAction.Id] = timerAction task := newTask(ctx) - task.kind = dedup.KindTimer ctx.pendingTasks[timerAction.Id] = task - ctx.consumeBufferedResolution(dedup.KindTimer, timerAction.Id) return task } @@ -815,78 +779,14 @@ func (ctx *WorkflowContext) onTaskScheduled(taskID int32, ts *protos.TaskSchedul return nil } -// bufferResolution records a resolution event that arrived before the -// workflow scheduled the matching work in this execution, so it can be -// delivered when the matching pending entry is registered. True duplicates -// (the id was already resolved this execution, or an identical early -// resolution is already buffered) are dropped; runtime state dedup upstream -// makes those unreachable in practice, this is defense in depth. -func (ctx *WorkflowContext) bufferResolution(key resolutionKey, eventName string, redeliver func() error) { - if _, resolved := ctx.resolvedResolutions[key]; resolved { - ctx.logger.Debugf("%v: dropping duplicate %s for id %d: already resolved this execution", ctx.ID, eventName, key.id) - return - } - if _, buffered := ctx.bufferedResolutions[key]; buffered { - ctx.logger.Debugf("%v: dropping duplicate %s for id %d: already buffered this execution", ctx.ID, eventName, key.id) - return - } - ctx.logger.Debugf("%v: buffering early %s for id %d until the workflow schedules the matching work", ctx.ID, eventName, key.id) - ctx.bufferedResolutions[key] = bufferedResolution{ - redeliver: redeliver, - desc: fmt.Sprintf("%s for id %d", eventName, key.id), - } -} - -// consumeBufferedResolution delivers a buffered early resolution to the -// pending entry that was just registered for (kind, id). The pending action -// is left in place and emitted like any other: the backend records its -// scheduling event so the history stays replayable and the resolution keeps -// its match, while the applier sees the resolution already in the state and -// withholds the dispatch. -func (ctx *WorkflowContext) consumeBufferedResolution(kind dedup.Kind, id int32) { - key := resolutionKey{kind: kind, id: id} - br, ok := ctx.bufferedResolutions[key] - if !ok { - return - } - delete(ctx.bufferedResolutions, key) - ctx.logger.Debugf("%v: delivering buffered %s to newly scheduled work; the schedule is recorded without being dispatched", ctx.ID, br.desc) - // The handler cannot re-buffer on this path: the pending entry exists. - _ = br.redeliver() -} - -// warnUnconsumedResolutions surfaces resolution events that were buffered -// this execution but never matched any scheduled work. -func (ctx *WorkflowContext) warnUnconsumedResolutions() { - if len(ctx.bufferedResolutions) == 0 { - return - } - keys := make([]resolutionKey, 0, len(ctx.bufferedResolutions)) - for key := range ctx.bufferedResolutions { - keys = append(keys, key) - } - sort.Slice(keys, func(i, j int) bool { - if keys[i].kind != keys[j].kind { - return keys[i].kind < keys[j].kind - } - return keys[i].id < keys[j].id - }) - for _, key := range keys { - br := ctx.bufferedResolutions[key] - ctx.logger.Warnf("%v: %s arrived before the matching work was scheduled and was not consumed by the end of this execution; the event stays in history and is re-evaluated next turn, but if it never matches this indicates a non-deterministic workflow or an out-of-order history", ctx.ID, br.desc) - } -} - func (ctx *WorkflowContext) onTaskCompleted(tc *protos.TaskCompletedEvent) error { taskID := tc.TaskScheduledId - key := resolutionKey{kind: dedup.KindTask, id: taskID} task, ok := ctx.pendingTasks[taskID] - if !ok || task.kind != dedup.KindTask { - ctx.bufferResolution(key, "TaskCompleted", func() error { return ctx.onTaskCompleted(tc) }) + if !ok { + ctx.logger.Warnf("%v: ignoring TaskCompleted for id %d: no pending activity with that id", ctx.ID, taskID) return nil } delete(ctx.pendingTasks, taskID) - ctx.resolvedResolutions[key] = struct{}{} if tc.Result != nil { task.complete([]byte(tc.Result.Value)) @@ -898,14 +798,12 @@ func (ctx *WorkflowContext) onTaskCompleted(tc *protos.TaskCompletedEvent) error func (ctx *WorkflowContext) onTaskFailed(tf *protos.TaskFailedEvent) error { taskID := tf.TaskScheduledId - key := resolutionKey{kind: dedup.KindTask, id: taskID} task, ok := ctx.pendingTasks[taskID] - if !ok || task.kind != dedup.KindTask { - ctx.bufferResolution(key, "TaskFailed", func() error { return ctx.onTaskFailed(tf) }) + if !ok { + ctx.logger.Warnf("%v: ignoring TaskFailed for id %d: no pending activity with that id", ctx.ID, taskID) return nil } delete(ctx.pendingTasks, taskID) - ctx.resolvedResolutions[key] = struct{}{} // completing a task will resume the corresponding Await() call task.fail(tf.FailureDetails) @@ -933,14 +831,12 @@ func (ctx *WorkflowContext) onChildWorkflowScheduled(taskID int32, ts *protos.Ch func (ctx *WorkflowContext) onChildWorkflowCompleted(soc *protos.ChildWorkflowInstanceCompletedEvent) error { taskID := soc.TaskScheduledId - key := resolutionKey{kind: dedup.KindChild, id: taskID} task, ok := ctx.pendingTasks[taskID] - if !ok || task.kind != dedup.KindChild { - ctx.bufferResolution(key, "ChildWorkflowInstanceCompleted", func() error { return ctx.onChildWorkflowCompleted(soc) }) + if !ok { + ctx.logger.Warnf("%v: ignoring ChildWorkflowInstanceCompleted for id %d: no pending child workflow with that id", ctx.ID, taskID) return nil } delete(ctx.pendingTasks, taskID) - ctx.resolvedResolutions[key] = struct{}{} // completing a task will resume the corresponding Await() call if soc.Result != nil { @@ -953,14 +849,12 @@ func (ctx *WorkflowContext) onChildWorkflowCompleted(soc *protos.ChildWorkflowIn func (ctx *WorkflowContext) onChildWorkflowFailed(sof *protos.ChildWorkflowInstanceFailedEvent) error { taskID := sof.TaskScheduledId - key := resolutionKey{kind: dedup.KindChild, id: taskID} task, ok := ctx.pendingTasks[taskID] - if !ok || task.kind != dedup.KindChild { - ctx.bufferResolution(key, "ChildWorkflowInstanceFailed", func() error { return ctx.onChildWorkflowFailed(sof) }) + if !ok { + ctx.logger.Warnf("%v: ignoring ChildWorkflowInstanceFailed for id %d: no pending child workflow with that id", ctx.ID, taskID) return nil } delete(ctx.pendingTasks, taskID) - ctx.resolvedResolutions[key] = struct{}{} // completing a task will resume the corresponding Await() call task.fail(sof.FailureDetails) @@ -990,14 +884,12 @@ func (ctx *WorkflowContext) onTimerCreated(e *protos.HistoryEvent) error { func (ctx *WorkflowContext) onTimerFired(tf *protos.TimerFiredEvent) error { timerID := tf.TimerId - key := resolutionKey{kind: dedup.KindTimer, id: timerID} task, ok := ctx.pendingTasks[timerID] - if !ok || task.kind != dedup.KindTimer { - ctx.bufferResolution(key, "TimerFired", func() error { return ctx.onTimerFired(tf) }) + if !ok { + ctx.logger.Warnf("%v: ignoring TimerFired for id %d: no pending timer with that id", ctx.ID, timerID) return nil } delete(ctx.pendingTasks, timerID) - ctx.resolvedResolutions[key] = struct{}{} // completing a task will resume the corresponding Await() call task.complete(nil) @@ -1229,18 +1121,6 @@ func (ctx *WorkflowContext) dropOptionalExternalEventTimerAt(atID int32) bool { } ctx.sequenceNumber-- - - // The shift moved pending entries onto their history numbering, so a - // resolution buffered under a history id may now match a shifted entry; - // deliver any that do. bufferedResolutions keys are numbered by history - // events and are not shifted: this drop exists precisely to align the - // current ids to the history numbering. - for _, id := range taskIDs { - newID := id - 1 - if t, ok := ctx.pendingTasks[newID]; ok { - ctx.consumeBufferedResolution(t.kind, newID) - } - } return true } diff --git a/task/orchestrator_buffered_test.go b/task/orchestrator_buffered_test.go deleted file mode 100644 index 808fda2c..00000000 --- a/task/orchestrator_buffered_test.go +++ /dev/null @@ -1,820 +0,0 @@ -/* -Copyright 2026 The Dapr Authors -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - http://www.apache.org/licenses/LICENSE-2.0 -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package task - -import ( - "errors" - "fmt" - "strings" - "sync" - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - "google.golang.org/protobuf/types/known/timestamppb" - "google.golang.org/protobuf/types/known/wrapperspb" - - "github.com/dapr/durabletask-go/api/protos" - "github.com/dapr/durabletask-go/backend" - "github.com/dapr/durabletask-go/backend/runtimestate" -) - -// captureLogger records formatted log lines per level for assertions. -type captureLogger struct { - mu sync.Mutex - warns []string - debug []string -} - -func (c *captureLogger) log(dst *[]string, format string, v ...any) { - c.mu.Lock() - defer c.mu.Unlock() - *dst = append(*dst, fmt.Sprintf(format, v...)) -} - -func (c *captureLogger) Debug(v ...any) {} -func (c *captureLogger) Debugf(format string, v ...any) { c.log(&c.debug, format, v...) } -func (c *captureLogger) Info(v ...any) {} -func (c *captureLogger) Infof(format string, v ...any) {} -func (c *captureLogger) Warn(v ...any) {} -func (c *captureLogger) Warnf(format string, v ...any) { c.log(&c.warns, format, v...) } -func (c *captureLogger) Error(v ...any) {} -func (c *captureLogger) Errorf(format string, v ...any) {} - -func (c *captureLogger) warnsContaining(sub string) int { - c.mu.Lock() - defer c.mu.Unlock() - n := 0 - for _, w := range c.warns { - if strings.Contains(w, sub) { - n++ - } - } - return n -} - -func evExecutionStarted(name string) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_ExecutionStarted{ - ExecutionStarted: &protos.ExecutionStartedEvent{ - Name: name, - WorkflowInstance: &protos.WorkflowInstance{InstanceId: "buffered-test"}, - }, - }, - } -} - -func evTaskScheduled(id int32, name string) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: id, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_TaskScheduled{ - TaskScheduled: &protos.TaskScheduledEvent{Name: name}, - }, - } -} - -func evTaskCompleted(id int32, result string) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_TaskCompleted{ - TaskCompleted: &protos.TaskCompletedEvent{ - TaskScheduledId: id, - Result: wrapperspb.String(result), - }, - }, - } -} - -func evTaskFailed(id int32, execID string) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_TaskFailed{ - TaskFailed: &protos.TaskFailedEvent{ - TaskScheduledId: id, - TaskExecutionId: execID, - FailureDetails: &protos.TaskFailureDetails{ErrorType: "TestError", ErrorMessage: "injected failure"}, - }, - }, - } -} - -func evTimerFired(id int32) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_TimerFired{ - TimerFired: &protos.TimerFiredEvent{TimerId: id, FireAt: timestamppb.Now()}, - }, - } -} - -func evChildCompleted(id int32, result string) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_ChildWorkflowInstanceCompleted{ - ChildWorkflowInstanceCompleted: &protos.ChildWorkflowInstanceCompletedEvent{ - TaskScheduledId: id, - Result: wrapperspb.String(result), - }, - }, - } -} - -func evChildFailed(id int32) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_ChildWorkflowInstanceFailed{ - ChildWorkflowInstanceFailed: &protos.ChildWorkflowInstanceFailedEvent{ - TaskScheduledId: id, - FailureDetails: &protos.TaskFailureDetails{ErrorType: "TestError", ErrorMessage: "injected child failure"}, - }, - }, - } -} - -func evEventRaised(name string) *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_EventRaised{ - EventRaised: &protos.EventRaisedEvent{Name: name}, - }, - } -} - -func evSuspended() *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_ExecutionSuspended{ - ExecutionSuspended: &protos.ExecutionSuspendedEvent{}, - }, - } -} - -func evResumed() *protos.HistoryEvent { - return &protos.HistoryEvent{ - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_ExecutionResumed{ - ExecutionResumed: &protos.ExecutionResumedEvent{}, - }, - } -} - -func runBuffered(t *testing.T, registry *TaskRegistry, oldEvents, newEvents []*protos.HistoryEvent) ([]*protos.WorkflowAction, *captureLogger) { - t.Helper() - cl := &captureLogger{} - ctx := NewWorkflowContext(registry, "buffered-test", oldEvents, newEvents) - ctx.SetLogger(cl) - return ctx.start(), cl -} - -func completeAction(t *testing.T, actions []*protos.WorkflowAction) *protos.CompleteWorkflowAction { - t.Helper() - for _, a := range actions { - if co := a.GetCompleteWorkflow(); co != nil { - return co - } - } - return nil -} - -func countActions(actions []*protos.WorkflowAction, pred func(*protos.WorkflowAction) bool) int { - n := 0 - for _, a := range actions { - if pred(a) { - n++ - } - } - return n -} - -func isScheduleTask(a *protos.WorkflowAction) bool { return a.GetScheduleTask() != nil } - -// A schedule whose resolution was delivered from the buffer is still emitted: -// the backend must record its scheduling event so the history stays -// replayable, and the applier withholds the dispatch because the resolution -// is already in the state (see Test_BufferedResolution_ExecutorReplayRoundTrip -// and the applier tests in backend/runtimestate). -const emittedNotDispatched = "the resolved schedule must still be emitted; the applier withholds its dispatch" -func isCreateChild(a *protos.WorkflowAction) bool { return a.GetCreateChildWorkflow() != nil } - -// waitThenActivityRegistry registers a workflow that waits for the "go" event -// and then calls the "act" activity, returning the activity output. Sequence -// numbers: the WaitForSingleEvent synthetic timer takes id 0, the activity -// takes id 1. -func waitThenActivityRegistry(t *testing.T) *TaskRegistry { - t.Helper() - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - var out string - if err := ctx.CallActivity("act").Await(&out); err != nil { - return nil, err - } - return out, nil - })) - require.NoError(t, r.AddActivityN("act", func(ActivityContext) (any, error) { return "ran", nil })) - return r -} - -func Test_BufferedResolution_EarlyTaskCompleted(t *testing.T) { - actions, cl := runBuffered(t, waitThenActivityRegistry(t), nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(1, `"injected"`), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co, "workflow must complete using the early completion") - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"injected"`, co.GetResult().GetValue()) - assert.Equal(t, 1, countActions(actions, isScheduleTask), emittedNotDispatched) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_EarlyTaskFailed(t *testing.T) { - var gotExecID string - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - task := ctx.CallActivity("act") - err := task.Await(nil) - gotExecID = task.TaskExecutionId() - return nil, err - })) - require.NoError(t, r.AddActivityN("act", func(ActivityContext) (any, error) { return nil, nil })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskFailed(1, "exec-x"), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED, co.WorkflowStatus) - assert.Contains(t, co.GetFailureDetails().GetErrorMessage(), "injected failure") - assert.Equal(t, 1, countActions(actions, isScheduleTask), emittedNotDispatched) - assert.Equal(t, "exec-x", gotExecID) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_EarlyTimerFired(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - if err := ctx.CreateTimer(time.Hour).Await(nil); err != nil { - return nil, err - } - return "done", nil - })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTimerFired(1), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, 1, countActions(actions, func(a *protos.WorkflowAction) bool { - return a.GetCreateTimer() != nil && a.Id == 1 - }), emittedNotDispatched) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_EarlyChildCompleted(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - var out string - if err := ctx.CallChildWorkflow("child").Await(&out); err != nil { - return nil, err - } - return out, nil - })) - require.NoError(t, r.AddWorkflowN("child", func(ctx *WorkflowContext) (any, error) { return "child-ran", nil })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evChildCompleted(1, `"injected-child"`), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"injected-child"`, co.GetResult().GetValue()) - assert.Equal(t, 1, countActions(actions, isCreateChild), emittedNotDispatched) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_EarlyChildFailed(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - return nil, ctx.CallChildWorkflow("child").Await(nil) - })) - require.NoError(t, r.AddWorkflowN("child", func(ctx *WorkflowContext) (any, error) { return nil, nil })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evChildFailed(1), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED, co.WorkflowStatus) - assert.Equal(t, 1, countActions(actions, isCreateChild), emittedNotDispatched) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_EarlyTimerFiredForExternalEventTimer(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("a", -1).Await(nil); err != nil { - return nil, err - } - err := ctx.WaitForSingleEvent("b", time.Hour).Await(nil) - if errors.Is(err, ErrTaskCanceled) { - return "timedout", nil - } - return nil, err - })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTimerFired(1), - evEventRaised("a"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co, "the buffered TimerFired must cancel the wait for event b immediately") - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"timedout"`, co.GetResult().GetValue()) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_UnconsumedOrphanWarns(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - return nil, ctx.WaitForSingleEvent("go", -1).Await(nil) - })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(99, `"orphan"`), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, 1, cl.warnsContaining("TaskCompleted for id 99")) -} - -func Test_BufferedResolution_UnconsumedOrphanWarnsOnBlockedTurn(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - return nil, ctx.WaitForSingleEvent("go", -1).Await(nil) - })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(99, `"orphan"`), - }) - - assert.Nil(t, completeAction(t, actions), "workflow stays blocked on the external event") - assert.Equal(t, 1, cl.warnsContaining("TaskCompleted for id 99")) -} - -func Test_BufferedResolution_DuplicateAfterResolutionDropped(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - var out string - if err := ctx.CallActivity("act").Await(&out); err != nil { - return nil, err - } - return out, nil - })) - require.NoError(t, r.AddActivityN("act", func(ActivityContext) (any, error) { return nil, nil })) - - actions, cl := runBuffered(t, r, - []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskScheduled(0, "act"), - }, - []*protos.HistoryEvent{ - evTaskCompleted(0, `"first"`), - evTaskCompleted(0, `"second"`), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"first"`, co.GetResult().GetValue(), "the first resolution wins; the duplicate is dropped") - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_LateTaskScheduledAfterDelivery(t *testing.T) { - // Histories produced by the pre-fix stall can contain the orphan - // completion followed by a late TaskScheduled for the same id. The - // retained pending action must match it without a nondeterminism error. - actions, cl := runBuffered(t, waitThenActivityRegistry(t), nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(1, `"injected"`), - evEventRaised("go"), - evTaskScheduled(1, "act"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"injected"`, co.GetResult().GetValue()) - assert.Zero(t, countActions(actions, isScheduleTask)) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_KindMismatchNotDelivered(t *testing.T) { - // A TimerFired for id 1 must not resolve the activity task at id 1: the - // activity is dispatched as before and the unmatched timer resolution - // warns at the end of the turn. - actions, cl := runBuffered(t, waitThenActivityRegistry(t), nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTimerFired(1), - evEventRaised("go"), - }) - - assert.Nil(t, completeAction(t, actions)) - assert.Equal(t, 1, countActions(actions, isScheduleTask), "the activity dispatch is unaffected") - assert.Equal(t, 1, cl.warnsContaining("TimerFired for id 1")) -} - -func Test_BufferedResolution_SuspensionPrecedence(t *testing.T) { - actions, cl := runBuffered(t, waitThenActivityRegistry(t), nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evSuspended(), - evTaskCompleted(1, `"injected"`), - evEventRaised("go"), - evResumed(), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"injected"`, co.GetResult().GetValue()) - assert.Equal(t, 1, countActions(actions, isScheduleTask), emittedNotDispatched) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_FreshContextPerExecution(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - return nil, ctx.WaitForSingleEvent("go", -1).Await(nil) - })) - - _, cl1 := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(99, `"orphan"`), - evEventRaised("go"), - }) - require.Equal(t, 1, cl1.warnsContaining("TaskCompleted for id 99")) - - _, cl2 := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evEventRaised("go"), - }) - assert.Empty(t, cl2.warns, "a fresh execution must not inherit buffered resolutions") -} - -func Test_BufferedResolution_EarlyFailureWithRetryPolicy(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - return nil, ctx.CallActivity("act", WithActivityRetryPolicy(&RetryPolicy{ - MaxAttempts: 3, - InitialRetryInterval: time.Second, - BackoffCoefficient: 2, - })).Await(nil) - })) - require.NoError(t, r.AddActivityN("act", func(ActivityContext) (any, error) { return nil, nil })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskFailed(1, "exec-x"), - evEventRaised("go"), - }) - - // The buffered failure resolves attempt one at scheduling time and the - // retry wrapper immediately arms the backoff timer: the workflow blocks - // on the retry timer instead of completing, the failed attempt's - // ScheduleTask is emitted for the record, and the retry timer action is - // emitted. - assert.Nil(t, completeAction(t, actions)) - assert.Equal(t, 1, countActions(actions, isScheduleTask), emittedNotDispatched) - assert.Equal(t, 1, countActions(actions, func(a *protos.WorkflowAction) bool { - return a.GetCreateTimer() != nil && a.Id == 2 - }), "the retry backoff timer must be armed") - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_FanOutTwoEarlyCompletions(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - t1 := ctx.CallActivity("a") - t2 := ctx.CallActivity("b") - var out1, out2 string - if err := t1.Await(&out1); err != nil { - return nil, err - } - if err := t2.Await(&out2); err != nil { - return nil, err - } - return out1 + out2, nil - })) - require.NoError(t, r.AddActivityN("a", func(ActivityContext) (any, error) { return nil, nil })) - require.NoError(t, r.AddActivityN("b", func(ActivityContext) (any, error) { return nil, nil })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(2, `"two"`), - evTaskCompleted(1, `"one"`), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"onetwo"`, co.GetResult().GetValue()) - assert.Equal(t, 2, countActions(actions, isScheduleTask), emittedNotDispatched) - assert.Empty(t, cl.warns) -} - -func Test_BufferedResolution_TerminatedTurnEmitsOnlyCompletion(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - return nil, ctx.WaitForSingleEvent("go", -1).Await(nil) - })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(7, `"orphan"`), - { - EventId: -1, - Timestamp: timestamppb.Now(), - EventType: &protos.HistoryEvent_ExecutionTerminated{ - ExecutionTerminated: &protos.ExecutionTerminatedEvent{Input: wrapperspb.String(`"stop"`)}, - }, - }, - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_TERMINATED, co.WorkflowStatus) - for _, a := range actions { - assert.NotNil(t, a.GetCompleteWorkflow(), "a terminated turn must emit only the completion action") - } - assert.Equal(t, 1, cl.warnsContaining("TaskCompleted for id 7")) -} - -func Test_BufferedResolution_ContinueAsNewNoCarryover(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - ctx.ContinueAsNew(nil, WithKeepUnprocessedEvents()) - return nil, nil - })) - - actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(5, `"orphan"`), - evEventRaised("go"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW, co.WorkflowStatus) - for _, e := range co.GetCarryoverEvents() { - assert.Nil(t, e.GetTaskCompleted(), "buffered resolutions must not be carried into the next generation") - } - assert.Equal(t, 1, cl.warnsContaining("TaskCompleted for id 5")) -} - -func Test_BufferedResolution_ExecutorSurfacesWarning(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - var out string - if err := ctx.CallActivity("act").Await(&out); err != nil { - return nil, err - } - return out, nil - })) - require.NoError(t, r.AddActivityN("act", func(ActivityContext) (any, error) { return nil, nil })) - - cl := &captureLogger{} - ex := NewTaskExecutorWithLogger(r, cl) - - resp, err := ex.ExecuteWorkflow(t.Context(), "exec-test", nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(1, `"injected"`), - evEventRaised("go"), - }, backend.ExecuteOptions{}) - require.NoError(t, err) - assert.Equal(t, 1, countActions(resp.GetActions(), isScheduleTask), emittedNotDispatched) - assert.Empty(t, cl.warns) - - _, err = ex.ExecuteWorkflow(t.Context(), "exec-test", nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(42, `"orphan"`), - evEventRaised("go"), - }, backend.ExecuteOptions{}) - require.NoError(t, err) - assert.Equal(t, 1, cl.warnsContaining("TaskCompleted for id 42")) -} - - -// Test_BufferedResolution_ExecutorReplayRoundTrip drives the dapr shape of -// the bug end to end: the completion of the first activity reaches the -// workflow before its TaskScheduled was committed, the turn re-emits that -// schedule and dispatches the second activity, and the next turn must replay -// the recorded history without a non-determinism error. -func Test_BufferedResolution_ExecutorReplayRoundTrip(t *testing.T) { - r := NewTaskRegistry() - require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - if err := ctx.WaitForSingleEvent("go", -1).Await(nil); err != nil { - return nil, err - } - var out string - if err := ctx.CallActivity("act").Await(&out); err != nil { - return nil, err - } - if err := ctx.CallActivity("act2").Await(nil); err != nil { - return nil, err - } - return out, nil - })) - require.NoError(t, r.AddActivityN("act", func(ActivityContext) (any, error) { return nil, nil })) - require.NoError(t, r.AddActivityN("act2", func(ActivityContext) (any, error) { return nil, nil })) - ex := NewTaskExecutor(r) - applier := runtimestate.NewApplier("app", "ns") - - state := runtimestate.NewWorkflowRuntimeState("exec-test", nil, nil) - for _, e := range []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(1, `"injected"`), - evEventRaised("go"), - } { - require.NoError(t, runtimestate.AddEvent(state, e)) - } - resp, err := ex.ExecuteWorkflow(t.Context(), "exec-test", state.GetOldEvents(), state.GetNewEvents(), backend.ExecuteOptions{}) - require.NoError(t, err) - assert.Equal(t, 2, countActions(resp.GetActions(), isScheduleTask), emittedNotDispatched) - _, err = applier.Actions(state, resp.GetCustomStatus(), resp.GetActions(), nil, nil) - require.NoError(t, err) - - require.Len(t, state.GetPendingTasks(), 1, "only the unresolved activity is dispatched") - assert.Equal(t, int32(2), state.GetPendingTasks()[0].GetEventId()) - var scheduled, completed bool - for _, e := range state.GetNewEvents() { - scheduled = scheduled || (e.GetTaskScheduled() != nil && e.GetEventId() == 1) - completed = completed || (e.GetTaskCompleted() != nil && e.GetTaskCompleted().GetTaskScheduledId() == 1) - } - assert.True(t, scheduled, "TaskScheduled#1 must be recorded so the completion keeps its match") - assert.True(t, completed, "TaskCompleted#1 must be retained") - - // The next turn replays the committed history with the second - // activity's completion. - history := append(state.GetOldEvents(), state.GetNewEvents()...) - resp, err = ex.ExecuteWorkflow(t.Context(), "exec-test", history, []*protos.HistoryEvent{evTaskCompleted(2, `null`)}, backend.ExecuteOptions{}) - require.NoError(t, err) - co := completeAction(t, resp.GetActions()) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus, co.GetFailureDetails().GetErrorMessage()) - assert.Equal(t, `"injected"`, co.GetResult().GetValue()) - assert.Zero(t, countActions(resp.GetActions(), isScheduleTask)) -} - -func Test_CompletableTask_OnCompletedAfterCompletion(t *testing.T) { - task := newTask(newTestContext(t)) - task.complete([]byte("x")) - fired := false - task.onCompleted(func() { fired = true }) - assert.True(t, fired, "onCompleted on an already completed task must fire immediately") -} - -// Benchmark_ReplaySequentialActivities measures a full replay of a workflow -// with 50 sequential completed activities, the shape dominated by the -// per-event and per-schedule bookkeeping this file's feature adds to. -func Benchmark_ReplaySequentialActivities(b *testing.B) { - const n = 50 - r := NewTaskRegistry() - if err := r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { - for range n { - if err := ctx.CallActivity("act").Await(nil); err != nil { - return nil, err - } - } - return "done", nil - }); err != nil { - b.Fatal(err) - } - if err := r.AddActivityN("act", func(ActivityContext) (any, error) { return nil, nil }); err != nil { - b.Fatal(err) - } - - events := make([]*protos.HistoryEvent, 0, 2*n+1) - events = append(events, evExecutionStarted("wf")) - for i := range int32(n) { - events = append(events, evTaskScheduled(i, "act"), evTaskCompleted(i, `null`)) - } - - b.ReportAllocs() - for b.Loop() { - ctx := NewWorkflowContext(r, "bench", events, nil) - if actions := ctx.start(); len(actions) != 1 { - b.Fatalf("expected 1 action, got %d", len(actions)) - } - } -} - -func Test_BufferedResolution_KindGuardOnOccupiedID(t *testing.T) { - // A TaskCompleted whose id is occupied by a pending TIMER (here the - // synthetic external event timer at id 0) must buffer rather than - // complete the timer, which would wrongly cancel the external event wait. - actions, cl := runBuffered(t, waitThenActivityRegistry(t), nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(0, `"x"`), - evEventRaised("go"), - }) - - assert.Nil(t, completeAction(t, actions), - "the wait must complete via the event, not fail via a wrongly cancelled timer") - assert.Equal(t, 1, countActions(actions, isScheduleTask), - "the activity dispatch proceeds normally") - assert.Equal(t, 1, cl.warnsContaining("TaskCompleted for id 0")) -} - -func Test_BufferedResolution_PreSyntheticTimerMigration(t *testing.T) { - // A history produced before WaitForSingleEvent emitted its synthetic - // timer numbers the activity as id 0, which the current replay assigns - // to the synthetic timer. The early completion for id 0 must buffer past - // the timer (kind mismatch), and when the historical TaskScheduled(0) - // drops the optional timer and shifts the activity onto id 0, the - // buffered completion must be delivered to it. - actions, cl := runBuffered(t, waitThenActivityRegistry(t), nil, []*protos.HistoryEvent{ - evExecutionStarted("wf"), - evTaskCompleted(0, `"migrated"`), - evEventRaised("go"), - evTaskScheduled(0, "act"), - }) - - co := completeAction(t, actions) - require.NotNil(t, co) - assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) - assert.Equal(t, `"migrated"`, co.GetResult().GetValue()) - assert.Zero(t, countActions(actions, isScheduleTask)) - assert.Empty(t, cl.warns) -} diff --git a/task/task.go b/task/task.go index 78bdd927..86e2a78b 100644 --- a/task/task.go +++ b/task/task.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/dapr/durabletask-go/api/protos" - "github.com/dapr/durabletask-go/backend/runtimestate/dedup" ) // ErrTaskBlocked is not an error, but rather a control flow signal indicating that a workflow @@ -31,12 +30,6 @@ type completableTask struct { failureDetails *protos.TaskFailureDetails completedCallback func() taskExecutionId string - // kind is the resolution correlator family this task belongs to when it - // is registered in pendingTasks (task, timer or child). A resolution - // event only completes a pending entry of its own kind; anything else is - // buffered. Zero (KindNone) for tasks never held in pendingTasks, such - // as external event wait tasks. - kind dedup.Kind } func newTask(ctx *WorkflowContext) *completableTask { @@ -89,13 +82,6 @@ func (t *completableTask) TaskExecutionId() string { } func (t *completableTask) onCompleted(callback func()) { - // A task can already be completed at registration time when a buffered - // early resolution was delivered as the task was scheduled; fire the - // callback immediately so completion side effects are not lost. - if t.isCompleted { - callback() - return - } t.completedCallback = callback } From 9af17f0ddcf3694e592f09a5c630b49848464b55 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Mon, 5 Oct 2026 15:05:08 +0100 Subject: [PATCH 2/2] Review comments Signed-off-by: joshvanl --- task/early_resolution_test.go | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/task/early_resolution_test.go b/task/early_resolution_test.go index d4f08e5e..e40454e8 100644 --- a/task/early_resolution_test.go +++ b/task/early_resolution_test.go @@ -64,12 +64,18 @@ func Test_UnmatchedResolutionIsIgnored(t *testing.T) { early := &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_TaskCompleted{ TaskCompleted: &protos.TaskCompletedEvent{TaskScheduledId: 1, Result: wrapperspb.String(`"early"`)}, }} - run := func(newEvents ...*protos.HistoryEvent) ([]*protos.WorkflowAction, *warnLogger) { + scheduledEvent := &protos.HistoryEvent{EventId: 1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_TaskScheduled{ + TaskScheduled: &protos.TaskScheduledEvent{Name: "a"}, + }} + replay := func(oldEvents, newEvents []*protos.HistoryEvent) ([]*protos.WorkflowAction, *warnLogger) { l := new(warnLogger) - ctx := NewWorkflowContext(r, "id", []*protos.HistoryEvent{started}, newEvents) + ctx := NewWorkflowContext(r, "id", oldEvents, newEvents) ctx.SetLogger(l) return ctx.start(), l } + run := func(newEvents ...*protos.HistoryEvent) ([]*protos.WorkflowAction, *warnLogger) { + return replay([]*protos.HistoryEvent{started}, newEvents) + } scheduled := func(actions []*protos.WorkflowAction) (n int) { for _, a := range actions { if a.GetScheduleTask() != nil { @@ -102,4 +108,15 @@ func Test_UnmatchedResolutionIsIgnored(t *testing.T) { assert.Equal(t, `"early"`, completed(actions).GetResult().GetValue()) assert.Empty(t, l.warns) }) + + t.Run("persisted after its scheduling it resolves the step on replay", func(t *testing.T) { + nudge := &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_EventRaised{ + EventRaised: &protos.EventRaisedEvent{Name: "nudge"}, + }} + actions, l := replay([]*protos.HistoryEvent{started, goEvent, scheduledEvent, early}, []*protos.HistoryEvent{nudge}) + assert.Equal(t, 0, scheduled(actions), "the schedule is already in history") + require.NotNil(t, completed(actions)) + assert.Equal(t, `"early"`, completed(actions).GetResult().GetValue()) + assert.Empty(t, l.warns) + }) }