From d8f1138289b47c73395d878a179027b4033735f4 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 11 Nov 2026 13:33:04 -0700 Subject: [PATCH 1/4] Document coroutine orchestrators and add Managed compatibility tests Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 17 +++ README.md | 47 ++++++++ azure-functions-durable/CHANGELOG.md | 11 ++ azure-functions-durable/MIGRATION_GUIDE.md | 10 +- azure-functions-durable/README.md | 25 +++++ durabletask-azuremanaged/CHANGELOG.md | 10 ++ .../test_dts_coroutine_orchestrators.py | 102 ++++++++++++++++++ 7 files changed, 221 insertions(+), 1 deletion(-) create mode 100644 tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d2c46962..4e7f94f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,23 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- Added `async def` orchestrators with an automatically supplied +`AsyncOrchestrationContext`, cold coroutine operations, durable asyncio +timers, and supported task-group and gather composition. Generator +orchestrators retain their eager durable handles. `as_task_context()` and +`Task.wait()` enable explicit interoperability with those handles. +- Async event waits support owned cancellation and timeouts, withdrawing +abandoned subscriptions so they cannot consume later events. + +> [!WARNING] +> Async cleanup can run at every dispatch, including suspension; it is not +> durable workflow compensation. Converting running orchestrations or changing +> Python versions may alter replay-visible action order. See the +> [async contract](docs/async-orchestrators.md) for supported operations and +> replay and upgrade guidance. + FIXED - Async orchestrators wrapped with `functools.wraps` are now recognized even diff --git a/README.md b/README.md index fb947a98..85b25425 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,53 @@ The Azure Functions Durable provider requires Python 3.13 or later. ## Optional Features +### Async Orchestrators + +All three packages support `async def` orchestrators and nested async helpers. +Existing generator orchestrators continue to work. + +```python +import asyncio + +from durabletask import task + + +async def sequence(ctx: task.AsyncOrchestrationContext, value: int) -> list[int]: + first = await ctx.call_activity("plus_one", input=value, return_type=int) + async with asyncio.TaskGroup() as group: + left = group.create_task(ctx.call_activity("plus_one", input=first, return_type=int)) + right = group.create_task(ctx.call_activity("plus_one", input=value, return_type=int)) + return [left.result(), right.result()] +``` + +Register the function with `TaskHubGrpcWorker`, `DurableTaskSchedulerWorker`, +or the Azure Functions `orchestration_trigger` decorator as usual. Functions +also supports single-argument `async def` orchestrators using the compatibility +context and `await context.task_all(...)`. Async orchestrators automatically +receive cold coroutine operations accepted by `create_task` and task groups. +Generator orchestrators keep their eager, reusable durable handles. For +interoperability, `ctx.as_task_context()` exposes the eager context and +`handle.wait()` creates an independently cancellable borrower. + +> [!WARNING] +> This is replay-driven await syntax, not general-purpose asyncio execution. +> Compose durable work with `asyncio.gather`, local tasks/task groups, and +> durable `asyncio.sleep` / timeouts within the +> [supported contract](docs/async-orchestrators.md). +> Perform network, storage, and other nondeterministic I/O in activities. +> Completed foreign awaitables and synchronous side effects cannot reliably be +> detected by the replay engine. +> Python `finally` blocks and async context-manager exits can run at ordinary +> dispatch suspension; they are not workflow-lifetime compensation scopes. + +An async helper is not a durable sub-orchestration: use +`await ctx.call_sub_orchestrator(...)` for a separate orchestration instance. +An `async def` function containing `yield` is an async generator, not a supported +orchestrator. Converting an existing orchestrator must preserve durable action +order and payloads for replay compatibility. +See the contract for cancellation, per-dispatch async cleanup, unsupported +ordering-sensitive APIs and framework adapters. + ### Task Hub Query Filters Set `OrchestrationQuery.task_hub_names` to filter orchestration queries by one or diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 4dd19e87..444d4c51 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -12,6 +12,16 @@ ADDED - A best-effort background check warns when a Function App's requirements allow a future major version, which may introduce replay-breaking changes. +- With the corresponding core SDK, native and v1-style `async def` +orchestrators automatically receive async contexts with cold durable operations +and supported asyncio composition. The v1-style context preserves compatibility +argument names and provides async `task_all` and `task_any`. Generator +orchestrators retain their eager behavior. See the +[async contract](../docs/async-orchestrators.md) for dispatch cleanup and +replay/upgrade precautions. +- The `orchestrator_generator_wrapper` test helper accepts direct-await +coroutines. Tests involving asyncio scheduling still require a replay executor +or backend. - Added opt-in `HttpRetryOptions` on `context.call_http()` for durable initial request retries with configurable timing and status selection. Empty or omitted status lists select all non-2xx responses; exhausted retries fail the call. @@ -62,6 +72,7 @@ readers before enabling new writers; existing headerless blobs still require their original compression setting. See the [core compatibility guidance](../docs/features.md#cross-sdk-compatibility). + ## v2.0.0rc4 ADDED diff --git a/azure-functions-durable/MIGRATION_GUIDE.md b/azure-functions-durable/MIGRATION_GUIDE.md index 20d992da..7d28e07b 100644 --- a/azure-functions-durable/MIGRATION_GUIDE.md +++ b/azure-functions-durable/MIGRATION_GUIDE.md @@ -102,10 +102,18 @@ The single-argument orchestrator receives the compatibility Use this state to validate the infrastructure upgrade before changing application APIs. +In 2.x, a single-argument `async def` orchestrator instead receives +`AsyncDurableOrchestrationContext`, with the same v1-style argument names and +awaitable durable operations. Generator orchestrators retain their existing +eager task-handle behavior. See the +[async authoring contract](../docs/async-orchestrators.md) before converting +running orchestrations. + ## Migrate to native orchestration APIs Native orchestrators accept the durabletask context and input as separate -arguments: +arguments. Generator functions receive `task.OrchestrationContext`; +`async def` functions receive `task.AsyncOrchestrationContext`: ```python from typing import Any diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 29da298a..09694c3b 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -108,6 +108,30 @@ These compatibility paths remain supported. > Changing typing modes can change values returned during orchestration replay. > Validate existing payloads and drain affected orchestrations before switching. +## Async orchestrators + +Both authoring styles accept `async def` and `await` without an adapter loop. +Await durable operations and compose branches with `asyncio.gather` or task +groups; `asyncio.sleep` and timeouts use durable timers. Synchronous generator +orchestrators remain supported, including mixed-style sub-orchestrations. + +Async orchestrators automatically receive cold coroutine operations accepted by +`create_task` and task groups. Native orchestrators receive +`task.AsyncOrchestrationContext`; single-argument orchestrators receive +`df.AsyncDurableOrchestrationContext`, retaining v1-style argument names and +`await context.task_all(...)` / `await context.task_any(...)`. +`wait_for_external_event()` owns its registration and optional durable timeout, +so an abandoned receive cannot consume the next event. Generator contexts +retain eager, reusable durable handles; async code can access those through +`context.as_task_context()` and use `handle.wait()` to create a borrowed waiter. + +The [shared async contract](../docs/async-orchestrators.md) defines the supported +API subset, cancellation, per-dispatch `finally`/async-generator cleanup, +replay/migration rules, and framework adapter responsibilities. Arbitrary I/O, +threads, subprocesses, `asyncio.wait`/`as_completed`, and automatic recording of +ordinary async helpers are not supported. Put nondeterministic work in ordinary +activities, not orchestrators. + ## Durable HTTP failure retries Opt in to durable retries for the **initial** `context.call_http()` request with @@ -153,6 +177,7 @@ rather than returning the last selected response. Transport and other activity failures also use the policy regardless of the status list. The policy and status list are snapshotted when the request is scheduled. +For async orchestrators, scheduling occurs when the coroutine starts. Omitting `retry_options` preserves the existing request payload and behavior. Automatic 202/Location polling remains separate: subsequent GET polls inherit neither the failure retry policy nor its status classification. Poll HTTP diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 157a4604..d39bbbe2 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,16 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- With the corresponding core SDK, `DurableTaskSchedulerWorker` accepts +`async def` orchestrators with an automatically supplied async context, cold +durable operations, and supported asyncio composition. Generator orchestrators +retain their existing behavior. Event waits support owned cancellation and +timeouts; remote operations are not cancelled by local waiter cancellation. +See the [async contract](../docs/async-orchestrators.md) for dispatch cleanup +and replay/upgrade precautions. + FIXED - With the corresponding core SDK update, async dispatch cleanup no longer diff --git a/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py new file mode 100644 index 00000000..f8d9af6c --- /dev/null +++ b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py @@ -0,0 +1,102 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Exercise both workers over gRPC and optionally against the DTS emulator.""" + +import json +import asyncio +import os +import socket +from collections.abc import Generator +from datetime import timedelta +from typing import Any + +import pytest + +from durabletask import client, task, worker +from durabletask.azuremanaged.client import DurableTaskSchedulerClient +from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker +from durabletask.testing import create_test_backend + + +@pytest.mark.parametrize("transport", [ + "core", + "azuremanaged", + pytest.param("dts", marks=pytest.mark.dts), +]) +@pytest.mark.parametrize("coroutine_parent", [False, True], ids=["generator-parent", "coroutine-parent"]) +@pytest.mark.parametrize("borrowed", [False, True], ids=["coroutines", "borrowed-handles"]) +def test_mixed_orchestrators_over_grpc(transport: str, coroutine_parent: bool, borrowed: bool) -> None: + def increment(ctx: task.ActivityContext, value: int) -> int: + return value + 1 + + async def coroutine_child(ctx: task.AsyncOrchestrationContext, value: int) -> int: + if not borrowed: + await ctx.create_timer(timedelta(milliseconds=10)) + return await ctx.call_activity(increment, input=value, return_type=int) + await asyncio.sleep(0) + await asyncio.sleep(0.01) + return await ctx.as_task_context().call_activity(increment, input=value, return_type=int) + + def generator_child(ctx: task.OrchestrationContext, value: int) -> Generator[task.Task[Any], Any, int]: + return (yield ctx.call_activity(increment, input=value)) + + async def coroutine(ctx: task.AsyncOrchestrationContext, value: int) -> list[Any]: + if not borrowed: + async with asyncio.TaskGroup() as group: + first = group.create_task(ctx.call_sub_orchestrator(coroutine_child, input=value)) + second = group.create_task(ctx.call_sub_orchestrator(generator_child, input=value + 1)) + return [first.result(), second.result(), await ctx.wait_for_external_event("go")] + handles = ctx.as_task_context() + values = list(await asyncio.gather( + handles.call_sub_orchestrator(coroutine_child, input=value), + handles.call_sub_orchestrator(generator_child, input=value + 1), + )) + event = await ctx.wait_for_external_event("go") + return values + [event] + + def generator(ctx: task.OrchestrationContext, value: int) -> Generator[task.Task[Any], Any, list[Any]]: + values = yield task.when_all([ + ctx.call_sub_orchestrator(coroutine_child, input=value), + ctx.call_sub_orchestrator(generator_child, input=value + 1), + ]) + event = yield ctx.wait_for_external_event("go") + return values + [event] + + if transport == "dts": + backend = None + host = os.environ["ENDPOINT"] + taskhub = os.environ["TASKHUB"] + else: + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + port = listener.getsockname()[1] + backend = create_test_backend(port=port) + host = f"localhost:{port}" + taskhub = "test" + if transport != "core": + w = DurableTaskSchedulerWorker( + host_address=host, secure_channel=False, taskhub=taskhub, token_credential=None) + c = DurableTaskSchedulerClient( + host_address=host, secure_channel=False, taskhub=taskhub, token_credential=None) + else: + w = worker.TaskHubGrpcWorker(host_address=host) + c = client.TaskHubGrpcClient(host_address=host) + try: + with w, c: + parent = coroutine if coroutine_parent else generator + for fn in (parent, coroutine_child, generator_child): + w.add_orchestrator(fn) + w.add_activity(increment) + w.start() + instance_id = c.schedule_new_orchestration(parent, input=3) + c.raise_orchestration_event(instance_id, "go", data="approved") + state = c.wait_for_orchestration_completion(instance_id, timeout=15) + assert state is not None + assert state.runtime_status == client.OrchestrationStatus.COMPLETED, state.failure_details + assert state.serialized_output is not None + assert json.loads(state.serialized_output) == [4, 5, "approved"] + finally: + if backend is not None: + backend.stop() + backend.reset() From 818c9e28c27257698f5e0ab1f4e5e3a60ae1c3db Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 12 Nov 2026 13:21:43 -0700 Subject: [PATCH 2/4] Clarify async migration joins and cover in-memory provider cases Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- azure-functions-durable/CHANGELOG.md | 2 ++ azure-functions-durable/MIGRATION_GUIDE.md | 38 ++++++++++++++++++++-- azure-functions-durable/README.md | 5 +-- durabletask-azuremanaged/CHANGELOG.md | 2 ++ eng/templates/build.yml | 9 +++-- 5 files changed, 49 insertions(+), 7 deletions(-) diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 444d4c51..566e70fc 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -19,6 +19,8 @@ argument names and provides async `task_all` and `task_any`. Generator orchestrators retain their eager behavior. See the [async contract](../docs/async-orchestrators.md) for dispatch cleanup and replay/upgrade precautions. +Release the core async support first, then coordinate this provider's version +and core dependency minimum in a dedicated release PR. - The `orchestrator_generator_wrapper` test helper accepts direct-await coroutines. Tests involving asyncio scheduling still require a replay executor or backend. diff --git a/azure-functions-durable/MIGRATION_GUIDE.md b/azure-functions-durable/MIGRATION_GUIDE.md index 7d28e07b..11ce963d 100644 --- a/azure-functions-durable/MIGRATION_GUIDE.md +++ b/azure-functions-durable/MIGRATION_GUIDE.md @@ -156,8 +156,8 @@ def say_hello(name: str) -> str: | `context.call_activity_with_retry(...)` | `ctx.call_activity(..., retry_policy=policy)` | | `context.call_sub_orchestrator(name, input_)` | `ctx.call_sub_orchestrator(name, input=input_)` | | `call_sub_orchestrator_with_retry` | `ctx.call_sub_orchestrator(..., retry_policy=...)` | -| `context.task_all(tasks)` | `task.when_all(tasks)` | -| `context.task_any(tasks)` | `task.when_any(tasks)` | +| `context.task_all(tasks)` | `task.when_all(tasks)` (eager `task.Task` handles only) | +| `context.task_any(tasks)` | `task.when_any(tasks)` (eager `task.Task` handles only) | | `context.new_guid()` | `ctx.new_uuid()` | | `df.EntityId(name, key)` | `entities.EntityInstanceId(name, key)` | @@ -167,6 +167,40 @@ retain their names. Activity, sub-orchestration, and entity calls use keyword `input` instead of the 1.x `input_` or `operationInput` names. `continue_as_new` uses `new_input`. +### Compose async operations + +The grouping mappings above apply to generator orchestrators and explicit eager +durable handles, not the cold coroutines returned by +`task.AsyncOrchestrationContext`. Do not pass `ctx.call_activity(...)` coroutines +directly to `task.when_all` or `task.when_any`. To keep durable grouping semantics +in a native async orchestrator, obtain reusable handles through +`ctx.as_task_context()`: + +```python +async def greetings(ctx: task.AsyncOrchestrationContext, names: list[str]) -> list[str]: + handles = ctx.as_task_context() + operations = [ + handles.call_activity("say_hello", input=name, return_type=str) + for name in names + ] + return await task.when_all(operations) +``` + +`task.when_all` waits for every handle before raising the first observed +failure. `task.when_any` returns the winning durable handle; await it to obtain +its result or failure. Losing handles are not cancelled; retain and join them +when their completion matters. + +Cold coroutines instead support ordered `asyncio.gather` and +`asyncio.TaskGroup`. These are not interchangeable with the wait-all-on-failure +join: default `gather` reports the first failure without cancelling siblings, +while task groups cancel and drain local siblings on failure. Local +cancellation does not stop already-issued remote work. The single-argument +async compatibility context retains `task_all` (wait for all before raising) +and `task_any` (return the winning future without cancelling losers). +See the [async composition and ownership contract](../docs/async-orchestrators.md#azure-functions-compatibility) +before changing joins or cancellation behavior in running orchestrations. + ### Retry policy mapping `RetryOptions` accepted millisecond values. Native `RetryPolicy` uses diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 09694c3b..8b0684c0 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -322,8 +322,9 @@ agree on the store and reference encoding; Functions references are JSON strings > storage error can therefore become a terminal orchestration failure. Registered orchestration and entity handlers await the store's async methods -before and after execution. Orchestrators remain synchronous generators, and -orchestration replay and entity code run on execution threads with their +before and after execution. Orchestrators may be synchronous generators or +native `async def` functions; both use synchronous replay execution. +Orchestration replay and entity code run on execution threads with their invocation logging context preserved. Their payload downloads and uploads do not occupy those threads, and serialization does not access storage during replay. Custom stores must implement genuinely nonblocking async methods to diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index d39bbbe2..6d76b806 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -16,6 +16,8 @@ retain their existing behavior. Event waits support owned cancellation and timeouts; remote operations are not cancelled by local waiter cancellation. See the [async contract](../docs/async-orchestrators.md) for dispatch cleanup and replay/upgrade precautions. +Release the core async support first, then coordinate this provider's version +and core dependency minimum in a dedicated release PR. FIXED diff --git a/eng/templates/build.yml b/eng/templates/build.yml index 655833d8..67fd7e1c 100644 --- a/eng/templates/build.yml +++ b/eng/templates/build.yml @@ -110,9 +110,12 @@ jobs: displayName: "pytest: durabletask (unit tests, no emulators)" workingDirectory: tests/durabletask - # `tests/durabletask-azuremanaged/` is NOT exercised here. - # Those tests require the Durable Task Scheduler emulator which is - # not available in the network-isolated pools. + # Exercise the in-memory gRPC compatibility cases without an emulator. + # DTS-backed cases remain in the GitHub Actions emulator matrix. + - script: pytest test_dts_coroutine_orchestrators.py -m "not dts" --verbose + displayName: "pytest: durabletask-azuremanaged (coroutines, no emulator)" + workingDirectory: tests/durabletask-azuremanaged + - script: | set -e python -P -c "import durabletask.azuremanaged; from durabletask.azuremanaged.client import DurableTaskSchedulerClient; from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker; print('durabletask.azuremanaged smoke import OK')" From e1ddad278ab504828b94ee10b5eeef8b50b23877 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 12 Nov 2026 16:24:52 -0700 Subject: [PATCH 3/4] Consolidate async changelog entries and gate Managed coroutine tests Use one ADDED entry for async orchestrators in the core and Managed changelogs and state the Managed core-release requirement in user terms. Run the in-memory Managed coroutine cases in the azuremanaged nox session, read emulator settings with defaults, reuse the credential and secure-channel pattern, and pick ports with the shared IPv6-aware helper. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 20 ++++---- durabletask-azuremanaged/CHANGELOG.md | 5 +- noxfile.py | 7 +++ .../test_dts_coroutine_orchestrators.py | 50 ++++++++++++++----- 4 files changed, 58 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 810c753b..d830fa92 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,18 +9,20 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED -- Added `async def` orchestrators with an automatically supplied -`AsyncOrchestrationContext`, cold coroutine operations, durable asyncio -timers, and supported task-group and gather composition. Generator -orchestrators retain their eager durable handles. `as_task_context()` and -`Task.wait()` enable explicit interoperability with those handles. -- Async event waits support owned cancellation and timeouts, withdrawing -abandoned subscriptions so they cannot consume later events. +- Added `async def` orchestrators. They receive an `AsyncOrchestrationContext` +whose durable operations are cold coroutines, support durable asyncio timers +and ordered composition such as `asyncio.gather`, `asyncio.TaskGroup`, and +`asyncio.timeout`, and offer event waits with owned cancellation and timeouts. +Generator orchestrators are unchanged; `as_task_context()` and `Task.wait()` +interoperate with their eager durable handles. Pass `return_type` to type and +reconstruct async sub-orchestration results. `asyncio.wait` and +`asyncio.as_completed` are not supported. > [!WARNING] > Async cleanup can run at every dispatch, including suspension; it is not -> durable workflow compensation. Converting running orchestrations or changing -> Python versions may alter replay-visible action order. See the +> durable workflow compensation. Converting running orchestrations can alter +> replay-visible action order, and recorded async histories depend on the +> Python minor version, so keep it fixed for in-flight instances. See the > [async contract](docs/async-orchestrators.md) for supported operations and > replay and upgrade guidance. diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 8b7091ff..54f86a1c 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -15,9 +15,8 @@ durable operations, and supported asyncio composition. Generator orchestrators retain their existing behavior. Event waits support owned cancellation and timeouts; remote operations are not cancelled by local waiter cancellation. See the [async contract](../docs/async-orchestrators.md) for dispatch cleanup -and replay/upgrade precautions. -Release the core async support first, then coordinate this provider's version -and core dependency minimum in a dedicated release PR. +and replay/upgrade precautions. Requires a `durabletask` release with async +orchestrator support. FIXED diff --git a/noxfile.py b/noxfile.py index 15cab4be..929ee5c4 100644 --- a/noxfile.py +++ b/noxfile.py @@ -457,6 +457,13 @@ def azuremanaged_tests(session: nox.Session) -> None: "pytest", *_pytest_warning_arguments(session), *arguments, "-m", "dts", "--verbose", ) + if not session.posargs: + # In-memory gRPC cases need no emulator but still gate PRs here. + session.run( + "pytest", *_pytest_warning_arguments(session), + "tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py", + "-m", "not dts", "--verbose", + ) finally: _stop_dts_emulator(container_name) diff --git a/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py index f8d9af6c..c2bbca90 100644 --- a/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py +++ b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py @@ -3,13 +3,14 @@ """Exercise both workers over gRPC and optionally against the DTS emulator.""" -import json import asyncio +import importlib.util +import json import os -import socket -from collections.abc import Generator +from collections.abc import Callable, Generator from datetime import timedelta -from typing import Any +from pathlib import Path +from typing import Any, cast import pytest @@ -19,6 +20,30 @@ from durabletask.testing import create_test_backend +def _load_find_free_port() -> Callable[[], int]: + # pytest imports this hyphenated directory's modules top-level, so the shared + # core test helper cannot be imported through the `tests` package here. + path = Path(__file__).resolve().parents[1] / "durabletask" / "_port_utils.py" + spec = importlib.util.spec_from_file_location("_durabletask_test_port_utils", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return cast(Callable[[], int], module.find_free_port) + + +find_free_port = _load_find_free_port() +taskhub_name = os.getenv("TASKHUB", "default") +endpoint = os.getenv("ENDPOINT", "http://localhost:8080") + + +def _get_credential(): + """Returns DefaultAzureCredential if endpoint is https, otherwise None (for emulator).""" + if endpoint.startswith("https://"): + from azure.identity import DefaultAzureCredential + return DefaultAzureCredential() + return None + + @pytest.mark.parametrize("transport", [ "core", "azuremanaged", @@ -65,20 +90,21 @@ def generator(ctx: task.OrchestrationContext, value: int) -> Generator[task.Task if transport == "dts": backend = None - host = os.environ["ENDPOINT"] - taskhub = os.environ["TASKHUB"] + host = endpoint + taskhub = taskhub_name + credential = _get_credential() else: - with socket.socket() as listener: - listener.bind(("127.0.0.1", 0)) - port = listener.getsockname()[1] + port = find_free_port() backend = create_test_backend(port=port) - host = f"localhost:{port}" + # The in-memory backend has no TLS; the http:// scheme selects an insecure channel. + host = f"http://localhost:{port}" taskhub = "test" + credential = None if transport != "core": w = DurableTaskSchedulerWorker( - host_address=host, secure_channel=False, taskhub=taskhub, token_credential=None) + host_address=host, secure_channel=True, taskhub=taskhub, token_credential=credential) c = DurableTaskSchedulerClient( - host_address=host, secure_channel=False, taskhub=taskhub, token_credential=None) + host_address=host, secure_channel=True, taskhub=taskhub, token_credential=credential) else: w = worker.TaskHubGrpcWorker(host_address=host) c = client.TaskHubGrpcClient(host_address=host) From fe1487ae7908948b4338049a240384b2f1c3100a Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 12 Nov 2026 17:04:01 -0700 Subject: [PATCH 4/4] Document the native async shape and call_http for Functions Describe two-parameter async orchestrators with AsyncOrchestrationContext, the unsupported single-parameter async shape, asyncio fan-out and as_task_context races, and the df.call_http helper. Remove the async task_all and task_any documentation and consolidate the changelog entry. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- README.md | 8 +-- azure-functions-durable/CHANGELOG.md | 18 ++--- azure-functions-durable/MIGRATION_GUIDE.md | 62 ++++++++-------- azure-functions-durable/README.md | 82 ++++++++++++++++++---- docs/async-orchestrators.md | 63 ++++++++--------- 5 files changed, 142 insertions(+), 91 deletions(-) diff --git a/README.md b/README.md index 85b25425..8bd214fd 100644 --- a/README.md +++ b/README.md @@ -52,10 +52,10 @@ async def sequence(ctx: task.AsyncOrchestrationContext, value: int) -> list[int] ``` Register the function with `TaskHubGrpcWorker`, `DurableTaskSchedulerWorker`, -or the Azure Functions `orchestration_trigger` decorator as usual. Functions -also supports single-argument `async def` orchestrators using the compatibility -context and `await context.task_all(...)`. Async orchestrators automatically -receive cold coroutine operations accepted by `create_task` and task groups. +or the Azure Functions `orchestration_trigger` decorator as usual. All three +packages use this two-argument shape for async orchestrators. Async +orchestrators automatically receive cold coroutine operations accepted by +`create_task` and task groups. Generator orchestrators keep their eager, reusable durable handles. For interoperability, `ctx.as_task_context()` exposes the eager context and `handle.wait()` creates an independently cancellable borrower. diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 566e70fc..8c0feec1 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -11,19 +11,14 @@ ADDED - A best-effort background check warns when a Function App's requirements allow a future major version, which may introduce replay-breaking changes. - -- With the corresponding core SDK, native and v1-style `async def` -orchestrators automatically receive async contexts with cold durable operations -and supported asyncio composition. The v1-style context preserves compatibility -argument names and provides async `task_all` and `task_any`. Generator -orchestrators retain their eager behavior. See the +- Added `async def` orchestrators with the core two-argument shape, +`async def orchestrator(ctx: task.AsyncOrchestrationContext, input)`, and +`df.call_http()` for durable HTTP requests from durabletask orchestration +contexts. Single-parameter `async def` orchestrators are not supported and +raise `TypeError` when registered. Generator orchestrators are unchanged. +Requires a `durabletask` release with async orchestrator support. See the [async contract](../docs/async-orchestrators.md) for dispatch cleanup and replay/upgrade precautions. -Release the core async support first, then coordinate this provider's version -and core dependency minimum in a dedicated release PR. -- The `orchestrator_generator_wrapper` test helper accepts direct-await -coroutines. Tests involving asyncio scheduling still require a replay executor -or backend. - Added opt-in `HttpRetryOptions` on `context.call_http()` for durable initial request retries with configurable timing and status selection. Empty or omitted status lists select all non-2xx responses; exhausted retries fail the call. @@ -74,7 +69,6 @@ readers before enabling new writers; existing headerless blobs still require their original compression setting. See the [core compatibility guidance](../docs/features.md#cross-sdk-compatibility). - ## v2.0.0rc4 ADDED diff --git a/azure-functions-durable/MIGRATION_GUIDE.md b/azure-functions-durable/MIGRATION_GUIDE.md index 11ce963d..c9c1f34d 100644 --- a/azure-functions-durable/MIGRATION_GUIDE.md +++ b/azure-functions-durable/MIGRATION_GUIDE.md @@ -102,12 +102,10 @@ The single-argument orchestrator receives the compatibility Use this state to validate the infrastructure upgrade before changing application APIs. -In 2.x, a single-argument `async def` orchestrator instead receives -`AsyncDurableOrchestrationContext`, with the same v1-style argument names and -awaitable durable operations. Generator orchestrators retain their existing -eager task-handle behavior. See the -[async authoring contract](../docs/async-orchestrators.md) before converting -running orchestrations. +Single-argument orchestrators must be generator or regular functions. To write +an `async def` orchestrator, use the native two-argument shape described next; +the `orchestration_trigger` decorator raises `TypeError` for a single-argument +`async def` orchestrator. ## Migrate to native orchestration APIs @@ -159,46 +157,54 @@ def say_hello(name: str) -> str: | `context.task_all(tasks)` | `task.when_all(tasks)` (eager `task.Task` handles only) | | `context.task_any(tasks)` | `task.when_any(tasks)` (eager `task.Task` handles only) | | `context.new_guid()` | `ctx.new_uuid()` | +| `context.call_http(method, uri, ...)` | `df.call_http(ctx, method, uri, ...)` | | `df.EntityId(name, key)` | `entities.EntityInstanceId(name, key)` | Methods including `create_timer`, `wait_for_external_event`, `continue_as_new`, `set_custom_status`, `call_entity`, and `signal_entity` retain their names. Activity, sub-orchestration, and entity calls use keyword `input` instead of the 1.x `input_` or `operationInput` names. -`continue_as_new` uses `new_input`. +`continue_as_new` uses `new_input`. `df.call_http` takes the request options +as keyword arguments. ### Compose async operations The grouping mappings above apply to generator orchestrators and explicit eager durable handles, not the cold coroutines returned by `task.AsyncOrchestrationContext`. Do not pass `ctx.call_activity(...)` coroutines -directly to `task.when_all` or `task.when_any`. To keep durable grouping semantics -in a native async orchestrator, obtain reusable handles through -`ctx.as_task_context()`: +directly to `task.when_all` or `task.when_any`. + +For fan-out in an async orchestrator, use ordered `asyncio.gather` or +`asyncio.TaskGroup`: ```python async def greetings(ctx: task.AsyncOrchestrationContext, names: list[str]) -> list[str]: - handles = ctx.as_task_context() - operations = [ - handles.call_activity("say_hello", input=name, return_type=str) + return list(await asyncio.gather(*( + ctx.call_activity("say_hello", input=name, return_type=str) for name in names - ] - return await task.when_all(operations) + ))) +``` + +These are not interchangeable with the generator `task.when_all` join, which +waits for every task before raising the first observed failure: default +`gather` reports the first failure without cancelling siblings, while task +groups cancel and drain local siblings on failure. Local cancellation does not +stop already-issued remote work. + +For races, obtain eager handles through `ctx.as_task_context()` and pass them to +`task.when_any`, which returns the identical winning handle: + +```python +async def approval(ctx: task.AsyncOrchestrationContext, _: Any) -> bool: + handles = ctx.as_task_context() + approved = handles.wait_for_external_event("approval", data_type=bool) + deadline = handles.create_timer(timedelta(hours=1)) + winner = await task.when_any([approved, deadline]) + return winner is approved and approved.get_result() ``` -`task.when_all` waits for every handle before raising the first observed -failure. `task.when_any` returns the winning durable handle; await it to obtain -its result or failure. Losing handles are not cancelled; retain and join them -when their completion matters. - -Cold coroutines instead support ordered `asyncio.gather` and -`asyncio.TaskGroup`. These are not interchangeable with the wait-all-on-failure -join: default `gather` reports the first failure without cancelling siblings, -while task groups cancel and drain local siblings on failure. Local -cancellation does not stop already-issued remote work. The single-argument -async compatibility context retains `task_all` (wait for all before raising) -and `task_any` (return the winning future without cancelling losers). -See the [async composition and ownership contract](../docs/async-orchestrators.md#azure-functions-compatibility) +Losing handles are not cancelled; retain and join them when their completion +matters. See the [async authoring contract](../docs/async-orchestrators.md) before changing joins or cancellation behavior in running orchestrations. ### Retry policy mapping diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 8b0684c0..dc5c0852 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -110,20 +110,50 @@ These compatibility paths remain supported. ## Async orchestrators -Both authoring styles accept `async def` and `await` without an adapter loop. -Await durable operations and compose branches with `asyncio.gather` or task -groups; `asyncio.sleep` and timeouts use durable timers. Synchronous generator -orchestrators remain supported, including mixed-style sub-orchestrations. - -Async orchestrators automatically receive cold coroutine operations accepted by -`create_task` and task groups. Native orchestrators receive -`task.AsyncOrchestrationContext`; single-argument orchestrators receive -`df.AsyncDurableOrchestrationContext`, retaining v1-style argument names and -`await context.task_all(...)` / `await context.task_any(...)`. -`wait_for_external_event()` owns its registration and optional durable timeout, -so an abandoned receive cannot consume the next event. Generator contexts -retain eager, reusable durable handles; async code can access those through -`context.as_task_context()` and use `handle.wait()` to create a borrowed waiter. +Async orchestrators use the same two-argument shape as the core SDK and Durable +Task Scheduler provider. They receive `task.AsyncOrchestrationContext` and the +orchestration input: + +```python +import asyncio +from datetime import timedelta +from typing import Any + +import azure.durable_functions as df +from durabletask import task + +app = df.DFApp() + + +@app.orchestration_trigger(context_name="context") +async def fan_out(ctx: task.AsyncOrchestrationContext, names: list[str]) -> list[str]: + return list(await asyncio.gather(*( + ctx.call_activity("say_hello", input=name, return_type=str) + for name in names + ))) + + +@app.orchestration_trigger(context_name="context") +async def approval(ctx: task.AsyncOrchestrationContext, _: Any) -> bool: + handles = ctx.as_task_context() + approved = handles.wait_for_external_event("approval", data_type=bool) + deadline = handles.create_timer(timedelta(hours=1)) + winner = await task.when_any([approved, deadline]) + return winner is approved and approved.get_result() +``` + +Durable operations on the async context are cold coroutines. Compose branches +with `asyncio.gather` or `asyncio.TaskGroup`; `asyncio.sleep` and timeouts use +durable timers. For races, schedule eager handles through +`ctx.as_task_context()` and pass them to `task.when_any`, which returns the +identical winning handle. `handle.wait()` creates an independently cancellable +waiter for an eager handle. Generator orchestrators remain supported, including +mixed-style sub-orchestrations. + +Single-argument `async def` orchestrators are not supported. The +`orchestration_trigger` decorator raises `TypeError` for them. The +single-argument `df.DurableOrchestrationContext` remains available for +generator orchestrators. The [shared async contract](../docs/async-orchestrators.md) defines the supported API subset, cancellation, per-dispatch `finally`/async-generator cleanup, @@ -132,9 +162,31 @@ threads, subprocesses, `asyncio.wait`/`as_completed`, and automatic recording of ordinary async helpers are not supported. Put nondeterministic work in ordinary activities, not orchestrators. +## Durable HTTP calls + +`df.call_http` makes durable HTTP requests from durabletask orchestration +contexts. With `task.AsyncOrchestrationContext`, it returns a cold coroutine to +`await`. With a generator's `task.OrchestrationContext`, it schedules the +request immediately and returns a task to `yield`. The result is a +`df.DurableHttpResponse`: + +```python +@app.orchestration_trigger(context_name="context") +async def check_status(ctx: task.AsyncOrchestrationContext, url: str) -> int: + response = await df.call_http(ctx, "GET", url) + return response.status_code +``` + +`df.call_http` accepts the same `content`, `headers`, `token_source`, +`is_raw_str`, and `retry_options` options as `context.call_http()` on +`df.DurableOrchestrationContext`, as keyword arguments. Both follow `202 +Accepted` responses with a `Location` header until the operation completes and +return unselected non-2xx responses rather than raising. + ## Durable HTTP failure retries -Opt in to durable retries for the **initial** `context.call_http()` request with +Opt in to durable retries for the **initial** `context.call_http()` or +`df.call_http()` request with `HttpRetryOptions`. It uses the core activity retry engine and durable timers, not an in-process HTTP retry loop: diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md index 8de9b64d..55794d71 100644 --- a/docs/async-orchestrators.md +++ b/docs/async-orchestrators.md @@ -4,8 +4,8 @@ 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. +`orchestration_trigger` decorator. Async orchestrators use the same +two-argument `(ctx, input)` shape in all three packages. ```python import asyncio @@ -156,47 +156,47 @@ 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 +## Azure Functions -Single-argument async Functions orchestrators receive -`df.AsyncDurableOrchestrationContext`, with the v1-style method and argument -names. Two-argument native orchestrators receive `task.AsyncOrchestrationContext`. +Functions async orchestrators use the two-argument shape and receive +`task.AsyncOrchestrationContext`, as in the other packages. The second argument +is the orchestration input. Single-argument `async def` orchestrators are not +supported: the `orchestration_trigger` decorator raises `TypeError` for them. +The single-argument `df.DurableOrchestrationContext` remains available for +generator orchestrators. + +Durable HTTP requests use `df.call_http`, which accepts the async context and +returns a cold coroutine: ```python +import asyncio + import azure.durable_functions as df +from durabletask import task 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 - ]) +async def greetings( + ctx: task.AsyncOrchestrationContext, names: list[str]) -> dict[str, object]: + messages = await asyncio.gather(*( + ctx.call_activity("say_hello", input=name, return_type=str) + for name in names + )) + response = await df.call_http(ctx, "GET", "https://example.com/status") + return {"greetings": messages, "status": response.status_code} ``` -`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. +Compose branches with `asyncio.gather` or `asyncio.TaskGroup`, and use +`ctx.as_task_context()` handles with `task.when_any` for races, as described +above. ## 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 +where supported by the provider. Generator and coroutine parents/children may call one another through durable sub-orchestrations. For sub-orchestration results, pass `return_type=Result` to reconstruct a custom @@ -409,8 +409,7 @@ instances and validate histories before upgrades. All workers processing an instance must use that pinned environment; 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. +The Functions `orchestrator_generator_wrapper` helper drives generator +orchestrators only. Use the real replay executor/provider or the in-memory +backend to test async orchestrators, including task groups, gather, and durable +timeouts.