Skip to content

Commit acae0c8

Browse files
andystaplesCopilot
andcommitted
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>
1 parent ba1476e commit acae0c8

5 files changed

Lines changed: 101 additions & 46 deletions

File tree

‎CHANGELOG.md‎

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

1010
FIXED
1111

12-
- Coroutine task-group cancellation now schedules shared-Future callbacks
13-
deterministically across replays, including when a dispatch resumes on another
14-
worker using the same SDK and Python runtime. Previously recorded cancellation
15-
histories may replay differently; validate or drain them before upgrading.
1612
- Activity and sub-orchestration `retry_timeout` decisions now use replay-safe
1713
orchestration time, preventing retries from expiring solely because history is
1814
replayed later. Retry calculations also accept timezone-aware timestamps.

‎docs/async-orchestrators.md‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -187,16 +187,20 @@ Create branches from ordered inputs. Ready work follows FIFO scheduling;
187187
history completions are delivered in history order, draining ready work after
188188
each completion. Task-group cancellation wakeups are ordered by local task
189189
creation, not set/object-hash iteration. Local tasks are dispatch resources,
190-
not new durable instances.
190+
not new durable instances. Create them with `asyncio.create_task` or a task
191+
group; do not construct `asyncio.Task` directly, because such tasks are not
192+
tracked by the dispatch and are not cancelled or cleaned up with it.
191193

192194
`loop.time()` uses seconds from a fixed UTC epoch and the current orchestration
193195
clock, never machine monotonic time. `loop.call_at` and `call_later` schedule
194196
durable timers for future finite deadlines. Timer conversion rounds up when
195197
necessary so a durable timer does not precede its floating-point deadline.
196198
Core/Functions long timers reuse the existing chunking; Azure Managed retains
197199
its native long timers. Past deadlines are ready callbacks. `sleep(0)` yields
198-
to ready coroutines without a durable timer. Clock advancement and timers
199-
require history events, not elapsed time in the worker.
200+
to ready coroutines without a durable timer. Delays below the resolution of
201+
`loop.time()` (about 238 nanoseconds at current dates) are effectively zero.
202+
Clock advancement and timers require history events, not elapsed time in the
203+
worker.
200204

201205
> [!WARNING]
202206
> This is not a general-purpose asyncio event loop. External network/storage

‎durabletask-azuremanaged/CHANGELOG.md‎

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

1010
FIXED
1111

12-
- With the corresponding core SDK update, coroutine task-group cancellation
13-
schedules shared-Future callbacks deterministically when replayed on another
14-
worker using the same SDK and Python runtime. See the
15-
[core upgrade guidance](../CHANGELOG.md#unreleased) for existing cancellation
16-
histories.
1712
- With the corresponding core SDK update, activity and sub-orchestration
1813
`retry_timeout` decisions use replay-safe orchestration time rather than
1914
wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased)

‎durabletask/internal/asyncio_loop.py‎

