diff --git a/task/orchestrator.go b/task/orchestrator.go index fe2ae4c..9757223 100644 --- a/task/orchestrator.go +++ b/task/orchestrator.go @@ -2,6 +2,7 @@ package task import ( "container/list" + "errors" "fmt" "math" "reflect" @@ -246,6 +247,26 @@ func (ctx *WorkflowContext) processNextEvent() (bool, error) { return true, nil } +// awaitUntil processes history events one at a time until [done] reports true, returning nil once +// it does. It returns any error processNextEvent reports along the way. If history runs out before +// [done] is satisfied, awaitUntil panics with ErrTaskBlocked, the same control-flow signal Await +// uses: this is normal for workflow functions, which must never recover from it. +func (ctx *WorkflowContext) awaitUntil(done func() bool) error { + for !done() { + ok, err := ctx.processNextEvent() + if err != nil { + return err + } + if !ok { + // TODO: Need a rule about using "defer" in workflows because planned panics will invoke them unexpectedly + // TODO: @joshvanl: remove panic- panic is something that should + // _never_ be called in normal operation. + panic(ErrTaskBlocked) + } + } + return nil +} + func (ctx *WorkflowContext) getNextHistoryEvent() (*protos.HistoryEvent, bool) { var historyList []*protos.HistoryEvent index := ctx.historyIndex @@ -375,9 +396,9 @@ func (ctx *WorkflowContext) CallActivity(activity interface{}, opts ...CallActiv activityName := helpers.GetTaskFunctionName(activity) if options.retryPolicy != nil { - return ctx.internalScheduleTaskWithRetries(activityName+"-retry", ctx.CurrentTimeUtc, func(taskExecutionId string, _ bool) Task { + return ctx.internalScheduleTaskWithRetries(activityName+"-retry", ctx.CurrentTimeUtc, func(taskExecutionId string, _ bool) *completableTask { return ctx.internalScheduleActivity(activityName, taskExecutionId, options) - }, *options.retryPolicy, 0, uuid.NewString(), func(a *protos.CreateTimerAction, execID string) { + }, *options.retryPolicy, func(a *protos.CreateTimerAction, execID string) { a.Origin = &protos.CreateTimerAction_ActivityRetry{ ActivityRetry: &protos.TimerOriginActivityRetry{ TaskExecutionId: execID, @@ -389,7 +410,7 @@ func (ctx *WorkflowContext) CallActivity(activity interface{}, opts ...CallActiv return ctx.internalScheduleActivity(activityName, uuid.NewString(), options) } -func (ctx *WorkflowContext) internalScheduleActivity(activityName, taskExecutionId string, options *callActivityOptions) Task { +func (ctx *WorkflowContext) internalScheduleActivity(activityName, taskExecutionId string, options *callActivityOptions) *completableTask { scheduleTaskAction := &protos.WorkflowAction{ Id: ctx.getNextSequenceNumber(), WorkflowActionType: &protos.WorkflowAction_ScheduleTask{ @@ -441,7 +462,7 @@ func (ctx *WorkflowContext) CallChildWorkflow(workflow interface{}, opts ...Chil if firstInstanceID == "" { firstInstanceID = helpers.GenerateChildWorkflowInstanceID(string(ctx.ID), ctx.sequenceNumber) } - return ctx.internalScheduleTaskWithRetries(workflowName+"-retry", ctx.CurrentTimeUtc, func(_ string, isRetry bool) Task { + return ctx.internalScheduleTaskWithRetries(workflowName+"-retry", ctx.CurrentTimeUtc, func(_ string, isRetry bool) *completableTask { // On retry attempts (2nd onward) carry the first attempt's instance // ID so the runtime can record it on the resulting // ChildWorkflowInstanceCreatedEvent and consumers can correlate the @@ -450,7 +471,7 @@ func (ctx *WorkflowContext) CallChildWorkflow(workflow interface{}, opts ...Chil return ctx.internalCallChildWorkflow(workflowName, options, &firstInstanceID) } return ctx.internalCallChildWorkflow(workflowName, options, nil) - }, *options.retryPolicy, 0, uuid.NewString(), func(a *protos.CreateTimerAction, _ string) { + }, *options.retryPolicy, func(a *protos.CreateTimerAction, _ string) { a.Origin = &protos.CreateTimerAction_ChildWorkflowRetry{ ChildWorkflowRetry: &protos.TimerOriginChildWorkflowRetry{ InstanceId: firstInstanceID, @@ -466,7 +487,7 @@ func (ctx *WorkflowContext) CallChildWorkflow(workflow interface{}, opts ...Chil // when non-nil, is the instance ID of the first attempt in this child's retry // chain; it is set only on retry attempts (2nd onward) and is persisted by the // runtime onto the ChildWorkflowInstanceCreatedEvent for retry-chain correlation. -func (ctx *WorkflowContext) internalCallChildWorkflow(workflowName string, options *callChildWorkflowOptions, retryParentInstanceID *string) Task { +func (ctx *WorkflowContext) internalCallChildWorkflow(workflowName string, options *callChildWorkflowOptions, retryParentInstanceID *string) *completableTask { createChildWorkflow := &protos.CreateChildWorkflowAction{ Name: workflowName, Input: options.rawInput, @@ -501,40 +522,91 @@ func (ctx *WorkflowContext) internalCallChildWorkflow(workflowName string, optio return task } -func (ctx *WorkflowContext) internalScheduleTaskWithRetries(name string, initialAttempt time.Time, schedule func(taskExecutionId string, isRetry bool) Task, policy RetryPolicy, retryCount int, taskExecutionId string, setTimerOrigin func(*protos.CreateTimerAction, string)) Task { - return &taskWrapper{ - delegate: schedule(taskExecutionId, retryCount > 0), - onAwaitResult: func(v any, taskExecutionId string, err error) error { - if err == nil { - return nil - } +// internalScheduleTaskWithRetries schedules [schedule]'s first attempt immediately and returns a +// plain *completableTask for the whole retry chain. The chain advances -- deciding whether a failed +// attempt retries, arming its backoff timer, scheduling the next attempt -- only when the returned +// task is polled (by Await or Select), never as a side effect of history alone: an attempt nobody +// observes stays failed and retries no further. A decode error from Await(v) on a successful +// attempt is returned as-is and never triggers a retry, and polling a chain that has already +// completed or failed (a second Await, or a Select that still includes it) is a no-op that never +// starts another chain. Both differ from the Await-driven design this replaced, and histories it +// recorded in either state do not replay. +func (ctx *WorkflowContext) internalScheduleTaskWithRetries(name string, initialAttempt time.Time, schedule func(taskExecutionId string, isRetry bool) *completableTask, policy RetryPolicy, setTimerOrigin func(*protos.CreateTimerAction, string)) Task { + outer := newTask(ctx) + taskExecutionId := uuid.NewString() + current := schedule(taskExecutionId, false) + retryCount := 0 + var timer *completableTask + + // giveUp completes outer with the current attempt's terminal state. A canceled attempt has no + // failureDetails, so it must propagate as a cancellation rather than as a nil-error success. + giveUp := func() { + if current.isCanceled { + outer.cancel() + return + } + outer.fail(current.failureDetails) + } - if retryCount+1 >= policy.MaxAttempts { - // next try will exceed the max attempts, dont continue - return err + outer.advance = func() { + if outer.isCompleted { + return + } + for { + if timer != nil { + if !timer.isCompleted { + return + } + // Observing the fired timer is itself what schedules the next attempt, and its + // action's sequence number is assigned here -- at the same point Await/Select's + // poll notices the timer fired -- matching where the previous, Await-driven retry + // design allocated it, so replay stays compatible with histories recorded before + // retries became pollable. + current = schedule(taskExecutionId, true) + timer = nil + retryCount++ + continue } - nextDelay := computeNextDelay(ctx.CurrentTimeUtc, policy, retryCount, initialAttempt, err) - if nextDelay == 0 { - return err + if !current.isCompleted { + return } - task, action := ctx.createTimerInternal(&name, nextDelay) - setTimerOrigin(action, taskExecutionId) - timerErr := task.Await(nil) - if timerErr != nil { - // TODO use errors.Join when updating golang - return fmt.Errorf("%v %w", timerErr, err) + + // TaskExecutionId reflects what's recorded on this attempt's TaskScheduled/ + // TaskCompleted/TaskFailed history events, which is what must be threaded into the + // next attempt and its backoff timer's origin so replay stays deterministic. A + // canceled attempt has no execution id of its own (nothing cancels activity/child + // tasks today, so this is currently unreachable, but cancellation must not erase the + // chain's carried id for whichever future attempt comes next). + if id := current.TaskExecutionId(); id != "" { + taskExecutionId = id } - t := ctx.internalScheduleTaskWithRetries(name, initialAttempt, schedule, policy, retryCount+1, taskExecutionId, setTimerOrigin) - err = t.Await(v) + err := current.completionError() if err == nil { - return nil + outer.complete(current.rawResult) + return } + outer.taskExecutionId = taskExecutionId - return err - }, + if retryCount+1 >= policy.MaxAttempts { + giveUp() + return + } + nextDelay := computeNextDelay(ctx.CurrentTimeUtc, policy, retryCount, initialAttempt, err) + if nextDelay == 0 { + giveUp() + return + } + + var action *protos.CreateTimerAction + timer, action = ctx.createTimerInternal(&name, nextDelay) + setTimerOrigin(action, taskExecutionId) + return + } } + + return outer } func computeNextDelay(currentTimeUtc time.Time, policy RetryPolicy, attempt int, firstAttempt time.Time, err error) time.Duration { @@ -627,6 +699,16 @@ func (ctx *WorkflowContext) createExternalEventTimerInternal(eventName string, f // Workflows can wait for the same event name multiple times, so waiting for multiple events with the same name // is allowed. Each event received by an workflow will complete just one task returned by this method. // +// A task returned by this method that is passed to [WorkflowContext.Select] but loses -- another +// candidate completes first -- is not retired: it remains queued for its event name. If the +// workflow then calls WaitForSingleEvent again for that same name (for example, re-selecting in a +// loop after handling the winner) the two tasks queue in call order, and the next matching event +// completes whichever of them is oldest, not necessarily the one just created. A loop over Select +// that discards losing tasks between iterations can therefore end up permanently waiting on a task +// no later iteration still holds a reference to. To race the same event name across iterations +// safely, carry every losing task forward into the next Select call instead of creating a new one +// for a name still pending. +// // Note that event names are case-insensitive. func (ctx *WorkflowContext) WaitForSingleEvent(eventName string, timeout time.Duration) Task { task := newTask(ctx) @@ -681,6 +763,77 @@ func (ctx *WorkflowContext) WaitForSingleEvent(eventName string, timeout time.Du return task } +// Select blocks until the first of the given [tasks] completes and returns its index. Once Select +// returns, callers should call Await on the task at the returned index to obtain its result or +// error; the remaining tasks are left pending and may still be selected or awaited later (for +// example, in a loop that repeatedly selects over the tasks that have not yet completed) -- doing so +// is required, not optional, for a losing [WorkflowContext.WaitForSingleEvent] task: see its doc +// comment for why discarding one instead of carrying it forward can hang a later Select on the same +// event name. +// +// Select polls the given tasks' completion state after each history event is processed, so if more +// than one of them is found completed at the same time -- whether because they were already +// completed before Select was called, or because a single history event completed several of them +// at once (possible when a batch of events buffered during a suspended execution is replayed) -- +// the one with the lowest index wins; this is the only tie-break rule Select ever applies. When two +// or more of the given tasks are retry-configured and both become newly failed within the same +// poll, their backoff timers are still allocated in argument order, not simultaneously. +// +// Select requires at least one task and returns an error if no tasks are given, if any task is nil, +// or if any task was not obtained from this same WorkflowContext (e.g. via CallActivity, CreateTimer, +// or WaitForSingleEvent, with or without a retry policy) -- a task from a different WorkflowContext +// can never complete from this context's point of view, which would otherwise block the workflow +// indefinitely with no diagnostic. A Task not created by a WorkflowContext method is rejected with +// ErrTaskNotSelectable. +// +// Like Await, Select may panic with ErrTaskBlocked as the panic value when none of the tasks have +// completed and there is no further history to process. This is normal control flow for workflow +// functions, which must never recover from such panics. +func (ctx *WorkflowContext) Select(tasks ...Task) (int, error) { + if len(tasks) == 0 { + return -1, errors.New("Select requires at least one task") + } + + completable := make([]*completableTask, len(tasks)) + for i, t := range tasks { + if t == nil { + return -1, fmt.Errorf("task at index %d is nil", i) + } + // Every Task this package hands out, including retry-configured ones (see + // internalScheduleTaskWithRetries), is backed by *completableTask; this only fails for a + // Task implementation from outside the package. + ct, ok := t.(*completableTask) + if !ok { + return -1, fmt.Errorf("task at index %d: %w", i, ErrTaskNotSelectable) + } + if ct.workflowCtx != ctx { + return -1, fmt.Errorf("task at index %d belongs to a different WorkflowContext", i) + } + completable[i] = ct + } + + // Every candidate is polled on every check, not just until the first completed one is found: + // skipping a later candidate's pollCompleted call would skip its advance too (see + // internalScheduleTaskWithRetries), leaving a retry candidate's backoff timer allocated on + // whatever later poll happens to reach it instead of here -- making its action sequence number + // depend on which candidate happened to win, rather than only on argument order and history, the + // same way plain tasks already behave. A single history event can also complete more than one + // candidate at once (e.g. onExecutionResumed replaying a batch buffered during a suspension), so + // ties still go to the lowest index. + winner := -1 + if err := ctx.awaitUntil(func() bool { + for i, ct := range completable { + if ct.pollCompleted() && winner < 0 { + winner = i + } + } + return winner >= 0 + }); err != nil { + return -1, err + } + return winner, nil +} + func (ctx *WorkflowContext) ContinueAsNew(newInput any, options ...ContinueAsNewOption) { ctx.continuedAsNew = true ctx.continuedAsNewInput = newInput @@ -805,9 +958,12 @@ func (ctx *WorkflowContext) onTaskFailed(tf *protos.TaskFailedEvent) error { } delete(ctx.pendingTasks, taskID) + // Set before fail() so TaskExecutionId() already reflects this event's value for anything that + // observes the task's completion afterward (e.g. a retry chain's advance, polled later by + // Await/Select). + task.taskExecutionId = tf.TaskExecutionId // completing a task will resume the corresponding Await() call task.fail(tf.FailureDetails) - task.taskExecutionId = tf.TaskExecutionId return nil } diff --git a/task/orchestrator_buffered_test.go b/task/orchestrator_buffered_test.go new file mode 100644 index 0000000..f0d7d9c --- /dev/null +++ b/task/orchestrator_buffered_test.go @@ -0,0 +1,673 @@ +/* +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" + "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" +) + +// captureLogger records formatted log lines per level for assertions. +type captureLogger struct { + mu sync.Mutex + warns []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) {} +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 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 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 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()}, + }, + } +} + +// evRetryTimerCreated is the TimerCreated event the backend records for a retry backoff timer +// action, as emitted by both the Await-driven retry design and the pollable one. +func evRetryTimerCreated(id int32, execID string) *protos.HistoryEvent { + return &protos.HistoryEvent{ + EventId: id, + Timestamp: timestamppb.Now(), + EventType: &protos.HistoryEvent_TimerCreated{ + TimerCreated: &protos.TimerCreatedEvent{ + FireAt: timestamppb.Now(), + Origin: &protos.TimerCreatedEvent_ActivityRetry{ + ActivityRetry: &protos.TimerOriginActivityRetry{TaskExecutionId: execID}, + }, + }, + }, + } +} + +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 } + +func Test_RetryPolicy_UnobservedAttemptNeverRetries(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + _ = ctx.CallActivity("A", WithActivityRetryPolicy(&RetryPolicy{ + MaxAttempts: 3, + InitialRetryInterval: time.Second, + BackoffCoefficient: 2, + })) + b := ctx.CallActivity("B") + return nil, b.Await(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"), + evTaskFailed(0, "exec-a"), + evTaskCompleted(1, `null`), + }) + + co := completeAction(t, actions) + require.NotNil(t, co, "the workflow must complete once its only awaited task (B) resolves") + assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) + assert.Zero(t, countActions(actions, func(a *protos.WorkflowAction) bool { return a.GetCreateTimer() != nil }), + "A's failure was never observed, so no retry backoff timer may be armed for it") + assert.Empty(t, cl.warns) +} + +// Test_RetryPolicy_TimerSequenceIdMatchesAwaitObservationPoint is a regression test for replay +// compatibility: the retry backoff timer's action id (sequence number) must match where the +// workflow function actually observes the failed attempt by calling Await, not where the attempt +// happens to fail in history. A workflow that schedules C between observing A's failure and +// awaiting A again must see C take the sequence id that would otherwise go to A's retry timer, and +// the retry timer take the next id after that -- exactly as if the retry decision were made inline +// in Await itself, which is what histories recorded before retries became pollable assume. +func Test_RetryPolicy_TimerSequenceIdMatchesAwaitObservationPoint(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(&RetryPolicy{ + MaxAttempts: 3, + InitialRetryInterval: time.Second, + BackoffCoefficient: 2, + })) // id 0 + b := ctx.CallActivity("B") // id 1 + if err := b.Await(nil); err != nil { + return nil, err + } + c := ctx.CallActivity("C") // id 2: scheduled before A's failure is ever observed + // A's retry timer is armed here, once Await actually polls A. The call then blocks by + // panicking with ErrTaskBlocked (history has run out), so nothing after it runs. + _ = a.Await(nil) + return nil, c.Await(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 })) + require.NoError(t, r.AddActivityN("C", func(ActivityContext) (any, error) { return nil, nil })) + + actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ + evExecutionStarted("wf"), + evTaskFailed(0, "exec-a"), + evTaskCompleted(1, `null`), + }) + + require.Equal(t, 3, countActions(actions, isScheduleTask), "A's first attempt, B, and C are each scheduled on this turn") + require.True(t, countActions(actions, func(a *protos.WorkflowAction) bool { + return a.GetScheduleTask().GetName() == "C" && a.Id == 2 + }) == 1, "C must take sequence id 2, the id it would get if A's retry were still decided inside Await rather than eagerly on TaskFailed") + require.True(t, countActions(actions, func(a *protos.WorkflowAction) bool { + return a.GetCreateTimer() != nil && a.Id == 3 + }) == 1, "A's retry backoff timer must take the next id after C, since Await only observes A's failure after C is scheduled") + assert.Empty(t, cl.warns) +} + +// Test_RetryPolicy_SecondAwaitDoesNotReRunChain is a regression test: once a retry-configured +// task's outer task has failed, polling it again (a second Await, or a Select that still includes +// it) must be a no-op -- it must not consult policy.Handle again, and it must not arm another +// backoff timer for a task the caller has already observed fail. MaxAttempts is left high enough +// that retries are nominally still available; policy.Handle itself declines the retry, which is the +// only way to reach the failure path without exhausting MaxAttempts, so a bug that re-runs this +// branch on every subsequent poll is caught by handleCalls incrementing again, not masked by +// retryCount already being at its limit. +func Test_RetryPolicy_SecondAwaitDoesNotReRunChain(t *testing.T) { + r := NewTaskRegistry() + var handleCalls int + require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + policy := &RetryPolicy{ + MaxAttempts: 3, + InitialRetryInterval: time.Second, + BackoffCoefficient: 2, + Handle: func(err error) bool { + handleCalls++ + return false + }, + } + flaky := ctx.CallActivity("Flaky", WithActivityRetryPolicy(policy)) // id 0 + err1 := flaky.Await(nil) + // The chain already failed (Handle declined the only attempt, resolved from buffered + // history below): this second Await must be a pure no-op that returns the same error. + err2 := flaky.Await(nil) + if err1 == nil || err2 == nil || err1.Error() != err2.Error() { + return nil, fmt.Errorf("expected both Awaits to return the same error, got %v and %v", err1, err2) + } + return nil, nil + })) + require.NoError(t, r.AddActivityN("Flaky", func(ActivityContext) (any, error) { return nil, nil })) + + actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ + evExecutionStarted("wf"), + evTaskFailed(0, "exec-1"), + }) + + co := completeAction(t, actions) + require.NotNil(t, co, "the workflow must complete once the retry chain fails and is observed twice") + assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) + assert.Zero(t, countActions(actions, func(a *protos.WorkflowAction) bool { return a.GetCreateTimer() != nil }), + "Handle declined the only attempt; no backoff timer may ever be armed, including on the second Await") + assert.Equal(t, 1, handleCalls, "policy.Handle must run exactly once, for the one real failure, not again on the second Await") + assert.Empty(t, cl.warns) +} + +// Test_RetryPolicy_TaskExecutionIdReflectsFailedAttempt pins the TaskExecutionId contract of a +// retry-configured task: it is the id recorded on the most recently observed failed attempt, and +// empty while no attempt has failed. The Await-driven design delegated to the first attempt, which +// only ever learned its id from a TaskFailed event, so this is what callers saw before retries +// became pollable. +func Test_RetryPolicy_TaskExecutionIdReflectsFailedAttempt(t *testing.T) { + policy := func(maxAttempts int) *RetryPolicy { + return &RetryPolicy{MaxAttempts: maxAttempts, InitialRetryInterval: time.Second, BackoffCoefficient: 2} + } + awaitThenReport := func(maxAttempts int) func(ctx *WorkflowContext) (any, error) { + return func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(policy(maxAttempts))) + _ = a.Await(nil) + return a.TaskExecutionId(), nil + } + } + + for name, tc := range map[string]struct { + wf func(ctx *WorkflowContext) (any, error) + history []*protos.HistoryEvent + want string + }{ + "retried then succeeded": { + wf: awaitThenReport(3), + history: []*protos.HistoryEvent{ + evTaskScheduled(0, "A"), evTaskFailed(0, "exec-1"), evRetryTimerCreated(1, "exec-1"), + evTimerFired(1), evTaskScheduled(2, "A"), evTaskCompleted(2, `null`), + }, + want: `"exec-1"`, + }, + "retries exhausted": { + wf: awaitThenReport(2), + history: []*protos.HistoryEvent{ + evTaskScheduled(0, "A"), evTaskFailed(0, "exec-1"), evRetryTimerCreated(1, "exec-1"), + evTimerFired(1), evTaskScheduled(2, "A"), evTaskFailed(2, "exec-1"), + }, + want: `"exec-1"`, + }, + "first attempt succeeded": { + wf: awaitThenReport(3), + history: []*protos.HistoryEvent{evTaskScheduled(0, "A"), evTaskCompleted(0, `null`)}, + want: `""`, + }, + "observed through Select": { + wf: func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(policy(3))) + b := ctx.CallActivity("B") + if i, err := ctx.Select(a, b); err != nil || i != 0 { + return nil, fmt.Errorf("expected A (index 0) to win, got %d, %v", i, err) + } + return a.TaskExecutionId(), nil + }, + history: []*protos.HistoryEvent{ + evTaskScheduled(0, "A"), evTaskScheduled(1, "B"), evTaskFailed(0, "exec-1"), + evRetryTimerCreated(2, "exec-1"), evTimerFired(2), evTaskScheduled(3, "A"), evTaskCompleted(3, `null`), + }, + want: `"exec-1"`, + }, + } { + t.Run(name, func(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddWorkflowN("wf", tc.wf)) + 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, append([]*protos.HistoryEvent{evExecutionStarted("wf")}, tc.history...)) + + co := completeAction(t, actions) + require.NotNil(t, co) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus, co.GetFailureDetails().GetErrorMessage()) + assert.Equal(t, tc.want, co.GetResult().GetValue()) + assert.Empty(t, cl.warns) + }) + } +} + +// Test_RetryPolicy_CanceledAttemptPropagatesCancel guards a latent path: nothing cancels an +// activity or child task today, but a canceled attempt has nil failureDetails, so if the chain +// gave up with outer.fail(nil) the caller would see a nil-error success. The attempt is canceled +// directly on the pending task here, since no history event can do it. +func Test_RetryPolicy_CanceledAttemptPropagatesCancel(t *testing.T) { + t.Run("give up surfaces ErrTaskCanceled", func(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(&RetryPolicy{MaxAttempts: 1, InitialRetryInterval: time.Second})) + ctx.pendingTasks[0].cancel() + err := a.Await(nil) + return errors.Is(err, ErrTaskCanceled), nil + })) + require.NoError(t, r.AddActivityN("A", func(ActivityContext) (any, error) { return nil, nil })) + + actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{evExecutionStarted("wf")}) + + co := completeAction(t, actions) + require.NotNil(t, co) + assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) + assert.Equal(t, `true`, co.GetResult().GetValue(), "Await must return ErrTaskCanceled, not nil") + assert.Empty(t, cl.warns) + }) + + t.Run("retries remaining consults Handle with ErrTaskCanceled", func(t *testing.T) { + r := NewTaskRegistry() + var handled error + require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(&RetryPolicy{ + MaxAttempts: 3, + InitialRetryInterval: time.Second, + BackoffCoefficient: 2, + Handle: func(err error) bool { handled = err; return true }, + })) + ctx.pendingTasks[0].cancel() + return nil, a.Await(nil) + })) + require.NoError(t, r.AddActivityN("A", func(ActivityContext) (any, error) { return nil, nil })) + + actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{evExecutionStarted("wf")}) + + assert.Nil(t, completeAction(t, actions), "the workflow must block on the backoff timer") + assert.ErrorIs(t, handled, ErrTaskCanceled) + var timerExecID string + for _, act := range actions { + if ct := act.GetCreateTimer(); ct != nil { + timerExecID = ct.GetActivityRetry().GetTaskExecutionId() + } + } + assert.Equal(t, 1, countActions(actions, func(a *protos.WorkflowAction) bool { return a.GetCreateTimer() != nil })) + assert.NotEmpty(t, timerExecID, + "a canceled attempt has no execution id of its own; the retry timer must still carry the chain's real id, not overwrite it with an empty one") + assert.Empty(t, cl.warns) + }) +} + +// Test_Select_PollsEveryCandidateRegardlessOfWinner is a regression test: Select must poll every +// candidate on every check, not stop at the first one found completed. A candidate Select never +// polls because an earlier one already won never gets its own advance run (see +// internalScheduleTaskWithRetries), so a retry candidate's backoff timer is left for whenever +// something later happens to poll it again -- here, scheduling D right after Select -- instead of +// being allocated during the Select call itself. That makes the timer's action sequence number land +// after D's rather than before it, breaking replay for any history recorded before this fix. +// +// The shape here is load-bearing and deliberate: c.Await(nil) is what lets A's failure and B's +// completion land from already-present history before Select is ever called, and D is what gives +// the fix something to prove its ordering against. A bare comparison of Select(a, b) vs Select(b, a) +// with no intervening task would NOT catch this bug -- awaitUntil's very first poll runs before any +// history event is processed, so on that first call neither candidate is resolved yet regardless of +// argument order, and the short-circuit never gets a chance to matter. +func Test_Select_PollsEveryCandidateRegardlessOfWinner(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(&RetryPolicy{ // id 0 + MaxAttempts: 3, + InitialRetryInterval: time.Second, + BackoffCoefficient: 2, + })) + b := ctx.CallActivity("B") // id 1 + c := ctx.CallActivity("C") // id 2 + // A fails and B completes while waiting here; neither a nor b (the retry chain's outer + // task and the plain task) has been polled yet. + if err := c.Await(nil); err != nil { + return nil, err + } + + // B wins at once. If Select stopped polling once it found B's winning completion, A would + // never be polled this call, so its backoff timer would not be allocated here. + winner, err := ctx.Select(b, a) + if err != nil { + return nil, err + } + if winner != 0 { + return nil, fmt.Errorf("expected B (index 0) to win Select(b, a), got index %d", winner) + } + + if err := ctx.CallActivity("D").Await(nil); err != nil { // must get sequence id 4, after A's timer + return nil, err + } + return nil, a.Await(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 })) + require.NoError(t, r.AddActivityN("C", func(ActivityContext) (any, error) { return nil, nil })) + require.NoError(t, r.AddActivityN("D", func(ActivityContext) (any, error) { return nil, nil })) + + actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ + evExecutionStarted("wf"), + evTaskFailed(0, "exec-a"), + evTaskCompleted(1, `null`), + evTaskCompleted(2, `null`), + }) + + require.Equal(t, 1, countActions(actions, func(a *protos.WorkflowAction) bool { + return a.GetCreateTimer() != nil && a.Id == 3 + }), "A's retry timer must be allocated during the Select call that observed its failure (id 3), before D") + require.Equal(t, 1, countActions(actions, func(act *protos.WorkflowAction) bool { + return act.GetScheduleTask().GetName() == "D" && act.Id == 4 + }), "D must come after A's retry timer, not before it") + assert.Empty(t, cl.warns) +} + +// Test_Select_RejectsNilTask verifies that Select reports a nil task explicitly, rather than +// misattributing it to ErrTaskNotSelectable, the error for an unrelated, unsupported Task +// implementation. +func Test_Select_RejectsNilTask(t *testing.T) { + ctx := newTestContext(t) + other := newTask(ctx) + + _, err := ctx.Select(other, nil) + require.EqualError(t, err, "task at index 1 is nil") +} + +// Test_Select_RejectsTaskFromDifferentContext is a regression test: a *completableTask obtained +// from a different WorkflowContext passes the type assertion in Select's validation loop, so +// without this check Select would poll a task that can never complete from this context's point of +// view (its workflow execution belongs to a different ctx, whose history this ctx never processes), +// blocking the workflow indefinitely with no diagnostic. +func Test_Select_RejectsTaskFromDifferentContext(t *testing.T) { + ctx := newTestContext(t) + other := newTestContext(t) + + own := newTask(ctx) + foreign := newTask(other) + + _, err := ctx.Select(own, foreign) + require.EqualError(t, err, "task at index 1 belongs to a different WorkflowContext") +} + +// Test_Select_DoesNotRegisterOnLosingTasks confirms Select's polling implementation never touches +// a losing candidate's completedCallback: repeatedly Selecting a still-pending task against an +// already-completed one must leave the pending task's callback field untouched (nil), since Select +// only reads isCompleted -- it never calls onCompleted. +func Test_Select_DoesNotRegisterOnLosingTasks(t *testing.T) { + ctx := newTestContext(t) + + done := newTask(ctx) + done.complete([]byte(`null`)) + + pending := newTask(ctx) + + const iterations = 5 + for i := 0; i < iterations; i++ { + winner, err := ctx.Select(pending, done) + require.NoError(t, err) + require.Equal(t, 1, winner) + require.Nil(t, pending.completedCallback, + "Select must never register a callback on a task it only polls") + } +} + +// Test_Select_TieBreak_LowestIndexWinsAcrossSuspendedBatch is a regression test for Select's +// tie-break rule in the one scenario where it's actually observable: several candidates completing +// within a single processNextEvent call. onExecutionResumed replays a whole batch of events +// buffered during a suspension in one such call, so if the event for the higher-index candidate +// (EventB, index 1) appears in that batch before the event for the lower-index one (EventA, index +// 0), Select must still return index 0 -- history order must not win over argument order. +func Test_Select_TieBreak_LowestIndexWinsAcrossSuspendedBatch(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + taskA := ctx.WaitForSingleEvent("EventA", -1) + taskB := ctx.WaitForSingleEvent("EventB", -1) + + winner, err := ctx.Select(taskA, taskB) + if err != nil { + return nil, err + } + return winner, nil + })) + + actions, cl := runBuffered(t, r, nil, []*protos.HistoryEvent{ + evExecutionStarted("wf"), + evSuspended(), + // EventB (Select index 1) is raised first in the suspended batch; EventA (Select index 0) + // second. If Select's tie-break preferred history order, it would return 1 here. + evEventRaised("EventB"), + evEventRaised("EventA"), + evResumed(), + }) + + co := completeAction(t, actions) + require.NotNil(t, co) + assert.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, co.WorkflowStatus) + assert.Equal(t, `0`, co.GetResult().GetValue(), "the lowest-index candidate (EventA) must win regardless of which event appeared first in the replayed batch") + assert.Empty(t, cl.warns) +} + +// 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)) + } + } +} + +// Benchmark_SelectFanOut measures a full replay of a workflow that fans out 50 activities and then +// repeatedly Selects over whichever ones haven't completed yet until all 50 have -- the shape +// dominated by Select's own per-call cost, as opposed to Benchmark_ReplaySequentialActivities, +// which is dominated by Await/CallActivity's per-schedule bookkeeping. +func Benchmark_SelectFanOut(b *testing.B) { + const n = 50 + r := NewTaskRegistry() + if err := r.AddWorkflowN("wf", func(ctx *WorkflowContext) (any, error) { + pending := make([]Task, n) + for i := range pending { + pending[i] = ctx.CallActivity("act") + } + for len(pending) > 0 { + winner, err := ctx.Select(pending...) + if err != nil { + return nil, err + } + if err := pending[winner].Await(nil); err != nil { + return nil, err + } + pending = append(pending[:winner], pending[winner+1:]...) + } + 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")) + } + for i := range int32(n) { + events = append(events, 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)) + } + } +} diff --git a/task/retry_replay_compat_test.go b/task/retry_replay_compat_test.go new file mode 100644 index 0000000..8c389eb --- /dev/null +++ b/task/retry_replay_compat_test.go @@ -0,0 +1,239 @@ +/* +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 ( + "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" +) + +// Test_RetryPolicy_ReplayCompatWithAwaitDrivenHistories replays histories recorded by the +// Await-driven retry design (retries decided inside taskWrapper.Await) against the pollable retry +// chain. Each history was captured from that design by running the same workflow function and +// feeding one completion per turn; the action sequence numbers embedded in it must line up with +// what this implementation emits, or replay fails with a nondeterminism error instead of reaching +// the expected CompleteWorkflow action. This is what lets a workflow mid-retry at upgrade time +// finish under the new code. +// +// The helpers below are self-contained so this file can be dropped into a checkout of either +// design and run unchanged. +func Test_RetryPolicy_ReplayCompatWithAwaitDrivenHistories(t *testing.T) { + policy := func(maxAttempts int) *RetryPolicy { + return &RetryPolicy{MaxAttempts: maxAttempts, InitialRetryInterval: time.Second, BackoffCoefficient: 2} + } + awaitAll := func(tasks ...Task) error { + for _, tk := range tasks { + if err := tk.Await(nil); err != nil { + return err + } + } + return nil + } + + for name, tc := range map[string]struct { + wf func(ctx *WorkflowContext) (any, error) + history []*protos.HistoryEvent + completeID int32 + status protos.OrchestrationStatus + }{ + "single activity retried twice": { + wf: func(ctx *WorkflowContext) (any, error) { + return nil, ctx.CallActivity("A", WithActivityRetryPolicy(policy(5))).Await(nil) + }, + history: []*protos.HistoryEvent{ + cEvTaskScheduled(0, "A"), cEvTaskFailed(0), cEvRetryTimerCreated(1), cEvTimerFired(1), + cEvTaskScheduled(2, "A"), cEvTaskFailed(2), cEvRetryTimerCreated(3), cEvTimerFired(3), + cEvTaskScheduled(4, "A"), cEvTaskCompleted(4), + }, + completeID: 5, + status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, + }, + "retries exhausted": { + wf: func(ctx *WorkflowContext) (any, error) { + return nil, ctx.CallActivity("A", WithActivityRetryPolicy(policy(3))).Await(nil) + }, + history: []*protos.HistoryEvent{ + cEvTaskScheduled(0, "A"), cEvTaskFailed(0), cEvRetryTimerCreated(1), cEvTimerFired(1), + cEvTaskScheduled(2, "A"), cEvTaskFailed(2), cEvRetryTimerCreated(3), cEvTimerFired(3), + cEvTaskScheduled(4, "A"), cEvTaskFailed(4), + }, + completeID: 5, + status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED, + }, + "fan-out awaited out of declaration order": { + // A fails once, B twice, C once; awaited as C, A, B. Every failure lands before + // anything is awaited, so each retry timer's id is decided by Await order alone. + wf: func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(policy(5))) + b := ctx.CallActivity("B", WithActivityRetryPolicy(policy(5))) + c := ctx.CallActivity("C", WithActivityRetryPolicy(policy(5))) + return nil, awaitAll(c, a, b) + }, + history: []*protos.HistoryEvent{ + cEvTaskScheduled(0, "A"), cEvTaskScheduled(1, "B"), cEvTaskScheduled(2, "C"), + cEvTaskFailed(0), cEvTaskFailed(1), cEvTaskFailed(2), + cEvRetryTimerCreated(3), cEvTimerFired(3), cEvTaskScheduled(4, "C"), cEvTaskCompleted(4), + cEvRetryTimerCreated(5), cEvTimerFired(5), cEvTaskScheduled(6, "A"), cEvTaskCompleted(6), + cEvRetryTimerCreated(7), cEvTimerFired(7), cEvTaskScheduled(8, "B"), cEvTaskFailed(8), + cEvRetryTimerCreated(9), cEvTimerFired(9), cEvTaskScheduled(10, "B"), cEvTaskCompleted(10), + }, + completeID: 11, + status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, + }, + "failure lands before a plain task is awaited": { + // A fails while the workflow is awaiting B; C is scheduled before A's failure is + // observed, so C takes id 2 and A's first retry timer id 3. + wf: func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(policy(5))) + b := ctx.CallActivity("B") + if err := b.Await(nil); err != nil { + return nil, err + } + c := ctx.CallActivity("C") + return nil, awaitAll(a, c) + }, + history: []*protos.HistoryEvent{ + cEvTaskScheduled(0, "A"), cEvTaskScheduled(1, "B"), cEvTaskFailed(0), cEvTaskCompleted(1), + cEvTaskScheduled(2, "C"), cEvRetryTimerCreated(3), cEvTaskCompleted(2), cEvTimerFired(3), + cEvTaskScheduled(4, "A"), cEvTaskFailed(4), cEvRetryTimerCreated(5), cEvTimerFired(5), + cEvTaskScheduled(6, "A"), cEvTaskCompleted(6), + }, + completeID: 7, + status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, + }, + "retry interleaved with a plain timer": { + wf: func(ctx *WorkflowContext) (any, error) { + a := ctx.CallActivity("A", WithActivityRetryPolicy(policy(5))) + if err := ctx.CreateTimer(time.Second).Await(nil); err != nil { + return nil, err + } + b := ctx.CallActivity("B") + return nil, awaitAll(a, b) + }, + history: []*protos.HistoryEvent{ + cEvTaskScheduled(0, "A"), cEvPlainTimerCreated(1), cEvTaskFailed(0), cEvTimerFired(1), + cEvTaskScheduled(2, "B"), cEvRetryTimerCreated(3), cEvTaskCompleted(2), cEvTimerFired(3), + cEvTaskScheduled(4, "A"), cEvTaskFailed(4), cEvRetryTimerCreated(5), cEvTimerFired(5), + cEvTaskScheduled(6, "A"), cEvTaskCompleted(6), + }, + completeID: 7, + status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, + }, + "child workflow retried twice": { + wf: func(ctx *WorkflowContext) (any, error) { + return nil, ctx.CallChildWorkflow("Child", WithChildWorkflowRetryPolicy(policy(4))).Await(nil) + }, + history: []*protos.HistoryEvent{ + cEvChildCreated(0, "Child"), cEvChildFailed(0), cEvChildRetryTimerCreated(1), cEvTimerFired(1), + cEvChildCreated(2, "Child"), cEvChildFailed(2), cEvChildRetryTimerCreated(3), cEvTimerFired(3), + cEvChildCreated(4, "Child"), cEvChildCompleted(4), + }, + completeID: 5, + status: protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, + }, + } { + t.Run(name, func(t *testing.T) { + r := NewTaskRegistry() + require.NoError(t, r.AddWorkflowN("wf", tc.wf)) + for _, a := range []string{"A", "B", "C"} { + require.NoError(t, r.AddActivityN(a, func(ActivityContext) (any, error) { return nil, nil })) + } + require.NoError(t, r.AddWorkflowN("Child", func(*WorkflowContext) (any, error) { return nil, nil })) + + ctx := NewWorkflowContext(r, "compat", append([]*protos.HistoryEvent{cEvExecutionStarted("wf")}, tc.history...), nil) + actions := ctx.start() + + require.Len(t, actions, 1, "a fully recorded history must leave only the completion to emit, got %v", actions) + co := actions[0].GetCompleteWorkflow() + require.NotNil(t, co, "expected CompleteWorkflow, got %v", actions[0]) + assert.Equal(t, tc.completeID, actions[0].Id) + assert.Equal(t, tc.status, co.WorkflowStatus, co.GetFailureDetails().GetErrorMessage()) + if tc.status == protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED { + assert.NotContains(t, co.GetFailureDetails().GetErrorMessage(), "a previous execution called") + } + }) + } +} + +func cEvExecutionStarted(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: "compat"}}, + }} +} + +func cEvTaskScheduled(id int32, name string) *protos.HistoryEvent { + return &protos.HistoryEvent{EventId: id, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_TaskScheduled{ + TaskScheduled: &protos.TaskScheduledEvent{Name: name}, + }} +} + +func cEvTaskCompleted(id int32) *protos.HistoryEvent { + return &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_TaskCompleted{ + TaskCompleted: &protos.TaskCompletedEvent{TaskScheduledId: id, Result: wrapperspb.String(`null`)}, + }} +} + +func cEvTaskFailed(id int32) *protos.HistoryEvent { + return &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_TaskFailed{ + TaskFailed: &protos.TaskFailedEvent{TaskScheduledId: id, TaskExecutionId: "exec", FailureDetails: &protos.TaskFailureDetails{ErrorType: "TestError", ErrorMessage: "injected"}}, + }} +} + +func cEvTimerFired(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 cEvTimerCreated(id int32, tc *protos.TimerCreatedEvent) *protos.HistoryEvent { + tc.FireAt = timestamppb.Now() + return &protos.HistoryEvent{EventId: id, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_TimerCreated{TimerCreated: tc}} +} + +func cEvPlainTimerCreated(id int32) *protos.HistoryEvent { + return cEvTimerCreated(id, &protos.TimerCreatedEvent{Origin: &protos.TimerCreatedEvent_CreateTimer{CreateTimer: &protos.TimerOriginCreateTimer{}}}) +} + +func cEvRetryTimerCreated(id int32) *protos.HistoryEvent { + return cEvTimerCreated(id, &protos.TimerCreatedEvent{Origin: &protos.TimerCreatedEvent_ActivityRetry{ActivityRetry: &protos.TimerOriginActivityRetry{TaskExecutionId: "exec"}}}) +} + +func cEvChildRetryTimerCreated(id int32) *protos.HistoryEvent { + return cEvTimerCreated(id, &protos.TimerCreatedEvent{Origin: &protos.TimerCreatedEvent_ChildWorkflowRetry{ChildWorkflowRetry: &protos.TimerOriginChildWorkflowRetry{}}}) +} + +func cEvChildCreated(id int32, name string) *protos.HistoryEvent { + return &protos.HistoryEvent{EventId: id, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_ChildWorkflowInstanceCreated{ + ChildWorkflowInstanceCreated: &protos.ChildWorkflowInstanceCreatedEvent{Name: name}, + }} +} + +func cEvChildCompleted(id int32) *protos.HistoryEvent { + return &protos.HistoryEvent{EventId: -1, Timestamp: timestamppb.Now(), EventType: &protos.HistoryEvent_ChildWorkflowInstanceCompleted{ + ChildWorkflowInstanceCompleted: &protos.ChildWorkflowInstanceCompletedEvent{TaskScheduledId: id, Result: wrapperspb.String(`null`)}, + }} +} + +func cEvChildFailed(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"}}, + }} +} diff --git a/task/task.go b/task/task.go index 86e2a78..599aaee 100644 --- a/task/task.go +++ b/task/task.go @@ -16,6 +16,10 @@ var ErrTaskBlocked = errors.New("the current task is blocked") // when configured timeouts expire. var ErrTaskCanceled = errors.New("the task was canceled") // CONSIDER: More specific info about the task +// ErrTaskNotSelectable is returned by [WorkflowContext.Select] when one of the given tasks was not +// created by a WorkflowContext method. +var ErrTaskNotSelectable = errors.New("task does not support Select") + // Task is an interface for asynchronous durable tasks. A task is conceptually similar to a future. type Task interface { Await(v any) error @@ -30,6 +34,22 @@ type completableTask struct { failureDetails *protos.TaskFailureDetails completedCallback func() taskExecutionId string + // advance, when set, runs on every poll (see pollCompleted) before isCompleted is read. Used by + // a retry chain's outer task (see internalScheduleTaskWithRetries) to drive itself forward -- + // start the next attempt once a backoff timer fires, fail or complete once an attempt's outcome + // is known -- only when something is actually polling for completion, not as a side effect of + // history alone. + advance func() +} + +// pollCompleted runs t's advance hook, if any, then reports whether t is completed. Await and +// Select's poll both call this instead of reading isCompleted directly, so that observing a task is +// what drives a retry chain forward. +func (t *completableTask) pollCompleted() bool { + if t.advance != nil { + t.advance() + } + return t.isCompleted } func newTask(ctx *WorkflowContext) *completableTask { @@ -48,39 +68,41 @@ func newTask(ctx *WorkflowContext) *completableTask { // of any kind. However, workflow functions must never attempt to recover from such panics to ensure that // the workflow execution can proceed normally. func (t *completableTask) Await(v any) error { - for { - if t.isCompleted { - if t.failureDetails != nil { - return fmt.Errorf("task failed with an error: %v", t.failureDetails.ErrorMessage) - } else if t.isCanceled { - return ErrTaskCanceled - } - if v != nil && len(t.rawResult) > 0 { - if err := unmarshalData(t.rawResult, v); err != nil { - return fmt.Errorf("failed to decode task result: %w", err) - } - } - return nil - } - - ok, err := t.workflowCtx.processNextEvent() - if err != nil { - return err - } - if !ok { - break + if err := t.workflowCtx.awaitUntil(t.pollCompleted); err != nil { + return err + } + if err := t.completionError(); err != nil { + return err + } + if v != nil && len(t.rawResult) > 0 { + if err := unmarshalData(t.rawResult, v); err != nil { + return fmt.Errorf("failed to decode task result: %w", err) } } - // TODO: Need a rule about using "defer" in workflows because planned panics will invoke them unexpectedly - // TODO: @joshvanl: remove panic- panic is something that should - // _never_ be called in normal operation. - panic(ErrTaskBlocked) + return nil } func (t *completableTask) TaskExecutionId() string { return t.taskExecutionId } +// completionError returns the error a completed task represents -- nil on success, the formatted +// failure on a failed task, or ErrTaskCanceled on a canceled one -- without touching the task's raw +// result. It must only be called once t.isCompleted is true. +func (t *completableTask) completionError() error { + if t.failureDetails != nil { + return fmt.Errorf("task failed with an error: %v", t.failureDetails.ErrorMessage) + } + if t.isCanceled { + return ErrTaskCanceled + } + return nil +} + +// onCompleted registers [callback] to run when the task completes. Only one callback may be +// registered on a task at a time; this package's only current caller, WaitForSingleEvent's internal +// timer plumbing, registers exactly one, on a task scoped to that single registration and never +// reused for another. func (t *completableTask) onCompleted(callback func()) { t.completedCallback = callback } @@ -106,19 +128,3 @@ func (t *completableTask) completeInternal() { t.completedCallback() } } - -type taskWrapper struct { - delegate Task - onAwaitResult func(any, string, error) error -} - -var _ Task = &taskWrapper{} - -func (t *taskWrapper) Await(v any) error { - err := t.delegate.Await(v) - return t.onAwaitResult(v, t.delegate.TaskExecutionId(), err) -} - -func (t *taskWrapper) TaskExecutionId() string { - return t.delegate.TaskExecutionId() -} diff --git a/tests/grpc/grpc_test.go b/tests/grpc/grpc_test.go index bbf5a0d..8e5909c 100644 --- a/tests/grpc/grpc_test.go +++ b/tests/grpc/grpc_test.go @@ -26,13 +26,15 @@ import ( "github.com/dapr/durabletask-go/client" "github.com/dapr/durabletask-go/task" "github.com/dapr/durabletask-go/tests/utils" + "github.com/dapr/durabletask-go/workflow" "go.opentelemetry.io/otel" ) var ( - grpcClient *client.TaskHubGrpcClient - ctx = context.Background() - tracer = otel.Tracer("grpc-test") + grpcClient *client.TaskHubGrpcClient + workflowClient *workflow.Client + ctx = context.Background() + tracer = otel.Tracer("grpc-test") ) // TestMain is the entry point for the test suite. We use this to set up a gRPC server and client instance @@ -79,6 +81,7 @@ func TestMain(m *testing.M) { } defer conn.Close() grpcClient = client.NewTaskHubGrpcClient(conn, logger) + workflowClient = workflow.NewClientWithLogger(conn, logger) // Run the test exitCode exitCode := m.Run() @@ -105,6 +108,12 @@ func startGrpcListener(t *testing.T, r *task.TaskRegistry) context.CancelFunc { return cancel } +func startWorkflowListener(t *testing.T, r *workflow.Registry) context.CancelFunc { + cancelCtx, cancel := context.WithCancel(ctx) + require.NoError(t, workflowClient.StartWorker(cancelCtx, r)) + return cancel +} + func Test_Grpc_WaitForInstanceStart_Timeout(t *testing.T) { r := task.NewTaskRegistry() r.AddWorkflowN("WaitForInstanceStartThrowsException", func(ctx *task.WorkflowContext) (any, error) { @@ -664,3 +673,117 @@ func Test_Grpc_PatchedWorkflow(t *testing.T) { assert.Equal(t, []bool{false}, patches1Found) assert.Equal(t, []bool{true}, patches2Found) } + +// Test_Workflow_Select_ExternalEventRace demonstrates the scenario from +// https://github.com/dapr/dapr/issues/10447 using only the public workflow package: a workflow +// that must respond to whichever of several named external events arrives first, using +// workflow.WorkflowContext.Select instead of polling. +func Test_Workflow_Select_ExternalEventRace(t *testing.T) { + r := workflow.NewRegistry() + require.NoError(t, r.AddWorkflowN("SelectRaceWorkflow", func(ctx *workflow.WorkflowContext) (any, error) { + approve := ctx.WaitForExternalEvent("Approve", -1) + reject := ctx.WaitForExternalEvent("Reject", -1) + abort := ctx.WaitForExternalEvent("Abort", -1) + + winner, err := ctx.Select(approve, reject, abort) + if err != nil { + return nil, err + } + + tasks := []workflow.Task{approve, reject, abort} + names := []string{"Approve", "Reject", "Abort"} + + var v string + if err := tasks[winner].Await(&v); err != nil { + return nil, err + } + return names[winner] + ":" + v, nil + })) + + cancel := startWorkflowListener(t, r) + defer cancel() + + id, err := workflowClient.ScheduleWorkflow(ctx, "SelectRaceWorkflow") + require.NoError(t, err) + + _, err = workflowClient.WaitForWorkflowStart(ctx, id) + require.NoError(t, err) + + // Only raise the event that should win the race; the workflow must not need the other two to + // ever be raised. + require.NoError(t, workflowClient.RaiseEvent(ctx, id, "Reject", workflow.WithEventPayload("nope"))) + + timeoutCtx, cancelTimeout := context.WithTimeout(ctx, 30*time.Second) + defer cancelTimeout() + metadata, err := workflowClient.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"Reject:nope"`, metadata.Output.Value) +} + +// Test_Workflow_ErrTaskCanceledIsCheckable verifies that a consumer using only the public workflow +// package can check a timed-out WaitForExternalEvent's error against workflow.ErrTaskCanceled with +// errors.Is, since task.ErrTaskCanceled itself isn't reachable without importing the lower-level +// task package. +func Test_Workflow_ErrTaskCanceledIsCheckable(t *testing.T) { + r := workflow.NewRegistry() + require.NoError(t, r.AddWorkflowN("EventTimeoutWorkflow", func(ctx *workflow.WorkflowContext) (any, error) { + err := ctx.WaitForExternalEvent("NeverRaised", 50*time.Millisecond).Await(nil) + if !errors.Is(err, workflow.ErrTaskCanceled) { + return "wrong error: " + fmt.Sprint(err), nil + } + return "ok", nil + })) + + cancel := startWorkflowListener(t, r) + defer cancel() + + id, err := workflowClient.ScheduleWorkflow(ctx, "EventTimeoutWorkflow") + require.NoError(t, err) + + timeoutCtx, cancelTimeout := context.WithTimeout(ctx, 30*time.Second) + defer cancelTimeout() + metadata, err := workflowClient.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"ok"`, metadata.Output.Value) +} + +// foreignWorkflowTask is a workflow.Task implementation from outside the task/workflow packages, +// used to exercise the ErrTaskNotSelectable path: only *completableTask (unexported, from package +// task) is selectable. +type foreignWorkflowTask struct{} + +func (foreignWorkflowTask) Await(v any) error { return nil } +func (foreignWorkflowTask) TaskExecutionId() string { return "" } + +// Test_Workflow_Select_ErrTaskNotSelectableIsCheckable verifies that a consumer using only the +// public workflow package (not importing task directly) can check a Select error against +// workflow.ErrTaskNotSelectable with errors.Is, since task.ErrTaskNotSelectable itself isn't +// reachable without importing the lower-level task package. +func Test_Workflow_Select_ErrTaskNotSelectableIsCheckable(t *testing.T) { + r := workflow.NewRegistry() + require.NoError(t, r.AddWorkflowN("SelectNotSelectableWorkflow", func(ctx *workflow.WorkflowContext) (any, error) { + _, err := ctx.Select(foreignWorkflowTask{}) + if err == nil { + return "no error", nil + } + if !errors.Is(err, workflow.ErrTaskNotSelectable) { + return "wrong error: " + err.Error(), nil + } + return "ok", nil + })) + + cancel := startWorkflowListener(t, r) + defer cancel() + + id, err := workflowClient.ScheduleWorkflow(ctx, "SelectNotSelectableWorkflow") + require.NoError(t, err) + + timeoutCtx, cancelTimeout := context.WithTimeout(ctx, 30*time.Second) + defer cancelTimeout() + metadata, err := workflowClient.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"ok"`, metadata.Output.Value) +} diff --git a/tests/orchestrations_test.go b/tests/orchestrations_test.go index f2e5624..1d1c7f8 100644 --- a/tests/orchestrations_test.go +++ b/tests/orchestrations_test.go @@ -360,6 +360,62 @@ func Test_ActivityRetries(t *testing.T) { ) } +// Test_ActivityRetries_FanOutAwaitedOutOfOrder fans out several retry-configured activities and +// awaits them in a different order than they were scheduled, replaying across multiple turns as +// each one fails, backs off, and retries, and checks that action sequence numbers stay +// consistent across replay. +func Test_ActivityRetries_FanOutAwaitedOutOfOrder(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("ActivityRetriesFanOut", func(ctx *task.WorkflowContext) (any, error) { + retryPolicy := &task.RetryPolicy{ + MaxAttempts: 2, + InitialRetryInterval: 10 * time.Millisecond, + } + // Scheduled in order A, B, C; B never fails, A and C each fail once then succeed. + a := ctx.CallActivity("FlakyActivity", task.WithActivityInput("A"), task.WithActivityRetryPolicy(retryPolicy)) + b := ctx.CallActivity("FlakyActivity", task.WithActivityInput("B"), task.WithActivityRetryPolicy(retryPolicy)) + c := ctx.CallActivity("FlakyActivity", task.WithActivityInput("C"), task.WithActivityRetryPolicy(retryPolicy)) + + // Awaited out of declaration order: C, then A, then B. + var results []string + for _, tk := range []task.Task{c, a, b} { + var v string + if err := tk.Await(&v); err != nil { + return nil, err + } + results = append(results, v) + } + return results, nil + }) + var failedOnce sync.Map + r.AddActivityN("FlakyActivity", func(ctx task.ActivityContext) (any, error) { + var name string + if err := ctx.GetInput(&name); err != nil { + return nil, err + } + if name != "B" { + if _, alreadyFailed := failedOnce.LoadOrStore(name, true); !alreadyFailed { + return nil, fmt.Errorf("%s: not yet", name) + } + } + return name + "-done", nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "ActivityRetriesFanOut") + require.NoError(t, err) + + timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `["C-done","A-done","B-done"]`, metadata.Output.Value) +} + func Test_ActivityFanOut(t *testing.T) { // Registration r := task.NewTaskRegistry() @@ -2079,7 +2135,6 @@ func Test_StartedAt_AfterExecution(t *testing.T) { "StartedAt %v should be >= CreatedAt %v", startedAt, metadata.CreatedAt.AsTime()) } - func Test_StartedAt_WithScheduleTime(t *testing.T) { r := task.NewTaskRegistry() r.AddWorkflowN("StartedAtAfterExec", func(ctx *task.WorkflowContext) (any, error) { diff --git a/tests/select_test.go b/tests/select_test.go new file mode 100644 index 0000000..7b47637 --- /dev/null +++ b/tests/select_test.go @@ -0,0 +1,400 @@ +package tests + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/dapr/durabletask-go/api" + "github.com/dapr/durabletask-go/api/protos" + "github.com/dapr/durabletask-go/task" +) + +// Test_Select_ExternalEventRace exercises the scenario from +// https://github.com/dapr/dapr/issues/10447: waiting on whichever of several +// named external events arrives first, without polling. +func Test_Select_ExternalEventRace(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("SelectRaceWorkflow", func(ctx *task.WorkflowContext) (any, error) { + approve := ctx.WaitForSingleEvent("Approve", -1) + reject := ctx.WaitForSingleEvent("Reject", -1) + abort := ctx.WaitForSingleEvent("Abort", -1) + + winner, err := ctx.Select(approve, reject, abort) + if err != nil { + return nil, err + } + + switch winner { + case 0: + var v string + if err := approve.Await(&v); err != nil { + return nil, err + } + return "Approve:" + v, nil + case 1: + var v string + if err := reject.Await(&v); err != nil { + return nil, err + } + return "Reject:" + v, nil + default: + var v string + if err := abort.Await(&v); err != nil { + return nil, err + } + return "Abort:" + v, nil + } + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "SelectRaceWorkflow") + require.NoError(t, err) + + _, err = client.WaitForWorkflowStart(ctx, id) + require.NoError(t, err) + + // Only raise the event that should win the race; the workflow must not + // need the other two to ever be raised. + require.NoError(t, client.RaiseEvent(ctx, id, "Reject", api.WithEventPayload("nope"))) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"Reject:nope"`, metadata.Output.Value) +} + +// Test_Select_LoopOverRemaining models a workflow that must observe every one +// of several events, in whatever order they arrive, by repeatedly Selecting +// over the tasks that have not yet completed. +func Test_Select_LoopOverRemaining(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("SelectLoopWorkflow", func(ctx *task.WorkflowContext) (any, error) { + pending := []task.Task{ + ctx.WaitForSingleEvent("First", -1), + ctx.WaitForSingleEvent("Second", -1), + ctx.WaitForSingleEvent("Third", -1), + } + var order []string + for len(pending) > 0 { + winner, err := ctx.Select(pending...) + if err != nil { + return nil, err + } + var v string + if err := pending[winner].Await(&v); err != nil { + return nil, err + } + order = append(order, v) + pending = append(pending[:winner], pending[winner+1:]...) + } + return order, nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "SelectLoopWorkflow") + require.NoError(t, err) + _, err = client.WaitForWorkflowStart(ctx, id) + require.NoError(t, err) + + // Raise out of declaration order to prove Select isn't just returning + // index 0 every time. + require.NoError(t, client.RaiseEvent(ctx, id, "Third", api.WithEventPayload("c"))) + require.NoError(t, client.RaiseEvent(ctx, id, "First", api.WithEventPayload("a"))) + require.NoError(t, client.RaiseEvent(ctx, id, "Second", api.WithEventPayload("b"))) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `["c","a","b"]`, metadata.Output.Value) +} + +// Test_Select_AlreadyCompletedWins verifies that a task which is already +// complete by the time Select is called (e.g. a timer created earlier in the +// same execution that has since fired) is picked immediately, without +// needing to process any further history. +func Test_Select_AlreadyCompletedWins(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("SelectAlreadyCompletedWorkflow", func(ctx *task.WorkflowContext) (any, error) { + immediate := ctx.CreateTimer(0) + if err := immediate.Await(nil); err != nil { + return nil, err + } + + neverFires := ctx.WaitForSingleEvent("NeverSent", -1) + + winner, err := ctx.Select(neverFires, immediate) + if err != nil { + return nil, err + } + return winner, nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "SelectAlreadyCompletedWorkflow") + require.NoError(t, err) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `1`, metadata.Output.Value) +} + +// Test_Select_NoTasks verifies that calling Select with no tasks reports an +// error instead of panicking or blocking forever. +func Test_Select_NoTasks(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("SelectNoTasksWorkflow", func(ctx *task.WorkflowContext) (any, error) { + _, err := ctx.Select() + if err == nil { + return nil, nil + } + return err.Error(), nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "SelectNoTasksWorkflow") + require.NoError(t, err) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"Select requires at least one task"`, metadata.Output.Value) +} + +// Test_Select_RetryWrappedTaskWins verifies that a task returned by CallActivity with a retry +// policy is a real, selectable task: Select's own polling drives its retries forward, so it can +// win a Select once its retries succeed, racing normally against a plain task. +func Test_Select_RetryWrappedTaskWins(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("SelectRetryWrappedWorkflow", func(ctx *task.WorkflowContext) (any, error) { + retried := ctx.CallActivity("FlakyActivity", task.WithActivityRetryPolicy(&task.RetryPolicy{ + MaxAttempts: 3, + InitialRetryInterval: 10 * time.Millisecond, + })) + neverFires := ctx.CreateTimer(1 * time.Hour) + + winner, err := ctx.Select(retried, neverFires) + if err != nil { + return nil, err + } + if winner != 0 { + return nil, fmt.Errorf("expected the retried activity (index 0) to win, got index %d", winner) + } + + var v string + if err := retried.Await(&v); err != nil { + return nil, err + } + return v, nil + }) + var attempts int32 + r.AddActivityN("FlakyActivity", func(ctx task.ActivityContext) (any, error) { + if atomic.AddInt32(&attempts, 1) < 2 { + return nil, errors.New("not yet") + } + return "eventually succeeded", nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "SelectRetryWrappedWorkflow") + require.NoError(t, err) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"eventually succeeded"`, metadata.Output.Value) + assert.Equal(t, int32(2), atomic.LoadInt32(&attempts)) +} + +// Test_Select_ActivityRacesTimer covers the most common WhenAny shape: a plain (non-retry) +// CallActivity racing a CreateTimer. The activity blocks on a channel the test only closes after +// the timer has had a chance to win, so the timer wins deterministically rather than by a wall-clock +// margin that could be eaten by the sqlite backend's own polling latency. +func Test_Select_ActivityRacesTimer(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("SelectActivityRacesTimerWorkflow", func(ctx *task.WorkflowContext) (any, error) { + slowActivity := ctx.CallActivity("SlowActivity") + timer := ctx.CreateTimer(50 * time.Millisecond) + + winner, err := ctx.Select(slowActivity, timer) + if err != nil { + return nil, err + } + if winner != 1 { + return nil, fmt.Errorf("expected the timer (index 1) to win, got index %d", winner) + } + if err := timer.Await(nil); err != nil { + return nil, err + } + return "timer won", nil + }) + // Block until the test is done instead of racing a wall-clock sleep against the timer: with a + // fixed sleep, the sqlite backend's poll backoff could delay TimerFired past the activity's + // own completion under load, making the activity win and the test fail. + release := make(chan struct{}) + r.AddActivityN("SlowActivity", func(ctx task.ActivityContext) (any, error) { + <-release + return "too slow", nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + // Registered after worker.Shutdown's defer, so it runs first (LIFO) and unblocks the activity + // before Shutdown's StopAndDrain waits on it. + defer close(release) + + id, err := client.ScheduleNewWorkflow(ctx, "SelectActivityRacesTimerWorkflow") + require.NoError(t, err) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"timer won"`, metadata.Output.Value) +} + +// Test_Select_WaitForSingleEventTimeoutWins verifies that when a WaitForSingleEvent task times out +// before its event is ever raised, Select returns that task as the winner and Await on it surfaces +// ErrTaskCanceled, exactly as it would outside of Select. +func Test_Select_WaitForSingleEventTimeoutWins(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("SelectEventTimeoutWorkflow", func(ctx *task.WorkflowContext) (any, error) { + neverRaised := ctx.WaitForSingleEvent("NeverRaised", 50*time.Millisecond) + neverFires := ctx.CreateTimer(1 * time.Hour) + + winner, err := ctx.Select(neverRaised, neverFires) + if err != nil { + return nil, err + } + if winner != 0 { + return nil, fmt.Errorf("expected the timed-out event wait (index 0) to win, got index %d", winner) + } + + err = neverRaised.Await(nil) + if !errors.Is(err, task.ErrTaskCanceled) { + return nil, fmt.Errorf("expected ErrTaskCanceled, got %v", err) + } + return "timed out as expected", nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "SelectEventTimeoutWorkflow") + require.NoError(t, err) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `"timed out as expected"`, metadata.Output.Value) +} + +// Test_Select_ApprovalGateLoop_CarriesLosingWaitForward demonstrates the approval-gate pattern from +// https://github.com/dapr/dapr/issues/10447 across multiple iterations of the same event names: on +// each iteration the workflow races WaitForSingleEvent("Approve", ...) against +// WaitForSingleEvent("Reject", ...) via Select, and must carry the losing task into the next +// iteration's Select call rather than discarding it and creating a fresh WaitForSingleEvent for the +// same name -- see WaitForSingleEvent's doc comment for why discarding it would hang. +// +// Iteration 1 is won by "Approve", leaving "Reject" as a genuine loser that must be carried +// forward; iteration 2 is then won by a fresh "Reject". This specifically exercises the hazard: if +// the carried-forward "Reject" from iteration 1 were discarded and a new one created instead, the +// stale task would still be the oldest queued waiter for that name and would steal iteration 2's +// "Reject" event out from under the new task, hanging the workflow. Racing two iterations both won +// by the same name (e.g. "Reject" then "Reject") would not catch this: nothing would have lost yet +// for that name to go stale. +func Test_Select_ApprovalGateLoop_CarriesLosingWaitForward(t *testing.T) { + r := task.NewTaskRegistry() + r.AddWorkflowN("ApprovalGateLoopWorkflow", func(ctx *task.WorkflowContext) (any, error) { + approve := ctx.WaitForSingleEvent("Approve", -1) + reject := ctx.WaitForSingleEvent("Reject", -1) + + var decisions []string + for i := 0; i < 2; i++ { + winner, err := ctx.Select(approve, reject) + if err != nil { + return nil, err + } + + var v string + switch winner { + case 0: + if err := approve.Await(&v); err != nil { + return nil, err + } + decisions = append(decisions, "Approve:"+v) + // Carry the losing "Reject" wait forward; a fresh WaitForSingleEvent("Reject", ...) + // here would leave this one permanently queued and stealing the next "Reject" event. + approve = ctx.WaitForSingleEvent("Approve", -1) + case 1: + if err := reject.Await(&v); err != nil { + return nil, err + } + decisions = append(decisions, "Reject:"+v) + reject = ctx.WaitForSingleEvent("Reject", -1) + } + } + return decisions, nil + }) + + ctx := context.Background() + client, worker := initTaskHubWorker(ctx, r) + defer worker.Shutdown(ctx) + + id, err := client.ScheduleNewWorkflow(ctx, "ApprovalGateLoopWorkflow") + require.NoError(t, err) + _, err = client.WaitForWorkflowStart(ctx, id) + require.NoError(t, err) + + // Iteration 1 is won by "Approve", so the original "Reject" task loses and must be carried + // forward. Iteration 2 is then resolved by a second "Reject" event: if the carry-forward were + // wrong (losing task discarded, a fresh one created instead), the stale iteration-1 "Reject" + // task would still be the oldest queued waiter for that name and would steal this event, + // leaving iteration 2's Select -- and the test -- hanging until the timeout below. + require.NoError(t, client.RaiseEvent(ctx, id, "Approve", api.WithEventPayload("first"))) + require.NoError(t, client.RaiseEvent(ctx, id, "Reject", api.WithEventPayload("second"))) + + timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + metadata, err := client.WaitForWorkflowCompletion(timeoutCtx, id) + require.NoError(t, err) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, metadata.RuntimeStatus) + assert.Equal(t, `["Approve:first","Reject:second"]`, metadata.Output.Value) +} diff --git a/workflow/task.go b/workflow/task.go index ec7c0f2..5e04572 100644 --- a/workflow/task.go +++ b/workflow/task.go @@ -5,3 +5,11 @@ import "github.com/dapr/durabletask-go/task" // Task is an interface for asynchronous durable tasks. A task is conceptually // similar to a future. type Task task.Task + +// ErrTaskCanceled is returned by [Task.Await] when the task was canceled, for example because a +// timeout configured via [WorkflowContext.WaitForExternalEvent] expired. +var ErrTaskCanceled = task.ErrTaskCanceled + +// ErrTaskNotSelectable is returned by [WorkflowContext.Select] when one of the given tasks isn't +// selectable -- see Select's doc comment for the specific cases. +var ErrTaskNotSelectable = task.ErrTaskNotSelectable diff --git a/workflow/workflow.go b/workflow/workflow.go index 54fb350..7a25d92 100644 --- a/workflow/workflow.go +++ b/workflow/workflow.go @@ -91,11 +91,28 @@ func (w *WorkflowContext) CreateTimer(delay time.Duration, opts ...CreateTimerOp // multiple events with the same name is allowed. Each event received by an // workflow will complete just one task returned by this method. // +// A task returned by this method that loses a [WorkflowContext.Select] stays +// queued for its event name and must be carried forward into the next Select +// rather than discarded; see [task.WorkflowContext.WaitForSingleEvent] for why. +// // Note that event names are case-insensitive. func (w *WorkflowContext) WaitForExternalEvent(eventName string, timeout time.Duration) Task { return w.oc.WaitForSingleEvent(eventName, timeout) } +// Select blocks until the first of the given [tasks] completes and returns its index. Once Select +// returns, callers should call Await on the task at the returned index to obtain its result or +// error; the remaining tasks are left pending and may still be selected or awaited later. See +// [task.WorkflowContext.Select] for the full behavior, including its tie-break rule, validation +// errors, and [task.ErrTaskBlocked] panic semantics. +func (w *WorkflowContext) Select(tasks ...Task) (int, error) { + otasks := make([]task.Task, len(tasks)) + for i, t := range tasks { + otasks[i] = t + } + return w.oc.Select(otasks...) +} + func (w *WorkflowContext) ContinueAsNew(newInput any, options ...ContinueAsNewOption) { oopts := make([]task.ContinueAsNewOption, len(options)) for i, o := range options {