From 5da67cafe9581df91ce3d64735dbedb3cf3dec70 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 07:33:01 +0000 Subject: [PATCH 1/4] Expose the active test and thread stacks when Python CI stalls --- .github/workflows/ci.yml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6381a07..ec2e734 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -135,7 +135,10 @@ jobs: with: python-version: ${{ matrix.python-version }} - run: pip install -e '.[dev]' - - run: pytest tests/ -m "not integration" -q + - name: Run unit tests with stalled-thread diagnostics + run: >- + pytest tests/ -m "not integration" -vv + -o faulthandler_timeout=120 --durations=20 package: runs-on: ubuntu-latest From 43976985645b60c394f9b76394c32820bbb77d23 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 08:05:02 +0000 Subject: [PATCH 2/4] Retain bounded stalled-test diagnostics in GitHub annotations --- .github/workflows/ci.yml | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ec2e734..e6ba402 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -136,9 +136,26 @@ jobs: python-version: ${{ matrix.python-version }} - run: pip install -e '.[dev]' - name: Run unit tests with stalled-thread diagnostics - run: >- - pytest tests/ -m "not integration" -vv - -o faulthandler_timeout=120 --durations=20 + id: unit_tests + shell: bash + run: | + set -o pipefail + # Leave room within the 15-minute job for bounded failure diagnostics. + timeout --signal=INT --kill-after=10s 13m \ + pytest tests/ -m "not integration" -vv \ + -o faulthandler_timeout=120 --durations=20 \ + 2>&1 | tee "$RUNNER_TEMP/sdk-unit-tests.log" + - name: Attach bounded failure diagnostics to the check + if: ${{ always() && steps.unit_tests.outcome == 'failure' }} + shell: bash + run: | + if [[ -f "$RUNNER_TEMP/sdk-unit-tests.log" ]]; then + phase_unit_tail="$(tail -c 12000 "$RUNNER_TEMP/sdk-unit-tests.log")" + phase_unit_tail="${phase_unit_tail//'%'/'%25'}" + phase_unit_tail="${phase_unit_tail//$'\r'/'%0D'}" + phase_unit_tail="${phase_unit_tail//$'\n'/'%0A'}" + printf '::error title=Unit-test failure diagnostics::%s\n' "$phase_unit_tail" + fi package: runs-on: ubuntu-latest From 73db146de03da6bf5afaeb0f9dae50118dc2f52f Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 08:14:30 +0000 Subject: [PATCH 3/4] Reproduce lease maintenance surviving a swallowed cancellation --- tests/test_activity_lease_worker.py | 59 ++++++++++++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) diff --git a/tests/test_activity_lease_worker.py b/tests/test_activity_lease_worker.py index b021bd9..42debe7 100644 --- a/tests/test_activity_lease_worker.py +++ b/tests/test_activity_lease_worker.py @@ -11,7 +11,7 @@ import pytest from durable_workflow import Client, Worker, activity, serializer -from durable_workflow._activity_lease import current_activity_authority +from durable_workflow._activity_lease import ActivityLeaseAuthority, current_activity_authority @pytest.mark.parametrize("capability", [True, False, None, "true"]) @@ -148,6 +148,63 @@ async def handler() -> str: assert not any(path.endswith("/heartbeat") for path in server.paths) +async def test_settled_callback_stops_maintenance_when_inflight_renewal_swallows_cancellation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + server = LeaseServer(lease=1) + authority = ActivityLeaseAuthority.delivery(object(), "17", server.task, owner="worker-1", + task_queue="queue-1", server_time=timestamp(server.started), request_started=server.monotonic_started) + renewal_started = asyncio.Event() + calls = 0 + swallowed = False + + async def renewal(_self: ActivityLeaseAuthority) -> bool: + nonlocal calls, swallowed + calls += 1 + renewal_started.set() + try: + await asyncio.get_running_loop().create_future() + except asyncio.CancelledError: + if not swallowed: + # Model an I/O wrapper completing concurrently with cancel(). + swallowed = True + return True + raise + raise AssertionError("The controlled renewal should remain pending until cancellation") + + async def operation() -> str: + await renewal_started.wait() + return "completed once" + + monkeypatch.setattr(ActivityLeaseAuthority, "renew", renewal) + running = asyncio.create_task(authority.run(operation)) + try: + done, _ = await asyncio.wait({running}, timeout=0.8) + assert running in done, "Settled callbacks must not await an indefinitely renewing helper" + assert await running == "completed once" + assert swallowed is True + assert calls == 1, "A settled operation must not start a subsequent renewal" + finally: + running.cancel() + await asyncio.gather(running, return_exceptions=True) + + +async def test_loading_supervision_can_settle_before_the_same_claim_runs_its_callback() -> None: + server = LeaseServer(lease=1) + authority = ActivityLeaseAuthority.delivery(object(), "17", server.task, owner="worker-1", + task_queue="queue-1", server_time=timestamp(server.started), request_started=server.monotonic_started) + + async def loading() -> str: + await asyncio.sleep(0) + return "loaded" + + assert await authority.run(loading) == "loaded" + original_deadline = authority.lease_until + assert await authority.run(loading) == "loaded" + assert authority.lease_until == original_deadline + assert authority.remaining() > 0 + + @pytest.mark.parametrize("loss", ["denied", "slow-renewal", "business-deadline"]) async def test_lost_claim_cannot_publish_a_late_result_or_failure(loss: str) -> None: server = LeaseServer(deadline=0.13 if loss == "business-deadline" else None) From d2939d4ee8a931e7183017e8b78262062d137bc0 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 08:25:04 +0000 Subject: [PATCH 4/4] Stop settled activity lease maintenance independently of task cancellation --- src/durable_workflow/_activity_lease.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/durable_workflow/_activity_lease.py b/src/durable_workflow/_activity_lease.py index 6b289a7..cd74b9b 100644 --- a/src/durable_workflow/_activity_lease.py +++ b/src/durable_workflow/_activity_lease.py @@ -278,19 +278,22 @@ async def renew(self) -> bool: self.lost_reason = "activity_renewal_unavailable" raise ActivityLeaseAuthorityLost(self.task_id, self.lost_reason) from error - async def _maintain(self) -> None: + async def _maintain(self, stopped: asyncio.Event) -> None: confirmed = True - while True: + while not stopped.is_set(): delay = min(30.0 if confirmed else 1.0, max(0.0, self.lease_until - time.monotonic()) / 3) if self.heartbeat_timeout is not None: delay = min(delay, self.heartbeat_timeout / 3) await asyncio.sleep(min(delay, self.remaining())) + if stopped.is_set(): + return confirmed = await self.renew() async def run(self, operation: Callable[[], Awaitable[_T]]) -> _T: self.remaining() + stopped = asyncio.Event() execution = asyncio.ensure_future(operation()) - maintenance = asyncio.create_task(self._maintain()) + maintenance = asyncio.create_task(self._maintain(stopped)) try: while True: done, _ = await asyncio.wait({execution, maintenance}, timeout=self.remaining(), @@ -306,6 +309,9 @@ async def run(self, operation: Callable[[], Awaitable[_T]]) -> _T: self.lost_reason = self.lost_reason or "activity_execution_abandoned" raise finally: + # An I/O wrapper may complete concurrently with cancel() and consume + # that cancellation. The settled scope must still stop renewing. + stopped.set() # A Python thread may continue after cancellation. Its existing # executor keeps the physical slot until it actually exits, while # this context independently fences every late HTTP publication.