Skip to content

Commit 89ab221

Browse files
andystaplesCopilot
andcommitted
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>
1 parent 1e5bf2f commit 89ab221

5 files changed

Lines changed: 145 additions & 3 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,10 @@ 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.
1216
- Activity and sub-orchestration `retry_timeout` decisions now use replay-safe
1317
orchestration time, preventing retries from expiring solely because history is
1418
replayed later. Retry calculations also accept timezone-aware timestamps.

‎docs/async-orchestrators.md‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,12 @@ automatic recording or interception of arbitrary await expressions.
261261
`is_replaying` is not a substitute for recording results. Activities remain
262262
at-least-once work: side effects require appropriate idempotency/deduplication.
263263

264+
Successive orchestration dispatches and their activities can run on different
265+
workers. Replay reconstructs local tasks and futures from ordered durable
266+
history; it does not rely on task identities or memory retained by a previous
267+
worker. Activity completion order is the order recorded in history, not worker
268+
placement.
269+
264270
Framework adapters can compose supported durable operations using async
265271
helpers. Adapters remain responsible for registering ordinary model/tool
266272
activities, encoding/decoding framework payloads, and supplying replay-stable
@@ -279,7 +285,8 @@ construction order is not execution order.
279285

280286
Supporting several Python versions does not guarantee recorded-history
281287
portability between them. Pin the runtime and dependencies for active
282-
instances and validate histories before upgrades. In particular, Python 3.11
288+
instances and validate histories before upgrades. All workers processing an
289+
instance must use that pinned environment. In particular, Python 3.11
283290
and 3.12+ schedule coroutine operands of `asyncio.wait_for()` differently, which
284291
can change durable action order inside concurrent branches. The async context
285292
does not normalize arbitrary standard-library scheduling.

‎durabletask-azuremanaged/CHANGELOG.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,11 @@ 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.
1217
- With the corresponding core SDK update, activity and sub-orchestration
1318
`retry_timeout` decisions use replay-safe orchestration time rather than
1419
wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased)

‎durabletask/internal/asyncio_loop.py‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,19 @@ def detach(self) -> None:
3535

3636

