Skip to content

Commit 95e411c

Browse files
andystaplesCopilot
andcommitted
Merge andystaples-async-stack-3-dispatch-cleanup into andystaples-async-stack-4-async-context
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
2 parents 704b30b + e918345 commit 95e411c

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
@@ -16,10 +16,6 @@ unchanged.
1616
- Async dispatch cleanup no longer hangs when async-generator finalizers
1717
continually start successor generators; exceeding the cleanup budget is reported
1818
without changing the snapshotted response.
19-
- Coroutine task-group cancellation now schedules shared-Future callbacks
20-
deterministically across replays, including when a dispatch resumes on another
21-
worker using the same SDK and Python runtime. Previously recorded cancellation
22-
histories may replay differently; validate or drain them before upgrading.
2319
- Activity and sub-orchestration `retry_timeout` decisions now use replay-safe
2420
orchestration time, preventing retries from expiring solely because history is
2521
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
@@ -193,16 +193,20 @@ Create branches from ordered inputs. Ready work follows FIFO scheduling;
193193
history completions are delivered in history order, draining ready work after
194194
each completion. Task-group cancellation wakeups are ordered by local task
195195
creation, not set/object-hash iteration. Local tasks are dispatch resources,
196-
not new durable instances.
196+
not new durable instances. Create them with `asyncio.create_task` or a task
197+
group; do not construct `asyncio.Task` directly, because such tasks are not
198+
tracked by the dispatch and are not cancelled or cleaned up with it.
197199

198200
`loop.time()` uses seconds from a fixed UTC epoch and the current orchestration
199201
clock, never machine monotonic time. `loop.call_at` and `call_later` schedule
200202
durable timers for future finite deadlines. Timer conversion rounds up when
201203
necessary so a durable timer does not precede its floating-point deadline.
202204
Core/Functions long timers reuse the existing chunking; Azure Managed retains
203205
its native long timers. Past deadlines are ready callbacks. `sleep(0)` yields
204-
to ready coroutines without a durable timer. Clock advancement and timers
205-
require history events, not elapsed time in the worker.
206+
to ready coroutines without a durable timer. Delays below the resolution of
207+
`loop.time()` (about 238 nanoseconds at current dates) are effectively zero.
208+
Clock advancement and timers require history events, not elapsed time in the
209+
worker.
206210

