From 68285d47cab0cfe06d046325923423043d2d5703 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 11 Nov 2026 12:48:26 -0700 Subject: [PATCH 1/4] Add asyncio composition and durable timers for coroutine orchestrators Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- durabletask/internal/asyncio_loop.py | 166 ++++- durabletask/task.py | 6 +- .../durabletask/test_asyncio_orchestrators.py | 588 ++++++++++++++++++ 3 files changed, 747 insertions(+), 13 deletions(-) create mode 100644 tests/durabletask/test_asyncio_orchestrators.py diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index 8865123b..3b0bea9e 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -7,11 +7,12 @@ import asyncio import contextvars +import math import logging from collections import deque from collections.abc import Callable, Coroutine, Generator -from datetime import datetime -from typing import Any, TypeVar +from datetime import datetime, timedelta +from typing import Any, TypeVar, cast from durabletask import task @@ -33,11 +34,67 @@ def detach(self) -> None: self._notify_parent() +class _DurableAsyncTask(asyncio.Task[T]): + def __await__(self) -> Generator[Any, None, T]: + try: + return (yield from super().__await__()) + finally: + self._retire_if_observed() + + __iter__ = __await__ + + def result(self) -> T: + try: + return super().result() + finally: + self._retire_if_observed() + + def exception(self) -> BaseException | None: + try: + return super().exception() + finally: + self._retire_if_observed() + + def _retire_if_observed(self) -> None: + loop = self.get_loop() + assert isinstance(loop, DurableEventLoop) + loop.retire_task(self) + + def cancel(self, msg: Any = None) -> bool: + loop = self.get_loop() + assert isinstance(loop, DurableEventLoop) + previous = loop.cancelling_task + loop.cancelling_task = self + try: + return super().cancel(msg) + finally: + loop.cancelling_task = previous + + +class _DurableTimerHandle(asyncio.TimerHandle): + def __eq__(self, other: object) -> bool: + return self is other + + __hash__ = object.__hash__ + + def __init__( + self, when: float, callback: Callable[..., Any], args: tuple[Any, ...], + loop: DurableEventLoop, context: contextvars.Context | None, + ) -> None: + super().__init__(when, callback, args, loop, context=context) + self.durable_timer: task.TimerTask | None = None + + def cancel(self) -> None: + super().cancel() + if self.durable_timer is not None: + self.durable_timer.cancel() + + class DurableEventLoop(asyncio.AbstractEventLoop): """A FIFO microtask loop; it never polls a selector or advances wall time. - Only history events can complete durable waits. asyncio owns coroutine - stepping; this loop owns the ordering of ready callbacks. + Only history events can complete durable waits/timers. asyncio owns coroutine + stepping, gather and structured cancellation; this loop owns their ordering. """ def __init__( @@ -48,10 +105,14 @@ def __init__( self._is_complete = is_complete self._logger = logger or logging.getLogger(__name__) self._ready: deque[asyncio.Handle] = deque() - self._tasks: list[asyncio.Task[Any]] = [] + self._cancel_ready: list[tuple[int, int, asyncio.Handle]] = [] + self._tasks: dict[asyncio.Task[Any], int] = {} + self._task_sequence = 0 + self.cancelling_task: asyncio.Task[Any] | None = None self._waits: dict[int, tuple[task.Task[Any], list[asyncio.Future[Any]]]] = {} self._observers: dict[int, _CompletionObserver] = {} self._completions: deque[Callable[[], None]] = deque() + self._timers: dict[_DurableTimerHandle, _CompletionObserver] = {} self._closed = False self._closing = False self._fatal: BaseException | None = None @@ -86,18 +147,75 @@ def create_task( if kwargs: coro.close() raise TypeError("Custom/eager task execution is not supported in durable orchestrators.") - future = asyncio.Task(coro, loop=self, name=name, context=context) - self._tasks.append(future) + future = _DurableAsyncTask(coro, loop=self, name=name, context=context) + self._tasks[future] = self._task_sequence + self._task_sequence += 1 + future.add_done_callback(self.retire_task) return future + def retire_task(self, future: asyncio.Task[Any]) -> None: + # A failed task may be observed after its done callbacks have run. + # Inspect without retrieving exceptions so unobserved failures still log. + if not self._closing and future.done() and not getattr(future, "_log_traceback", False): + self._tasks.pop(future, None) + def call_soon( self, callback: Callable[..., Any], *args: Any, context: contextvars.Context | None = None, ) -> asyncio.Handle: handle = asyncio.Handle(callback, args, self, context=context) - self._ready.append(handle) + if self.cancelling_task is not None: + # TaskGroup cancels a set of tasks. Do not let object hashes + # determine the order in which their finally blocks execute. + owner = getattr(callback, "__self__", None) + index = self._tasks[self.cancelling_task] + if isinstance(owner, asyncio.Task): + index = self._tasks.get(cast(asyncio.Task[Any], owner), index) + self._cancel_ready.append((index, len(self._cancel_ready), handle)) + else: + self._ready.append(handle) return handle + def call_later( + self, delay: float, callback: Callable[..., Any], *args: Any, + context: contextvars.Context | None = None, + ) -> asyncio.TimerHandle: + if not math.isfinite(delay): + raise ValueError("Durable delays must be finite.") + return self.call_at(self.time() + max(0, delay), callback, *args, context=context) + + def call_at( + self, when: float, callback: Callable[..., Any], *args: Any, + context: contextvars.Context | None = None, + ) -> asyncio.TimerHandle: + if not math.isfinite(when): + raise ValueError("Durable deadlines must be finite.") + handle = _DurableTimerHandle(when, callback, args, self, context) + if when <= self.time(): + self._ready.append(handle) + else: + if self._closing: + raise RuntimeError("Dispatch cleanup cannot schedule durable timers.") + deadline = datetime(1970, 1, 1) + timedelta(seconds=when) + if (deadline - datetime(1970, 1, 1)).total_seconds() < when: + deadline += timedelta(microseconds=1) + handle.durable_timer = self.ctx.create_timer(deadline) + self._timers[handle] = _CompletionObserver( + handle.durable_timer, + lambda: self._completions.append(lambda: self._complete_timer(handle))) + return handle + + def _timer_handle_cancelled(self, handle: asyncio.TimerHandle) -> None: + if isinstance(handle, _DurableTimerHandle): + observer = self._timers.pop(handle, None) + if observer is not None: + observer.detach() + + def _complete_timer(self, handle: _DurableTimerHandle) -> None: + self._timers.pop(handle, None) + if not handle.cancelled(): + self._ready.append(handle) + async def wait_durable(self, durable: task.Task[T]) -> T: if durable.is_complete: return durable.get_result() @@ -154,7 +272,7 @@ def resume(self) -> None: "Await durable activities, events or timers; external I/O is unsupported.") def _has_durable_work(self) -> bool: - return any(not durable.is_complete for durable, _ in self._waits.values()) + return any(not durable.is_complete for durable, _ in self._waits.values()) or bool(self._timers) def _drain(self) -> None: previous_loop = asyncio._get_running_loop() @@ -163,7 +281,9 @@ def _drain(self) -> None: asyncio.tasks._leave_task(previous_loop, previous_task) asyncio._set_running_loop(self) try: - while self._ready: + while self._ready or self._cancel_ready: + self._ready.extend(handle for _, _, handle in sorted(self._cancel_ready)) + self._cancel_ready.clear() handle = self._ready.popleft() if not handle.cancelled(): handle._run() @@ -217,9 +337,33 @@ def close(self) -> None: finally: self._closed = True self._ready.clear() - for observer in self._observers.values(): + self._cancel_ready.clear() + for observer in (*self._observers.values(), *self._timers.values()): observer.detach() self._completions.clear() self._observers.clear() self._waits.clear() + self._timers.clear() self._tasks.clear() + + def run_in_executor(self, executor: Any, func: Any, *args: Any) -> asyncio.Future[Any]: + raise RuntimeError("Threads/executors are unsupported in orchestrators; use an activity.") + + def call_soon_threadsafe(self, callback: Any, *args: Any, context: Any = None) -> asyncio.Handle: + raise RuntimeError("Cross-thread scheduling is unsupported in orchestrators; use an activity.") + + async def _unsupported_io(self, *args: Any, **kwargs: Any) -> Any: + raise RuntimeError("External I/O is unsupported in orchestrators; move it to an ordinary activity.") + + create_connection = _unsupported_io + create_server = _unsupported_io + create_datagram_endpoint = _unsupported_io + connect_read_pipe = _unsupported_io + connect_write_pipe = _unsupported_io + getaddrinfo = _unsupported_io + getnameinfo = _unsupported_io + sock_connect = _unsupported_io + sock_recv = _unsupported_io + sock_sendall = _unsupported_io + subprocess_exec = _unsupported_io + subprocess_shell = _unsupported_io diff --git a/durabletask/task.py b/durabletask/task.py index 79132ea9..483b4767 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -633,8 +633,10 @@ def __await__(self) -> Generator[Task[T], Any, T]: async def wait(self) -> T: """Borrow this operation's result through a fresh native coroutine. - Calling this method does not schedule another durable operation. Each - call creates an independent waiter. + Suitable for ``asyncio.create_task`` and task groups. Calling this + method does not schedule another durable operation. Each call creates + an independent waiter; cancelling it does not cancel the underlying + operation, timer, or event subscription, or any other waiter. """ return await self diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py new file mode 100644 index 00000000..7570437b --- /dev/null +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -0,0 +1,588 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Deterministic asyncio behavior through the real history executor.""" + +import asyncio +import gc +import json +from contextvars import ContextVar +from dataclasses import dataclass +from datetime import timedelta +from typing import Any +from weakref import ReferenceType, ref + +import pytest + +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb +from tests.durabletask.test_coroutine_orchestrators import ( + NOW, + _completion, # pyright: ignore[reportPrivateUsage] + _execute, # pyright: ignore[reportPrivateUsage] + _start, # pyright: ignore[reportPrivateUsage] +) + + +@pytest.mark.parametrize("fail_branch", [False, True]) +def test_task_group_branches_inherit_context_and_replay_stable_actions(fail_branch: bool) -> None: + context: ContextVar[task.OrchestrationContext] = ContextVar("orchestration") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + token = context.set(ctx) + try: + results: dict[str, Any] = {} + + async def branch(name: str) -> None: + current = context.get() + assert current is ctx + results[name] = await current.call_activity(name, input=current.instance_id) + + async with asyncio.TaskGroup() as group: + group.create_task(branch("first")) + group.create_task(branch("second")) + return await context.get().call_activity("finish", input=[results["first"], results["second"]]) + finally: + context.reset(token) + + def replay(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]): + first = _execute(orchestrator, old, new) + assert _execute(orchestrator, old, new).actions == first.actions + with pytest.raises(LookupError): + context.get() + return first + + start = _start() + initial = replay([], start) + assert [action.scheduleTask.name for action in initial.actions] == ["first", "second"] + assert all(json.loads(action.scheduleTask.input.value) == "coroutine-test" for action in initial.actions) + old = start + [helpers.new_task_scheduled_event(1, "first"), + helpers.new_task_scheduled_event(2, "second")] + second_done = [helpers.new_task_completed_event(2, '"second result"')] + assert replay(old, second_done).actions == [] + first_done = [helpers.new_task_failed_event(1, ValueError("recorded branch failure"))] if fail_branch else [ + helpers.new_task_completed_event(1, '"first result"')] + result = replay(old + second_done, first_done) + if fail_branch: + complete = _completion(result) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "recorded branch failure" in str(complete.failureDetails) + else: + assert len(result.actions) == 1 + assert result.actions[0].scheduleTask.name == "finish" + assert json.loads(result.actions[0].scheduleTask.input.value) == ["first result", "second result"] + complete = _completion(replay(old + second_done + first_done, [ + helpers.new_task_scheduled_event(3, "finish"), + helpers.new_task_completed_event(3, '"done"')])) + assert json.loads(complete.result.value) == "done" + + +def test_gather_resumes_each_branch_without_waiting_for_batch() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[Any]: + async def branch(name: str) -> Any: + value: Any = await ctx.call_activity(name) + await asyncio.sleep(0) + return await ctx.call_activity(name + "-next", input=value) + + return list(await asyncio.gather(branch("a"), branch("b"))) + + start = _start() + actions = _execute(orchestrator, [], start).actions + assert [a.scheduleTask.name for a in actions] == ["a", "b"] + old = start + [helpers.new_task_scheduled_event(i, name) for i, name in [(1, "a"), (2, "b")]] + second = [helpers.new_task_completed_event(2, '"b"')] + actions = _execute(orchestrator, old, second).actions + assert [(a.id, a.scheduleTask.name) for a in actions] == [(3, "b-next")] + third = [helpers.new_task_scheduled_event(3, "b-next"), + helpers.new_task_completed_event(1, '"a"'), + helpers.new_task_scheduled_event(4, "a-next"), + helpers.new_task_completed_event(4, '"a-done"'), + helpers.new_task_completed_event(3, '"b-done"')] + assert json.loads(_completion(_execute(orchestrator, old + second, third)).result.value) == [ + "a-done", "b-done"] + + +@pytest.mark.parametrize("return_exceptions", [False, True]) +def test_gather_failure_does_not_cancel_sibling(return_exceptions: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + async def sibling() -> str: + await ctx.call_activity("slow") + return await ctx.call_activity("continued", return_type=str) + + failure = asyncio.create_task(ctx.call_activity("fail", return_type=str).wait()) + pending = asyncio.create_task(sibling()) + try: + values = await asyncio.gather( + failure, pending, return_exceptions=return_exceptions) + return [isinstance(values[0], task.TaskFailedError), values[1]] + except task.TaskFailedError: + assert not pending.cancelled() + return await pending + + old = _start() + [helpers.new_task_scheduled_event(1, "fail"), + helpers.new_task_scheduled_event(2, "slow")] + failed = [helpers.new_task_failed_event(1, ValueError("boom"))] + assert _execute(orchestrator, old, failed).actions == [] + completed = [helpers.new_task_completed_event(2, "null")] + assert _execute(orchestrator, old + failed, completed).actions[0].scheduleTask.name == "continued" + result = _completion(_execute(orchestrator, old + failed + completed, [ + helpers.new_task_scheduled_event(3, "continued"), helpers.new_task_completed_event(3, '"ok"')])) + assert json.loads(result.result.value) == ([True, "ok"] if return_exceptions else "ok") + + +def test_task_group_failure_cancels_waiters_not_durable_work() -> None: + cleaned: list[str] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def branch(name: str) -> None: + try: + await ctx.call_activity(name) + finally: + await asyncio.sleep(0) + cleaned.append(name) + + async with asyncio.TaskGroup() as group: + group.create_task(branch("fail")) + group.create_task(branch("slow")) + + old = _start() + [helpers.new_task_scheduled_event(1, "fail"), + helpers.new_task_scheduled_event(2, "slow")] + result = _completion(_execute(orchestrator, old, [helpers.new_task_failed_event(1, ValueError("boom"))])) + assert result.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert cleaned == ["fail", "slow"] + + +@pytest.mark.parametrize("kind", ["sleep", "wait_for", "timeout"]) +def test_durable_timer_and_timeout(kind: str) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + try: + if kind == "sleep": + await asyncio.sleep(2) + elif kind == "wait_for": + await asyncio.wait_for(ctx.wait_for_external_event("never"), timeout=2) + else: + async with asyncio.timeout(2): + await ctx.wait_for_external_event("never") + except TimeoutError: + return "timed out" + return "slept" + + actions = _execute(orchestrator, [], _start()).actions + assert len(actions) == 1 + deadline = actions[0].createTimer.fireAt.ToDatetime() + assert deadline == NOW + timedelta(seconds=2) + old = _start() + [helpers.new_timer_created_event(1, deadline)] + new = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(1, deadline)] + assert json.loads(_completion(_execute(orchestrator, old, new)).result.value) == ( + "slept" if kind == "sleep" else "timed out") + + +def test_sleep_zero_fifo_without_timer() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[int]: + observed: list[int] = [] + + async def branch(index: int) -> None: + observed.append(index) + await asyncio.sleep(0) + observed.append(index + 2) + + await asyncio.gather(branch(0), branch(1)) + return observed + + result = _execute(orchestrator, [], _start()) + assert len(result.actions) == 1 + assert json.loads(_completion(result).result.value) == [0, 1, 2, 3] + + +def test_discarded_gather_batches_release_decoded_results_during_replay() -> None: + references: list[ReferenceType[Any]] = [] + + @dataclass + class Payload: + value: int + + def __post_init__(self) -> None: + references.append(ref(self)) + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + for batch in range(8): + await asyncio.gather(*(ctx.call_activity("work", return_type=Payload) for _ in range(4))) + await asyncio.sleep(0) + gc.collect() + assert len(references) == (batch + 1) * 4 + # The history handler still holds the most recent decoded result + # while resume() drains this batch's callbacks. + assert sum(reference() is not None for reference in references) <= 1 + return "done" + + history = _start() + for batch in range(8): + ids = range(batch * 4 + 1, batch * 4 + 5) + history += [helpers.new_task_scheduled_event(index, "work") for index in ids] + history += [helpers.new_task_completed_event(index, json.dumps({"value": index})) for index in ids] + complete = _completion(_execute(orchestrator, history[:-1], history[-1:])) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, complete.failureDetails + assert json.loads(complete.result.value) == "done" + gc.collect() + assert all(reference() is None for reference in references) + + +@pytest.mark.parametrize("observe", [ + "await", "gather", "gather-exceptions", "result", "exception", "task-group", +]) +def test_observed_failures_release_traceback_payloads_during_replay(observe: str) -> None: + references: list[ReferenceType[Any]] = [] + + @dataclass + class Payload: + value: int + + def __post_init__(self) -> None: + references.append(ref(self)) + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + async def branch() -> None: + payload = await ctx.call_activity("work", return_type=Payload) + raise ValueError(f"Failed with payload {payload.value}") + + for batch in range(8): + if observe == "task-group": + group = asyncio.TaskGroup() + try: + async with group: + group.create_task(branch()) + except* ValueError: + pass + del group + else: + pending = asyncio.create_task(branch()) + await asyncio.sleep(0) + # Observe only after the failed task's initial retirement callback. + await ctx.call_activity("gate") + if observe == "gather-exceptions": + await asyncio.gather(pending, return_exceptions=True) + elif observe == "exception": + assert isinstance(pending.exception(), ValueError) + else: + try: + if observe == "await": + await pending + elif observe == "gather": + await asyncio.gather(pending) + else: + pending.result() + except ValueError: + pass + del pending + await asyncio.sleep(0) + gc.collect() + assert len(references) == batch + 1 + assert sum(reference() is not None for reference in references) <= 1 + return "done" + + history = _start() + sequence = 0 + for batch in range(8): + sequence += 1 + work_id = sequence + history.append(helpers.new_task_scheduled_event(work_id, "work")) + if observe != "task-group": + sequence += 1 + history.append(helpers.new_task_scheduled_event(sequence, "gate")) + history.append(helpers.new_task_completed_event(work_id, json.dumps({"value": batch}))) + if observe != "task-group": + history.append(helpers.new_task_completed_event(sequence, "null")) + complete = _completion(_execute(orchestrator, history[:-1], history[-1:])) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, complete.failureDetails + assert json.loads(complete.result.value) == "done" + gc.collect() + assert all(reference() is None for reference in references) + + +def test_retired_tasks_do_not_reuse_cancellation_order() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + async def completed() -> None: + pass + + async def branch(index: int) -> None: + try: + await ctx.wait_for_external_event(str(index)) + finally: + await ctx.call_activity("compensate", input=index) + + asyncio.create_task(completed()) + first = asyncio.create_task(branch(0)) + await asyncio.sleep(0) + await asyncio.sleep(0) + second = asyncio.create_task(branch(1)) + await asyncio.sleep(0) + second.cancel() + first.cancel() + await asyncio.gather(first, second, return_exceptions=True) + return "done" + + initial = _execute(orchestrator, [], _start()) + assert [json.loads(action.scheduleTask.input.value) for action in initial.actions] == [0, 1] + assert _execute(orchestrator, [], _start()).actions == initial.actions + old = _start() + [helpers.new_task_scheduled_event(index, "compensate") for index in (1, 2)] + new = [helpers.new_task_completed_event(index, "null") for index in (1, 2)] + assert json.loads(_completion(_execute(orchestrator, old, new)).result.value) == "done" + + +def test_shared_operation_survives_local_waiter_cancellation() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + durable = ctx.call_activity("shared", return_type=str) + + async def wait() -> Any: + return await durable + + first = asyncio.create_task(wait()) + second = asyncio.create_task(wait()) + await asyncio.sleep(0) + first.cancel() + await asyncio.gather(first, return_exceptions=True) + return await second + + old = _start() + [helpers.new_task_scheduled_event(1, "shared")] + assert json.loads(_completion(_execute(orchestrator, old, [ + helpers.new_task_completed_event(1, '"ok"')])).result.value) == "ok" + + +def test_local_durable_cancellation_wakes_waiter_without_history() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + timer = ctx.create_timer(timedelta(days=1)) + + async def waiter() -> None: + await timer + + pending = asyncio.create_task(waiter()) + await asyncio.sleep(0) + timer.cancel() + try: + await pending + except task.TaskCancelledError: + return "cancelled" + return "unexpected" + + result = _execute(orchestrator, [], _start()) + assert len(result.actions) == 1 + assert json.loads(_completion(result).result.value) == "cancelled" + + +def test_unhandled_local_root_cancellation_fails_not_abandons() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + raise asyncio.CancelledError + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType.endswith("TaskCancelledError") + + +@pytest.mark.parametrize("delay", [0.0000003, 0.000001, 0.0000013, 2.25]) +def test_timer_rounding_never_precedes_loop_deadline(delay: float) -> None: + observed: list[float] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + loop = asyncio.get_running_loop() + observed.append(loop.time() + delay) + await asyncio.sleep(delay) + + actions = _execute(orchestrator, [], _start()).actions + assert len(actions) == 1 + deadline = actions[0].createTimer.fireAt.ToDatetime() + from datetime import datetime + assert (deadline - datetime(1970, 1, 1)).total_seconds() >= observed[0] + + +def test_async_sleep_uses_shared_long_timer_chunking() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + await asyncio.sleep(7 * 86400) + return "done" + + old = _start() + for timer_id, days in [(1, 3), (2, 6), (3, 7)]: + result = _execute(orchestrator, [], old) + assert len(result.actions) == 1 + action = result.actions[0] + assert action.id == timer_id + deadline = NOW + timedelta(days=days) + assert action.createTimer.fireAt.ToDatetime() == deadline + old += [helpers.new_timer_created_event(timer_id, deadline), + helpers.new_orchestrator_started_event(deadline), + helpers.new_timer_fired_event(timer_id, deadline)] + assert json.loads(_completion(_execute(orchestrator, old[:-1], old[-1:])).result.value) == "done" + + +@pytest.mark.parametrize("api", ["thread", "io", "foreign-loop"]) +def test_unsupported_work_fails_actionably(api: str) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + if api == "thread": + await asyncio.to_thread(lambda: None) + elif api == "io": + await asyncio.open_connection("localhost", 1) + else: + foreign = asyncio.new_event_loop() + try: + await foreign.create_future() + finally: + foreign.close() + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert any(text in complete.failureDetails.errorMessage for text in ( + "use an activity", "ordinary activity", "different loop")) + + +def test_task_group_cancellation_durable_cleanup_has_stable_action_order() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def branch(index: int) -> None: + try: + await ctx.call_activity("work", input=index) + finally: + # This is real task-group cancellation during replay, not + # dispatch teardown: durable compensation is still allowed. + await ctx.call_activity("compensate", input=index) + + async with asyncio.TaskGroup() as group: + for index in range(4): + group.create_task(branch(index)) + + old = _start() + [helpers.new_task_scheduled_event(index, "work") for index in range(1, 5)] + failure = [helpers.new_task_failed_event(1, ValueError("boom"))] + # The failing branch first enters its durable finally, so deliver that + # completion before task-group sibling cancellation starts. + first = _execute(orchestrator, old, failure) + assert [(a.id, a.scheduleTask.name, json.loads(a.scheduleTask.input.value)) for a in first.actions] == [ + (5, "compensate", 0)] + old += failure + [helpers.new_task_scheduled_event(5, "compensate"), + helpers.new_task_completed_event(5, "null")] + expected = [helpers.new_schedule_task_action(index + 6, "compensate", json.dumps(index + 1), None) + for index in range(3)] + for _ in range(10): + assert _execute(orchestrator, [], old).actions == expected + + +def test_cancelling_last_loop_timer_cannot_hide_foreign_wait() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + handle = asyncio.get_running_loop().call_later(1, lambda: None) + handle.cancel() + await asyncio.Future() + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "blocked without durable work" in complete.failureDetails.errorMessage + + +@pytest.mark.parametrize("observed", [False, True]) +@pytest.mark.parametrize("probe_pending", [False, True]) +def test_background_failure_is_reported_unless_observed( + observed: bool, probe_pending: bool, caplog: pytest.LogCaptureFixture) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + async def fail() -> None: + raise ValueError("background failure") + + pending = asyncio.create_task(fail()) + if probe_pending: + with pytest.raises(asyncio.InvalidStateError): + pending.result() + with pytest.raises(asyncio.InvalidStateError): + pending.exception() + await asyncio.sleep(0) + if observed: + await asyncio.gather(pending, return_exceptions=True) + del pending + await asyncio.sleep(0) + gc.collect() + return "done" + + assert json.loads(_completion(_execute(orchestrator, [], _start())).result.value) == "done" + assert ("background failure" in caplog.text) is not observed + + +@pytest.mark.parametrize("count", [128, 4096]) +def test_fanout_waiter_inspections_scale_linearly( + count: int, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None: + inspections = 0 + + def is_complete(durable: task.Task[Any]) -> bool: + nonlocal inspections + inspections += 1 + return durable._is_complete # pyright: ignore[reportPrivateUsage] + + monkeypatch.setattr(task.Task, "is_complete", property(is_complete)) + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[Any]: + async def branch(index: int) -> Any: + return await ctx.call_activity("work", input=index) + + return await asyncio.gather(*(branch(index) for index in range(count))) + + result = _execute(orchestrator, [], _start()) + assert [action.id for action in result.actions] == list(range(1, count + 1)) + assert inspections < count * 20 + assert not caplog.records + + +@pytest.mark.parametrize("timer", [False, True]) +def test_callback_stop_iteration_fails_instead_of_completing_orchestrator(timer: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + def fail() -> None: + raise StopIteration("callback stopped") + + loop = asyncio.get_running_loop() + if timer: + loop.call_later(1, fail) + else: + loop.call_soon(fail) + await ctx.wait_for_external_event("never") + + start = _start() + if timer: + deadline = NOW + timedelta(seconds=1) + old = start + [helpers.new_timer_created_event(1, deadline)] + new = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(1, deadline)] + else: + old, new = [], start + complete = _completion(_execute(orchestrator, old, new)) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType == "builtins.RuntimeError" + assert complete.failureDetails.innerFailure.errorType == "builtins.StopIteration" + assert complete.failureDetails.innerFailure.errorMessage == "callback stopped" + + +@pytest.mark.parametrize("cancelled", [None, 0, 1]) +def test_identical_timer_callbacks_complete_and_cancel_independently(cancelled: int | None) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> int: + loop = asyncio.get_running_loop() + result = loop.create_future() + completed: list[None] = [] + expected = 2 if cancelled is None else 1 + + def callback() -> None: + completed.append(None) + if len(completed) == expected: + result.set_result(len(completed)) + + handles = [loop.call_at(loop.time() + 1, callback) for _ in range(2)] + assert handles[0] != handles[1] + assert len(set(handles)) == 2 + if cancelled is not None: + handles[cancelled].cancel() + return await result + + start = _start() + initial = _execute(orchestrator, [], start) + remaining = [index + 1 for index in range(2) if index != cancelled] + assert [action.id for action in initial.actions] == remaining + deadline = NOW + timedelta(seconds=1) + old = start + [helpers.new_timer_created_event(index, deadline) for index in remaining] + for offset, timer_id in enumerate(remaining): + new = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(timer_id, deadline)] + result = _execute(orchestrator, old, new) + assert _execute(orchestrator, old, new).actions == result.actions + if offset < len(remaining) - 1: + assert result.actions == [] + else: + assert json.loads(_completion(result).result.value) == len(remaining) + old += new From 638f8d998b19205753a288de0735a327f02f547e Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 12 Nov 2026 10:47:09 -0700 Subject: [PATCH 2/4] Fix same-runtime task-group cancellation replay order Use immutable dispatch-local task creation hashes so native TaskGroup traversal cannot depend on worker-local object addresses before scheduling shared-Future callbacks. Preserve existing identity equality and cancellation-wakeup ordering. Cover direct and gather-wrapped shared futures, out-of-order activity completion, fresh dispatch executors, and replay in separate worker processes. Document pinned worker environments and existing-history upgrade risks. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 4 + docs/async-orchestrators.md | 9 +- durabletask-azuremanaged/CHANGELOG.md | 5 + durabletask/internal/asyncio_loop.py | 18 ++- .../durabletask/test_asyncio_orchestrators.py | 112 ++++++++++++++++++ 5 files changed, 145 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 15f56206..3604c296 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,10 @@ FIXED when an intermediate async wrapper decorates a synchronous function, including callable instances and partials. Cyclic wrapper chains fail instead of hanging worker dispatch. +- Coroutine task-group cancellation now schedules shared-Future callbacks +deterministically across replays, including when a dispatch resumes on another +worker using the same SDK and Python runtime. Previously recorded cancellation +histories may replay differently; validate or drain them before upgrading. - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe orchestration time, preventing retries from expiring solely because history is replayed later. Retry calculations also accept timezone-aware timestamps. diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md index 03ae3eb7..50cbf8a5 100644 --- a/docs/async-orchestrators.md +++ b/docs/async-orchestrators.md @@ -266,6 +266,12 @@ automatic recording or interception of arbitrary await expressions. `is_replaying` is not a substitute for recording results. Activities remain at-least-once work: side effects require appropriate idempotency/deduplication. +Successive orchestration dispatches and their activities can run on different +workers. Replay reconstructs local tasks and futures from ordered durable +history; it does not rely on task identities or memory retained by a previous +worker. Activity completion order is the order recorded in history, not worker +placement. + Framework adapters can compose supported durable operations using async helpers. Adapters remain responsible for registering ordinary model/tool activities, encoding/decoding framework payloads, and supplying replay-stable @@ -284,7 +290,8 @@ construction order is not execution order. Supporting several Python versions does not guarantee recorded-history portability between them. Pin the runtime and dependencies for active -instances and validate histories before upgrades. In particular, Python 3.11 +instances and validate histories before upgrades. All workers processing an +instance must use that pinned environment. In particular, Python 3.11 and 3.12+ schedule coroutine operands of `asyncio.wait_for()` differently, which can change durable action order inside concurrent branches. The async context does not normalize arbitrary standard-library scheduling. diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 1b327beb..11b59946 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -9,6 +9,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). FIXED +- With the corresponding core SDK update, coroutine task-group cancellation +schedules shared-Future callbacks deterministically when replayed on another +worker using the same SDK and Python runtime. See the +[core upgrade guidance](../CHANGELOG.md#unreleased) for existing cancellation +histories. - With the corresponding core SDK update, activity and sub-orchestration `retry_timeout` decisions use replay-safe orchestration time rather than wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased) diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index 3b0bea9e..621127d2 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -35,6 +35,19 @@ def detach(self) -> None: class _DurableAsyncTask(asyncio.Task[T]): + def __init__( + self, coro: Coroutine[Any, Any, T] | Generator[Any, None, T], *, + sequence: int, loop: DurableEventLoop, name: str | None, + context: contextvars.Context | None, + ) -> None: + self._sequence = sequence + super().__init__(coro, loop=loop, name=name, context=context) + + def __hash__(self) -> int: + # TaskGroup traverses a set before cancellation schedules shared-Future + # callbacks. Stabilize that traversal, not just the resulting wakeups. + return self._sequence + def __await__(self) -> Generator[Any, None, T]: try: return (yield from super().__await__()) @@ -147,9 +160,10 @@ def create_task( if kwargs: coro.close() raise TypeError("Custom/eager task execution is not supported in durable orchestrators.") - future = _DurableAsyncTask(coro, loop=self, name=name, context=context) - self._tasks[future] = self._task_sequence + sequence = self._task_sequence self._task_sequence += 1 + future = _DurableAsyncTask(coro, sequence=sequence, loop=self, name=name, context=context) + self._tasks[future] = sequence future.add_done_callback(self.retire_task) return future diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 7570437b..044b2af2 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -6,9 +6,13 @@ import asyncio import gc import json +import os +import subprocess +import sys from contextvars import ContextVar from dataclasses import dataclass from datetime import timedelta +from functools import partial from typing import Any from weakref import ReferenceType, ref @@ -463,6 +467,114 @@ async def branch(index: int) -> None: assert _execute(orchestrator, [], old).actions == expected +async def _shared_future_cancellation_orchestrator( + ctx: task.OrchestrationContext, _: Any, *, direct_future: bool, gather_wait: bool, +) -> str: + loop = asyncio.get_running_loop() + futures: list[asyncio.Future[None]] = [ + asyncio.Future[None]() if direct_future else loop.create_future() for _ in range(2)] + + def on_cancelled(index: int, future: asyncio.Future[None]) -> None: + assert future.cancelled() + ctx.call_activity(f"cancelled-{index}") + + for index, future in enumerate(futures): + future.add_done_callback(partial(on_cancelled, index)) + + async def wait(future: asyncio.Future[None]) -> None: + if gather_wait: + await asyncio.gather(future) + else: + await future + + async def fail() -> None: + await asyncio.sleep(0) + raise ValueError("cancel siblings") + + try: + async with asyncio.TaskGroup() as group: + for future in futures * 2: + group.create_task(wait(future)) + group.create_task(fail()) + except* ValueError: + pass + return await ctx.call_activity("join", return_type=str) + + +@pytest.mark.parametrize("direct_future", [False, True]) +@pytest.mark.parametrize("gather_wait", [False, True]) +def test_shared_future_cancellation_callbacks_replay_on_fresh_executors( + direct_future: bool, gather_wait: bool, +) -> None: + orchestrator = partial( + _shared_future_cancellation_orchestrator, direct_future=direct_future, gather_wait=gather_wait) + start = _start() + initial = _execute(orchestrator, [], start) + assert sorted(a.scheduleTask.name for a in initial.actions[:-1]) == ["cancelled-0", "cancelled-1"] + assert initial.actions[-1].scheduleTask.name == "join" + recorded = start + [ + helpers.new_task_scheduled_event(action.id, action.scheduleTask.name) + for action in initial.actions + ] + # Activity workers may complete in any order. Each dispatch reconstructs + # all local tasks and futures from history, as on a different worker. + completed = [ + helpers.new_task_completed_event(initial.actions[1].id, "null"), + helpers.new_task_completed_event(initial.actions[0].id, "null"), + ] + for _ in range(30): + assert _execute(orchestrator, [], start).actions == initial.actions + assert _execute(orchestrator, recorded, completed).actions == [] + complete = _completion(_execute( + orchestrator, recorded + completed, + [helpers.new_task_completed_event(initial.actions[-1].id, '"joined"')])) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, complete.failureDetails + assert json.loads(complete.result.value) == "joined" + + +def test_shared_future_cancellation_history_replays_in_other_worker_processes() -> None: + orchestrator = partial( + _shared_future_cancellation_orchestrator, direct_future=True, gather_wait=False) + start = _start() + initial = _execute(orchestrator, [], start) + history = start + [ + helpers.new_task_scheduled_event(action.id, action.scheduleTask.name) + for action in initial.actions + ] + history += [ + helpers.new_task_completed_event(initial.actions[1].id, "null"), + helpers.new_task_completed_event(initial.actions[0].id, "null"), + ] + payload = json.dumps({ + "initial": [action.SerializeToString().hex() for action in initial.actions], + "history": [event.SerializeToString().hex() for event in history], + "join_id": initial.actions[-1].id, + }) + code = """ +import json +import sys +from functools import partial +from durabletask.internal import helpers, orchestrator_service_pb2 as pb +from tests.durabletask.test_asyncio_orchestrators import _shared_future_cancellation_orchestrator +from tests.durabletask.test_coroutine_orchestrators import _completion, _execute, _start + +data = json.load(sys.stdin) +orchestrator = partial(_shared_future_cancellation_orchestrator, direct_future=True, gather_wait=False) +initial = _execute(orchestrator, [], _start()) +assert [action.SerializeToString().hex() for action in initial.actions] == data["initial"] +history = [pb.HistoryEvent.FromString(bytes.fromhex(event)) for event in data["history"]] +result = _completion(_execute( + orchestrator, history, [helpers.new_task_completed_event(data["join_id"], '"joined"')])) +assert result.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, result.failureDetails +assert json.loads(result.result.value) == "joined" +""" + for seed in ("0", "999"): + process = subprocess.run( + [sys.executable, "-c", code], input=payload, capture_output=True, text=True, + env={**os.environ, "PYTHONHASHSEED": seed}, timeout=30) + assert process.returncode == 0, process.stdout + process.stderr + + def test_cancelling_last_loop_timer_cannot_hide_foreign_wait() -> None: async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: handle = asyncio.get_running_loop().call_later(1, lambda: None) From 44eccfc79a33e91c5f44fb18d8b02ce5d17b4ba6 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 12 Nov 2026 11:03:26 -0700 Subject: [PATCH 3/4] Use exact microsecond ceiling for durable call_at deadlines Convert loop deadlines with exact rational arithmetic so durable timers never fire before the requested loop time, and add exact-oracle regression tests. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- durabletask/internal/asyncio_loop.py | 16 +++-- .../durabletask/test_asyncio_orchestrators.py | 60 +++++++++++++++++++ 2 files changed, 71 insertions(+), 5 deletions(-) diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index 621127d2..695054c9 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -12,12 +12,16 @@ from collections import deque from collections.abc import Callable, Coroutine, Generator from datetime import datetime, timedelta +from fractions import Fraction from typing import Any, TypeVar, cast from durabletask import task T = TypeVar("T") +_EPOCH = datetime(1970, 1, 1) +_MICROSECOND = timedelta(microseconds=1) + class _CompletionObserver(task.CompositeTask[None]): def __init__(self, durable: task.Task[Any], callback: Callable[[], None]) -> None: @@ -133,7 +137,7 @@ def __init__( def time(self) -> float: # A fixed UTC epoch avoids local-time and platform timestamp differences. - return (self.ctx.current_utc_datetime - datetime(1970, 1, 1)).total_seconds() + return (self.ctx.current_utc_datetime - _EPOCH).total_seconds() def get_debug(self) -> bool: return False @@ -205,14 +209,16 @@ def call_at( if not math.isfinite(when): raise ValueError("Durable deadlines must be finite.") handle = _DurableTimerHandle(when, callback, args, self, context) - if when <= self.time(): + # Exact rational ceiling: float round-trips can emit a deadline that + # precedes the requested one by a fraction of a microsecond. + deadline_us = math.ceil(Fraction(when) * 1_000_000) + now_us = (self.ctx.current_utc_datetime - _EPOCH) // _MICROSECOND + if when <= self.time() or deadline_us <= now_us: self._ready.append(handle) else: if self._closing: raise RuntimeError("Dispatch cleanup cannot schedule durable timers.") - deadline = datetime(1970, 1, 1) + timedelta(seconds=when) - if (deadline - datetime(1970, 1, 1)).total_seconds() < when: - deadline += timedelta(microseconds=1) + deadline = _EPOCH + timedelta(microseconds=deadline_us) handle.durable_timer = self.ctx.create_timer(deadline) self._timers[handle] = _CompletionObserver( handle.durable_timer, diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 044b2af2..d0c2db8f 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -12,6 +12,7 @@ from contextvars import ContextVar from dataclasses import dataclass from datetime import timedelta +from decimal import ROUND_CEILING, Decimal, localcontext from functools import partial from typing import Any from weakref import ReferenceType, ref @@ -399,6 +400,65 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: assert (deadline - datetime(1970, 1, 1)).total_seconds() >= observed[0] +@pytest.mark.parametrize("offset", [0.000003, 0.0000003, 0.000001, 0.0000013, 1.0, 2.25]) +def test_call_at_deadline_is_exact_microsecond_ceiling(offset: float) -> None: + requested: list[float] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + loop = asyncio.get_running_loop() + when = loop.time() + offset + requested.append(when) + fired = loop.create_future() + loop.call_at(when, fired.set_result, None) + await fired + + actions = _execute(orchestrator, [], _start()).actions + assert len(actions) == 1 + fire_at = actions[0].createTimer.fireAt + assert fire_at.nanos % 1000 == 0 + emitted_us = fire_at.seconds * 1_000_000 + fire_at.nanos // 1000 + # Decimal(float) is exact, so this oracle shares no float arithmetic with the loop. + with localcontext() as exact: + exact.prec = 400 + requested_seconds = Decimal(requested[0]) + oracle_us = int((requested_seconds * 1_000_000).to_integral_value(rounding=ROUND_CEILING)) + assert emitted_us == oracle_us + assert Decimal(emitted_us) / 1_000_000 >= requested_seconds + + +def test_call_at_known_float_deadline_rounds_up_to_next_microsecond() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + loop = asyncio.get_running_loop() + fired = loop.create_future() + loop.call_at(loop.time() + 0.000003, fired.set_result, None) + await fired + + actions = _execute(orchestrator, [], _start()).actions + assert actions[0].createTimer.fireAt.ToDatetime() == NOW + timedelta(microseconds=4) + + +@pytest.mark.parametrize("offset", [0.0, -1.0]) +def test_call_at_due_deadline_runs_without_durable_timer(offset: float) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + loop = asyncio.get_running_loop() + fired = loop.create_future() + loop.call_at(loop.time() + offset, fired.set_result, "fired") + return await fired + + result = _execute(orchestrator, [], _start()) + assert json.loads(_completion(result).result.value) == "fired" + + +@pytest.mark.parametrize("when", [1e20, float(sys.float_info.max)]) +def test_call_at_out_of_range_deadline_fails(when: float) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + asyncio.get_running_loop().call_at(when, lambda: None) + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType.endswith("OverflowError") + + def test_async_sleep_uses_shared_long_timer_chunking() -> None: async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: await asyncio.sleep(7 * 86400) From acae0c87319bb40a3dcc9fd475268ba2f9eaa002 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 12 Nov 2026 15:57:29 -0700 Subject: [PATCH 4/4] Harden durable loop API coverage and task bookkeeping Report every unimplemented asyncio loop API through one actionable error table, tolerate waiters finalized after dispatch close, use replay-safe lookups for cancellation ordering, normalize timezone-aware clocks for loop timers, and document why task hashing is load-bearing and which private asyncio hooks the loop relies on. Document timer resolution and that tasks must be created with asyncio.create_task. Remove changelog entries for unreleased behavior. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 4 - docs/async-orchestrators.md | 10 +- durabletask-azuremanaged/CHANGELOG.md | 5 - durabletask/internal/asyncio_loop.py | 97 +++++++++++++------ .../durabletask/test_asyncio_orchestrators.py | 31 +++++- 5 files changed, 101 insertions(+), 46 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 233a9bb9..40aa2e88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,10 +9,6 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). FIXED -- Coroutine task-group cancellation now schedules shared-Future callbacks -deterministically across replays, including when a dispatch resumes on another -worker using the same SDK and Python runtime. Previously recorded cancellation -histories may replay differently; validate or drain them before upgrading. - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe orchestration time, preventing retries from expiring solely because history is replayed later. Retry calculations also accept timezone-aware timestamps. diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md index 56720974..eb4d9bdd 100644 --- a/docs/async-orchestrators.md +++ b/docs/async-orchestrators.md @@ -187,7 +187,9 @@ Create branches from ordered inputs. Ready work follows FIFO scheduling; history completions are delivered in history order, draining ready work after each completion. Task-group cancellation wakeups are ordered by local task creation, not set/object-hash iteration. Local tasks are dispatch resources, -not new durable instances. +not new durable instances. Create them with `asyncio.create_task` or a task +group; do not construct `asyncio.Task` directly, because such tasks are not +tracked by the dispatch and are not cancelled or cleaned up with it. `loop.time()` uses seconds from a fixed UTC epoch and the current orchestration clock, never machine monotonic time. `loop.call_at` and `call_later` schedule @@ -195,8 +197,10 @@ durable timers for future finite deadlines. Timer conversion rounds up when necessary so a durable timer does not precede its floating-point deadline. Core/Functions long timers reuse the existing chunking; Azure Managed retains its native long timers. Past deadlines are ready callbacks. `sleep(0)` yields -to ready coroutines without a durable timer. Clock advancement and timers -require history events, not elapsed time in the worker. +to ready coroutines without a durable timer. Delays below the resolution of +`loop.time()` (about 238 nanoseconds at current dates) are effectively zero. +Clock advancement and timers require history events, not elapsed time in the +worker. > [!WARNING] > This is not a general-purpose asyncio event loop. External network/storage diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 11b59946..1b327beb 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -9,11 +9,6 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). FIXED -- With the corresponding core SDK update, coroutine task-group cancellation -schedules shared-Future callbacks deterministically when replayed on another -worker using the same SDK and Python runtime. See the -[core upgrade guidance](../CHANGELOG.md#unreleased) for existing cancellation -histories. - With the corresponding core SDK update, activity and sub-orchestration `retry_timeout` decisions use replay-safe orchestration time rather than wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased) diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index acb0cb85..b371cbb1 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -15,15 +15,22 @@ - ``asyncio.Handle._run``: run a ready callback with asyncio's own context and exception-handler semantics. - ``asyncio.Task._log_traceback``: report unobserved local-task failures during - dispatch cleanup without marking them as retrieved. + dispatch cleanup without marking them as retrieved, and keep such tasks + tracked until then. +- ``__self__`` of Task step/wakeup callbacks (including the C implementation's + step wrapper and bound ``task_wakeup``): identify which local task a + callback resumes, so task-group cancellation wakeups run in task-creation + order. +- ``AbstractEventLoop._timer_handle_cancelled``: the hook ``TimerHandle.cancel`` + calls on its loop, used to withdraw the durable timer observer. """ from __future__ import annotations import asyncio import contextvars -import math import logging +import math from collections import deque from collections.abc import Callable, Coroutine, Generator from datetime import UTC, datetime, timedelta @@ -63,8 +70,10 @@ def __init__( super().__init__(coro, loop=loop, name=name, context=context) def __hash__(self) -> int: - # TaskGroup traverses a set before cancellation schedules shared-Future - # callbacks. Stabilize that traversal, not just the resulting wakeups. + # Load-bearing for replay determinism: TaskGroup keeps its children in a + # set and iterates it to cancel them. Hashing by creation sequence makes + # that traversal, and the cancellation wakeups it schedules, follow task + # creation order instead of memory addresses that differ across workers. return self._sequence def __await__(self) -> Generator[Any, None, T]: @@ -150,12 +159,15 @@ def __init__( self._fatal: BaseException | None = None self._root: asyncio.Task[Any] | None = None - def time(self) -> float: - # A fixed UTC epoch avoids local-time and platform timestamp differences. + def _now(self) -> datetime: now = self.ctx.current_utc_datetime if now.tzinfo is not None: now = now.astimezone(UTC).replace(tzinfo=None) - return (now - _EPOCH).total_seconds() + return now + + def time(self) -> float: + # A fixed UTC epoch avoids local-time and platform timestamp differences. + return (self._now() - _EPOCH).total_seconds() def get_debug(self) -> bool: return False @@ -204,7 +216,8 @@ def call_soon( # TaskGroup cancels a set of tasks. Do not let object hashes # determine the order in which their finally blocks execute. owner = getattr(callback, "__self__", None) - index = self._tasks[self.cancelling_task] + index = self._tasks.get( + self.cancelling_task, getattr(self.cancelling_task, "_sequence", self._task_sequence)) if isinstance(owner, asyncio.Task): index = self._tasks.get(cast(asyncio.Task[Any], owner), index) self._cancel_ready.append((index, len(self._cancel_ready), handle)) @@ -230,7 +243,7 @@ def call_at( # Exact rational ceiling: float round-trips can emit a deadline that # precedes the requested one by a fraction of a microsecond. deadline_us = math.ceil(Fraction(when) * 1_000_000) - now_us = (self.ctx.current_utc_datetime - _EPOCH) // _MICROSECOND + now_us = (self._now() - _EPOCH) // _MICROSECOND if when <= self.time() or deadline_us <= now_us: self._ready.append(handle) else: @@ -270,7 +283,10 @@ async def wait_durable(self, durable: task.Task[T]) -> T: entry[1].remove(future) if not entry[1]: self._waits.pop(id(durable), None) - self._observers.pop(id(durable)).detach() + # Dispatch close may already have detached and cleared observers. + observer = self._observers.pop(id(durable), None) + if observer is not None: + observer.detach() def _complete_wait(self, durable: task.Task[Any]) -> None: entry = self._waits.get(id(durable)) @@ -384,24 +400,43 @@ def close(self) -> None: self._timers.clear() self._tasks.clear() - def run_in_executor(self, executor: Any, func: Any, *args: Any) -> asyncio.Future[Any]: - raise RuntimeError("Threads/executors are unsupported in orchestrators; use an activity.") - - def call_soon_threadsafe(self, callback: Any, *args: Any, context: Any = None) -> asyncio.Handle: - raise RuntimeError("Cross-thread scheduling is unsupported in orchestrators; use an activity.") - - async def _unsupported_io(self, *args: Any, **kwargs: Any) -> Any: - raise RuntimeError("External I/O is unsupported in orchestrators; move it to an ordinary activity.") - - create_connection = _unsupported_io - create_server = _unsupported_io - create_datagram_endpoint = _unsupported_io - connect_read_pipe = _unsupported_io - connect_write_pipe = _unsupported_io - getaddrinfo = _unsupported_io - getnameinfo = _unsupported_io - sock_connect = _unsupported_io - sock_recv = _unsupported_io - sock_sendall = _unsupported_io - subprocess_exec = _unsupported_io - subprocess_shell = _unsupported_io + +_IO_HINT = "External I/O is unsupported in orchestrators; move it to an ordinary activity." +_THREAD_HINT = "Threads/executors are unsupported in orchestrators; use an activity." +_LOOP_HINT = "The worker owns the durable loop's lifecycle and configuration." + +# Every asyncio loop API this loop does not implement fails with one actionable +# error instead of AbstractEventLoop's bare NotImplementedError. +_UNSUPPORTED_LOOP_APIS: dict[str, str] = { + **dict.fromkeys(( + "create_connection", "create_server", "create_datagram_endpoint", + "create_unix_connection", "create_unix_server", "connect_accepted_socket", + "connect_read_pipe", "connect_write_pipe", "start_tls", "sendfile", + "getaddrinfo", "getnameinfo", "sock_connect", "sock_accept", "sock_recv", + "sock_recv_into", "sock_recvfrom", "sock_recvfrom_into", "sock_sendall", + "sock_sendto", "sock_sendfile", "subprocess_exec", "subprocess_shell", + "add_reader", "remove_reader", "add_writer", "remove_writer", + "add_signal_handler", "remove_signal_handler", + ), _IO_HINT), + **dict.fromkeys(( + "run_in_executor", "set_default_executor", "shutdown_default_executor", + "call_soon_threadsafe", + ), _THREAD_HINT), + **dict.fromkeys(( + "run_forever", "run_until_complete", "stop", "set_task_factory", + "set_exception_handler", + ), _LOOP_HINT), +} + + +def _unsupported_loop_api(name: str, hint: str) -> Callable[..., Any]: + def unsupported(self: DurableEventLoop, *args: Any, **kwargs: Any) -> Any: + raise RuntimeError(f"loop.{name}() is not supported in durable orchestrators. {hint}") + + unsupported.__name__ = unsupported.__qualname__ = name + return unsupported + + +for _name, _hint in _UNSUPPORTED_LOOP_APIS.items(): + if _name not in vars(DurableEventLoop): + setattr(DurableEventLoop, _name, _unsupported_loop_api(_name, _hint)) diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index d0c2db8f..45bfde6b 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -11,10 +11,11 @@ import sys from contextvars import ContextVar from dataclasses import dataclass -from datetime import timedelta +from datetime import datetime, timedelta from decimal import ROUND_CEILING, Decimal, localcontext from functools import partial -from typing import Any +from types import SimpleNamespace +from typing import Any, cast from weakref import ReferenceType, ref import pytest @@ -22,6 +23,7 @@ from durabletask import task from durabletask.internal import helpers from durabletask.internal import orchestrator_service_pb2 as pb +from durabletask.internal.asyncio_loop import _UNSUPPORTED_LOOP_APIS, DurableEventLoop # pyright: ignore[reportPrivateUsage] from tests.durabletask.test_coroutine_orchestrators import ( NOW, _completion, # pyright: ignore[reportPrivateUsage] @@ -396,7 +398,6 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: actions = _execute(orchestrator, [], _start()).actions assert len(actions) == 1 deadline = actions[0].createTimer.fireAt.ToDatetime() - from datetime import datetime assert (deadline - datetime(1970, 1, 1)).total_seconds() >= observed[0] @@ -498,6 +499,30 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: "use an activity", "ordinary activity", "different loop")) +@pytest.mark.parametrize("name", sorted(_UNSUPPORTED_LOOP_APIS)) +def test_every_unsupported_loop_api_fails_actionably(name: str) -> None: + ctx = cast(task.OrchestrationContext, SimpleNamespace(current_utc_datetime=NOW)) + loop = DurableEventLoop(ctx, is_complete=lambda: False) + with pytest.raises(RuntimeError, match=rf"loop\.{name}\(\) is not supported") as error: + getattr(loop, name)() + assert str(error.value).endswith(_UNSUPPORTED_LOOP_APIS[name]) + + +def test_untracked_task_finalized_after_dispatch_close_is_quiet(monkeypatch: pytest.MonkeyPatch) -> None: + unraisable: list[Any] = [] + monkeypatch.setattr(sys, "unraisablehook", unraisable.append) + untracked: list[asyncio.Task[Any]] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + untracked.append(asyncio.Task(ctx.call_activity("background").wait())) + await ctx.call_activity("work") + + _execute(orchestrator, [], _start()) + untracked.clear() + gc.collect() + assert not [hook for hook in unraisable if isinstance(hook.exc_value, KeyError)] + + def test_task_group_cancellation_durable_cleanup_has_stable_action_order() -> None: async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: async def branch(index: int) -> None: