Skip to content

Commit b4e4f97

Browse files
andystaplesCopilot
andcommitted
Prototype replay-driven async orchestrators across providers
Make durable tasks awaitable, drive native coroutines through the existing replay engine, and preserve generator compatibility. Include provider smoke coverage, prototype documentation, and coordinated-release notes. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
1 parent d9cf9e2 commit b4e4f97

13 files changed

Lines changed: 629 additions & 30 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
77

88
## Unreleased
99

10+
ADDED
11+
12+
- Prototype support for `async def` orchestrators that `await` durable tasks,
13+
including activities, timers, external events, and `when_all` / `when_any`.
14+
Generator orchestrators remain supported. This is replay-driven syntax, not
15+
support for asyncio I/O inside orchestrators.
16+
1017
## v1.12.0
1118

1219
ADDED

‎README.md‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,44 @@ and [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/
2727

2828
## Optional Features
2929

30+
### Async Orchestrators (Prototype)
31+
32+
This branch prototypes `async def` orchestrators across all three packages.
33+
Existing generator orchestrators continue to work.
34+
35+
```python
36+
from durabletask import task
37+
38+
39+
async def sequence(ctx: task.OrchestrationContext, value: int) -> list[int]:
40+
first = await ctx.call_activity("plus_one", input=value, return_type=int)
41+
pending: list[task.Task[int]] = [
42+
ctx.call_activity("plus_one", input=first, return_type=int),
43+
ctx.call_activity("plus_one", input=value, return_type=int),
44+
]
45+
return await task.when_all(pending)
46+
```
47+
48+
Register the function with `TaskHubGrpcWorker`, `DurableTaskSchedulerWorker`,
49+
or the Azure Functions `orchestration_trigger` decorator as usual. Functions
50+
also supports single-argument `async def` orchestrators using the compatibility
51+
context and `await context.task_all(...)`.
52+
53+
> [!WARNING]
54+
> This is replay-driven await syntax, not general-purpose asyncio execution.
55+
> Await only durable tasks or async helpers that themselves await durable tasks.
56+
> Use durable timers instead of `asyncio.sleep`, and `task.when_all` /
57+
> `task.when_any` instead of `asyncio.gather`, `create_task`, or `TaskGroup`.
58+
> Perform network, storage, and other nondeterministic I/O in activities.
59+
> Completed foreign awaitables and synchronous side effects cannot reliably be
60+
> detected by the replay engine.
61+
62+
An async helper is not a durable sub-orchestration: use
63+
`await ctx.call_sub_orchestrator(...)` for a separate orchestration instance.
64+
An `async def` function containing `yield` is an async generator, not a supported
65+
orchestrator. Converting an existing orchestrator must preserve durable action
66+
order and payloads for replay compatibility.
67+
3068
### Large Payload Externalization
3169

3270
Install the `azure-blob-payloads` extra to automatically offload oversized orchestration payloads to

‎azure-functions-durable/CHANGELOG.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
77

88
## Unreleased
99

10+
ADDED
11+
12+
- With the corresponding core SDK prototype, native and v1-style orchestrators
13+
can use `async def` and await durable tasks. The
14+
`orchestrator_generator_wrapper` test helper also accepts coroutines.
15+
Requires a coordinated release after the core SDK is published; dependency
16+
minimums are intentionally unchanged in this spike.
17+
1018
## v2.0.0rc3
1119

1220
ADDED

‎azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,17 @@
11
# Copyright (c) Microsoft Corporation.
22
# Licensed under the MIT License.
33

4-
"""Replay simulation helper for unit-testing orchestrator generators."""
4+
"""Replay simulation helper for unit-testing generator or coroutine orchestrators."""
55

6-
from typing import Any, Generator
6+
from collections.abc import Coroutine, Generator
7+
from typing import Any
78

89

910
def orchestrator_generator_wrapper(
10-
generator: Generator[Any, Any, Any]) -> Generator[Any, None, None]:
11-
"""Drive a user orchestrator generator the way the Durable replay engine does.
11+
generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any]) -> Generator[Any, None, None]:
12+
"""Drive a user orchestrator the way the Durable replay engine does.
1213
13-
A Durable Functions orchestrator is a generator that yields task objects and
14+
A Durable Functions orchestrator yields or awaits durable task objects and
1415
is resumed with each task's result until it returns the orchestration
1516
output. In a real execution the Durable Task runtime performs this drive
1617
loop during replay. This wrapper reproduces that loop in-process so an
@@ -22,8 +23,8 @@ def orchestrator_generator_wrapper(
2223
2324
Parameters
2425
----------
25-
generator: Generator[Any, Any, Any]
26-
The user orchestrator generator. It is expected to yield task objects
26+
generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any]
27+
The user orchestrator generator or coroutine. It must suspend on tasks
2728
that expose a ``result`` attribute and to be resumed with those results
2829
until it returns the orchestration output.
2930
@@ -34,13 +35,12 @@ def orchestrator_generator_wrapper(
3435
orchestrator's final return value.
3536
"""
3637
try:
37-
previous = next(generator)
38+
previous = generator.send(None)
3839
except StopIteration as e:
3940
# An orchestrator that returns immediately without yielding any task
4041
# completes right away (mirroring the runtime, which finishes the
4142
# orchestration on the first return). Surface its output as the final
42-
# value and stop -- ``next`` raising here otherwise leaks as an
43-
# unhandled ``StopIteration``.
43+
# value and stop -- letting StopIteration escape would become RuntimeError.
4444
yield e.value
4545
return
4646
yield previous

‎durabletask-azuremanaged/CHANGELOG.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
77

88
## Unreleased
99

10+
ADDED
11+
12+
- With the corresponding core SDK prototype, `DurableTaskSchedulerWorker`
13+
accepts `async def` orchestrators that await durable tasks alongside generator
14+
orchestrators. Requires a coordinated release after the core SDK is published;
15+
dependency minimums are intentionally unchanged in this spike.
16+
1017
## v1.12.0
1118

1219
ADDED

‎durabletask/task.py‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
import logging
88
import math
99
from abc import ABC, abstractmethod
10-
from collections.abc import Callable, Generator, Sequence
10+
from collections.abc import Callable, Coroutine, Generator, Sequence
1111
from dataclasses import dataclass
1212
from datetime import datetime, timedelta, timezone
1313
from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeVar, cast, overload
@@ -631,6 +631,16 @@ def __init__(self) -> None:
631631
# Dict insertion order preserves subscription order for shared races.
632632
self._parents = {}
633633

634+
def __await__(self) -> Generator[Task[T], Any, T]:
635+
"""Wait for this task in a replay-driven ``async def`` orchestrator.
636+
637+
Durable tasks are not asyncio futures. Only the orchestration replay
638+
engine may drive a coroutine that awaits them.
639+
"""
640+
if not self.is_complete:
641+
yield self
642+
return self.get_result()
643+
634644
@property
635645
def is_complete(self) -> bool:
636646
"""Returns True if the task has completed, False otherwise."""
@@ -961,8 +971,11 @@ def task_id(self) -> int:
961971
return self._task_id
962972

963973

964-
# Orchestrators are generators that yield tasks, receive any type, and return TOutput
965-
Orchestrator: TypeAlias = Callable[[OrchestrationContext, TInput], Generator[Task[Any], Any, TOutput] | TOutput]
974+
# Orchestrators yield or await durable tasks, or return an immediate result.
975+
Orchestrator: TypeAlias = Callable[
976+
[OrchestrationContext, TInput],
977+
Generator[Task[Any], Any, TOutput] | Coroutine[Any, Any, TOutput] | TOutput,
978+
]
966979

967980
# Activities are simple functions that can be scheduled by orchestrators
968981
Activity: TypeAlias = Callable[[ActivityContext, TInput], TOutput]

‎durabletask/worker.py‎

Lines changed: 25 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,12 @@
88
import logging
99
import os
1010
import time
11-
from collections.abc import Callable, Generator, Sequence
11+
from collections.abc import Callable, Coroutine, Generator, Sequence
1212
from concurrent.futures import ThreadPoolExecutor
1313
from dataclasses import dataclass, field
1414
from datetime import datetime, timedelta, timezone
1515
from threading import Event, Lock, Thread
16-
from types import GeneratorType
16+
from types import CoroutineType, GeneratorType
1717
from enum import Enum
1818
from typing import Any, TypeVar, cast, overload
1919
import uuid
@@ -1503,7 +1503,7 @@ def _cancel_entity_batch(
15031503

15041504

15051505
class _RuntimeOrchestrationContext(task.OrchestrationContext):
1506-
_generator: Generator[task.Task[Any], Any, Any] | None
1506+
_generator: Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any] | None
15071507
_previous_task: task.Task[Any] | None
15081508

15091509
def __init__(self,
@@ -1549,21 +1549,26 @@ def __init__(self,
15491549
self._exception_properties_provider = exception_properties_provider
15501550
self._logger = logger
15511551

1552-
def run(self, generator: Generator[task.Task[Any], Any, Any]) -> None:
1552+
def run(self, generator: Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any]) -> None:
15531553
self._generator = generator
1554-
# TODO: Do something with this task
1555-
task = next(generator) # this starts the generator
1556-
# TODO: Check if the task is null?
1557-
self._previous_task = task
1554+
self._set_next_task(generator.send(None))
1555+
1556+
def _set_next_task(self, next_task: Any) -> None:
1557+
if not isinstance(next_task, task.Task):
1558+
raise TypeError(
1559+
"Orchestrators must only yield or await durabletask.Task objects. "
1560+
"Use durable timers and task.when_all/when_any instead of asyncio APIs."
1561+
)
1562+
self._previous_task = next_task
15581563

15591564
def resume(self) -> None:
15601565
if self._generator is None:
15611566
# This is never expected unless maybe there's an issue with the history
15621567
raise TypeError(
1563-
"The orchestrator generator is not initialized! Was the orchestration history corrupted?"
1568+
"The orchestrator is not initialized! Was the orchestration history corrupted?"
15641569
)
15651570

1566-
# We can resume the generator only if the previously yielded task
1571+
# We can resume the generator or coroutine only if the previous task
15671572
# has reached a completed state. The only time this won't be the
15681573
# case is if the user yielded on a WhenAll task and there are still
15691574
# outstanding child tasks that need to be completed.
@@ -1578,9 +1583,7 @@ def resume(self) -> None:
15781583
# This will either return a Task or raise StopIteration if it's done.
15791584
next_task = self._generator.send(self._previous_task.get_result())
15801585

1581-
if not isinstance(next_task, task.Task):
1582-
raise TypeError("The orchestrator generator yielded a non-Task object")
1583-
self._previous_task = next_task
1586+
self._set_next_task(next_task)
15841587

15851588
def set_complete(
15861589
self,
@@ -2521,10 +2524,15 @@ def process_event(
25212524

25222525
result = fn(
25232526
ctx, input
2524-
) # this does not execute the generator, only creates it
2525-
if isinstance(result, GeneratorType):
2526-
# Start the orchestrator's generator function
2527-
ctx.run(cast(Generator[task.Task[Any], Any, Any], result))
2527+
) # this creates the generator or coroutine without executing it
2528+
if isinstance(result, (GeneratorType, CoroutineType)):
2529+
# Both protocols suspend on durable tasks and finish with StopIteration.
2530+
ctx.run(cast(Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any], result))
2531+
elif inspect.isasyncgen(result):
2532+
raise TypeError(
2533+
"Async generator orchestrators are not supported. "
2534+
"Use async def with await, not yield."
2535+
)
25282536
else:
25292537
# This is an orchestrator that doesn't schedule any tasks
25302538
ctx.set_complete(result, pb.ORCHESTRATION_STATUS_COMPLETED)

‎tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,15 @@ def activity_chain(ctx: task.OrchestrationContext, _: Any):
3535
return [first, second, third]
3636

3737

38+
@bp.orchestration_trigger(context_name="context")
39+
async def async_await_probe(ctx: task.OrchestrationContext, _: Any) -> dict[str, Any]:
40+
greeting = await ctx.call_activity("say_hello", input="Tokyo", return_type=str)
41+
values = await task.when_all([ctx.call_activity("square", input=i, return_type=int) for i in (2, 3)])
42+
await ctx.create_timer(timedelta(milliseconds=10))
43+
child: list[str] = await ctx.call_sub_orchestrator("activity_chain")
44+
return {"greeting": greeting, "total": sum(values), "child": child}
45+
46+
3847
@bp.orchestration_trigger(context_name="context")
3948
def logging_filtered(ctx: task.OrchestrationContext, _: Any):
4049
_LOGGER.info("worker-info-filter-anchor %s", ctx.instance_id)

‎tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
"""
1414

1515
from datetime import timedelta
16+
from typing import Any
1617

1718
import azure.durable_functions as df
1819

@@ -31,6 +32,15 @@ def activity_chain(context: df.DurableOrchestrationContext):
3132
return [first, second, third]
3233

3334

35+
@bp.orchestration_trigger(context_name="context")
36+
async def async_await_probe(context: df.DurableOrchestrationContext) -> dict[str, Any]:
37+
greeting = await context.call_activity("say_hello", "Tokyo")
38+
values = await context.task_all([context.call_activity("square", i) for i in (2, 3)])
39+
await context.create_timer(context.current_utc_datetime + timedelta(milliseconds=10))
40+
child = await context.call_sub_orchestrator("activity_chain")
41+
return {"greeting": greeting, "total": sum(values), "child": child}
42+
43+
3444
@bp.orchestration_trigger(context_name="context")
3545
def fan_out_fan_in(context: df.DurableOrchestrationContext):
3646
count = context.get_input() or 5
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
# Copyright (c) Microsoft Corporation.
2+
# Licensed under the MIT License.
3+
4+
"""Real-host smoke coverage for both async orchestrator authoring styles."""
5+
6+
import pytest
7+
8+
from ._harness import FunctionApp
9+
10+
11+
pytestmark = pytest.mark.functions_e2e
12+
13+
14+
@pytest.mark.parametrize("app_fixture", ["dtask_app", "v1_app"])
15+
def test_async_await_orchestrator(request: pytest.FixtureRequest, app_fixture: str) -> None:
16+
app: FunctionApp = request.getfixturevalue(app_fixture)
17+
instance_id = app.start_orchestration("async_await_probe")
18+
status = app.wait_for_completion(instance_id)
19+
assert status["runtimeStatus"].upper() == "COMPLETED", status
20+
assert status["output"] == {
21+
"greeting": "Hello Tokyo!",
22+
"total": 13,
23+
"child": ["Hello Tokyo!", "Hello Seattle!", "Hello London!"],
24+
}

0 commit comments

Comments
 (0)