Lines changed: 66 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,22 @@
1515
- ``asyncio.Handle._run``: run a ready callback with asyncio's own context and
1616
exception-handler semantics.
1717
- ``asyncio.Task._log_traceback``: report unobserved local-task failures during
18-
dispatch cleanup without marking them as retrieved.
18+
dispatch cleanup without marking them as retrieved, and keep such tasks
19+
tracked until then.
20+
- ``__self__`` of Task step/wakeup callbacks (including the C implementation's
21+
step wrapper and bound ``task_wakeup``): identify which local task a
22+
callback resumes, so task-group cancellation wakeups run in task-creation
23+
order.
24+
- ``AbstractEventLoop._timer_handle_cancelled``: the hook ``TimerHandle.cancel``
25+
calls on its loop, used to withdraw the durable timer observer.
1926
"""
2027

2128
from __future__ import annotations
2229

2330
import asyncio
2431
import contextvars
25-
import math
2632
import logging
33+
import math
2734
from collections import deque
2835
from collections.abc import Callable, Coroutine, Generator
2936
from datetime import UTC, datetime, timedelta
@@ -63,8 +70,10 @@ def __init__(
6370
super().__init__(coro, loop=loop, name=name, context=context)
6471

6572
def __hash__(self) -> int:
66-
# TaskGroup traverses a set before cancellation schedules shared-Future
67-
# callbacks. Stabilize that traversal, not just the resulting wakeups.
73+
# Load-bearing for replay determinism: TaskGroup keeps its children in a
74+
# set and iterates it to cancel them. Hashing by creation sequence makes
75+
# that traversal, and the cancellation wakeups it schedules, follow task
76+
# creation order instead of memory addresses that differ across workers.
6877
return self._sequence
6978

7079
def __await__(self) -> Generator[Any, None, T]:
@@ -150,12 +159,15 @@ def __init__(
150159
self._fatal: BaseException | None = None
151160
self._root: asyncio.Task[Any] | None = None
152161

153-
def time(self) -> float:
154-
# A fixed UTC epoch avoids local-time and platform timestamp differences.
162+
def _now(self) -> datetime:
155163
now = self.ctx.current_utc_datetime
156164
if now.tzinfo is not None:
157165
now = now.astimezone(UTC).replace(tzinfo=None)
158-
return (now - _EPOCH).total_seconds()
166+
return now
167+
168+
def time(self) -> float:
169+
# A fixed UTC epoch avoids local-time and platform timestamp differences.
170+
return (self._now() - _EPOCH).total_seconds()
159171

160172
def get_debug(self) -> bool:
161173
return False
@@ -204,7 +216,8 @@ def call_soon(
204216
# TaskGroup cancels a set of tasks. Do not let object hashes
205217
# determine the order in which their finally blocks execute.
206218
owner = getattr(callback, "__self__", None)
207-
index = self._tasks[self.cancelling_task]
219+
index = self._tasks.get(
220+
self.cancelling_task, getattr(self.cancelling_task, "_sequence", self._task_sequence))
208221
if isinstance(owner, asyncio.Task):
209222
index = self._tasks.get(cast(asyncio.Task[Any], owner), index)
210223
self._cancel_ready.append((index, len(self._cancel_ready), handle))
@@ -230,7 +243,7 @@ def call_at(
230243
# Exact rational ceiling: float round-trips can emit a deadline that
231244
# precedes the requested one by a fraction of a microsecond.
232245
deadline_us = math.ceil(Fraction(when) * 1_000_000)
233-
now_us = (self.ctx.current_utc_datetime - _EPOCH) // _MICROSECOND
246+
now_us = (self._now() - _EPOCH) // _MICROSECOND
234247
if when <= self.time() or deadline_us <= now_us:
235248
self._ready.append(handle)
236249
else:
@@ -270,7 +283,10 @@ async def wait_durable(self, durable: task.Task[T]) -> T:
270283
entry[1].remove(future)
271284
if not entry[1]:
272285
self._waits.pop(id(durable), None)
273-
self._observers.pop(id(durable)).detach()
286+
# Dispatch close may already have detached and cleared observers.
287+
observer = self._observers.pop(id(durable), None)
288+
if observer is not None:
289+
observer.detach()
274290

275291
def _complete_wait(self, durable: task.Task[Any]) -> None:
276292
entry = self._waits.get(id(durable))
@@ -384,24 +400,43 @@ def close(self) -> None:
384400
self._timers.clear()
385401
self._tasks.clear()
386402

387-
def run_in_executor(self, executor: Any, func: Any, *args: Any) -> asyncio.Future[Any]:
388-
raise RuntimeError("Threads/executors are unsupported in orchestrators; use an activity.")
389-
390-
def call_soon_threadsafe(self, callback: Any, *args: Any, context: Any = None) -> asyncio.Handle:
391-
raise RuntimeError("Cross-thread scheduling is unsupported in orchestrators; use an activity.")
392-
393-
async def _unsupported_io(self, *args: Any, **kwargs: Any) -> Any:
394-
raise RuntimeError("External I/O is unsupported in orchestrators; move it to an ordinary activity.")
395-
396-
create_connection = _unsupported_io
397-
create_server = _unsupported_io
398-
create_datagram_endpoint = _unsupported_io
399-
connect_read_pipe = _unsupported_io
400-
connect_write_pipe = _unsupported_io
401-
getaddrinfo = _unsupported_io
402-
getnameinfo = _unsupported_io
403-
sock_connect = _unsupported_io
404-
sock_recv = _unsupported_io
405-
sock_sendall = _unsupported_io
406-
subprocess_exec = _unsupported_io
407-
subprocess_shell = _unsupported_io
403+
404+
_IO_HINT = "External I/O is unsupported in orchestrators; move it to an ordinary activity."
405+
_THREAD_HINT = "Threads/executors are unsupported in orchestrators; use an activity."
406+
_LOOP_HINT = "The worker owns the durable loop's lifecycle and configuration."
407+
408+
# Every asyncio loop API this loop does not implement fails with one actionable
409+
# error instead of AbstractEventLoop's bare NotImplementedError.
410+
_UNSUPPORTED_LOOP_APIS: dict[str, str] = {
411+
**dict.fromkeys((
412+
"create_connection", "create_server", "create_datagram_endpoint",
413+
"create_unix_connection", "create_unix_server", "connect_accepted_socket",
414+
"connect_read_pipe", "connect_write_pipe", "start_tls", "sendfile",
415+
"getaddrinfo", "getnameinfo", "sock_connect", "sock_accept", "sock_recv",
416+
"sock_recv_into", "sock_recvfrom", "sock_recvfrom_into", "sock_sendall",
417+
"sock_sendto", "sock_sendfile", "subprocess_exec", "subprocess_shell",
418+
"add_reader", "remove_reader", "add_writer", "remove_writer",
419+
"add_signal_handler", "remove_signal_handler",
420+
), _IO_HINT),
421+
**dict.fromkeys((
422+
"run_in_executor", "set_default_executor", "shutdown_default_executor",
423+
"call_soon_threadsafe",
424+
), _THREAD_HINT),
425+
**dict.fromkeys((
426+
"run_forever", "run_until_complete", "stop", "set_task_factory",
427+
"set_exception_handler",
428+
), _LOOP_HINT),
429+
}
430+
431+
432+
def _unsupported_loop_api(name: str, hint: str) -> Callable[..., Any]:
433+
def unsupported(self: DurableEventLoop, *args: Any, **kwargs: Any) -> Any:
434+
raise RuntimeError(f"loop.{name}() is not supported in durable orchestrators. {hint}")
435+
436+
unsupported.__name__ = unsupported.__qualname__ = name
437+
return unsupported
438+
439+
440+
for _name, _hint in _UNSUPPORTED_LOOP_APIS.items():
441+
if _name not in vars(DurableEventLoop):
442+
setattr(DurableEventLoop, _name, _unsupported_loop_api(_name, _hint))

‎tests/durabletask/test_asyncio_orchestrators.py‎

Lines changed: 28 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,17 +11,19 @@
1111
import sys
1212
from contextvars import ContextVar
1313
from dataclasses import dataclass
14-
from datetime import timedelta
14+
from datetime import datetime, timedelta
1515
from decimal import ROUND_CEILING, Decimal, localcontext
1616
from functools import partial
17-
from typing import Any
17+
from types import SimpleNamespace
18+
from typing import Any, cast
1819
from weakref import ReferenceType, ref
1920

2021
import pytest
2122

2223
from durabletask import task
2324
from durabletask.internal import helpers
2425
from durabletask.internal import orchestrator_service_pb2 as pb
26+
from durabletask.internal.asyncio_loop import _UNSUPPORTED_LOOP_APIS, DurableEventLoop # pyright: ignore[reportPrivateUsage]
2527
from tests.durabletask.test_coroutine_orchestrators import (
2628
NOW,
2729
_completion, # pyright: ignore[reportPrivateUsage]
@@ -396,7 +398,6 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
396398
actions = _execute(orchestrator, [], _start()).actions
397399
assert len(actions) == 1
398400
deadline = actions[0].createTimer.fireAt.ToDatetime()
399-
from datetime import datetime
400401
assert (deadline - datetime(1970, 1, 1)).total_seconds() >= observed[0]
401402

402403

@@ -498,6 +499,30 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
498499
"use an activity", "ordinary activity", "different loop"))
499500

500501

502+
@pytest.mark.parametrize("name", sorted(_UNSUPPORTED_LOOP_APIS))
503+
def test_every_unsupported_loop_api_fails_actionably(name: str) -> None:
504+
ctx = cast(task.OrchestrationContext, SimpleNamespace(current_utc_datetime=NOW))
505+
loop = DurableEventLoop(ctx, is_complete=lambda: False)
506+
with pytest.raises(RuntimeError, match=rf"loop\.{name}\(\) is not supported") as error:
507+
getattr(loop, name)()
508+
assert str(error.value).endswith(_UNSUPPORTED_LOOP_APIS[name])
509+
510+
511+
def test_untracked_task_finalized_after_dispatch_close_is_quiet(monkeypatch: pytest.MonkeyPatch) -> None:
512+
unraisable: list[Any] = []
513+
monkeypatch.setattr(sys, "unraisablehook", unraisable.append)
514+
untracked: list[asyncio.Task[Any]] = []
515+
516+
async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
517+
untracked.append(asyncio.Task(ctx.call_activity("background").wait()))
518+
await ctx.call_activity("work")
519+
520+
_execute(orchestrator, [], _start())
521+
untracked.clear()
522+
gc.collect()
523+
assert not [hook for hook in unraisable if isinstance(hook.exc_value, KeyError)]
524+
525+
501526
def test_task_group_cancellation_durable_cleanup_has_stable_action_order() -> None:
502527
async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
503528
async def branch(index: int) -> None:

0 commit comments

Comments
 (0)