diff --git a/CHANGELOG.md b/CHANGELOG.md index 16194e95..cf3f4536 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- Prototype support for `async def` orchestrators that `await` durable tasks, +including activities, timers, external events, and `when_all` / `when_any`. +Generator orchestrators remain supported. This is replay-driven syntax, not +support for asyncio I/O inside orchestrators. + ## v1.12.0 ADDED diff --git a/README.md b/README.md index 2bd8e7fb..efc47a06 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,44 @@ and [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/ ## Optional Features +### Async Orchestrators (Prototype) + +This branch prototypes `async def` orchestrators across all three packages. +Existing generator orchestrators continue to work. + +```python +from durabletask import task + + +async def sequence(ctx: task.OrchestrationContext, value: int) -> list[int]: + first = await ctx.call_activity("plus_one", input=value, return_type=int) + pending: list[task.Task[int]] = [ + ctx.call_activity("plus_one", input=first, return_type=int), + ctx.call_activity("plus_one", input=value, return_type=int), + ] + return await task.when_all(pending) +``` + +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(...)`. + +> [!WARNING] +> This is replay-driven await syntax, not general-purpose asyncio execution. +> Await only durable tasks or async helpers that themselves await durable tasks. +> Use durable timers instead of `asyncio.sleep`, and `task.when_all` / +> `task.when_any` instead of `asyncio.gather`, `create_task`, or `TaskGroup`. +> 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. + +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. + ### Large Payload Externalization Install the `azure-blob-payloads` extra to automatically offload oversized orchestration payloads to diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index b17a3654..8b9ba663 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +ADDED + +- With the corresponding core SDK prototype, native and v1-style orchestrators +can use `async def` and await durable tasks. The +`orchestrator_generator_wrapper` test helper also accepts coroutines. +Requires a coordinated release after the core SDK is published; dependency +minimums are intentionally unchanged in this spike. + ## v2.0.0rc3 ADDED diff --git a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py index 43daf0c5..d516a697 100644 --- a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py +++ b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py @@ -1,16 +1,17 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -"""Replay simulation helper for unit-testing orchestrator generators.""" +"""Replay simulation helper for unit-testing generator or coroutine orchestrators.""" -from typing import Any, Generator +from collections.abc import Coroutine, Generator +from typing import Any def orchestrator_generator_wrapper( - generator: Generator[Any, Any, Any]) -> Generator[Any, None, None]: - """Drive a user orchestrator generator the way the Durable replay engine does. + generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any]) -> Generator[Any, None, None]: + """Drive a user orchestrator the way the Durable replay engine does. - A Durable Functions orchestrator is a generator that yields task objects and + A Durable Functions orchestrator yields or awaits durable task objects and is resumed with each task's result until it returns the orchestration output. In a real execution the Durable Task runtime performs this drive loop during replay. This wrapper reproduces that loop in-process so an @@ -22,8 +23,8 @@ def orchestrator_generator_wrapper( Parameters ---------- - generator: Generator[Any, Any, Any] - The user orchestrator generator. It is expected to yield task objects + generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any] + The user orchestrator generator or coroutine. It must suspend on tasks that expose a ``result`` attribute and to be resumed with those results until it returns the orchestration output. @@ -34,13 +35,12 @@ def orchestrator_generator_wrapper( orchestrator's final return value. """ try: - previous = next(generator) + previous = generator.send(None) except StopIteration as e: # An orchestrator that returns immediately without yielding any task # completes right away (mirroring the runtime, which finishes the # orchestration on the first return). Surface its output as the final - # value and stop -- ``next`` raising here otherwise leaks as an - # unhandled ``StopIteration``. + # value and stop -- letting StopIteration escape would become RuntimeError. yield e.value return yield previous diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 338b183f..b441a795 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- With the corresponding core SDK prototype, `DurableTaskSchedulerWorker` +accepts `async def` orchestrators that await durable tasks alongside generator +orchestrators. Requires a coordinated release after the core SDK is published; +dependency minimums are intentionally unchanged in this spike. + ## v1.12.0 ADDED diff --git a/durabletask/task.py b/durabletask/task.py index b2ca7b32..278527e6 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -7,7 +7,7 @@ import logging import math from abc import ABC, abstractmethod -from collections.abc import Callable, Generator, Sequence +from collections.abc import Callable, Coroutine, Generator, Sequence from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeVar, cast, overload @@ -631,6 +631,16 @@ def __init__(self) -> None: # Dict insertion order preserves subscription order for shared races. self._parents = {} + def __await__(self) -> Generator[Task[T], Any, T]: + """Wait for this task in a replay-driven ``async def`` orchestrator. + + Durable tasks are not asyncio futures. Only the orchestration replay + engine may drive a coroutine that awaits them. + """ + if not self.is_complete: + yield self + return self.get_result() + @property def is_complete(self) -> bool: """Returns True if the task has completed, False otherwise.""" @@ -961,8 +971,11 @@ def task_id(self) -> int: return self._task_id -# Orchestrators are generators that yield tasks, receive any type, and return TOutput -Orchestrator: TypeAlias = Callable[[OrchestrationContext, TInput], Generator[Task[Any], Any, TOutput] | TOutput] +# Orchestrators yield or await durable tasks, or return an immediate result. +Orchestrator: TypeAlias = Callable[ + [OrchestrationContext, TInput], + Generator[Task[Any], Any, TOutput] | Coroutine[Any, Any, TOutput] | TOutput, +] # Activities are simple functions that can be scheduled by orchestrators Activity: TypeAlias = Callable[[ActivityContext, TInput], TOutput] diff --git a/durabletask/worker.py b/durabletask/worker.py index 25010a17..ef3cf3b3 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -8,12 +8,12 @@ import logging import os import time -from collections.abc import Callable, Generator, Sequence +from collections.abc import Callable, Coroutine, Generator, Sequence from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from threading import Event, Lock, Thread -from types import GeneratorType +from types import CoroutineType, GeneratorType from enum import Enum from typing import Any, TypeVar, cast, overload import uuid @@ -1503,7 +1503,7 @@ def _cancel_entity_batch( class _RuntimeOrchestrationContext(task.OrchestrationContext): - _generator: Generator[task.Task[Any], Any, Any] | None + _generator: Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any] | None _previous_task: task.Task[Any] | None def __init__(self, @@ -1549,21 +1549,26 @@ def __init__(self, self._exception_properties_provider = exception_properties_provider self._logger = logger - def run(self, generator: Generator[task.Task[Any], Any, Any]) -> None: + def run(self, generator: Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any]) -> None: self._generator = generator - # TODO: Do something with this task - task = next(generator) # this starts the generator - # TODO: Check if the task is null? - self._previous_task = task + self._set_next_task(generator.send(None)) + + def _set_next_task(self, next_task: Any) -> None: + if not isinstance(next_task, task.Task): + raise TypeError( + "Orchestrators must only yield or await durabletask.Task objects. " + "Use durable timers and task.when_all/when_any instead of asyncio APIs." + ) + self._previous_task = next_task def resume(self) -> None: if self._generator is None: # This is never expected unless maybe there's an issue with the history raise TypeError( - "The orchestrator generator is not initialized! Was the orchestration history corrupted?" + "The orchestrator is not initialized! Was the orchestration history corrupted?" ) - # We can resume the generator only if the previously yielded task + # We can resume the generator or coroutine only if the previous task # has reached a completed state. The only time this won't be the # case is if the user yielded on a WhenAll task and there are still # outstanding child tasks that need to be completed. @@ -1578,9 +1583,7 @@ def resume(self) -> None: # This will either return a Task or raise StopIteration if it's done. next_task = self._generator.send(self._previous_task.get_result()) - if not isinstance(next_task, task.Task): - raise TypeError("The orchestrator generator yielded a non-Task object") - self._previous_task = next_task + self._set_next_task(next_task) def set_complete( self, @@ -2521,10 +2524,15 @@ def process_event( result = fn( ctx, input - ) # this does not execute the generator, only creates it - if isinstance(result, GeneratorType): - # Start the orchestrator's generator function - ctx.run(cast(Generator[task.Task[Any], Any, Any], result)) + ) # this creates the generator or coroutine without executing it + if isinstance(result, (GeneratorType, CoroutineType)): + # Both protocols suspend on durable tasks and finish with StopIteration. + ctx.run(cast(Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any], result)) + elif inspect.isasyncgen(result): + raise TypeError( + "Async generator orchestrators are not supported. " + "Use async def with await, not yield." + ) else: # This is an orchestrator that doesn't schedule any tasks ctx.set_complete(result, pb.ORCHESTRATION_STATUS_COMPLETED) diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py index 34b44d98..347bce46 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py @@ -35,6 +35,15 @@ def activity_chain(ctx: task.OrchestrationContext, _: Any): return [first, second, third] +@bp.orchestration_trigger(context_name="context") +async def async_await_probe(ctx: task.OrchestrationContext, _: Any) -> dict[str, Any]: + greeting = await ctx.call_activity("say_hello", input="Tokyo", return_type=str) + values = await task.when_all([ctx.call_activity("square", input=i, return_type=int) for i in (2, 3)]) + await ctx.create_timer(timedelta(milliseconds=10)) + child: list[str] = await ctx.call_sub_orchestrator("activity_chain") + return {"greeting": greeting, "total": sum(values), "child": child} + + @bp.orchestration_trigger(context_name="context") def logging_filtered(ctx: task.OrchestrationContext, _: Any): _LOGGER.info("worker-info-filter-anchor %s", ctx.instance_id) diff --git a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py index a96467c7..d4766d87 100644 --- a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py @@ -13,6 +13,7 @@ """ from datetime import timedelta +from typing import Any import azure.durable_functions as df @@ -31,6 +32,15 @@ def activity_chain(context: df.DurableOrchestrationContext): return [first, second, third] +@bp.orchestration_trigger(context_name="context") +async def async_await_probe(context: df.DurableOrchestrationContext) -> dict[str, Any]: + greeting = await context.call_activity("say_hello", "Tokyo") + values = await context.task_all([context.call_activity("square", i) for i in (2, 3)]) + await context.create_timer(context.current_utc_datetime + timedelta(milliseconds=10)) + child = await context.call_sub_orchestrator("activity_chain") + return {"greeting": greeting, "total": sum(values), "child": child} + + @bp.orchestration_trigger(context_name="context") def fan_out_fan_in(context: df.DurableOrchestrationContext): count = context.get_input() or 5 diff --git a/tests/azure-functions-durable/e2e/test_coroutine_e2e.py b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py new file mode 100644 index 00000000..06215119 --- /dev/null +++ b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py @@ -0,0 +1,24 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Real-host smoke coverage for both async orchestrator authoring styles.""" + +import pytest + +from ._harness import FunctionApp + + +pytestmark = pytest.mark.functions_e2e + + +@pytest.mark.parametrize("app_fixture", ["dtask_app", "v1_app"]) +def test_async_await_orchestrator(request: pytest.FixtureRequest, app_fixture: str) -> None: + app: FunctionApp = request.getfixturevalue(app_fixture) + instance_id = app.start_orchestration("async_await_probe") + status = app.wait_for_completion(instance_id) + assert status["runtimeStatus"].upper() == "COMPLETED", status + assert status["output"] == { + "greeting": "Hello Tokyo!", + "total": 13, + "child": ["Hello Tokyo!", "Hello Seattle!", "Hello London!"], + } diff --git a/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py new file mode 100644 index 00000000..f264499c --- /dev/null +++ b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py @@ -0,0 +1,106 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Coroutine replay through native and compatibility Functions entry points.""" + +import base64 +import json +from collections.abc import Callable +from typing import Any + +import pytest + +import azure.durable_functions as df +from azure.durable_functions.orchestrator import Orchestrator +from azure.durable_functions.testing import orchestrator_generator_wrapper +from azure.durable_functions.worker import DurableFunctionsWorker +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb + + +@pytest.mark.parametrize("native", [False, True], ids=["compat", "native"]) +@pytest.mark.parametrize("entry", ["worker-sync", "worker-async", "decorator", "v1-handle"]) +async def test_coroutine_replay_through_functions_entry_points(native: bool, entry: str) -> None: + async def modern(ctx: task.OrchestrationContext, value: int) -> list[Any]: + return await task.when_all([ + ctx.call_activity("work", input=value, return_type=str), + ctx.call_activity("work", input=value + 1, return_type=str), + ]) + + async def compatible(ctx: df.DurableOrchestrationContext) -> list[Any]: + value = ctx.get_input() + return await ctx.task_all([ + ctx.call_activity("work", value), + ctx.call_activity("work", value + 1), + ]) + + fn: Callable[..., Any] = modern if native else compatible + worker = DurableFunctionsWorker() + handle = Orchestrator.create(fn) + app = df.DFApp() + builder = app.orchestration_trigger(context_name="ctx")(fn) + registered = builder.build().get_user_function() + instance_id = "async-functions" + + async def invoke(request: pb.OrchestratorRequest) -> pb.OrchestratorResponse: + encoded = base64.b64encode(request.SerializeToString()).decode() + if entry == "worker-sync": + output = worker.execute_orchestration_request(fn, encoded) + elif entry == "worker-async": + output = await worker.execute_orchestration_request_async(fn, encoded) + elif entry == "decorator": + output = await registered(ctx=encoded) + else: + output = handle(encoded) + response = pb.OrchestratorResponse() + response.ParseFromString(base64.b64decode(output)) + return response + + start = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event(fn.__name__, instance_id, "3"), + ] + first = await invoke(pb.OrchestratorRequest(instanceId=instance_id, newEvents=start)) + assert [a.scheduleTask.name for a in first.actions] == ["work", "work"] + assert [json.loads(a.scheduleTask.input.value) for a in first.actions] == [3, 4] + assert [a.id for a in first.actions] == [1, 2] + + response = await invoke(pb.OrchestratorRequest( + instanceId=instance_id, + pastEvents=start + [helpers.new_task_scheduled_event(i, "work") for i in (1, 2)], + newEvents=[ + helpers.new_task_completed_event(2, '"second"'), + helpers.new_task_completed_event(1, '"first"'), + ], + )) + assert len(response.actions) == 1 + complete = response.actions[0].completeOrchestration + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert json.loads(complete.result.value) == ["first", "second"] + + +@pytest.mark.parametrize("fail", [False, True]) +def test_testing_wrapper_drives_coroutine(fail: bool) -> None: + pending = task.CompletableTask[str]() + + async def orchestrator() -> str: + try: + return await pending + except task.TaskFailedError: + return "caught" + + values = orchestrator_generator_wrapper(orchestrator()) + assert next(values) is pending + if fail: + pending.fail("boom", ValueError("boom")) + else: + pending.complete("done") + assert list(values) == ["caught" if fail else "done"] + + +def test_testing_wrapper_drives_immediate_coroutine() -> None: + async def orchestrator() -> str: + return "done" + + assert list(orchestrator_generator_wrapper(orchestrator())) == ["done"] 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..42fd4e3a --- /dev/null +++ b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py @@ -0,0 +1,90 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Exercise both workers over gRPC and optionally against the DTS emulator.""" + +import json +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"]) +def test_mixed_orchestrators_over_grpc(transport: str, coroutine_parent: bool) -> None: + def increment(ctx: task.ActivityContext, value: int) -> int: + return value + 1 + + async def coroutine_child(ctx: task.OrchestrationContext, value: int) -> int: + await ctx.create_timer(timedelta(milliseconds=10)) + return await ctx.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.OrchestrationContext, value: int) -> list[Any]: + values = await task.when_all([ + ctx.call_sub_orchestrator(coroutine_child, input=value), + ctx.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() diff --git a/tests/durabletask/test_coroutine_orchestrators.py b/tests/durabletask/test_coroutine_orchestrators.py new file mode 100644 index 00000000..2eaef122 --- /dev/null +++ b/tests/durabletask/test_coroutine_orchestrators.py @@ -0,0 +1,279 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Replay-level coverage for the async/await orchestration prototype.""" + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator, Generator +from datetime import datetime, timedelta +from typing import Any + +import pytest + +from durabletask import entities, task, worker +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb +from durabletask.serialization import JsonDataConverter + + +INSTANCE_ID = "coroutine-test" +NOW = datetime(2026, 1, 1) + + +def _start() -> list[pb.HistoryEvent]: + return [ + helpers.new_orchestrator_started_event(NOW), + helpers.new_execution_started_event("test", INSTANCE_ID, json.dumps(3)), + ] + + +def _execute( + fn: task.Orchestrator[Any, Any], + old: list[pb.HistoryEvent], + new: list[pb.HistoryEvent], +) -> worker.ExecutionResults: + registry = worker._Registry() # pyright: ignore[reportPrivateUsage] + registry.add_named_orchestrator("test", fn) + executor = worker._OrchestrationExecutor( # pyright: ignore[reportPrivateUsage] + registry, logging.getLogger(__name__), JsonDataConverter()) + return executor.execute(INSTANCE_ID, old, new) + + +def _completion(result: worker.ExecutionResults) -> pb.CompleteOrchestrationAction: + completions = [a.completeOrchestration for a in result.actions if a.HasField("completeOrchestration")] + assert len(completions) == 1 + return completions[0] + + +def test_coroutine_immediate_return() -> None: + async def orchestrator(ctx: task.OrchestrationContext, value: int) -> int: + return value + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert json.loads(complete.result.value) == 3 + + +def test_await_is_concrete_and_completed_tasks_do_not_suspend() -> None: + assert "__await__" not in task.Task.__abstractmethods__ + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[int]: + completed = task.CompletableTask[int]() + completed.complete(7) + return [await completed, await completed] + + assert json.loads(_completion(_execute(orchestrator, [], _start())).result.value) == [7, 7] + + +def test_precompleted_failure_and_cancellation_are_catchable() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[str]: + failed = task.CompletableTask[int]() + failed.fail("failed", ValueError("boom")) + cancelled = task.TimerTask() + cancelled.cancel() + results: list[str] = [] + try: + await failed + except task.TaskFailedError: + results.append("failed") + try: + await cancelled + except task.TaskCancelledError: + results.append("cancelled") + return results + + assert json.loads(_completion(_execute(orchestrator, [], _start())).result.value) == [ + "failed", "cancelled"] + + +def test_nested_async_helpers_replay_with_generator_action_parity() -> None: + async def helper(ctx: task.OrchestrationContext, value: int) -> int: + return await ctx.call_activity("increment", input=value, return_type=int) + + async def coroutine(ctx: task.OrchestrationContext, value: int) -> int: + first = await helper(ctx, value) + return await helper(ctx, first) + + def generator(ctx: task.OrchestrationContext, value: int) -> Generator[task.Task[Any], Any, int]: + first = yield ctx.call_activity("increment", input=value, return_type=int) + return (yield ctx.call_activity("increment", input=first, return_type=int)) + + batches = [ + _start(), + [helpers.new_task_scheduled_event(1, "increment"), + helpers.new_task_completed_event(1, json.dumps(4))], + [helpers.new_task_scheduled_event(2, "increment"), + helpers.new_task_completed_event(2, json.dumps(5))], + ] + old: list[pb.HistoryEvent] = [] + result: worker.ExecutionResults | None = None + for new in batches: + result = _execute(coroutine, old, new) + assert result.actions == _execute(generator, old, new).actions + old += new + assert result is not None + assert json.loads(_completion(result).result.value) == 5 + + +def test_fan_out_waits_for_all_and_preserves_input_order() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[str]: + pending: list[task.Task[str]] = [ + ctx.call_activity("work", input=i, return_type=str) for i in range(2)] + return await task.when_all(pending) + + old = _start() + assert len(_execute(orchestrator, [], old).actions) == 2 + old += [helpers.new_task_scheduled_event(i, "work") for i in (1, 2)] + second = [helpers.new_task_completed_event(2, '"second"')] + assert _execute(orchestrator, old, second).actions == [] + complete = _completion(_execute( + orchestrator, old + second, [helpers.new_task_completed_event(1, '"first"')])) + assert json.loads(complete.result.value) == ["first", "second"] + + +@pytest.mark.parametrize("caught", [False, True]) +def test_activity_failure_is_thrown_into_coroutine(caught: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + try: + await ctx.call_activity("boom") + except task.TaskFailedError: + if not caught: + raise + return "caught" + return "unexpected" + + old = _start() + [helpers.new_task_scheduled_event(1, "boom")] + complete = _completion(_execute( + orchestrator, old, [helpers.new_task_failed_event(1, ValueError("boom"))])) + assert complete.orchestrationStatus == ( + pb.ORCHESTRATION_STATUS_COMPLETED if caught else pb.ORCHESTRATION_STATUS_FAILED) + if caught: + assert json.loads(complete.result.value) == "caught" + else: + assert complete.failureDetails.innerFailure.errorMessage == "boom" + + +@pytest.mark.parametrize("event_wins", [False, True]) +def test_when_any_timer_and_external_event(event_wins: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + event = ctx.wait_for_external_event("approval") + timer = ctx.create_timer(timedelta(seconds=1)) + winner = await task.when_any([event, timer]) + if winner is event: + timer.cancel() + return await event + return "timeout" + + deadline = NOW + timedelta(seconds=1) + old = _start() + [helpers.new_timer_created_event(1, deadline)] + new = [helpers.new_event_raised_event("approval", '"approved"')] if event_wins else [ + helpers.new_timer_fired_event(1, deadline)] + complete = _completion(_execute(orchestrator, old, new)) + assert json.loads(complete.result.value) == ("approved" if event_wins else "timeout") + + +def test_sub_orchestration_and_replay_flags() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[Any]: + before = ctx.is_replaying + value = await ctx.call_sub_orchestrator("child", instance_id="child-test", return_type=str) + return [value, before, ctx.is_replaying] + + old = _start() + [helpers.new_sub_orchestration_created_event(1, "child", "child-test")] + complete = _completion(_execute( + orchestrator, old, [helpers.new_sub_orchestration_completed_event(1, '"done"')])) + assert json.loads(complete.result.value) == ["done", True, False] + + +def test_retry_timer_resumes_same_coroutine() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + return await ctx.call_activity( + "retry", retry_policy=task.RetryPolicy( + first_retry_interval=timedelta(seconds=1), max_number_of_attempts=2)) + + old = _start() + [helpers.new_task_scheduled_event(1, "retry")] + failure = [helpers.new_task_failed_event(1, ValueError("retry me"))] + retry = _execute(orchestrator, old, failure).actions + assert len(retry) == 1 + assert retry[0].id == 2 + deadline = retry[0].createTimer.fireAt.ToDatetime() + old += failure + [helpers.new_timer_created_event(2, deadline)] + fired = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(2, deadline)] + scheduled = _execute(orchestrator, old, fired).actions + assert len(scheduled) == 1 + assert scheduled[0].id == 1 + assert scheduled[0].scheduleTask.name == "retry" + old += fired + [helpers.new_task_scheduled_event(1, "retry")] + complete = _completion(_execute( + orchestrator, old, [helpers.new_task_completed_event(1, '"success"')])) + assert json.loads(complete.result.value) == "success" + + +def test_entity_lock_and_call_suspend_coroutine() -> None: + entity = entities.EntityInstanceId("counter", "one") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + with await ctx.lock_entities([entity]): + return await ctx.call_entity(entity, "get") + + old = _start() + lock = _execute(orchestrator, [], old).actions[0].sendEntityMessage.entityLockRequested + request = json.dumps({"id": lock.criticalSectionId}) + new = [ + helpers.new_event_sent_event(1, str(entity), request), + helpers.new_event_raised_event(lock.criticalSectionId, None), + ] + actions = _execute(orchestrator, old, new).actions + assert len(actions) == 1 + call = actions[0].sendEntityMessage.entityOperationCalled + assert call.targetInstanceId.value == str(entity) + old += new + [helpers.new_event_sent_event(2, str(entity), json.dumps({"id": call.requestId}))] + complete = _completion(_execute( + orchestrator, old, [helpers.new_event_raised_event(call.requestId, json.dumps({"result": "42"}))])) + assert json.loads(complete.result.value) == 42 + + +def test_continue_as_new_from_coroutine() -> None: + async def orchestrator(ctx: task.OrchestrationContext, value: int) -> None: + await ctx.wait_for_external_event("next") + ctx.continue_as_new(value + 1) + + complete = _completion(_execute( + orchestrator, _start(), [helpers.new_event_raised_event("next", None)])) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + assert json.loads(complete.result.value) == 4 + + +def test_nondeterminism_still_detected_for_coroutines() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + return await ctx.call_activity("changed") + + complete = _completion(_execute( + orchestrator, _start(), [helpers.new_task_scheduled_event(1, "original")])) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType.endswith("NonDeterminismError") + + +@pytest.mark.parametrize("after_activity", [False, True]) +def test_non_durable_suspension_is_rejected(after_activity: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + if after_activity: + await ctx.call_activity("work") + await asyncio.sleep(0) + + old = _start() + [helpers.new_task_scheduled_event(1, "work")] if after_activity else [] + new = [helpers.new_task_completed_event(1, "null")] if after_activity else _start() + complete = _completion(_execute(orchestrator, old, new)) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType == "builtins.TypeError" + assert "must only yield or await durabletask.Task" in complete.failureDetails.errorMessage + + +def test_async_generator_is_rejected() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> AsyncGenerator[task.Task[Any], None]: + yield ctx.call_activity("work", return_type=int) + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "Async generator orchestrators are not supported" in complete.failureDetails.errorMessage