perf: event-driven work item completion, drop per-item waiter goroutine

[JoshVanL]

The task worker spawned one goroutine per work item that parked inside WaitForWorkflowTaskCompletion / WaitForActivityCompletion for the entire app roundtrip, pairing every in-flight item with a second parked goroutine (2.0 goroutines per item, invariant across all 33 collapse dumps). At collapse that waiter population made goroutine count a linear function of backlog: stack scanning plus runtime.malg accounted for 17.9 percent of collapsed live heap and about half of markroot/scanstack CPU.

Completions now run event-driven on the goroutine that already delivers them (the CompleteWorkflowTask / CompleteActivityTask RPC path):

• Backend gains an optional CompletionCallbackBackend interface that registers a completion callback instead of blocking a waiter; the local TasksBackend (and therefore sqlite/postgres) implements it.
• grpcExecutor gains executeWorkflowAsync / executeActivityAsync, which register the continuation, dispatch, and return; an asyncWait arbiter settles exactly one of backend delivery, cancellation, context error, or dispatch failure, and context.AfterFunc replaces the parked select for shutdown draining.
• The workflow and activity processors implement ProcessWorkItemAsync, carrying the post-wait work (applier, continue-as-new loop, spans, complete/abandon) as a callback chain with unchanged semantics.
• The worker runs the completion tail through a once-guarded finish closure, so the parallelLock slot is held for the full logical lifetime of the item and released exactly once on completion, error, abandon, and shutdown paths; StopAndDrain still waits for in-flight items.

Because the async callbacks are keyed by instance (workflow) or instance+task (activity), a completion from a superseded dispatch of the same key (a duplicate forward delivery, or a response parked by an aborted attempt) could settle a later dispatch with a response computed from an older history prefix. Each dispatch therefore carries a fresh completion token on the WorkItem: the worker echoes it back on WorkflowResponse and ActivityResponse, and the executor discards a mismatched-token response and re-registers for the current dispatch's token, draining any displaced parked payload. No wire change: the completionToken fields already exist in the protos, and workers that do not echo tokens send an empty one and keep uncorrelated matching. The pre-existing blocking ExecuteWorkflow path stays uncorrelated as before; that is acceptable because dapr consumes the async CompletionCallbackBackend path introduced here.

The pre-dispatch goroutine still exists but dies once the item is handed to the stream, taking goroutines per in-flight item from 2 to 1 and halving the count-vs-backlog slope. Backends without callback support (dapr ClusterTasksBackend today) keep the blocking path unchanged.

Backwards compatible with workers that do not echo CompletionToken: the stale-response guard only engages for a non-empty mismatched token (resp.GetCompletionToken() != "" && != token), so a legacy SDK's empty echo settles completions exactly as before this change. Completion routing is keyed on instance ID, never the token. Legacy workers therefore keep pre-change behavior without the new stale-dispatch protection, which they gain when their SDK starts echoing the token; the wire fields themselves predate this change.

[Copilot]

Pull request overview

Introduces event-driven work-item completion to reduce parked goroutines while preserving fallback blocking behavior.

Changes:

• Adds callback-based backend and processor interfaces.
• Implements asynchronous workflow/activity execution with completion tokens.
• Adds concurrency, shutdown, callback, and continue-as-new tests.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
backend/backend.go	Defines callback completion API.
backend/worker.go	Supports asynchronous processor lifetimes.
backend/executor.go	Adds asynchronous dispatch and token arbitration.
backend/orchestration.go	Adds callback-driven workflow processing.
backend/activity.go	Adds callback-driven activity processing.
backend/local/task.go	Implements local completion callbacks.
client/worker_grpc.go	Echoes dispatch completion tokens.
backend/local/task_test.go	Tests callback delivery and cancellation.
tests/worker_async_test.go	Tests worker concurrency and draining.
tests/grpc/grpc_test.go	Tests gRPC continue-as-new behavior.

Suppressed comments (1)

backend/local/task.go:164

• The workflow path has the same re-registration gap: LoadAndDelete removes the callback before a stale-token callback can install its replacement. A concurrent response for the current dispatch can fail with UnknownInstanceID during that window and strand the workflow. Keep delivery routable (or park the response) until re-registration is atomic.

	if pending.cb != nil {
		if res == nil {
			pending.cb(nil, api.ErrTaskCancelled)
		} else {
			pending.cb(res, nil)
		}
		return true

---

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

[Copilot]

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.

Suppressed comments (7)

client/worker_grpc.go:245

• The token check and cache mutation are not atomic. A newer dispatch can call noteDispatch after this check succeeds but before put/delete acquires the cache mutex, allowing the superseded handler to overwrite the newer prefix—the exact poisoning this guard is intended to prevent. Please perform the token comparison and cache mutation under one shared lock (for example, token-aware conditional put/delete methods).

	if err == nil && c.statefulHistoryEnabled() && historyCache.isLatestDispatch(iid, completionToken) {

client/worker_grpc.go:248

• Deleting the latest-token marker makes every older token valid again because isLatestDispatch returns true when no marker exists. If the latest handler completes the workflow and an older in-flight handler finishes afterward, that older handler can repopulate the cache with a stale prefix. Terminal completion needs to leave a fence/tombstone until older handlers can no longer commit.

			historyCache.forgetDispatch(iid)

backend/local/task.go:76

• A newer registration for the same activity key silently overwrites the previous callback. In the superseded-dispatch scenario this PR handles, the old async wait can then never receive a completion and holds its worker semaphore/WG slot until global shutdown. Replace registrations through a race-safe helper that also settles the displaced waiter with a cancellation/superseded error.

	pending := &pendingActivity{cb: cb}
	be.pendingActivities.Store(key, pending)

backend/local/task.go:121

• A newer registration for the same workflow instance silently overwrites the prior callback. The displaced async execution has no per-item context timeout, so it can never settle and permanently retains its worker slot until shutdown. Registration replacement must race-safely notify the displaced waiter of cancellation/supersession.

	pending := &pendingWorkflow{cb: cb}
	be.pendingWorkflows.Store(key, pending)

backend/executor.go:317

• This unconditional delete can erase a newer activity attempt's pending entry when a superseded attempt settles later, leaving the current attempt invisible to stream-disconnect/shutdown cancellation. Capture the value stored for this attempt and use CompareAndDelete(key, pending).

		g.pendingActivities.Delete(key)

client/worker_history.go:127

• This token map is not covered by the cache's TTL or LRU limits. Any workflow that remains nonterminal (for example, a long-lived workflow waiting indefinitely) leaves its instance ID and token here for the lifetime of the listener even after its history entry is evicted, so memory grows with every distinct active workflow. Add bounded lifecycle cleanup that still preserves fencing for in-flight stale handlers.

	latestTokens  sync.Map

backend/backend.go:185

• The “exactly once” callback contract contradicts the non-consuming delivery contract below and the local implementation/tests, which intentionally invoke a registration multiple times so stale responses can be ignored. This public interface documentation should state that callbacks may receive multiple deliveries until deregistration and that consumers must arbitrate them.

// WaitForActivityCompletion. The callback is invoked exactly once, on the