diff --git a/src/rigplane/core/tx_safety.py b/src/rigplane/core/tx_safety.py index 0e2a4b75..1aa33af7 100644 --- a/src/rigplane/core/tx_safety.py +++ b/src/rigplane/core/tx_safety.py @@ -146,6 +146,8 @@ class TxSafetySnapshot: release_last_error: str | None active_attempt: ProviderAttempt | None watchdog_deadline_monotonic: float | None + # Configured *and* driven: a watchdog nothing ticks cannot fire, and this + # field must never advertise one (MOR-1191). watchdog_enabled: bool external_conflict: bool @@ -249,6 +251,7 @@ def __init__( self._active: ProviderAttempt | None = None self._cancel_pending: CancelProviderAttempt | None = None self._watchdog_deadline: float | None = None + self._driven = False @property def snapshot(self) -> TxSafetySnapshot: @@ -290,7 +293,7 @@ def snapshot(self) -> TxSafetySnapshot: release_last_error=self._release.error if self._release else None, active_attempt=self._active, watchdog_deadline_monotonic=self._watchdog_deadline, - watchdog_enabled=self._watchdog_seconds is not None, + watchdog_enabled=self._watchdog_seconds is not None and self._driven, external_conflict=managed_on and not confirmed, ) @@ -470,6 +473,7 @@ def settle_attempt( return self._result(TxOutcome.APPLIED) def tick(self) -> TxTransition: + self._driven = True now = self._clock() effects: tuple[TxEffect, ...] = () if ( diff --git a/src/rigplane/runtime/managed_radio_runtime.py b/src/rigplane/runtime/managed_radio_runtime.py index 80069377..067a4728 100644 --- a/src/rigplane/runtime/managed_radio_runtime.py +++ b/src/rigplane/runtime/managed_radio_runtime.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import logging import time from collections.abc import Awaitable, Callable from dataclasses import dataclass, replace @@ -18,6 +19,8 @@ TxTransition, ) +logger = logging.getLogger(__name__) + TxService = Callable[[TxSafetySupervisor, TxTransition], Awaitable[None]] ProviderRelease = Callable[[], Awaitable[None]] _PttObserver = Callable[[ProviderPttObservation], None] @@ -62,9 +65,13 @@ def __init__( clock: Clock | None = None, id_factory: IdFactory | None = None, shutdown_timeout_seconds: float = 3.0, + tick_interval_seconds: float = 0.25, ) -> None: - if not 0 < shutdown_timeout_seconds < float("inf"): - raise ValueError("shutdown_timeout_seconds must be finite-positive") + if not all( + 0 < seconds < float("inf") + for seconds in (shutdown_timeout_seconds, tick_interval_seconds) + ): + raise ValueError("shutdown timeout and tick interval must be positive") self.target_id, self._clock = target_id, clock or time.monotonic self._tx_safety = TxSafetySupervisor(clock=self._clock, id_factory=id_factory) self._provider_lifecycle, self._provider_generation = provider_lifecycle, 0 @@ -78,6 +85,8 @@ def __init__( self._observation_version = 0 self._shutdown_task: _ShutdownTask | None = None self._shutdown_pending, self._shutdown_timeout = False, shutdown_timeout_seconds + self._tick_interval = tick_interval_seconds + self._tick_task: asyncio.Task[None] | None = None self._effect_host = _ManagedTxEffectHost( self._clock, self._host_write, self._host_read, self._host_retire ) @@ -91,6 +100,31 @@ async def _service_effects(self, transition: TxTransition) -> None: if transition.effects: await self._service(self._tx_safety, transition) + async def _tick_loop(self) -> None: + """The production driver of ``tick``: max key-down plus the timed retry. + + Only ``_lifecycle_lock`` is taken, and only around the reducer call: it + is the lock that guards supervisor mutation, while ``_lifecycle_change`` + guards provider identity, which a tick never changes. Waiting on the + latter would park the watchdog behind a retirement — exactly when a + keyed rig most needs it. The loop retires itself once the lease is gone + so an idle target costs nothing and leaves no task behind. + """ + try: + while True: + async with self._lifecycle_lock: + if self._shutdown_pending or self.tx_snapshot.lease_id is None: + self._tick_task = None + return + transition = self._tx_safety.tick() + try: + await self._service_effects(transition) + except Exception: + logger.warning("managed TX tick service failed", exc_info=True) + await asyncio.sleep(self._tick_interval) + except asyncio.CancelledError: + pass + def _not_ready(self) -> TxTransition: return TxTransition(TxOutcome.NOT_READY, self.tx_snapshot) @@ -290,6 +324,8 @@ async def request_on(self, owner: TxOwner) -> TxTransition: ): return self._not_ready() transition = self._tx_safety.request_on(owner) + if self._tick_task is None and transition.snapshot.lease_id is not None: + self._tick_task = asyncio.create_task(self._tick_loop()) await self._service_effects(transition) return transition @@ -338,6 +374,10 @@ async def _complete_shutdown( self, transition: TxTransition, release_provider: ProviderRelease ) -> tuple[TxTransition, BaseException | None]: error: BaseException | None = None + if (ticker := self._tick_task) is not None: + self._tick_task = None + ticker.cancel() + await asyncio.gather(ticker, return_exceptions=True) try: if transition.outcome is not TxOutcome.NOOP: await asyncio.wait_for( diff --git a/tests/test_managed_tx_watchdog_ticker.py b/tests/test_managed_tx_watchdog_ticker.py new file mode 100644 index 00000000..4b6028ea --- /dev/null +++ b/tests/test_managed_tx_watchdog_ticker.py @@ -0,0 +1,152 @@ +"""MOR-1191: the managed TX max-key-down watchdog needs a production driver. + +``TxSafetySupervisor.tick`` is the only path to the watchdog and to the timed +retry of a failed OFF, and nothing under ``src`` ever called it: a rig keyed +through the managed path stayed keyed with no bound at all. No test here calls +``tick`` -- the whole defect is that nothing did -- so each one advances a fake +clock, lets the loop run, and watches the wire. + +A real supervisor, the real effect service and a hand-rolled provider drive +every case: a scripted double would answer whatever it was told, and a +``MagicMock`` satisfies a ``runtime_checkable`` protocol on 3.11 but not on +3.12+ (gh-102433). ``_Provider`` is imported rather than copied so both suites +watch the same wire. +""" + +from __future__ import annotations + +import asyncio +import time +from collections.abc import Callable + +from rigplane.core.tx_safety import ( + TxOwner, + TxPhase, + TxReleaseReason, + TxSafetySupervisor, + TxSource, + TxTransition, +) +from rigplane.runtime.managed_radio_runtime import ManagedRadioRuntime, TxService +from rigplane.runtime.managed_tx_effect_service import managed_tx_effect_service +from test_web_recovery_durable_off import _Provider + +_OWNER = TxOwner(TxSource.WEBSOCKET, "ws-1") +_TICK = 0.002 + + +class _Clock: + """Monotonic only when the test says so.""" + + def __init__(self) -> None: + self.now = 1_000.0 + + def __call__(self) -> float: + return self.now + + +class _Managed: + """A managed runtime plus everything the tests need to watch it.""" + + def __init__(self) -> None: + self.clock, self.log = _Clock(), [] + self.serviced: list[TxTransition] = [] + self.provider = _Provider(self.log) + self.runtime = ManagedRadioRuntime( + "watchdog", + service_factory=self._factory, + provider_lifecycle=self.provider, + clock=self.clock, + tick_interval_seconds=_TICK, + ) + + def _factory(self, host: object) -> TxService: + inner = managed_tx_effect_service(host) + + async def service(sup: TxSafetySupervisor, moved: TxTransition) -> None: + self.serviced.append(moved) + await inner(sup, moved) + + return service + + +async def _armed(*, key: bool = True) -> _Managed: + """Bring the provider up, seed the OFF ``request_on`` demands, then key.""" + managed = _Managed() + await managed.runtime.replace_provider(ready=True) + await managed.runtime.request_fresh_ptt() + if key: + assert (await managed.runtime.request_on(_OWNER)).snapshot.lease_id + managed.log.clear() + managed.serviced.clear() + return managed + + +async def _settles(predicate: Callable[[], bool], timeout: float = 2.0) -> None: + """Let real tick intervals elapse until the driver has done its work.""" + deadline = time.monotonic() + timeout + while not predicate(): + assert time.monotonic() < deadline, "the ticker never got there" + await asyncio.sleep(_TICK) + + +async def test_a_lease_held_past_max_key_down_is_dekeyed_on_the_wire() -> None: + """Acceptance 1: the OFF reaches the provider, with no call from the test.""" + managed = await _armed() + assert managed.runtime.tx_snapshot.phase is TxPhase.KEYED + + managed.clock.now += 181.0 + await _settles(lambda: managed.runtime.tx_snapshot.phase is TxPhase.IDLE) + + assert managed.log == ["ptt(off)", "read_ptt"] + assert len(managed.serviced) == 1 # the effects reached the provider, not a void + reason = managed.serviced[0].snapshot.release_reason + assert reason is TxReleaseReason.BACKEND_MAX_KEY_DOWN + # Nothing left to watch: the driver retires instead of idling forever. + await _settles(lambda: managed.runtime._tick_task is None) + + +async def test_a_refused_off_retries_on_its_own_schedule() -> None: + """Acceptance 2: the clock brings it back -- no reconnect, no second call.""" + managed = await _armed() + managed.provider.write_failures = 1 + + lease = managed.runtime.tx_snapshot.lease_id or "" + await managed.runtime.request_off(_OWNER, lease) + assert managed.log == ["ptt(off)"] + assert managed.runtime.tx_snapshot.phase is TxPhase.FAULTED + + await asyncio.sleep(_TICK * 20) # many ticks, but the retry is not due yet + assert managed.log == ["ptt(off)"] + + managed.clock.now += 0.25 # retry_schedule_seconds[0] + await _settles(lambda: managed.runtime.tx_snapshot.phase is TxPhase.IDLE) + assert managed.log == ["ptt(off)", "ptt(off)", "read_ptt"] + + +async def test_the_ticker_stops_at_shutdown_and_never_fires_again() -> None: + """Acceptance 3: shutdown owns the last word and leaves no live task.""" + managed = await _armed() + ticker = managed.runtime._tick_task + assert ticker is not None + + await managed.runtime.shutdown(release_provider=lambda: asyncio.sleep(0)) + + assert managed.log == ["ptt(off)", "read_ptt"] + assert managed.runtime._tick_task is None and ticker.done() + managed.clock.now += 10_000.0 + await asyncio.sleep(_TICK * 20) + assert managed.log == ["ptt(off)", "read_ptt"] + + +async def test_nothing_ticks_until_a_lease_exists_and_the_signal_says_so() -> None: + """Acceptance 4 and 5: no lease, no driver, no cost, no watchdog claimed.""" + managed = await _armed(key=False) + + await asyncio.sleep(_TICK * 20) + assert managed.runtime._tick_task is None and managed.log == [] + assert not managed.runtime.tx_snapshot.watchdog_enabled + + await managed.runtime.request_on(_OWNER) + await _settles(lambda: managed.runtime.tx_snapshot.watchdog_enabled) + assert managed.runtime._tick_task is not None