207211
> [!WARNING]
208212
> 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
@@ -12,11 +12,6 @@ FIXED
1212
- With the corresponding core SDK update, async dispatch cleanup no longer
1313
hangs when async-generator finalizers continually start successor generators.
1414
Exceeding the cleanup budget is reported without changing the snapshotted response.
15-
- With the corresponding core SDK update, coroutine task-group cancellation
16-
schedules shared-Future callbacks deterministically when replayed on another
17-
worker using the same SDK and Python runtime. See the
18-
[core upgrade guidance](../CHANGELOG.md#unreleased) for existing cancellation
19-
histories.
2015
- With the corresponding core SDK update, activity and sub-orchestration
2116
`retry_timeout` decisions use replay-safe orchestration time rather than
2217
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
import sys
2835
from collections import deque
2936
from collections.abc import Callable, Coroutine, Generator
@@ -72,8 +79,10 @@ def __init__(
7279
super().__init__(coro, loop=loop, name=name, context=context)
7380

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

7988
def __await__(self) -> Generator[Any, None, T]:
@@ -162,12 +171,15 @@ def __init__(
162171
self._cleanup_failure: BaseException | None = None
163172
self._root: asyncio.Task[Any] | None = None
164173

165-
def time(self) -> float:
166-
# A fixed UTC epoch avoids local-time and platform timestamp differences.
174+
def _now(self) -> datetime:
167175
now = self.ctx.current_utc_datetime
168176
if now.tzinfo is not None:
169177
now = now.astimezone(UTC).replace(tzinfo=None)
170-
return (now - _EPOCH).total_seconds()
178+
return now
179+
180+
def time(self) -> float:
181+
# A fixed UTC epoch avoids local-time and platform timestamp differences.
182+
return (self._now() - _EPOCH).total_seconds()
171183

172184
def get_debug(self) -> bool:
173185
return False
@@ -216,7 +228,8 @@ def call_soon(
216228
# TaskGroup cancels a set of tasks. Do not let object hashes
217229
# determine the order in which their finally blocks execute.
218230
owner = getattr(callback, "__self__", None)
219-
index = self._tasks[self.cancelling_task]
231+
index = self._tasks.get(
232+
self.cancelling_task, getattr(self.cancelling_task, "_sequence", self._task_sequence))
220233
if isinstance(owner, asyncio.Task):
221234
index = self._tasks.get(cast(asyncio.Task[Any], owner), index)
222235
self._cancel_ready.append((index, len(self._cancel_ready), handle))
@@ -242,7 +255,7 @@ def call_at(
242255
# Exact rational ceiling: float round-trips can emit a deadline that
243256
# precedes the requested one by a fraction of a microsecond.
244257
deadline_us = math.ceil(Fraction(when) * 1_000_000)
245-
now_us = (self.ctx.current_utc_datetime - _EPOCH) // _MICROSECOND
258+
now_us = (self._now() - _EPOCH) // _MICROSECOND
246259
if when <= self.time() or deadline_us <= now_us:
247260
self._ready.append(handle)
248261
else:
@@ -284,7 +297,10 @@ async def wait_durable(self, durable: task.Task[T]) -> T:
284297
entry[1].remove(future)
285298
if not entry[1]:
286299
self._waits.pop(id(durable), None)
287-
self._observers.pop(id(durable)).detach()
300+
# Dispatch close may already have detached and cleared observers.
301+
observer = self._observers.pop(id(durable), None)
302+
if observer is not None:
303+
observer.detach()
288304

289305
def _complete_wait(self, durable: task.Task[Any]) -> None:
290306
entry = self._waits.get(id(durable))
@@ -473,24 +489,43 @@ def close(self) -> None:
473489
self._tasks.clear()
474490
self._asyncgens.clear()
475491

476-
def run_in_executor(self, executor: Any, func: Any, *args: Any) -> asyncio.Future[Any]:
477-
raise RuntimeError("Threads/executors are unsupported in orchestrators; use an activity.")
478-
479-
def call_soon_threadsafe(self, callback: Any, *args: Any, context: Any = None) -> asyncio.Handle:
480-
raise RuntimeError("Cross-thread scheduling is unsupported in orchestrators; use an activity.")
481-
482-
async def _unsupported_io(self, *args: Any, **kwargs: Any) -> Any:
483-
raise RuntimeError("External I/O is unsupported in orchestrators; move it to an ordinary activity.")
484-
485-
create_connection = _unsupported_io
486-
create_server = _unsupported_io
487-
create_datagram_endpoint = _unsupported_io
488-
connect_read_pipe = _unsupported_io
489-
connect_write_pipe = _unsupported_io
490-
getaddrinfo = _unsupported_io
491-
getnameinfo = _unsupported_io
492-
sock_connect = _unsupported_io
493-
sock_recv = _unsupported_io
494-
sock_sendall = _unsupported_io
495-
subprocess_exec = _unsupported_io
496-
subprocess_shell = _unsupported_io
492+
493+
_IO_HINT = "External I/O is unsupported in orchestrators; move it to an ordinary activity."
494+
_THREAD_HINT = "Threads/executors are unsupported in orchestrators; use an activity."
495+
_LOOP_HINT = "The worker owns the durable loop's lifecycle and configuration."
496+
497+
# Every asyncio loop API this loop does not implement fails with one actionable
498+
# error instead of AbstractEventLoop's bare NotImplementedError.
499+
_UNSUPPORTED_LOOP_APIS: dict[str, str] = {
500+
**dict.fromkeys((
501+
"create_connection", "create_server", "create_datagram_endpoint",
502+
"create_unix_connection", "create_unix_server", "connect_accepted_socket",
503+
"connect_read_pipe", "connect_write_pipe", "start_tls", "sendfile",
504+
"getaddrinfo", "getnameinfo", "sock_connect", "sock_accept", "sock_recv",
505+
"sock_recv_into", "sock_recvfrom", "sock_recvfrom_into", "sock_sendall",
506+
"sock_sendto", "sock_sendfile", "subprocess_exec", "subprocess_shell",
507+
"add_reader", "remove_reader", "add_writer", "remove_writer",
508+
"add_signal_handler", "remove_signal_handler",
509+
), _IO_HINT),
510+
**dict.fromkeys((
511+
"run_in_executor", "set_default_executor", "shutdown_default_executor",
512+
"call_soon_threadsafe",
513+
), _THREAD_HINT),
514+
**dict.fromkeys((
515+
"run_forever", "run_until_complete", "stop", "set_task_factory",
516+
"set_exception_handler",
517+
), _LOOP_HINT),
518+
}
519+
520+
521+
def _unsupported_loop_api(name: str, hint: str) -> Callable[..., Any]:
522+
def unsupported(self: DurableEventLoop, *args: Any, **kwargs: Any) -> Any:
523+
raise RuntimeError(f"loop.{name}() is not supported in durable orchestrators. {hint}")
524+
525+
unsupported.__name__ = unsupported.__qualname__ = name
526+
return unsupported
527+
528+
529+
for _name, _hint in _UNSUPPORTED_LOOP_APIS.items():
530+
if _name not in vars(DurableEventLoop):
531+
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
@@ -13,17 +13,19 @@
1313
from contextlib import asynccontextmanager
1414
from contextvars import ContextVar
1515
from dataclasses import dataclass
16-
from datetime import timedelta
16+
from datetime import datetime, timedelta
1717
from decimal import ROUND_CEILING, Decimal, localcontext
1818
from functools import partial
19-
from typing import Any
19+
from types import SimpleNamespace
20+
from typing import Any, cast
2021
from weakref import ReferenceType, ref
2122

2223
import pytest
2324

2425
from durabletask import entities, task
2526
from durabletask.internal import helpers
2627
from durabletask.internal import orchestrator_service_pb2 as pb
28+
from durabletask.internal.asyncio_loop import _UNSUPPORTED_LOOP_APIS, DurableEventLoop # pyright: ignore[reportPrivateUsage]
2729
from tests.durabletask.test_coroutine_orchestrators import (
2830
NOW,
2931
_completion, # pyright: ignore[reportPrivateUsage]
@@ -492,7 +494,6 @@ async def orchestrator(ctx: task.AsyncOrchestrationContext, _: Any) -> None:
492494
actions = _execute(orchestrator, [], _start()).actions
493495
assert len(actions) == 1
494496
deadline = actions[0].createTimer.fireAt.ToDatetime()
495-
from datetime import datetime
496497
assert (deadline - datetime(1970, 1, 1)).total_seconds() >= observed[0]
497498

498499

@@ -630,6 +631,30 @@ async def resource():
630631
assert "cleanup failed" in caplog.text
631632

632633

634+
@pytest.mark.parametrize("name", sorted(_UNSUPPORTED_LOOP_APIS))
635+
def test_every_unsupported_loop_api_fails_actionably(name: str) -> None:
636+
ctx = cast(task.OrchestrationContext, SimpleNamespace(current_utc_datetime=NOW))
637+
loop = DurableEventLoop(ctx, is_complete=lambda: False)
638+
with pytest.raises(RuntimeError, match=rf"loop\.{name}\(\) is not supported") as error:
639+
getattr(loop, name)()
640+
assert str(error.value).endswith(_UNSUPPORTED_LOOP_APIS[name])
641+
642+
643+
def test_untracked_task_finalized_after_dispatch_close_is_quiet(monkeypatch: pytest.MonkeyPatch) -> None:
644+
unraisable: list[Any] = []
645+
monkeypatch.setattr(sys, "unraisablehook", unraisable.append)
646+
untracked: list[asyncio.Task[Any]] = []
647+
648+
async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
649+
untracked.append(asyncio.Task(ctx.call_activity("background").wait()))
650+
await ctx.call_activity("work")
651+
652+
_execute(orchestrator, [], _start())
653+
untracked.clear()
654+
gc.collect()
655+
assert not [hook for hook in unraisable if isinstance(hook.exc_value, KeyError)]
656+
657+
633658
def test_task_group_cancellation_durable_cleanup_has_stable_action_order() -> None:
634659
async def orchestrator(ctx: task.AsyncOrchestrationContext, _: Any) -> None:
635660
async def branch(index: int) -> None:

0 commit comments

Comments
 (0)