diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index 6c74c622d9..6ae68ddf2c 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -1019,6 +1019,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: observer=enclosure_permit_observer, kernel=deps, name_to_id=enclosure_permit_ids, + enclosure_projection_registry=enclosure_only_registry, + startup_timeout_seconds=( + settings.enclosure_permit_monitor_startup_timeout_seconds + ), ), run_supervisor_lifespan( deps, diff --git a/apps/api/src/cora/enclosure/_monitor.py b/apps/api/src/cora/enclosure/_monitor.py index 15a56afa3a..185ce41713 100644 --- a/apps/api/src/cora/enclosure/_monitor.py +++ b/apps/api/src/cora/enclosure/_monitor.py @@ -29,6 +29,34 @@ `reconnect_delay_seconds` then re-subscribes. A single bad observation is logged and skipped (the subscription survives). Cancellation (lifespan shutdown) propagates out of the loop. + +## Startup race + +Between process start and the monitor's first settled read, `permit_status` +still holds whatever a prior process instance last wrote, so a `start_run` / +`start_procedure` preflight landing in that window can pass on a stale +`Permitted` the current process has not confirmed. `enclosure_permit_monitor_lifespan` +narrows most of that window, in two steps, before yielding to the rest of +boot: + +1. It waits, up to a bounded timeout, for every configured enclosure to + produce one settled observation (a real reading or the observer's own + `Unknown` on a dead PV). A PV that never answers does not block boot past + the timeout; the loop keeps retrying in the background and that one + enclosure's stale reading stands, gated the same as any other stale + reading. +2. It then drains the enclosure projection once, because step 1 only proves + an `EnclosurePermitObserved` event was appended (or an append attempt was + swallowed on failure); `permit_status` is a denormalized read-model + column the `ProjectionWorker` catches up to on its own poll cadence, and + the preflight reads that column, never the event stream directly. Like + step 1, this is bounded: a slow catch-up is logged and boot proceeds + rather than blocking or failing. + +Neither step closes the window to zero: CORA can only observe that a +settlement happened, never that no more are pending, and a PV that never +answers leaves that one enclosure's preflight reading exactly as stale as +before this fix. """ from __future__ import annotations @@ -53,6 +81,7 @@ from cora.enclosure.ports.enclosure_observer import EnclosureObserverScope from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.logging import get_logger +from cora.infrastructure.projection import ProjectionDrainTimeoutError, drain_projections from cora.infrastructure.routing import SYSTEM_PRINCIPAL_ID from cora.shared.identity import MonitorSourceId @@ -64,10 +93,18 @@ EnclosureObserver, ) from cora.infrastructure.kernel import Kernel + from cora.infrastructure.projection import ProjectionRegistry _STREAM_TYPE = "Enclosure" _COMMAND_NAME = "ObserveEnclosureStatus" _RECONNECT_DELAY_SECONDS = 5.0 +# A few seconds above EpicsCaControlPort's own _DEFAULT_TIMEOUT_S (5.0s): +# a dead PV's per-code settlement (pump reaching Unknown via a bounded +# connect attempt) needs room to finish before this outer bound gives up, +# or every dead-PV boot falls through to the blunter warn-and-proceed path +# instead of the cleaner "every code settled, including Unknown" one. +_STARTUP_TIMEOUT_SECONDS = 8.0 +_PROJECTION_DRAIN_DEADLINE_SECONDS = 5.0 # Stable monitor-source id for the enclosure permit monitor; stamped onto # EnclosurePermitObserved.triggered_by as the in-process adapter attribution. @@ -152,11 +189,22 @@ async def run_enclosure_permit_monitor( kernel: Kernel, name_to_id: Mapping[str, UUID], reconnect_delay_seconds: float = _RECONNECT_DELAY_SECONDS, + startup_ready: asyncio.Event | None = None, ) -> None: - """Drain the observer, recording each observation; re-subscribe on stream end.""" + """Drain the observer, recording each observation; re-subscribe on stream end. + + `startup_ready`, when given, is set as soon as every configured enclosure + code has produced one observation (a real reading or the observer's own + `Unknown` on a dead PV), or, failing that, once the current pass ends (the + observer's stream terminated or raised with codes still unheard from). The + second case fires on every reconnect attempt, not only the first, but + `Event.set` on an already-set event is a no-op, so the caller only ever + sees the earliest settlement. + """ if not name_to_id: return scope = EnclosureObserverScope(enclosure_codes=frozenset(name_to_id)) + pending_codes: set[str] = set(name_to_id) if startup_ready is not None else set() while True: try: async for observation in observer.observe(scope): @@ -169,10 +217,15 @@ async def run_enclosure_permit_monitor( "enclosure_monitor.record_failed", enclosure_code=observation.enclosure_code, ) + pending_codes.discard(observation.enclosure_code) + if startup_ready is not None and not pending_codes: + startup_ready.set() except asyncio.CancelledError: raise except Exception: _log.exception("enclosure_monitor.iteration_failed") + if startup_ready is not None: + startup_ready.set() await asyncio.sleep(reconnect_delay_seconds) @@ -182,20 +235,68 @@ async def enclosure_permit_monitor_lifespan( observer: EnclosureObserver, kernel: Kernel, name_to_id: Mapping[str, UUID], + enclosure_projection_registry: ProjectionRegistry | None = None, + startup_timeout_seconds: float = _STARTUP_TIMEOUT_SECONDS, ) -> AsyncGenerator[None]: """Run the permit monitor as a background task for the app's lifetime. No-op when `name_to_id` is empty (no enclosures configured): yields immediately without starting a task. Mirrors `projection_worker_lifespan`. + + Otherwise, before yielding: waits up to `startup_timeout_seconds` for + every configured enclosure's first settled observation, then, when + `enclosure_projection_registry` is given and `kernel.pool` is a real + pool, drains it once. Both steps exist because neither alone gets a + preflight landing right after boot to a status this process has + confirmed (see module docstring, "Startup race"): the wait only proves + an event was appended (or an append attempt failed and was logged); the + drain is what catches `permit_status` itself up to that event. A PV + that never settles does not delay boot past `startup_timeout_seconds`: + the background task keeps retrying and that enclosure's existing + reading stands, gated the same as any other stale reading. + `enclosure_projection_registry` is optional so tests that stub `kernel` + can omit it and skip the drain entirely. """ if not name_to_id: yield return + startup_ready = asyncio.Event() task = asyncio.create_task( - run_enclosure_permit_monitor(observer=observer, kernel=kernel, name_to_id=name_to_id), + run_enclosure_permit_monitor( + observer=observer, + kernel=kernel, + name_to_id=name_to_id, + startup_ready=startup_ready, + ), name="enclosure-permit-monitor", ) try: + try: + await asyncio.wait_for(startup_ready.wait(), timeout=startup_timeout_seconds) + except TimeoutError: + _log.warning( + "enclosure_monitor.startup_timeout", + startup_timeout_seconds=startup_timeout_seconds, + enclosure_codes=sorted(name_to_id), + ) + if enclosure_projection_registry is not None and kernel.pool is not None: + try: + await drain_projections( + kernel.pool, + enclosure_projection_registry, + deadline_seconds=_PROJECTION_DRAIN_DEADLINE_SECONDS, + ) + except ProjectionDrainTimeoutError: + # Same posture as the startup_timeout_seconds branch above: + # a slow catch-up degrades permit_status freshness, it must + # never abort boot. Letting this propagate would turn a + # read-honesty fix into an availability regression, which is + # a worse trade than the staleness this fix exists to narrow. + _log.warning( + "enclosure_monitor.projection_drain_timeout", + deadline_seconds=_PROJECTION_DRAIN_DEADLINE_SECONDS, + enclosure_codes=sorted(name_to_id), + ) yield finally: task.cancel() diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index d6aa3c7f47..f652b72c8c 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -626,6 +626,23 @@ class Settings(BaseSettings): # `cora.enclosure.adapters.control_port_enclosure_observer`. enclosure_permit_pvs: dict[str, str] = {} + # Bounds how long boot waits for the permit monitor's first settled read + # per configured enclosure before serving requests (the startup-race + # window documented in `cora.enclosure._monitor`, "Startup race"). A PV + # that has not settled by the deadline does not delay boot further: the + # monitor keeps retrying in the background and that enclosure's existing + # `permit_status` stands. Irrelevant when `enclosure_permit_pvs` is empty. + # + # `main.py` always passes this value explicitly, so it is the LIVE + # default; `_monitor._STARTUP_TIMEOUT_SECONDS` only covers a caller that + # skips Settings entirely. Keep both above EpicsCaControlPort's own + # `_DEFAULT_TIMEOUT_S` (5.0s): a dead PV's per-code settlement needs + # room to finish before this outer bound gives up, or a dead-PV boot + # always falls through to the blunter warn-and-proceed path. These two + # defaults drifted out of sync once already; if you change one, change + # both. + enclosure_permit_monitor_startup_timeout_seconds: float = 8.0 + # Beam-availability pre-flight (BEAM-1, beam-availability slice). # Role -> read-only PV for the run / procedure start gate. `fes` and # `sbs` are the front-end and station-shutter BeamBlockingM PVs diff --git a/apps/api/tests/integration/test_enclosure_permit_monitor.py b/apps/api/tests/integration/test_enclosure_permit_monitor.py index b47fc8358a..24ed511832 100644 --- a/apps/api/tests/integration/test_enclosure_permit_monitor.py +++ b/apps/api/tests/integration/test_enclosure_permit_monitor.py @@ -4,7 +4,10 @@ store (maps an observation to an EnclosurePermitObserved transition, status-change-only idempotency, unknown-code no-op). The retry loop + lifespan are covered for the empty-config no-op path and for a -fake-observer drive that records one observation end to end. +fake-observer drive that records one observation end to end. The +startup-race fix (module docstring "Startup race") is covered +separately: `startup_ready` semantics on the loop, and the lifespan's +bounded wait plus its still-yields-on-timeout fallback. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -19,8 +22,9 @@ import asyncpg import pytest +import structlog.testing -from cora.enclosure import seed_enclosures +from cora.enclosure import register_enclosure_projections, seed_enclosures from cora.enclosure._monitor import ( enclosure_permit_monitor_lifespan, record_observation, @@ -34,6 +38,7 @@ from cora.infrastructure.config import Settings from cora.infrastructure.kernel import Kernel from cora.infrastructure.ports.event_store import StoredEvent +from cora.infrastructure.projection import ProjectionDrainTimeoutError, ProjectionRegistry from tests.integration._helpers import build_postgres_deps _T = datetime(2026, 6, 17, 12, 0, 0, tzinfo=UTC) @@ -224,6 +229,196 @@ async def test_lifespan_nonempty_starts_and_cancels_monitor_task() -> None: # context exit cancels the task cleanly: no hang, no error surfaced +@pytest.mark.unit +async def test_run_monitor_startup_ready_waits_for_every_configured_code() -> None: + code_a, code_b = "hutch-a", "hutch-b" + release = asyncio.Event() + observer = _StaggeredObserver(_obs(code_a, "Garbage"), _obs(code_b, "Garbage"), release) + ready = asyncio.Event() + task = asyncio.create_task( + run_enclosure_permit_monitor( + observer=observer, + kernel=cast("Kernel", None), + name_to_id={code_a: uuid4(), code_b: uuid4()}, + reconnect_delay_seconds=3600.0, + startup_ready=ready, + ) + ) + try: + await asyncio.wait_for(observer.first_yielded.wait(), timeout=2.0) + assert not ready.is_set() # code_b has not answered yet + release.set() + await asyncio.wait_for(ready.wait(), timeout=2.0) + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + +@pytest.mark.unit +async def test_run_monitor_startup_ready_set_when_pass_yields_nothing() -> None: + # An empty scope (observer.observe ends without yielding) still settles: + # startup_ready must not wait forever for a monitor that has nothing to say. + ready = asyncio.Event() + task = asyncio.create_task( + run_enclosure_permit_monitor( + observer=_FakeObserver([]), + kernel=cast("Kernel", None), + name_to_id={"hutch-empty": uuid4()}, + reconnect_delay_seconds=3600.0, + startup_ready=ready, + ) + ) + try: + await asyncio.wait_for(ready.wait(), timeout=2.0) + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + +@pytest.mark.unit +async def test_run_monitor_startup_ready_set_on_iteration_failure() -> None: + # The observer raising outright (e.g. an immediate connect failure the + # observer itself could not turn into an Unknown) must still settle + # startup_ready, not hang it until the timeout on every reconnect pass. + ready = asyncio.Event() + observer = _BoomObserver() + task = asyncio.create_task( + run_enclosure_permit_monitor( + observer=observer, + kernel=cast("Kernel", None), + name_to_id={"hutch-iter-fail": uuid4()}, + reconnect_delay_seconds=3600.0, + startup_ready=ready, + ) + ) + try: + await asyncio.wait_for(ready.wait(), timeout=2.0) + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + +@pytest.mark.unit +async def test_lifespan_yields_once_ready_without_waiting_full_timeout() -> None: + # startup_timeout_seconds is set absurdly high; if the wait blocked for + # the full duration this test would hang past pytest-timeout instead of + # completing, so reaching the assertion proves readiness unblocked it. + name = "hutch-lifespan-ready" + async with enclosure_permit_monitor_lifespan( + observer=_FakeObserver([_obs(name, "Garbage")]), + kernel=cast("Kernel", None), + name_to_id={name: uuid4()}, + startup_timeout_seconds=3600.0, + ): + pass + + +@pytest.mark.unit +async def test_lifespan_gives_up_after_timeout_and_still_yields() -> None: + entered = False + with structlog.testing.capture_logs() as logs: + async with enclosure_permit_monitor_lifespan( + observer=_HangingObserver(), + kernel=cast("Kernel", None), + name_to_id={"hutch-hang": uuid4()}, + startup_timeout_seconds=0.05, + ): + entered = True + assert entered # boot proceeds even though the monitor never settled + events = [e.get("event") for e in logs] + assert "enclosure_monitor.startup_timeout" in events + + +@pytest.mark.unit +async def test_lifespan_gives_up_after_drain_timeout_and_still_yields( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # A slow projection catch-up must degrade the same way as a slow PV: log + # and proceed, never abort boot. `drain_projections` is monkeypatched + # (rather than driven against a real slow drain) so this stays a fast + # unit test pinning the CATCH, not drain_projections' own timeout logic. + async def _raising_drain(*_args: object, **_kwargs: object) -> None: + raise ProjectionDrainTimeoutError( + deadline_seconds=5.0, + subscribed_heads={"proj_enclosure_summary": 1}, + bookmarks={"proj_enclosure_summary": 0}, + ) + + monkeypatch.setattr("cora.enclosure._monitor.drain_projections", _raising_drain) + + name = "hutch-drain-timeout" + entered = False + with structlog.testing.capture_logs() as logs: + async with enclosure_permit_monitor_lifespan( + observer=_FakeObserver([_obs(name, "Garbage")]), + kernel=cast("Kernel", _PoolOnlyKernel()), + name_to_id={name: uuid4()}, + enclosure_projection_registry=ProjectionRegistry(), + startup_timeout_seconds=5.0, + ): + entered = True + assert entered # boot proceeds even though the projection drain timed out + events = [e.get("event") for e in logs] + assert "enclosure_monitor.projection_drain_timeout" in events + + +@pytest.mark.unit +async def test_lifespan_cancelled_during_startup_wait_still_cleans_up_task() -> None: + # Regression: entering the context manager used to create the background + # task, then await the startup wait with no enclosing try/finally, so a + # cancellation delivered during that wait (before __aenter__ ever + # returns) skipped cleanup entirely and leaked the task. + cm = enclosure_permit_monitor_lifespan( + observer=_HangingObserver(), + kernel=cast("Kernel", None), + name_to_id={"hutch-cancel-startup": uuid4()}, + startup_timeout_seconds=3600.0, # only our cancellation should end this + ) + entering = asyncio.ensure_future(cm.__aenter__()) + await asyncio.sleep(0) # let it reach the startup wait + entering.cancel() + with pytest.raises(asyncio.CancelledError): + await entering + await asyncio.sleep(0) # let the finally's task.cancel()/await settle + leaked = [ + t + for t in asyncio.all_tasks() + if t.get_name() == "enclosure-permit-monitor" and not t.done() + ] + assert not leaked + + +@pytest.mark.integration +async def test_lifespan_drains_projection_before_yielding(db_pool: asyncpg.Pool) -> None: + # startup_ready proves the event was appended, not that permit_status + # itself is caught up; this pins the second half of the fix, that the + # projection is drained before the context manager body ever runs. + name = f"hutch-drain-{uuid4().hex[:8]}" + deps = _deps_with(db_pool, permit_pvs={name: "pv"}) + name_to_id = await seed_enclosures(deps) + enclosure_id = name_to_id[name] + + registry = ProjectionRegistry() + register_enclosure_projections(registry, deps) + + async with enclosure_permit_monitor_lifespan( + observer=_FakeObserver([_obs(name, "Permitted")]), + kernel=deps, + name_to_id=name_to_id, + enclosure_projection_registry=registry, + startup_timeout_seconds=5.0, + ): + row = await db_pool.fetchrow( + "SELECT permit_status FROM proj_enclosure_summary WHERE enclosure_id = $1", + enclosure_id, + ) + assert row is not None + assert row["permit_status"] == "Permitted" + + class _FakeObserver: """Yields a fixed observation sequence once, then ends the stream.""" @@ -253,6 +448,64 @@ async def _drain(self) -> AsyncGenerator[EnclosureObservation]: yield # pragma: no cover - unreachable, marks this body an async generator +class _StaggeredObserver: + """Yields one observation, then gates the second behind `release`. + + `first_yielded` fires once the consumer has resumed this generator to + ask for the second item, which only happens after the consumer has + finished processing the first (including any `startup_ready` bookkeeping). + """ + + def __init__( + self, + first: EnclosureObservation, + second: EnclosureObservation, + release: asyncio.Event, + ) -> None: + self._first = first + self._second = second + self._release = release + self.first_yielded = asyncio.Event() + + def observe(self, scope: EnclosureObserverScope) -> AsyncGenerator[EnclosureObservation]: + return self._drain() + + async def _drain(self) -> AsyncGenerator[EnclosureObservation]: + yield self._first + self.first_yielded.set() + await self._release.wait() + yield self._second + + +class _HangingObserver: + """Never yields and never ends. + + Real adapters (`EpicsCaControlPort`) always bound a connect attempt and + resolve to a real `Unknown` observation, so this is not a faithful model + of an unreachable PV; it exercises the lifespan's `TimeoutError` fallback + in the abstract, for whatever future observer or bug might not settle. + """ + + def observe(self, scope: EnclosureObserverScope) -> AsyncGenerator[EnclosureObservation]: + return self._drain() + + async def _drain(self) -> AsyncGenerator[EnclosureObservation]: + await asyncio.Event().wait() # never released + yield _obs("unreachable", "Garbage") # pragma: no cover - dead code, never released + + +class _PoolOnlyKernel: + """Kernel double exposing a non-None `pool` sentinel, nothing else. + + Only used with a monkeypatched `drain_projections`, which never touches + `pool` for real; this just needs to satisfy the lifespan's + `kernel.pool is not None` gate before reaching the drain call. + """ + + def __init__(self) -> None: + self.pool = object() + + class _RaisingLoadKernel: """Kernel double whose event-store load raises, to drive the per-record branch."""