task: add Select for racing multiple tasks (WhenAny)

[gomitrah]

Summary

Adds WorkflowContext.Select(tasks ...Task) (int, error), a WhenAny/Select-style
combinator that resolves as soon as the first of several tasks completes.

Today task.Task only exposes Await/TaskExecutionId, so a workflow that needs to
race several external events (e.g. an approval gate waiting on "Approve", "Reject", or
"Abort", whichever comes first) has no way to do it other than polling: repeatedly
creating short-timeout WaitForSingleEvent waits for each candidate and checking for
ErrTaskCanceled. Each poll cycle creates a new timer/history entry, so a workflow
parked at a gate for any length of time generates continuous, unbounded history growth
purely from the workaround.

Select avoids all of that: it polls each candidate's completion state after every
history event, driven by the same awaitUntil loop Await uses, and returns the index
of the first one found completed. It creates no new timers, actions, or history entries
of its own -- it only observes tasks the caller already scheduled.

Fixes dapr/dapr#10447

Design notes

• Signature matches the one proposed in the issue:
  func (ctx *WorkflowContext) Select(tasks ...task.Task) (completedIndex int, err error).
• Once Select returns, the caller calls .Await() on the winning task to get its
  value/error, same as any other task. Losing tasks are left pending and can be awaited
  or selected again later (e.g. in a loop that keeps selecting over the tasks that
  haven't completed yet, to observe all of them in arrival order).
• If more than one task is found completed at the same time -- whether already
  completed before Select was called, or completed together by a single history
  event -- the lowest index wins; this is the only tie-break rule Select applies.
• Select polls every candidate, including tasks backed by a retry policy
  (CallActivity/CallChildWorkflow with WithActivityRetryPolicy/
  WithChildWorkflowRetryPolicy): polling now drives a retry chain forward the same
  way Await does (see the retry redesign below), so a retry-configured task can win a
  Select once it eventually succeeds. Select rejects a nil task, a Task not
  created by this WorkflowContext, or a Task implementation from outside this
  package, with a descriptive error or the new ErrTaskNotSelectable.
• Select itself creates no callbacks, timers, or history entries; it is a pure read
  of each candidate's completion state.

Retry chain redesign (lazy, poll-driven advancement)

Making retry-configured tasks selectable required changing how a retry chain
advances. Previously, Await drove a failed attempt's retry/backoff decision as a
direct side effect of processing that attempt's TaskFailed event. Select has no
equivalent single call site to hang that logic off of, since it polls many tasks
without "belonging" to any one of them.

The chain now advances lazily: internalScheduleTaskWithRetries schedules the first
attempt immediately, but only decides whether a failed attempt retries, arms its
backoff timer, or schedules the next attempt when something actually polls the
returned task -- via Await or Select, never as a side effect of history replay
alone. An attempt nobody ever observes stays failed and retries no further.

This fixes two correctness issues that existed before this PR even introduced
Select, caught during review:

• Unobserved attempts no longer retry in the background. A retry-configured task
  whose caller never calls Await on it again after it fails (e.g. the workflow
  moved on, awaiting something else) no longer schedules backoff timers and further
  attempts nobody is watching.
• A task's retry chain decision is deterministic regardless of poll/argument
  order. Because advancement only happens on poll, and Select polls every
  candidate on every check (not just until the first completed one is found), a
  retry candidate's backoff timer always lands at the same action sequence number
  regardless of which other candidates are in the same Select call or in what
  order.

Breaking changes for in-flight workflows

Two behavior changes in this redesign affect replay compatibility for workflow
instances with history recorded on main before this change, flagged in review:

• A second Await on an already-resolved retry-configured task no longer re-runs
  the retry chain. On main, every Await call on such a task re-ran the retry
  decision against its first attempt, so a second Await after an earlier chain had
  already succeeded or failed could schedule another backoff timer and attempt. This
  PR makes the outer task complete exactly once and stay completed, which is the
  correct behavior, but a history recorded under the old behavior that relies on a
  second Await producing a new TimerCreated/TaskScheduled pair will hit an
  event with no matching action on replay.
• A decode error from Await(v) on a successful attempt no longer triggers a
  retry. On main, a decode error (e.g. the activity's result doesn't unmarshal
  into the type Await was given) was passed into the retry policy's Handle the
  same as any other error, and the default Handle retries on every error. This PR
  treats a decode error as outside the retry chain's concern -- it's returned as-is
  from Await -- since retrying an already-successful attempt because the caller
  asked for the wrong type was never the intent. A history recorded under the old
  behavior in this situation will also hit a TimerCreated with no matching action.

Both are deliberate behavior fixes, not regressions, but should be called out in the
release notes as breaking for workflow instances already in flight when this ships.

Testing

task/orchestrator_buffered_test.go and tests/select_test.go cover:

• Select resolving to the lowest-index already-completed or simultaneously-completed
  candidate, including across a batch of events replayed after a suspension.
• Select rejecting a nil task, a task from a different WorkflowContext, and a Task
  implementation from outside the package.
• Select never registering a callback on a losing candidate (it only polls).
• Select polling every candidate on every check rather than short-circuiting at the
  first winner, with a regression test for the replay-breaking timer sequence-number
  bug that short-circuiting caused.
• A retry-configured task's backoff/attempt chain only advancing when polled, never
  as an unobserved side effect of history.
• A second Await/Select poll of an already-resolved retry chain being a no-op:
  it doesn't re-run the policy's Handle or arm another timer.
• Benchmark_ReplaySequentialActivities and Benchmark_SelectFanOut for the
  replay-cost shape of plain sequential awaiting versus a Select-driven fan-out.

go test ./task/... ./backend/... passes. The combined ./tests/... run has one
pre-existing, Windows-only flake unrelated to this change (a sqlite file-lock-on-delete
error in tests/backend_test.go's Test_CompleteWorkflow), reproducible identically on
an unmodified branch tip without this PR's changes.

Notes for reviewers

This branch has gone through several review rounds; the current design (pure polling,
no callback registration in Select, lazy poll-driven retry advancement) is the result
of addressing correctness and performance feedback from earlier rounds, including a
callback-leak fix, a performance regression from the original callback-based design,
two separate replay-breaking timer sequence-number bugs, and the background-retry and
decode-error-retry correctness issues described above.

[asalkeld]

Thanks for putting this together. This is exactly the primitive needed for external-event races such as approval gates.
One small API gap: Select is currently exposed only on the low-level task.WorkflowContext. Consumers using the public workflow package cannot call it, because workflow.WorkflowContext wraps its underlying *task.WorkflowContext privately.

Could you also add a forwarding method in workflow/workflow.go?

func (w *WorkflowContext) Select(tasks ...Task) (int, error) {
    taskTasks := make([]task.Task, len(tasks))
    for i, t := range tasks {
        taskTasks[i] = task.Task(t)
    }
    return w.oc.Select(taskTasks...)
}

A small workflow-package test covering an external-event race would also confirm public API works end-to-end.
Our workflow code uses github.com/dapr/durabletask-go/workflow, so this wrapper would let us replace short-timeout polling with Select directly.

[gomitrah]

Thanks for the review! Added a Select wrapper on workflow.WorkflowContext that forwards to the underlying task.WorkflowContext.Select, plus a new end-to-end test (tests/workflow/select_test.go) that races three external events through the public API over a real gRPC connection, similar to what you described. Let me know if you'd like anything adjusted.

[asalkeld]

Select registers a completion callback on every candidate task but has no deregistration path when another task wins — only completeInternal clears a task's callback list, which happens when that task itself completes, not when it loses a race. This leaks stale closures until the losing task eventually completes on its own.

Concretely: with done already completed and pending unresolved, calling Select(pending, done) registers a callback on pending, then returns index 1 (done wins). Repeat that call and each one appends another dead callback onto pending — nothing removes the previous ones. The same growth shows up in the pattern your own doc comment recommends (looping Select over a shrinking set of remaining tasks, as in Test_Select_LoopOverRemaining): each iteration's Select call leaves a stale callback on every task that hasn't won yet.

Not a correctness bug today — dead closures just no-op on a discarded winner variable — but it's unbounded per-turn growth in exactly the usage pattern the API is designed for.

Suggested fix: have onCompleted return an unregister function, and have Select defer cleanup on the losing registrations before returning. Worth locking this down with a regression test for repeated Select(pending, completed) calls asserting the callback count on pending stays bounded.

Separately, two small things:

1. fmt.Errorf("%w: task at index %d", ErrTaskNotSelectable, i) reads awkwardly doubled up ("task does not support Select: task at index 3") — consider "task at index %d: %w" instead.
2. tests/workflow/select_test.go's new test doesn't assert RuntimeStatus == COMPLETED before checking Output.Value, unlike its tests/select_test.go sibling — worth adding back so a workflow failure shows up as a clear status mismatch rather than a confusing output-value diff.

[gomitrah]

Thanks for catching that. it is now Fixed. onCompleted now returns an unregister function, backed by a map keyed by a monotonic id so a specific registration can be removed independent of others (also lets completion order stay deterministic via a sort, which matters if the same task is ever passed to Select twice). Select collects all the unregister funcs and runs them in a defer, so cleanup happens on every return path, including the ErrTaskBlocked panic path.

Added Test_Select_DoesNotLeakCallbacksOnLosingTasks, which calls Select(pending, done) in a loop and asserts zero leftover callbacks on pending after each call. I have verified it fails on the old code and passes now.

Also applied both smaller suggestions: the error is now "task at index %d: %w", and tests/workflow/select_test.go asserts RuntimeStatus == COMPLETED before checking Output.Value.

[asalkeld]

Thank you, looks great to me.

[Copilot]

🟡 Changes recommended

Select does not enforce its documented same-context requirement, allowing incorrect selection or indefinite blocking.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds a Select/WhenAny combinator for efficiently racing durable workflow tasks.

Changes:

• Adds WorkflowContext.Select to task and public workflow APIs.
• Supports multiple completion callbacks with cleanup.
• Adds unit and end-to-end coverage for selection behavior.

File summaries

File	Description
workflow/workflow.go	Exposes the public Select API.
tests/workflow/select_test.go	Tests public API behavior over gRPC.
tests/select_test.go	Tests selection scenarios and validation.
task/task.go	Adds selectable-task errors and callback registration.
task/orchestrator.go	Implements task selection and event processing.
task/orchestrator_buffered_test.go	Tests callback cleanup and unregistration.

Review details

• Files reviewed: 6/6 changed files
• Comments generated: 1
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[asalkeld]

@gomitrah let me know if you are happy for me to continue with this. I am keen to get this functionality in.

[gomitrah]

@asalkeld I'm still driving this, the remaining two points from JoshVanL (the polling rewrite and the struct-size regression) are implemented and verified locally; pushing shortly. No need to take over but appreciate the offer. Will ping when it's up for another look.

[cicoyle]

Also please resolve all addressed feedback 🙏🏻 It helps reviewers know what is addressed vs outstanding.

[JoshVanL]

Hi @gomitrah - I think we need these extra changes to maintain correctness. I have PR'd them into your branch if you want to take a look? gomitrah#1

[gomitrah]

merged, thanks for catching all three. also found a gap in the cancel fix while testing - a canceled attempt with retries remaining was wiping the chain's execution id instead of carrying it forward, same thing copilot flagged on your PR. added a one-line guard + strengthened the test to check the timer's exec id isn't empty, not just the count.

[JoshVanL]

Thank you for this work @gomitrah and being patient on the reviews!

Would you be able to update the Go workflow docs https://github.com/dapr/docs/ with examples of this new func? Happy to also pick this up if you don't have time.