Repository navigation
feat(worker): support async def activities - #320
KSouleymane wants to merge 1 commit into
Conversation
Registering an `async def` activity used to fail at runtime: the activity
executor called it without awaiting, and the data converter then tried to
serialize the coroutine ("Failed to serialize object of type 'coroutine'").
Coroutine activities now run on the worker's event loop instead of the
thread pool, so an I/O-bound activity no longer holds a worker thread while
it waits. Their concurrency stays bounded by
`maximum_concurrent_activity_work_items`. Synchronous activities keep their
existing path unchanged.
- `_ActivityExecutor.execute_async` awaits awaitable results; input
deserialization and output serialization are shared with `execute`.
- The worker dispatches coroutine activities to `_execute_activity_async`,
which keeps tracing, failure details and the activity execution hooks,
and offloads the blocking calls (payload store, `CompleteActivityTask`)
to a thread.
- `wrap_with_release` keeps coroutine handlers awaitable; it previously
wrapped every handler in a synchronous function, which would have sent
coroutine handlers to the thread pool unawaited.
- `call_activity()` gains an overload so an async activity's task is typed
with the awaited result rather than a coroutine.
- A synchronous activity that returns an awaitable now fails with a
`TypeError` asking for `async def`, instead of a serialization error.
Tests cover the executor, the activity hooks, handler selection, the activity
span on both paths, end-to-end runs on the in-memory backend (including
activities that must all be in flight at once to pass a barrier, with a
single pool thread) and the static result types.
|
KSouleymane please read the following Contributor License Agreement(CLA). If you agree with the CLA, please reply with the following information.
Contributor License AgreementContribution License AgreementThis Contribution License Agreement (“Agreement”) is agreed to by the party signing below (“You”),
|
There was a problem hiding this comment.
🟡 Changes recommended
Async cancellation can leave work items unresolved, blocking reconnection can stall all in-flight async activities, and callable-object activities remain unusable through the public API.
3 open issues
What changed in this PR
Adds async def activity support to the core worker, enabling I/O-bound activities to run without holding worker threads. This also extends support to the Azure Managed worker and addresses #319.
Changes:
- Dispatches async activities onto the worker event loop while preserving concurrency limits, tracing, and execution hooks.
- Adds awaited-result typing and clearer errors for synchronous activities returning awaitables.
- Adds regression tests, documentation, a fan-out example, and changelog entries.
| File | Description |
|---|---|
tests/durabletask/typing/activity_results.py |
Checks sync and async activity result types. |
tests/durabletask/test_worker_activity_hooks.py |
Tests async hooks and handler selection. |
tests/durabletask/test_tracing.py |
Tests activity spans on both execution paths. |
tests/durabletask/test_orchestration_e2e.py |
Tests async sequencing, failures, and concurrent execution. |
tests/durabletask/test_activity_executor.py |
Tests awaiting, callable detection, and awaitable rejection. |
examples/async_activities.py |
Demonstrates async activity fan-out with Azure Managed. |
durabletask/worker.py |
Implements async dispatch and shared activity execution helpers. |
durabletask/task.py |
Adds async activity typing and a scheduling overload. |
durabletask-azuremanaged/CHANGELOG.md |
Documents inherited async activity support. |
docs/features.md |
Explains async activities and event-loop constraints. |
CHANGELOG.md |
Records async support and changed awaitable error behavior. |
Review effort: Balanced
| result = await executor.execute_async( | ||
| instance_id, req.name, req.taskId, req.input.value | ||
| ) |
| except Exception as ex: | ||
| self._logger.exception( | ||
| f"Failed to deliver activity response for '{req.name}#{req.taskId}' of orchestration ID '{instance_id}' to sidecar: {ex}" | ||
| ) | ||
| res = self._new_activity_failure_response(req, completionToken, ex) |
| return inspect.iscoroutinefunction(fn) or inspect.iscoroutinefunction( | ||
| getattr(fn, "__call__", None) | ||
| ) |
|
Please do not open issues and immediately PRs to implement them. Maintainers should triage and discuss issues first. |
Bernd Verst (berndverst)
left a comment
Bernd Verst (berndverst)
left a comment
There was a problem hiding this comment.
Async activities address a real gap in #319, and the happy-path execution and result typing look good. I'm requesting changes before merging because the new execution path still has worker-wide liveness risks and introduces an avoidable compatibility break for synchronous activities.
| Priority | Concern |
|---|---|
| P1 | Reconnect setup, reader cleanup, and retry backoff block the same event loop that now hosts in-flight activities and their timeout handlers. |
| P1 | Cancelled activities release local capacity without completing or abandoning the backend work item; cancellation during a threaded completion RPC also needs an explicit uncertain-outcome policy. |
| P1 | Blocking completion/payload calls use the default asyncio executor shared with work-item reception and DNS, allowing saturation to stall otherwise unrelated work and bypassing the configured worker-pool limit. |
| P2 | The unconditional awaitable-result rejection breaks previously successful synchronous activities returning serializable awaitable data. |
| P2 | Input/output conversion remains synchronous on the shared loop, so large payloads and custom converters can stall dispatch and other async activities even when the activity body is nonblocking. |
| P2 | Callable-object support stops at detection: normal public registration/scheduling cannot name ordinary instances, and postponed call annotations are not reconstructed. |
The inline comments include focused reproductions and suggested regression coverage. In particular, I reproduced the synchronous-output behavior difference against main, cancellation with neither completion nor abandonment, default-executor/DNS starvation, reconnect-induced activity starvation, conversion-induced timer delays, and the callable registration/type-discovery gaps.
There is no new abstract-method requirement here; the compatibility concern is the changed treatment of existing valid synchronous outputs. Please preserve that behavior and cover the recovery, cancellation, saturation, large-payload, and public callable-object paths before merging.
| try: | ||
| result = await executor.execute_async( | ||
| instance_id, req.name, req.taskId, req.input.value | ||
| ) |
There was a problem hiding this comment.
[P1] Make reconnect handling nonblocking before hosting activities on this loop
The loop awaiting this activity also runs _async_run_loop, where connection setup calls synchronous Hello, connection invalidation joins the reader thread, and each retry uses the blocking self._shutdown.wait(delay). After a stream disconnect, a retrying backend therefore freezes every in-flight async activity, including already-ready I/O callbacks and timeout handlers. Moving activity-body I/O to async does not help when transport recovery owns the event-loop thread.
I reproduced this through _async_run_loop: an activity started await asyncio.sleep(0.01), the stream disconnected, and two failed reconnect attempts used 150 ms Hello calls with 50 ms backoff. The activity could not resume for 419 ms; it resumed only after the reconnect loop exited. With continued failures, that starvation continues for the outage. Please offload blocking connection/reader cleanup and use awaitable, shutdown-aware backoff. A regression should keep an async activity/heartbeat progressing while reconnect attempts continue, rather than only testing successful dispatch.
| self._logger.exception( | ||
| f"Failed to deliver activity response for '{req.name}#{req.taskId}' of orchestration ID '{instance_id}' to sidecar: {ex}" | ||
| ) | ||
| res = self._new_activity_failure_response(req, completionToken, ex) |
There was a problem hiding this comment.
[P1] Give cancelled async activities an explicit backend outcome
asyncio.CancelledError inherits from BaseException, so cancellation escapes both except Exception blocks here and the manager's _process_work_item() recovery handler. The completion hook, channel-release wrapper, and semaphore cleanup still run, but neither CompleteActivityTask nor AbandonTaskActivityWorkItem is sent. The worker considers its capacity released while the orchestration remains blocked until backend lease recovery/redelivery.
I reproduced this with an activity awaiting an already-cancelled future: zero completion RPCs, zero abandonment RPCs, and the semaphore permit returned. Please handle cancellation explicitly and test it through the manager/dispatch path. Also cover cancellation while awaiting asyncio.to_thread(...): cancelling that await does not stop the underlying thread, so releasing/retiring its channel or sending an abandonment concurrently with an already-running completion RPC can produce an uncertain outcome. The recovery path needs to distinguish cancellation during the activity from cancellation after completion delivery has started.
| ) | ||
| res = self._new_activity_failure_response(req, completionToken, ex) | ||
|
|
||
| await asyncio.to_thread(self._complete_activity_task, req, stub, res) |
There was a problem hiding this comment.
[P1] Isolate blocking activity I/O from the loop's default executor
asyncio.to_thread() uses the loop's default executor, which _async_run_loop also uses for work_item_queue.get() and async networking libraries use for DNS resolution. A burst of slow CompleteActivityTask calls (or payload-store calls above) can occupy that entire pool, preventing incoming work from being dispatched and unrelated async activities from resolving hosts until a blocking call returns. These RPCs have no deadline here, so the stall can be indefinite.
I reproduced this with maximum_thread_pool_workers=1: 32 completion calls ran concurrently, and both a default-executor dispatch probe and loop.getaddrinfo('localhost', 80) timed out while those calls were gated. The configured worker-pool limit does not bound this work. Please keep blocking SDK I/O on an explicitly bounded executor that cannot starve the stream reader/DNS, or use nonblocking APIs (the payload helpers already have async variants), and cover executor saturation in a regression test.
| fn, ctx, activity_input = self._prepare(orchestration_id, name, task_id, encoded_input) | ||
|
|
||
| # Execute the activity function | ||
| activity_output = fn(ctx, activity_input) |
There was a problem hiding this comment.
[P2] Preserve serializable awaitable values from synchronous activities
This changes more than the error for an accidentally returned coroutine: inspect.isawaitable() also matches valid data objects that the existing converter can serialize. For example, a dataclass Result(value: int) with a __await__ method is serialized as {"value": 7} by a synchronous activity on main, but the same activity fails here before its data converter is called. Awaitable dictionary subclasses and custom converters can be affected too.
This is intentionally returned data, not necessarily an accidentally unawaited operation; rewriting a previously synchronous activity as async def just to retain its old output also moves its user code onto the event loop. Please preserve successfully serializable synchronous outputs and narrow the improved diagnostic to unsupported awaitable/coroutine outputs. Add a compatibility regression covering a serializable awaitable value; documenting an error-message change does not capture this successful-to-failed behavior change.
|
|
||
| return self._serialize_output(orchestration_id, name, task_id, activity_output) | ||
|
|
||
| def _prepare( |
There was a problem hiding this comment.
[P2] Keep synchronous payload conversion off the shared worker loop
Both _prepare() and _serialize_output() run synchronously inside this coroutine, so deserialization, type reconstruction, to_json() hooks, and serialization now execute on the dispatch/event-loop thread. Offloading only the payload-store and completion calls does not make the SDK's own activity plumbing nonblocking.
With the default JsonDataConverter, three concurrent activities processing approximately 4.5 MB JSON payloads delayed a 10 ms loop timer to 426 ms in my reproduction. These are within the supported external-payload size limit. A synchronous custom converter taking 150 ms for each conversion delayed the timer to 301 ms even though the activity itself only awaited sleep(0). This stalls other async activities and work-item dispatch without the user putting any blocking work in their activity. Please isolate potentially expensive/blocking conversion from the shared loop and add coverage beyond the tiny string/integer results used here.
| """Returns True for ``async def`` functions and objects with an ``async def __call__``.""" | ||
| return inspect.iscoroutinefunction(fn) or inspect.iscoroutinefunction( | ||
| getattr(fn, "__call__", None) | ||
| ) |
There was a problem hiding this comment.
[P2] Wire async callable objects through registration, scheduling, and type discovery
The new detector recognizes an instance with async def __call__, but that support is unreachable through the normal public API: worker.add_activity(instance) and ctx.call_activity(instance) both use task.get_name(), which unconditionally reads fn.__name__. An ordinary callable instance has no such attribute; registration raises AttributeError before dispatch. The detector-only test does not exercise this.
There is a second gap after working around the name: with from __future__ import annotations, _resolve_hints(instance) cannot resolve the __call__ annotations. I verified that both activity_input_type(instance) and activity_output_type(instance) return None, while passing instance.__call__ discovers the dataclass correctly. Consequently, a typed activity can receive a dict instead of its annotated input and fail, or return an un-reconstructed result. Please either complete callable-object support with consistent naming and __call__ type discovery, plus a public registration-to-result regression test, or narrow the claimed support and tests to the callable forms that actually work.
| try: | ||
| return await handler(*args, **kwargs) | ||
| finally: | ||
| release() |
There was a problem hiding this comment.
[P2] Retain the channel through ordinary-error abandonment as well
This async wrapper releases its channel reference before _AsyncWorkerManager._process_work_item() invokes the fallback cancellation handler. If a reconnect has retired the channel and this is its last in-flight reference, release() closes it before _cancel_activity() sends AbandonTaskActivityWorkItem through the original stub.
This also happens without cancellation: for example, deexternalize_payloads can raise before _execute_activity_async() enters its activity-failure conversion block. The exception escapes this wrapper, the retired channel closes, and only then does the manager attempt abandonment. The existing sync wrapper has the same ordering; the new async wrapper carries that ownership gap forward.
Please retain the channel reference until execution and any fallback abandonment have finished, with a regression covering an async activity whose input payload download fails after its channel is retired. Handling CancelledError alone will not cover this ordinary-exception path.
Problem
Registering an
async defactivity fails at runtime._ActivityExecutor.executecalls the function without awaiting it, and the data converter then tries to serialize the coroutine (Failed to serialize object of type 'coroutine'). The only workaround is a sync activity callingasyncio.run(...), which holds a worker thread for the whole I/O wait.Fix
Coroutine activities now run on the worker's event loop instead of the thread pool. Their concurrency stays bounded by
maximum_concurrent_activity_work_items. Sync activities keep their existing path._ActivityExecutor.execute_asyncawaits the activity. Input deserialization and output serialization are shared withexecute.async def __call__) to a new_execute_activity_async. It keeps tracing, failure details and the activity execution hooks, and offloads the blocking calls (payload store,CompleteActivityTask) withasyncio.to_thread._execute_activityis split into small helpers (span, response building, completion) used by both paths. The sync path's behavior is unchanged.wrap_with_releasekeeps coroutine handlers awaitable. It previously wrapped every handler in a sync function, which would have sent coroutine handlers to the thread pool unawaited.call_activity()gains anAsyncActivityoverload inOrchestrationContextand_RuntimeOrchestrationContext, so an async activity's task is typedCompletableTask[TOutput]instead of a coroutine. No abstract member is added.TypeErrorasking forasync def, instead of the serialization error. This is listed under CHANGED in the changelog.Docs: a paragraph and an
httpxexample under "Activities" indocs/features.md, a newexamples/async_activities.py(fan-out on DTS), and changelog entries fordurabletaskanddurabletask-azuremanaged.Validation
tests/durabletask,-m "not e2e"), Python 3.11mainwithout the fixnox -s lintnox -s typecheck_core(strict Pyright)tests/durabletask/typing/activity_results.pywithout the new overloadCompletableTask[Coroutine[...]]) as expectedNew tests:
execute_asyncon sync activities, theTypeErrorfor awaitables returned by sync activities, async callable detection;emit_trace_spanson and off. The sync path was not covered before;asyncio.Barrier, withmaximum_thread_pool_workers=1. The test is deterministic and does not rely on timing.Not run: live DTS / emulator integration tests, the
durabletask-azuremanagedandazure-functions-durablesuites (no code change there), and Python versions above 3.11.Closes #319
Suggested reviewers: andystaples Bernd Verst (@berndverst)