3737
class _DurableAsyncTask(asyncio.Task[T]):
38+
def __init__(
39+
self, coro: Coroutine[Any, Any, T] | Generator[Any, None, T], *,
40+
sequence: int, loop: DurableEventLoop, name: str | None,
41+
context: contextvars.Context | None,
42+
) -> None:
43+
self._sequence = sequence
44+
super().__init__(coro, loop=loop, name=name, context=context)
45+
46+
def __hash__(self) -> int:
47+
# TaskGroup traverses a set before cancellation schedules shared-Future
48+
# callbacks. Stabilize that traversal, not just the resulting wakeups.
49+
return self._sequence
50+
3851
def __await__(self) -> Generator[Any, None, T]:
3952
try:
4053
return (yield from super().__await__())
@@ -147,9 +160,10 @@ def create_task(
147160
if kwargs:
148161
coro.close()
149162
raise TypeError("Custom/eager task execution is not supported in durable orchestrators.")
150-
future = _DurableAsyncTask(coro, loop=self, name=name, context=context)
151-
self._tasks[future] = self._task_sequence
163+
sequence = self._task_sequence
152164
self._task_sequence += 1
165+
future = _DurableAsyncTask(coro, sequence=sequence, loop=self, name=name, context=context)
166+
self._tasks[future] = sequence
153167
future.add_done_callback(self.retire_task)
154168
return future
155169

‎tests/durabletask/test_asyncio_orchestrators.py‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,13 @@
66
import asyncio
77
import gc
88
import json
9+
import os
10+
import subprocess
11+
import sys
912
from contextvars import ContextVar
1013
from dataclasses import dataclass
1114
from datetime import timedelta
15+
from functools import partial
1216
from typing import Any
1317
from weakref import ReferenceType, ref
1418

@@ -463,6 +467,114 @@ async def branch(index: int) -> None:
463467
assert _execute(orchestrator, [], old).actions == expected
464468

465469

470+
async def _shared_future_cancellation_orchestrator(
471+
ctx: task.OrchestrationContext, _: Any, *, direct_future: bool, gather_wait: bool,
472+
) -> str:
473+
loop = asyncio.get_running_loop()
474+
futures: list[asyncio.Future[None]] = [
475+
asyncio.Future[None]() if direct_future else loop.create_future() for _ in range(2)]
476+
477+
def on_cancelled(index: int, future: asyncio.Future[None]) -> None:
478+
assert future.cancelled()
479+
ctx.call_activity(f"cancelled-{index}")
480+
481+
for index, future in enumerate(futures):
482+
future.add_done_callback(partial(on_cancelled, index))
483+
484+
async def wait(future: asyncio.Future[None]) -> None:
485+
if gather_wait:
486+
await asyncio.gather(future)
487+
else:
488+
await future
489+
490+
async def fail() -> None:
491+
await asyncio.sleep(0)
492+
raise ValueError("cancel siblings")
493+
494+
try:
495+
async with asyncio.TaskGroup() as group:
496+
for future in futures * 2:
497+
group.create_task(wait(future))
498+
group.create_task(fail())
499+
except* ValueError:
500+
pass
501+
return await ctx.call_activity("join", return_type=str)
502+
503+
504+
@pytest.mark.parametrize("direct_future", [False, True])
505+
@pytest.mark.parametrize("gather_wait", [False, True])
506+
def test_shared_future_cancellation_callbacks_replay_on_fresh_executors(
507+
direct_future: bool, gather_wait: bool,
508+
) -> None:
509+
orchestrator = partial(
510+
_shared_future_cancellation_orchestrator, direct_future=direct_future, gather_wait=gather_wait)
511+
start = _start()
512+
initial = _execute(orchestrator, [], start)
513+
assert sorted(a.scheduleTask.name for a in initial.actions[:-1]) == ["cancelled-0", "cancelled-1"]
514+
assert initial.actions[-1].scheduleTask.name == "join"
515+
recorded = start + [
516+
helpers.new_task_scheduled_event(action.id, action.scheduleTask.name)
517+
for action in initial.actions
518+
]
519+
# Activity workers may complete in any order. Each dispatch reconstructs
520+
# all local tasks and futures from history, as on a different worker.
521+
completed = [
522+
helpers.new_task_completed_event(initial.actions[1].id, "null"),
523+
helpers.new_task_completed_event(initial.actions[0].id, "null"),
524+
]
525+
for _ in range(30):
526+
assert _execute(orchestrator, [], start).actions == initial.actions
527+
assert _execute(orchestrator, recorded, completed).actions == []
528+
complete = _completion(_execute(
529+
orchestrator, recorded + completed,
530+
[helpers.new_task_completed_event(initial.actions[-1].id, '"joined"')]))
531+
assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, complete.failureDetails
532+
assert json.loads(complete.result.value) == "joined"
533+
534+
535+
def test_shared_future_cancellation_history_replays_in_other_worker_processes() -> None:
536+
orchestrator = partial(
537+
_shared_future_cancellation_orchestrator, direct_future=True, gather_wait=False)
538+
start = _start()
539+
initial = _execute(orchestrator, [], start)
540+
history = start + [
541+
helpers.new_task_scheduled_event(action.id, action.scheduleTask.name)
542+
for action in initial.actions
543+
]
544+
history += [
545+
helpers.new_task_completed_event(initial.actions[1].id, "null"),
546+
helpers.new_task_completed_event(initial.actions[0].id, "null"),
547+
]
548+
payload = json.dumps({
549+
"initial": [action.SerializeToString().hex() for action in initial.actions],
550+
"history": [event.SerializeToString().hex() for event in history],
551+
"join_id": initial.actions[-1].id,
552+
})
553+
code = """
554+
import json
555+
import sys
556+
from functools import partial
557+
from durabletask.internal import helpers, orchestrator_service_pb2 as pb
558+
from tests.durabletask.test_asyncio_orchestrators import _shared_future_cancellation_orchestrator
559+
from tests.durabletask.test_coroutine_orchestrators import _completion, _execute, _start
560+
561+
data = json.load(sys.stdin)
562+
orchestrator = partial(_shared_future_cancellation_orchestrator, direct_future=True, gather_wait=False)
563+
initial = _execute(orchestrator, [], _start())
564+
assert [action.SerializeToString().hex() for action in initial.actions] == data["initial"]
565+
history = [pb.HistoryEvent.FromString(bytes.fromhex(event)) for event in data["history"]]
566+
result = _completion(_execute(
567+
orchestrator, history, [helpers.new_task_completed_event(data["join_id"], '"joined"')]))
568+
assert result.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, result.failureDetails
569+
assert json.loads(result.result.value) == "joined"
570+
"""
571+
for seed in ("0", "999"):
572+
process = subprocess.run(
573+
[sys.executable, "-c", code], input=payload, capture_output=True, text=True,
574+
env={**os.environ, "PYTHONHASHSEED": seed}, timeout=30)
575+
assert process.returncode == 0, process.stdout + process.stderr
576+
577+
466578
def test_cancelling_last_loop_timer_cannot_hide_foreign_wait() -> None:
467579
async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
468580
handle = asyncio.get_running_loop().call_later(1, lambda: None)

0 commit comments

Comments
 (0)