diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md new file mode 100644 index 00000000..77b0ff1f --- /dev/null +++ b/docs/async-orchestrators.md @@ -0,0 +1,360 @@ +# Async orchestrators + +All three packages accept native `async def` orchestrators alongside existing +synchronous generator orchestrators. Both styles use the same history, action, +retry, payload, and entity engine. Register the function normally with +`TaskHubGrpcWorker`, `DurableTaskSchedulerWorker`, or the Functions +`orchestration_trigger` decorator. Functions supports both native two-argument +and compatibility single-argument orchestrators. + +```python +import asyncio +from datetime import timedelta + +from durabletask import task + + +async def workflow(ctx: task.AsyncOrchestrationContext, values: list[int]) -> list[int]: + async def branch(value: int) -> int: + first = await ctx.call_activity("increment", input=value, return_type=int) + await ctx.create_timer(timedelta(seconds=1)) + return await ctx.call_activity("increment", input=first, return_type=int) + + async with asyncio.TaskGroup() as group: + branches = [group.create_task(branch(value)) for value in values] + return [branch.result() for branch in branches] +``` + +Each branch resumes as its own durable operation completes, including when +multiple completions arrive in one dispatch. It does not wait for an entire +batch before scheduling its next operation. + +> [!WARNING] +> Python scopes are not persisted workflow resources. `finally` blocks and async +> context-manager exits can run when an ordinary dispatch suspends, not only +> when the workflow finishes. Keep durable compensation in normal orchestration +> control flow; see [dispatch teardown](#dispatch-teardown-is-not-workflow-cancellation). + +## Motivation and scope + +Generator orchestrators cannot call `async def` helpers or use ordinary +asyncio composition, so async-only libraries need adapters or rewrites before +they can participate in a durable workflow +([#294](https://github.com/microsoft/durabletask-python/issues/294)). + +Goals: + +- Let orchestrators be written with `async def` and `await`, sharing the +existing replay engine, history format, and provider integrations. +- Support ordinary deterministic asyncio composition, such as +`asyncio.gather`, `asyncio.TaskGroup`, `asyncio.timeout`, and +`asyncio.wait_for`, with durable timers as the clock. +- Allow async helpers and libraries to compose durable operations, as long as +they use only supported asyncio APIs. +- Keep generator orchestrators and their recorded histories unchanged. + +Non-goals: + +- A general-purpose asyncio event loop. Real network, storage, or subprocess +I/O, threads, and executors remain activity work. +- Set-based composition such as `asyncio.wait` and `asyncio.as_completed`, +whose ordering is not replay-stable. +- Automatic recording of arbitrary `await` results or local activities. + +Alternatives considered: + +- **A minimal await-to-generator driver without an event loop.** Driving the +coroutine directly would support sequential awaits, but not asyncio tasks, +task groups, timeouts, or libraries that call `asyncio.get_running_loop()`. +- **Combinator-only APIs.** Extending `task.when_all` / `task.when_any` keeps the +generator model but still excludes async-only code and standard asyncio +composition. + +A deterministic, dispatch-local event loop supports the standard composition +primitives while deriving its clock and task ordering from durable history, +so replays reconstruct the same local scheduling decisions. + +## Cold operations and reusable handles + +An `async def` orchestrator automatically receives an +`AsyncOrchestrationContext`. Its awaited durable operations return cold, +single-use native coroutines. +Generator orchestrators still receive `OrchestrationContext` and keep their +eager durable task handles. Both contexts use the same replay engine. + +One-way commands `ctx.send_event(...)`, `ctx.signal_entity(...)`, and +`ctx.continue_as_new(...)` remain synchronous and eager: call them **without +`await`**. These commands take effect when invoked and return `None`. + +| Expression | Meaning | +|---|---| +| `ctx.call_activity(...)` | Create a single-use native coroutine; schedule when it starts. | +| `ctx.as_task_context().call_activity(...)` | Schedule eagerly and return a durable handle. | +| `handle.wait()` | Create a native coroutine borrowing an existing handle. | +| `ctx.wait_for_external_event(...)` | Withdraw an owned event wait on cancellation or timeout. | +| `ctx.create_timer(...)` | Own a timer, cancelling its pending local work on cancellation. | + +Use these cold coroutines with `asyncio.create_task`, `TaskGroup.create_task`, +and ordered `gather`. Constructing a coroutine does not validate the operation, +allocate action IDs, serialize input, or consume buffered events. Cancelling its local task before +its first step starts no durable operation. Operation inputs are not copied: +mutations made before the coroutine's first step are observed, and relative +timer delays start from that execution point. Keep or copy inputs accordingly. + +Cold coroutines are single-use, like ordinary Python coroutines. For an +explicitly started operation with multiple independent waiters, retain its +durable handle: + +```python +operation = ctx.as_task_context().call_activity("quote", input="order-1", return_type=int) +first = asyncio.create_task(operation.wait()) +second = asyncio.create_task(operation.wait()) +first.cancel() +value = await second +``` + +Cancelling either borrower does not cancel the underlying operation. Direct +`await operation` remains supported, but `create_task(operation)` is not valid: +an awaitable durable handle is not a coroutine. Do not retry that expression +assuming the eager operation was rolled back. + +The worker classifies each orchestrator once, when it is registered: + +1. A callable that is itself a coroutine function (including partials, bound +methods, and callable instances whose `__call__` is `async def`) is async. +2. A callable that is itself a generator function is a generator orchestrator. +3. Otherwise, a `functools.wraps` wrapper is classified by the function it +wraps, so a synchronous decorator around an async orchestrator remains async. +4. Any other callable is synchronous. + +Async orchestrators receive an `AsyncOrchestrationContext` and must return a +coroutine; synchronous orchestrators receive `OrchestrationContext`. An +unmarked synchronous wrapper returning a coroutine is rejected because it would +have received the generator context. Cyclic `__wrapped__` chains are rejected at +registration. Existing registration naming requirements still apply. + +## Azure Functions compatibility + +Single-argument async Functions orchestrators receive +`df.AsyncDurableOrchestrationContext`, with the v1-style method and argument +names. Two-argument native orchestrators receive `task.AsyncOrchestrationContext`. + +```python +import azure.durable_functions as df + +app = df.DFApp() + + +@app.orchestration_trigger(context_name="context") +async def greetings(context: df.AsyncDurableOrchestrationContext) -> list[str]: + names = context.get_input() + return await context.task_all([ + context.call_activity("say_hello", name) for name in names + ]) +``` + +`task_all` starts its inputs, waits for all of them, and returns results in +input order, including duplicate references. If an operation fails, the first +observed failure is raised after all have finished. `task_any` returns the +winning `asyncio.Future`, whether it succeeded, failed, or was cancelled, +without cancelling the losers. Await the winner to obtain its result or failure. +An existing future keeps its identity; a coroutine is wrapped in a new task. +Create tasks yourself when you need to identify the winner or later await the +losers. Repeated references within a group are scheduled only once. + +On group cancellation, coroutines started by that group are cancelled and +drained, while supplied futures are borrowed and continue running. Remote work +already issued by either kind of operation is not cancelled. The compatibility +context's `as_task_context()` returns its eager v1-style counterpart when +explicit durable handles are needed. + +## Supported authoring contract + +Await activities, timers, external events, sub-orchestrations, durable +operations via eager `task.when_all` / `task.when_any`, and entity calls/locks +where supported by the provider. Functions compatibility contexts also provide +async `task_all` / `task_any`. Generator and coroutine parents/children may call +one another through durable sub-orchestrations. + +The dispatch-local loop supports FIFO ready callbacks, local futures and tasks, +`asyncio.create_task`, `asyncio.gather`, `asyncio.sleep`, `asyncio.TaskGroup`, +and `asyncio.timeout` / `timeout_at`. +`asyncio.wait_for` uses the same task/future scheduling and durable timers. +Third-party async libraries are not part of this compatibility contract, even +when individual code paths happen to use supported asyncio primitives. + +Create branches from ordered inputs. Ready work follows FIFO scheduling; +history completions are delivered in history order, draining ready work after +each completion. Task-group cancellation wakeups are ordered by local task +creation, not set/object-hash iteration. Local tasks are dispatch resources, +not new durable instances. + +`loop.time()` uses seconds from a fixed UTC epoch and the current orchestration +clock, never machine monotonic time. `loop.call_at` and `call_later` schedule +durable timers for future finite deadlines. Timer conversion rounds up when +necessary so a durable timer does not precede its floating-point deadline. +Core/Functions long timers reuse the existing chunking; Azure Managed retains +its native long timers. Past deadlines are ready callbacks. `sleep(0)` yields +to ready coroutines without a durable timer. Clock advancement and timers +require history events, not elapsed time in the worker. + +> [!WARNING] +> This is not a general-purpose asyncio event loop. External network/storage +> I/O, threads, executors, subprocesses, custom/eager task factories, and +> cross-thread callbacks are outside the orchestrator contract. Perform +> nondeterministic work in ordinary activities. Detected unsupported operations +> raise errors, but the SDK cannot detect every unsafe operation, especially +> synchronous side effects or already-completed foreign awaitables. + +Other asyncio APIs are not implicitly supported because they happen to execute. +Do not iterate sets to schedule durable actions. + +> [!WARNING] +> `asyncio.wait` and `asyncio.as_completed` are **not supported** in async +> orchestrators. They are not rejected at run time, but `wait` returns +> unordered sets and `as_completed` converts its inputs to a set and can create +> tasks in hash-dependent order. Pre-completed or tied tasks and differences +> between Python implementations can change durable action order on replay, +> even when a particular example executes successfully. Use ordered `gather`, +> task groups, or durable `when_any` and explicitly handle the selected durable +> task. + +## Failures, timeouts, and cancellation + +Durable failures raise the existing SDK exceptions at the await, preserving +failure details, typed results, and retry policies. `task.when_all` still waits +for every durable child before surfacing failure. In contrast, +`asyncio.gather` with default options reports the first failure without +cancelling siblings. Retain and await siblings if their results matter. +`return_exceptions=True` collects their outcomes in input order. +Unobserved local background-task failures are reported during dispatch cleanup; +ordinary exceptions are logged, not silently discarded or automatically joined. +Discarded local-task results and observed-failure tracebacks are released during +replay rather than retained until dispatch cleanup. + +Task groups cancel and drain local sibling tasks on failure and use asyncio's +normal exception grouping rules. Python-version-specific exception grouping +and cancel-count behavior remain those of asyncio. + +Cancelling a borrower of an existing durable handle, including `handle.wait()`, +detaches that waiter; it does **not** cancel the activity, sub-orchestration, +entity operation, or external-event subscription. Another borrower can still +await the same handle. Keep the handle if you intend to await it again. +Use explicit SDK cancellation when abandoning a cancellable shared timer/event +wait and deliberately affecting every borrower. + +The async context owns its private timer and event registrations. For an event +with a deadline, use the owned receive instead of timing out a borrowed handle: + +```python +try: + approval = await ctx.wait_for_external_event( + "approval", timeout=timedelta(seconds=30), data_type=bool) +except TimeoutError: + approval = await ctx.wait_for_external_event("approval", data_type=bool) +``` + +The timed-out registration is withdrawn before the replacement is registered, +so it cannot consume the next approval. The receive also cancels its own +deadline timer when the event wins. A nonpositive timeout polls an already +buffered event and otherwise raises `TimeoutError` without creating a timer. +This ownership does not extend to an +activity, sub-orchestration, or entity call: cancellation cannot stop already +issued remote work. A task-group sibling failure can therefore leave remote +operations running. Retain and join their handles when completion is required +before compensation or releasing application-level capacity. + +An entity lock request can still be granted after its local waiter is cancelled. +Abandoning that waiter can leave locks held until the orchestration ends and +block subsequent critical sections and some entity operations. If a timeout is +needed, retain an acquisition task and shield it from the timeout; later await +it and exit the returned lock to release it. Do not start another critical +section while acquisition is pending. + +Cancelling a timer callback removes an uncommitted timer when possible, but +cannot undo a timer already committed to history. Neither local nor durable +cancellation implies external side effects were undone. + +An unhandled local root-coroutine cancellation fails the orchestration with +`TaskCancelledError`; it does not request scheduler termination or repeatedly +abandon the same deterministic replay. Use the client to terminate the durable +instance. Worker/dispatch cancellation and process-control exceptions continue +to follow the existing worker recovery policy. + +## Dispatch teardown is not workflow cancellation + +> [!WARNING] +> Dispatch teardown is not workflow cancellation. Nested `finally` blocks and +> async context-manager exits may run at **every dispatch boundary**, including +> dispatches that simply suspend while waiting for durable work, long before +> the workflow completes. + +Every dispatch reconstructs the coroutine and its local tasks from history. +After the response and entity-release obligations are snapshotted, teardown +cancels remaining local tasks, drains cooperative cleanup, closes tracked async +generators, and disposes the loop. This includes dispatches that suspend, +continue as new, encounter a history failure, or reject a version. + +Local cleanup can await ready work such as `sleep(0)`; it must not wait for new +durable or external work. +Scheduling/awaiting durable work during async teardown raises an actionable +error. Teardown must cooperate with cancellation; a non-quiescent cleanup +queue is bounded and reported, not allowed to run indefinitely. + +Ordinary teardown failures (including `CancelledError`) are logged without +changing the decided response or skipping other cooperative finalizers. +Unexpected non-cancellation `BaseException` +exits retain the existing terminal-failure/recovery policy; an existing primary +failure is preserved. Teardown status/actions do not replace the response. +Held entity-release obligations are captured before teardown for terminal +payload/failure handling. + +Keep workflow cleanup that must be durable in **normal orchestration control +flow**, before return or continue-as-new, not solely in a dispatch finalizer. +No cleanup is guaranteed after process crash or forced termination. +Async-generator *helpers* are closed; async-generator *orchestrators* +(`async def` containing `yield`) are rejected. + +## Replay and framework adapters + +> [!WARNING] +> Recorded histories are coupled to the Python minor version. Python 3.11 and +> 3.12+ schedule coroutine operands of `asyncio.wait_for()` differently, which +> can change durable action order inside concurrent branches, and the async +> context does not normalize arbitrary standard-library scheduling. Keep the +> Python minor version (and dependencies) fixed for in-flight instances; drain +> or version affected instances before upgrading. See +> [#337](https://github.com/microsoft/durabletask-python/issues/337). + +Ordinary async helpers rerun during replay. Their return values are not +automatically checkpointed. There are no local/virtual activities and no +automatic recording or interception of arbitrary await expressions. +`is_replaying` is not a substitute for recording results. Activities remain +at-least-once work: side effects require appropriate idempotency/deduplication. + +Framework adapters can compose supported durable operations using async +helpers. Adapters remain responsible for registering ordinary model/tool +activities, encoding/decoding framework payloads, and supplying replay-stable +IDs, clocks, and message metadata. These responsibilities are not handled by +SDK-wide interception. Libraries that require unsupported asyncio APIs or +nondeterministic operations need an appropriate adapter or activity boundary. + +Changing an in-flight generator to equivalent coroutine code is safe **only +if its durable history contract is preserved**: action IDs/order, names, +payloads, retries, timer deadlines/chunks, event consumption, and entity +release order. Introducing concurrency, a different join, or a new timer may +change that contract. Test recorded histories and version/drain affected +instances; basic sequential parity tests are not blanket migration safety. +Switching from eager handles to cold operations can also change action order: +construction order is not execution order. + +Supporting several Python versions does not guarantee recorded-history +portability between them. Pin the runtime and dependencies for active +instances and validate histories before upgrades; see the Python version +warning above. + +The Functions `orchestrator_generator_wrapper` helper supports direct durable +awaits in synchronous and async mock-based unit tests; it does not simulate the +asyncio scheduler. +Use the real replay executor/provider or the in-memory backend to test task +groups, gather, and durable timeouts. diff --git a/docs/features.md b/docs/features.md index 8573aa0d..b2fb5142 100644 --- a/docs/features.md +++ b/docs/features.md @@ -4,12 +4,13 @@ The following features are currently supported: ## Orchestrations -Orchestrators are implemented using ordinary Python functions that take an `OrchestrationContext` as -their first parameter. The `OrchestrationContext` provides APIs for starting child orchestrations, -scheduling activities, and waiting for external events, among other things. Orchestrations are -fault-tolerant and durable, meaning that they can automatically recover from failures and rebuild -their local execution state. Orchestrator functions must be deterministic, meaning that they must -always produce the same output given the same input. +Orchestrators can be generator functions taking an `OrchestrationContext`, or +`async def` functions taking an `AsyncOrchestrationContext`. Both provide APIs +for starting child orchestrations, scheduling activities, and waiting for +external events. Orchestrations are fault-tolerant and durable: they recover +from failures by rebuilding local state through replay. Orchestrator code must +be deterministic. See the [async authoring contract](async-orchestrators.md) +for supported asyncio operations and replay constraints. ### Orchestration versioning diff --git a/docs/supported-patterns.md b/docs/supported-patterns.md index b67df626..cf7c2b4c 100644 --- a/docs/supported-patterns.md +++ b/docs/supported-patterns.md @@ -2,6 +2,13 @@ The following orchestration patterns are currently supported. +Generator and native coroutine orchestrators share these durable patterns. +See the [async orchestrator contract](async-orchestrators.md) for supported +asyncio composition, durable clocks/timeouts, and dispatch-local cleanup. +Async orchestrators receive cold coroutine operations and owned event/timeouts +automatically. Generator orchestrators retain eager handles; async code can +obtain them through `ctx.as_task_context()` and share them via `handle.wait()`. + ## Function chaining An orchestration can chain a sequence of function calls using the following syntax: @@ -81,6 +88,10 @@ to nested exception groups containing these signals on Python 3.11 and later. The existing warning-only handling of direct `asyncio.CancelledError` during forced teardown is unchanged. Other cancellation or process-control exits propagate after the worker attempts to abandon the accepted dispatch. +Native coroutine authoring distinguishes a cancelled local root task: an +unhandled local cancellation becomes `TaskCancelledError` and fails the +orchestration instead of repeatedly abandoning replay. It does not terminate +the durable instance; see the [async cancellation contract](async-orchestrators.md#failures-timeouts-and-cancellation). If a completion or abandonment RPC raises, its server-side outcome may be unknown. The worker logs the uncertainty and closes the dispatch's originating