Three deadline tests fire the real deadline inside the held operation, not on wall time (BACKLOG #3871)

tests/_deadline_capture.py

@@ -0,0 +1,143 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# Copyright (C) 2026 MessageFoundry Foundation, LLC and contributors
+"""Fire the request deadline from inside the operation it is meant to cut off (vault BACKLOG #3871).
+
+A test that sets ``request_timeout_seconds`` to a fraction of a second races wall time against the
+request: the deadline counts from the start of the request, so on a loaded runner the auth
+dependencies, the body read or the file save alone could outlast it. The 503 then came before the
+route had started the operation, nothing was left running to outlive the request, and the test
+failed on a state the code under test never reached.
+
+:func:`capture_deadlines` replaces ``asyncio`` inside the deadline middleware's module only, so the
+:class:`DeadlineCapture` it returns keeps each deadline the middleware opens, and gives the app a
+deadline long enough never to fire on its own. The operation the test holds awaits
+:meth:`DeadlineCapture.cut_off`. That expires the middleware's own ``asyncio.Timeout`` on the loop's
+next pass, wherever the request is, so the 503 lands mid-operation and nowhere else, then holds the
+operation until :meth:`DeadlineCapture.held` lets it go.
+
+Engine PR 2438, open and not on ``main`` when this was written, carries a simpler copy of the
+stand-in in the web console's ``test_ui_dr_outlive.py``: it fires the latest deadline, with none of
+the checks below.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import types
+from collections.abc import AsyncIterator, Awaitable
+from typing import Any
+
+import httpx
+import pytest
+
+from messagefoundry.api import request_timeout
+from messagefoundry.api.outlive import OutlivingOperations
+
+__all__ = ["GENEROUS_DEADLINE_SECONDS", "DeadlineCapture", "capture_deadlines", "expect_cut_off"]
+
+#: Long enough that a deadline never fires on its own during a test: :meth:`DeadlineCapture.fire`
+#: fires it. Short enough that a test whose operation never fires it still fails on its own
+#: assertion inside the per-test timeout, after this, the drain below and its cancel grace.
+GENEROUS_DEADLINE_SECONDS = 15.0
+
+#: How long a test waits for the held operation to finish once it is let go.
+_DRAIN_SECONDS = 10.0
+
+
+class DeadlineCapture:
+    """Keeps each deadline the middleware opens, so a test can make the real one fire when it
+    chooses. :func:`capture_deadlines` builds and installs it."""
+
+    def __init__(self, outliving: OutlivingOperations) -> None:
+        self.deadlines: list[asyncio.Timeout] = []
+        self.fired = False
+        #: Why a fire went wrong. Recorded rather than raised: :meth:`fire` runs on the operation's
+        #: task, where the code under test may catch an error and turn it into its own outcome.
+        #: :func:`expect_cut_off` reports it on the test's task.
+        self.problem: str | None = None
+        self._outliving = outliving
+        self._released = asyncio.Event()
+
+    def timeout(self, delay: float | None) -> asyncio.Timeout:
+        deadline = asyncio.timeout(delay)
+        self.deadlines.append(deadline)
+        return deadline
+
+    def fire(self) -> None:
+        """Expire the one request deadline opened since this capture was installed, now.
+
+        Exactly one, still armed: a test installs the capture just before the request it cuts off,
+        so a second deadline means a request the test did not expect, and a disarmed one means the
+        response has already started, where firing would cut the body rather than the operation.
+        Anything else fires nothing: it is recorded, and the hold is let go so the request ends
+        and the test reports it, rather than every await waiting on the next."""
+        if len(self.deadlines) != 1:
+            self._refuse(f"expected one request deadline, saw {len(self.deadlines)}")
+            return
+        deadline = self.deadlines[0]
+        if deadline.when() is None:
+            self._refuse("the request had already started its response")
+            return
+        try:
+            deadline.reschedule(asyncio.get_running_loop().time())
+        except RuntimeError as exc:  # it had already expired, or its request had ended
+            self._refuse(f"the deadline could not be fired: {exc}")
+            return
+        self.fired = True
+
+    def _refuse(self, problem: str) -> None:
+        self.problem = problem
+        self._released.set()
+
+    async def cut_off(self) -> None:
+        """Awaited from inside the held operation: fire the deadline, then hold the operation until
+        :meth:`held` lets it go, so it cannot end before the 503 is in."""
+        self.fire()
+        await self._released.wait()
+
+    @contextlib.asynccontextmanager
+    async def held(self) -> AsyncIterator[None]:
+        """While the block runs, the operation stays held in :meth:`cut_off`. On the way out it is
+        let go and waited for, so it never runs on into the fixture's engine stop, and on a clean
+        exit it must have finished by itself rather than been cancelled by the drain. Whether it
+        succeeded is for the test's own assertions on its outcome rows."""
+        try:
+            yield
+        finally:
+            self._released.set()
+            cancelled = await self._outliving.drain(timeout=_DRAIN_SECONDS)
+        assert cancelled == [], f"never finished once let go: {cancelled}"
+
+
+def capture_deadlines(monkeypatch: pytest.MonkeyPatch, app: Any) -> DeadlineCapture:
+    """Install a :class:`DeadlineCapture` for ``app`` and give it a deadline that never fires on
+    its own; both are undone with ``monkeypatch``.
+
+    Call it after the sign-in and any proof, just before the request the test cuts off. The
+    middleware's module sees only ``timeout``, which is all it uses, so a new use fails loudly."""
+    capture = DeadlineCapture(app.state.outliving_operations)
+    monkeypatch.setattr(request_timeout, "asyncio", types.SimpleNamespace(timeout=capture.timeout))
+    monkeypatch.setattr(
+        app.state, request_timeout.TIMEOUT_STATE_ATTR, GENEROUS_DEADLINE_SECONDS, raising=False
+    )
+    return capture
+
+
+async def expect_cut_off(capture: DeadlineCapture, request: Awaitable[httpx.Response]) -> None:
+    """Await ``request`` and check the answer is the deadline middleware's own 503, fired by
+    ``capture`` from inside the operation: not a route's 503, and not the deadline running out.
+
+    The generous deadline bounds the wait: a deadline the operation never fires answers 503 there,
+    and the ``fired`` check fails. A fire that went wrong is reported first, even when the request
+    raised because of it."""
+    try:
+        r = await request
+    except Exception as exc:
+        if capture.problem is not None:
+            raise AssertionError(capture.problem) from exc
+        raise
+    assert capture.problem is None, capture.problem
+    assert r.status_code == 503, r.text
+    assert r.json() == {"detail": "the server timed out building a response"}
+    assert capture.fired, "the deadline was not fired from inside the operation"

tests/conftest.py

@@ -41,6 +41,8 @@
 # A helper module holding the content-free asserts several connector tests share. pytest rewrites
 # asserts only in test modules unless told, and an unrewritten assert vanishes under `python -O`.
 pytest.register_assert_rewrite("tests._content_free")
+# The same for the request-deadline helper the cut-off tests share (vault BACKLOG #3871).
+pytest.register_assert_rewrite("tests._deadline_capture")
 
 # ---------------------------------------------------------------------------------------------------
 # Per-PROCESS test slot.

tests/test_dr_outlive_deadline.py

@@ -11,9 +11,9 @@
 it (``api/outlive.py``), and the coordinator's own cancellation arms restore state and record an
 interrupted outcome for what can still cancel one, such as a shutdown.
 
-The HTTP tests patch the deadline down to a fraction of a second rather than wait 120 s; the
-mechanism under test is the cancellation, not the number. No test here measures how long a real
-activation takes on a restored store.
+The HTTP tests fire the middleware's own deadline from inside the operation rather than wait 120 s
+(``tests/_deadline_capture.py``, vault BACKLOG #3871); the mechanism under test is the cancellation,
+not the number. No test here measures how long a real activation takes on a restored store.
 """
 
 from __future__ import annotations
@@ -54,6 +54,7 @@
 from messagefoundry.store import MessageStore, Store
 from messagefoundry.store.crypto import generate_key, make_cipher
 from tests._admin_account import create_local_user_chosen
+from tests._deadline_capture import capture_deadlines, expect_cut_off
 
 PW = "Sup3rSecret!!DR-outlive"
 #: Synthetic, never real PHI (CLAUDE.md section 9).
@@ -168,8 +169,8 @@ async def seeded(tmp_path: Path) -> AsyncIterator[tuple[Engine, str]]:
 async def _admin_client(
     engine: Engine, deadline: float, *, provision: bool = True
 ) -> tuple[httpx.AsyncClient, Any]:
-    """A signed-in administrator client. The deadline is set AFTER the login, whose password hash
-    alone takes longer than the fractions of a second these tests use."""
+    """A signed-in administrator client, with ``deadline`` set AFTER the login. A test here that
+    cuts a request off then installs ``tests/_deadline_capture.py``, which replaces it."""
     service = AuthService(
         engine.store, AuthSettings(require_mfa=False, admin_write_min_interval_seconds=0)
     )
@@ -255,25 +256,29 @@ async def test_a_release_cut_off_by_the_deadline_still_hands_back_and_records_th
     depth = await _activate_directly(engine)
     rr = engine.registry_runner
     assert rr is not None
+    coord = _coordinator(engine)
 
-    # The drain gives up after 1.0 s; the request deadline fires at 0.3 s, well inside it -- the
-    # shape the 120 s pair had, where any non-draining row meant the deadline fired first.
-    monkeypatch.setattr(engine_module, "DR_RELEASE_DRAIN_TIMEOUT_SECONDS", 1.0)
-    client, app = await _admin_client(engine, deadline=0.3)
-    async with client:
-        r = await client.post("/dr/release")
-        assert r.status_code == 503
-        assert "timed out" in r.json()["detail"]
-
-        # The release goes on without its caller and completes the hand-back.
-        coord = _coordinator(engine)
-
-        async def released() -> bool:
-            return not coord.active and not app.state.outliving_operations.inflight
-
-        await _wait(released)
-
-    assert engine.dr_active is False
+    # The request deadline fires inside the drain, before the drain gives up -- the shape the
+    # 120 s pair had, where any non-draining row meant the deadline fired first. The drain is held
+    # until the 503 is in, then runs for real and gives up on the row that cannot drain.
+    monkeypatch.setattr(engine_module, "DR_RELEASE_DRAIN_TIMEOUT_SECONDS", 0.3)
+    real_drain = engine._drain_pipeline
+    client, app = await _admin_client(engine, deadline=30.0)
+    capture = capture_deadlines(monkeypatch, app)
+
+    async def drain_cut_off(**kwargs: Any) -> tuple[int, int]:
+        await capture.cut_off()
+        return await real_drain(**kwargs)
+
+    monkeypatch.setattr(engine, "_drain_pipeline", drain_cut_off)
+    outliving = app.state.outliving_operations
+    async with client, capture.held():
+        await expect_cut_off(capture, client.post("/dr/release"))
+        # Cut off in its drain: intake is already unbound, and the box is still active.
+        assert outliving.inflight == ["DR release"]
+        assert coord.active and not rr.inbound_running("file_in")
+    # Let go, the release went on without its caller and completed the hand-back.
+    assert not coord.active and engine.dr_active is False
     assert not rr.inbound_running("file_in")
     rows = await _rows(engine.store, "dr.release")
     assert len(rows) == 1
@@ -347,27 +352,25 @@ async def draining() -> bool:
 
 
 async def test_an_activation_cut_off_by_the_deadline_completes_and_a_retry_reads_its_outcome(
-    seeded: tuple[Engine, str],
+    seeded: tuple[Engine, str], monkeypatch: pytest.MonkeyPatch
 ) -> None:
     engine, _archive = seeded
     coord = _coordinator(engine)
     real_profile = coord._activate_profile
+    client, app = await _admin_client(engine, deadline=30.0)
+    capture = capture_deadlines(monkeypatch, app)
 
-    async def slow_profile() -> None:
-        await asyncio.sleep(0.6)  # outlasts the 0.2 s deadline below
+    async def profile_cut_off() -> None:
+        await capture.cut_off()  # the deadline fires in step 4
         await real_profile()
 
-    coord._activate_profile = slow_profile
-    client, app = await _admin_client(engine, deadline=0.2)
+    coord._activate_profile = profile_cut_off
+    outliving = app.state.outliving_operations
     async with client:
-        r = await client.post("/dr/activate", json={})
-        assert r.status_code == 503
-        assert "timed out" in r.json()["detail"]
-
-        async def settled() -> bool:
-            return not app.state.outliving_operations.inflight
-
-        await _wait(settled)
+        async with capture.held():
+            await expect_cut_off(capture, client.post("/dr/activate", json={}))
+            assert outliving.inflight == ["DR activation"]
+            assert await _rows(engine.store, "dr.activate") == []
         assert coord.active and engine.dr_active
         rows = await _rows(engine.store, "dr.activate")
         assert len(rows) == 1 and "recorded_late" not in rows[0]

tests/test_reload_cancellation.py

@@ -15,7 +15,7 @@
 
 import asyncio
 import json
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import Any
 
@@ -41,18 +41,11 @@
 from messagefoundry.store import MessageStore
 from tests._admin_account import create_local_user_chosen
 from tests._bound_proof import prove, reproving_transport
+from tests._deadline_capture import capture_deadlines, expect_cut_off
 
 PW = "Sup3rSecret!!reload-outlive"
 
 
-async def _wait(predicate: Callable[[], Awaitable[bool]], timeout: float = 10.0) -> None:
-    deadline = asyncio.get_running_loop().time() + timeout
-    while not await predicate():
-        if asyncio.get_running_loop().time() > deadline:
-            raise AssertionError("condition not met within timeout")
-        await asyncio.sleep(0.02)
-
-
 def _registry(tmp_path: Path, *, inbound: str = "file_in", directory: str = "in") -> Registry:
     inbox, outdir = tmp_path / directory, tmp_path / "out"
     inbox.mkdir(exist_ok=True)
@@ -172,7 +165,7 @@ async def engine(tmp_path: Path) -> AsyncIterator[Engine]:
     await eng.stop()
 
 
-async def _deployer(engine: Engine, deadline: float) -> tuple[httpx.AsyncClient, Any]:
+async def _deployer(engine: Engine) -> tuple[httpx.AsyncClient, Any]:
     # The shipped default: the action-bound proof is on, and reproving_transport re-proves for it
     # as a JSON client does (vault BACKLOG #3241). tests/test_bound_step_up_injection.py
     # pins the gate itself.
@@ -206,12 +199,10 @@ async def _deployer(engine: Engine, deadline: float) -> tuple[httpx.AsyncClient,
             "/auth/login", json={"username": "deployer", "password": PW, "provider": "local"}
         )
     assert r.status_code == 200, r.text
-    # The reload takes a proof bound to it (vault BACKLOG #2625). Mint it here, before the
-    # deadline: a re-proof hashes a password, which alone outlasts the deadlines these tests set.
+    # The reload takes a proof bound to it (vault BACKLOG #2625). Mint it here, up front: a
+    # re-proof is a second request, and the test cuts off the reload request alone.
     async with httpx.AsyncClient(transport=transport, base_url="http://t") as proving:
         proved = await prove(proving, r.json()["token"], PW, "config_reload")
-    # After the login: its password hash alone outlasts the deadline these tests set.
-    app.state.request_timeout_seconds = deadline
     client = httpx.AsyncClient(
         transport=transport,
         base_url="http://t",
@@ -228,22 +219,21 @@ async def test_a_reload_cut_off_by_the_deadline_finishes_and_writes_its_row(
     engine: Engine, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     registry = _registry(tmp_path)
+    client, app = await _deployer(engine)
+    # The middleware's own deadline, fired from inside the reload (tests/_deadline_capture.py).
+    capture = capture_deadlines(monkeypatch, app)
 
-    async def slow_reload(*_args: Any, **_kwargs: Any) -> ReloadOutcome:
-        await asyncio.sleep(0.6)  # outlasts the 0.2 s deadline below
+    async def reload_cut_off(*_args: Any, **_kwargs: Any) -> ReloadOutcome:
+        await capture.cut_off()
         return ReloadOutcome(registry=registry, applied=True, directory=tmp_path)
 
-    monkeypatch.setattr(engine, "reload_detail", slow_reload)
-    client, app = await _deployer(engine, deadline=0.2)
-    async with client:
-        r = await client.post("/config/reload", json={})
-    assert r.status_code == 503
-    assert "timed out" in r.json()["detail"]
-
-    async def settled() -> bool:
-        return not app.state.outliving_operations.inflight
-
-    await _wait(settled)
+    monkeypatch.setattr(engine, "reload_detail", reload_cut_off)
+    outliving = app.state.outliving_operations
+    async with client, capture.held():
+        await expect_cut_off(capture, client.post("/config/reload", json={}))
+        # Cut off while the reload was still running, and not cancelled with the request.
+        assert outliving.inflight == ["config reload"]
+        assert "config_reload" not in await _actions(engine.store)
     actions = await _actions(engine.store)
     assert "config_reload" in actions  # the reload's own row, written after its caller left
     assert "config_reload_interrupted" not in actions