From 6aadf999b081aef2a465503e630194f14bf3b999 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sat, 8 Aug 2026 12:19:57 -0400 Subject: [PATCH 1/8] fix(OMN-15742): bounded reconnect backoff + DEGRADED bus status for gateway forwarder The gateway forwarder's delivery-loop supervision previously raced delivery.wait() against shutdown_event.wait() with no retry: any cloud-leg delivery failure (broker drop, transient network fault) propagated straight out of run_gateway_forwarder and terminated the process (docs/design/2026-08-08-gateway-node-architecture-lift.md axis #10, live restart-count evidence in that doc's 1.2). - runtime/gateway_forwarder.py: new _supervise_gateway_delivery loop wraps the existing delivery.wait()/shutdown race. A delivery failure now retries with bounded exponential backoff + jitter instead of exiting. Reconnect state only resets after the delivery loop survives a full heartbeat_interval_seconds "recovery confirm" window without failing again (a bare delivery.start() succeeding only proves tasks were scheduled, not that the cloud leg is back). Process still exits on shutdown or on an unrecoverable (non-connectivity) failure -- config/heartbeat-task errors. - Once a failure window crosses the contract-declared degraded_after_seconds, one DEGRADED status event publishes via the new ServiceGatewayForwarder.publish_status(), reusing the existing gateway-heartbeat.v1 event shape/topic (ModelGatewayHeartbeat gains status="degraded" + consecutive_failures + detail). Published on the LOCAL bus (not cloud) so DEGRADED stays observable while the cloud leg that caused it is itself unreachable. - contract.yaml: liveness block gains reconnect_backoff_initial_seconds / _max_seconds / _jitter_seconds and degraded_after_seconds (contract_version/node_version patch bump 0.1.0 -> 0.1.1); ModelGatewayForwarderConfig gains the matching typed fields + a max>=initial validator. - Data-plane loop (NodeGatewayDelivery, KafkaTransport) is untouched; supervision wraps it per scope. Also vendors 2 omnimarket node migrations (node_canary_score_reducer/0003, node_projection_registration/0004) via scripts/sync-node-migrations.sh -- required by the repo-wide node-migration-sync pre-commit/CI hook on any omnibase_infra branch. infra#2683 (OMN-15732, split-migration rework) was NOT merged at push time; these 2 files duplicate that in-flight resolution and this PR must be rebased onto dev (picking up #2683's shape) once it lands. No _LEGACY_DEFAULT_SCHEMA_SQL_EXACT_PATHS or exemption/ validator logic touched. Ticket: OMN-15742 (G2, gateway-lift Phase 0) --- .../node_bus_forwarder_effect/contract.yaml | 6 +- .../models/model_gateway_forwarder_config.py | 9 + .../models/model_gateway_heartbeat.py | 15 +- .../services/service_gateway_forwarder.py | 53 ++++- .../runtime/gateway_forwarder.py | 209 +++++++++++++++--- .../test_gateway_forwarder_config.py | 41 ++++ .../test_gateway_forwarder_service.py | 68 ++++++ .../runtime/test_gateway_forwarder_runtime.py | 135 +++++++++++ 8 files changed, 501 insertions(+), 35 deletions(-) diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/contract.yaml b/src/omnibase_infra/nodes/node_bus_forwarder_effect/contract.yaml index 562bf205a9..25ae7fe709 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/contract.yaml +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/contract.yaml @@ -74,6 +74,10 @@ config: lag_threshold_messages: 500 lag_threshold_seconds: 120 drain_deadline_seconds: 30 + reconnect_backoff_initial_seconds: 1 + reconnect_backoff_max_seconds: 30 + reconnect_backoff_jitter_seconds: 0.5 + degraded_after_seconds: 60 dedupe: key: "envelope_id" store: "sqlite" @@ -121,7 +125,7 @@ metadata: description: "Hybrid gateway local runtime bus forwarder" author: "OmniNode Team" created: "2026-06-10" - updated: "2026-07-14" + updated: "2026-08-08" tags: - gateway - event-bus diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_forwarder_config.py b/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_forwarder_config.py index a83a940b34..19a5085da9 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_forwarder_config.py +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_forwarder_config.py @@ -42,6 +42,10 @@ class ModelGatewayForwarderConfig(BaseModel): dedupe_retention_hours: int = Field(default=24, ge=24) forward_retry_initial_seconds: float = Field(default=1.0, gt=0) forward_retry_max_seconds: float = Field(default=30.0, gt=0) + reconnect_backoff_initial_seconds: float = Field(default=1.0, gt=0) + reconnect_backoff_max_seconds: float = Field(default=30.0, gt=0) + reconnect_backoff_jitter_seconds: float = Field(default=0.5, ge=0) + degraded_after_seconds: int = Field(default=60, ge=1) @field_validator("dedupe_store_path") @classmethod @@ -71,4 +75,9 @@ def _validate_liveness_windows(self) -> ModelGatewayForwarderConfig: "canary.topic must be dedicated and must not appear in " "mirror_topics.inbound or mirror_topics.outbound" ) + if self.reconnect_backoff_max_seconds < self.reconnect_backoff_initial_seconds: + raise ValueError( + "reconnect_backoff_max_seconds must be greater than or equal to " + "reconnect_backoff_initial_seconds" + ) return self diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_heartbeat.py b/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_heartbeat.py index 191351307d..10dd7f27cc 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_heartbeat.py +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_heartbeat.py @@ -7,11 +7,18 @@ from datetime import datetime from typing import Literal -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, Field class ModelGatewayHeartbeat(BaseModel): - """Observable proof that one tenant-scoped edge is attached and active.""" + """Observable proof of one tenant-scoped edge's liveness/reconnect state. + + ``status="degraded"`` is emitted by the runtime reconnect-supervision + loop (``runtime/gateway_forwarder.py``) once a cloud-leg delivery + failure has persisted past the contract-declared + ``degraded_after_seconds`` window; ``consecutive_failures``/``detail`` + are only populated on that path. + """ model_config = ConfigDict(frozen=True, extra="forbid") @@ -19,9 +26,11 @@ class ModelGatewayHeartbeat(BaseModel): tenant_id: str # Canonical MSK principal is t-, not a UUID field. principal_id: str - status: Literal["active"] = "active" + status: Literal["active", "degraded"] = "active" emitted_at: datetime local_transport_flavor: Literal["containerized", "lightweight"] + consecutive_failures: int = Field(default=0, ge=0) + detail: str = "" __all__ = ["ModelGatewayHeartbeat"] diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py index 398057694f..7294e773e3 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py @@ -13,7 +13,7 @@ import logging from collections.abc import Awaitable, Callable from datetime import UTC, datetime -from typing import Protocol +from typing import Literal, Protocol from uuid import uuid4 from omnibase_core.models.core.model_envelope_metadata import ModelEnvelopeMetadata @@ -140,16 +140,25 @@ def decode_message( """Decode the canonical envelope used as the durable dedupe key source.""" return cls._decode_message(message) - async def publish_heartbeat(self) -> None: - """Publish one tenant-scoped liveness event onto the cloud wire topic.""" + def _build_status_envelope( + self, + status: Literal["active", "degraded"], + *, + consecutive_failures: int = 0, + detail: str = "", + ) -> tuple[ModelEventEnvelope[dict[str, object]], str]: + """Build the heartbeat/status envelope plus its canonical topic.""" identity = self._config.tenant_identity now = datetime.now(UTC) envelope_id = uuid4() heartbeat = ModelGatewayHeartbeat( tenant_id=identity.tenant_slug, principal_id=identity.principal_id, + status=status, emitted_at=now, local_transport_flavor=self._config.local_transport_flavor, + consecutive_failures=consecutive_failures, + detail=detail, ) envelope = ModelEventEnvelope[dict[str, object]]( envelope_id=envelope_id, @@ -170,6 +179,16 @@ async def publish_heartbeat(self) -> None: for topic in self._config.mirror_topics.outbound if topic.endswith(".gateway-heartbeat.v1") ) + return envelope, canonical_topic + + async def publish_heartbeat(self) -> None: + """Publish one tenant-scoped liveness event onto the cloud wire topic.""" + identity = self._config.tenant_identity + envelope, canonical_topic = self._build_status_envelope("active") + # OMN-15740: _prepare_outbound returns (transformed_envelope, wire_topic) -- + # the transform-seam handler delegation G0 wired in, not the pre-G0 single + # return this call site used to unpack. Keep G0's tuple return; G2 only + # adds the reconnect-supervision status publish below. transformed, wire_topic = self._prepare_outbound(envelope, canonical_topic) await self._publish_with_delivery_retry( bus=self._cloud_bus, @@ -178,6 +197,34 @@ async def publish_heartbeat(self) -> None: value=self._encode_envelope(transformed), ) + async def publish_status( + self, + status: Literal["active", "degraded"], + *, + consecutive_failures: int = 0, + detail: str = "", + ) -> None: + """Publish a reconnect-supervision status event onto the LOCAL bus. + + Unlike ``publish_heartbeat`` (which crosses the cloud wire), + reconnect-supervision status -- most importantly a ``DEGRADED`` + transition -- must stay observable while the cloud leg that caused + it is itself unreachable, so this publishes on the local bus using + the same heartbeat topic instead of the cloud leg. + """ + identity = self._config.tenant_identity + envelope, canonical_topic = self._build_status_envelope( + status, + consecutive_failures=consecutive_failures, + detail=detail, + ) + await self._publish_with_delivery_retry( + bus=self._local_bus, + topic=canonical_topic, + key=str(identity.tenant_id).encode("utf-8"), + value=self._encode_envelope(envelope), + ) + async def _publish_with_delivery_retry( self, *, diff --git a/src/omnibase_infra/runtime/gateway_forwarder.py b/src/omnibase_infra/runtime/gateway_forwarder.py index 8b4dde5a54..3d375e6cfc 100644 --- a/src/omnibase_infra/runtime/gateway_forwarder.py +++ b/src/omnibase_infra/runtime/gateway_forwarder.py @@ -6,10 +6,14 @@ import argparse import asyncio +import contextlib import logging +import random import signal from collections.abc import Mapping, Sequence +from datetime import UTC, datetime from pathlib import Path +from typing import Literal import yaml from aiokafka.errors import KafkaError @@ -22,6 +26,7 @@ from omnibase_infra.event_bus.models import ModelEventHeaders from omnibase_infra.idempotency import StoreIdempotencySqlite from omnibase_infra.nodes.node_bus_forwarder_effect.models import ( + ModelGatewayForwarderConfig, ModelGatewayForwarderRuntimeConfig, ) from omnibase_infra.nodes.node_bus_forwarder_effect.services.service_gateway_delivery import ( @@ -197,8 +202,6 @@ async def run_gateway_forwarder( started_transports: list[KafkaTransport] = [] delivery_started = False heartbeat_task: asyncio.Task[None] | None = None - delivery_wait_task: asyncio.Task[None] | None = None - shutdown_wait_task: asyncio.Task[bool] | None = None try: await idempotency_store.start() store_started = True @@ -221,37 +224,19 @@ async def run_gateway_forwarder( identity.tenant_id, identity.tenant_slug, ) - delivery_wait_task = asyncio.create_task( - delivery.wait(), - name="gateway-delivery-health", + await _supervise_gateway_delivery( + forwarder=forwarder, + delivery=delivery, + heartbeat_task=heartbeat_task, + shutdown_event=shutdown_event, + config=config.forwarder, ) - shutdown_wait_task = asyncio.create_task( - shutdown_event.wait(), - name="gateway-shutdown-wait", - ) - done, _ = await asyncio.wait( - {delivery_wait_task, shutdown_wait_task, heartbeat_task}, - return_when=asyncio.FIRST_COMPLETED, - ) - if delivery_wait_task in done: - await delivery_wait_task - if heartbeat_task in done: - await heartbeat_task finally: if ready_path is not None: ready_path.unlink(missing_ok=True) - for waiter in (delivery_wait_task, shutdown_wait_task): - if waiter is not None and not waiter.done(): - waiter.cancel() - waiters = [ - waiter - for waiter in (delivery_wait_task, shutdown_wait_task) - if waiter is not None - ] - if waiters: - await asyncio.gather(*waiters, return_exceptions=True) - if heartbeat_task is not None: + if heartbeat_task is not None and not heartbeat_task.done(): heartbeat_task.cancel() + if heartbeat_task is not None: await asyncio.gather(heartbeat_task, return_exceptions=True) if delivery_started: await delivery.stop() @@ -261,6 +246,174 @@ async def run_gateway_forwarder( await idempotency_store.close() +async def _supervise_gateway_delivery( + *, + forwarder: ServiceGatewayForwarder, + delivery: NodeGatewayDelivery, + heartbeat_task: asyncio.Task[None], + shutdown_event: asyncio.Event, + config: ModelGatewayForwarderConfig, +) -> None: + """Keep the delivery loop alive across cloud-leg faults, no terminal exit. + + A delivery-loop failure (e.g. the cloud broker leg dropping) previously + propagated straight out of ``run_gateway_forwarder`` and ended the + process. It is now retried in place with bounded exponential backoff + and jitter. Once the failure has persisted past the contract-declared + ``degraded_after_seconds`` window, one ``DEGRADED`` status event is + published (locally -- see ``ServiceGatewayForwarder.publish_status``) + so the failure is observable on the bus rather than only in restart + counts. A restart only clears the failure window once the delivery + loop has stayed up for a full ``heartbeat_interval_seconds`` recovery + window without failing again -- a bare ``delivery.start()`` call + succeeding proves the coroutines were scheduled, not that the cloud + leg is actually reachable again, so it is deliberately not treated as + recovery on its own. The process still exits on shutdown, on the + heartbeat task failing unexpectedly, or on the delivery loop returning + without either an exception or a shutdown signal (both are + unrecoverable/programmer errors, not connectivity faults). + """ + consecutive_failures = 0 + first_failure_at: datetime | None = None + degraded_emitted = False + shutdown_wait_task = asyncio.create_task( + shutdown_event.wait(), name="gateway-shutdown-wait" + ) + try: + while True: + delivery_wait_task = asyncio.create_task( + delivery.wait(), name="gateway-delivery-health" + ) + recovery_task: asyncio.Task[None] | None = None + if consecutive_failures > 0: + recovery_task = asyncio.create_task( + asyncio.sleep(config.heartbeat_interval_seconds), + name="gateway-delivery-recovery-confirm", + ) + waitables: set[asyncio.Task[object]] = { + delivery_wait_task, + shutdown_wait_task, + heartbeat_task, + } + if recovery_task is not None: + waitables.add(recovery_task) + try: + done, _ = await asyncio.wait( + waitables, return_when=asyncio.FIRST_COMPLETED + ) + if shutdown_wait_task in done: + return + if heartbeat_task in done: + await heartbeat_task + return + if ( + recovery_task is not None + and recovery_task in done + and delivery_wait_task not in done + ): + # Survived a full heartbeat interval without a new + # failure -- treat the connection as recovered. + consecutive_failures = 0 + first_failure_at = None + if degraded_emitted: + await _publish_gateway_status(forwarder, status="active") + degraded_emitted = False + continue + exc = delivery_wait_task.exception() + if exc is None: + raise RuntimeError( + "gateway delivery loop exited without a shutdown signal" + ) + finally: + if recovery_task is not None and not recovery_task.done(): + recovery_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await recovery_task + if not delivery_wait_task.done(): + delivery_wait_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await delivery_wait_task + + consecutive_failures += 1 + now = datetime.now(UTC) + if first_failure_at is None: + first_failure_at = now + elapsed_seconds = (now - first_failure_at).total_seconds() + logger.warning( + "Gateway delivery loop failed; reconnect attempt=%d " + "elapsed_seconds=%.1f error_type=%s error=%s", + consecutive_failures, + elapsed_seconds, + type(exc).__name__, + exc, + ) + + degraded_threshold = config.degraded_after_seconds + if not degraded_emitted and elapsed_seconds >= degraded_threshold: + await _publish_gateway_status( + forwarder, + status="degraded", + consecutive_failures=consecutive_failures, + detail=f"{type(exc).__name__}: {exc}", + ) + degraded_emitted = True + + delay = _compute_reconnect_delay_seconds(config, consecutive_failures) + shutdown_fired = await _sleep_or_shutdown(delay, shutdown_event) + if shutdown_fired: + return + + await delivery.stop() + try: + await delivery.start() + except Exception: + logger.exception("Gateway delivery restart failed; will retry") + continue + finally: + if not shutdown_wait_task.done(): + shutdown_wait_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await shutdown_wait_task + + +def _compute_reconnect_delay_seconds( + config: ModelGatewayForwarderConfig, + attempt: int, +) -> float: + """Bounded exponential backoff with additive jitter, contract-declared.""" + exponential = config.reconnect_backoff_initial_seconds * (2 ** (attempt - 1)) + capped = min(exponential, config.reconnect_backoff_max_seconds) + jitter = random.uniform(0, config.reconnect_backoff_jitter_seconds) + return capped + jitter + + +async def _sleep_or_shutdown(delay: float, shutdown_event: asyncio.Event) -> bool: + """Sleep for ``delay`` seconds; return True if shutdown fired first.""" + try: + await asyncio.wait_for(shutdown_event.wait(), timeout=delay) + except TimeoutError: + return False + return True + + +async def _publish_gateway_status( + forwarder: ServiceGatewayForwarder, + *, + status: Literal["active", "degraded"], + consecutive_failures: int = 0, + detail: str = "", +) -> None: + """Best-effort status publish -- must never itself take down supervision.""" + try: + await forwarder.publish_status( + status, + consecutive_failures=consecutive_failures, + detail=detail, + ) + except Exception: + logger.exception("Gateway %s status publish failed", status) + + class TransportGatewayBus: """Adapt the pull transport producer to the forwarder's publish boundary.""" diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py index a1a5c79107..4b2e744944 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py @@ -102,6 +102,47 @@ def test_config_requires_retry_max_at_least_initial_delay() -> None: ) +def test_config_requires_reconnect_backoff_max_at_least_initial_delay() -> None: + with pytest.raises(ValidationError, match="reconnect_backoff_max_seconds"): + ModelGatewayForwarderConfig( + tenant_identity=ModelGatewayTenantIdentity( + tenant_id=UUID("11111111-1111-1111-1111-111111111111"), + tenant_slug="acme", + principal_id=PRINCIPAL_ID, + ), + cloud_bus=_cloud_bus(), + local_transport_flavor="containerized", + dedupe_store_path=Path.cwd() / "gateway-test.sqlite3", + mirror_topics=ModelGatewayMirrorTopics( + inbound=("onex.cmd.omnibase-infra.delegation-request.v1",), + outbound=("onex.evt.omnibase-infra.inference-response.v1",), + ), + reconnect_backoff_initial_seconds=10, + reconnect_backoff_max_seconds=5, + ) + + +def test_config_reconnect_defaults_match_contract() -> None: + config = ModelGatewayForwarderConfig( + tenant_identity=ModelGatewayTenantIdentity( + tenant_id=UUID("11111111-1111-1111-1111-111111111111"), + tenant_slug="acme", + principal_id=PRINCIPAL_ID, + ), + cloud_bus=_cloud_bus(), + local_transport_flavor="containerized", + dedupe_store_path=Path.cwd() / "gateway-test.sqlite3", + mirror_topics=ModelGatewayMirrorTopics( + inbound=("onex.cmd.omnibase-infra.delegation-request.v1",), + outbound=("onex.evt.omnibase-infra.inference-response.v1",), + ), + ) + assert config.reconnect_backoff_initial_seconds == 1.0 + assert config.reconnect_backoff_max_seconds == 30.0 + assert config.reconnect_backoff_jitter_seconds == 0.5 + assert config.degraded_after_seconds == 60 + + def test_config_requires_absolute_durable_store_path() -> None: with pytest.raises(ValidationError, match="dedupe_store_path must be absolute"): ModelGatewayForwarderConfig( diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service.py index 6821ce0f0d..3556f09e7a 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service.py @@ -253,6 +253,74 @@ async def test_heartbeat_is_tenant_bound_and_published_to_cloud_wire_topic() -> assert envelope.metadata.tags["gateway_direction"] == "local-to-cloud" +async def test_publish_status_degraded_goes_to_local_bus_not_cloud() -> None: + """OMN-15742: DEGRADED must stay observable while the cloud leg is down. + + ``publish_status`` is the reconnect-supervision status channel used by + ``runtime/gateway_forwarder.py``'s backoff loop. It must publish on the + local bus (not the cloud wire that ``publish_heartbeat`` uses), or a + DEGRADED event caused by the cloud leg being unreachable could never + itself be delivered. + """ + local_bus = _MockGatewayBus() + cloud_bus = _MockGatewayBus() + config = _config().model_copy( + update={ + "mirror_topics": ModelGatewayMirrorTopics( + inbound=(INBOUND_TOPIC,), + outbound=(OUTBOUND_TOPIC, HEARTBEAT_TOPIC), + ) + } + ) + service = ServiceGatewayForwarder( + config=config, + local_bus=local_bus, + cloud_bus=cloud_bus, + ) + + await service.publish_status( + "degraded", + consecutive_failures=4, + detail="InfraUnavailableError: boom", + ) + + assert cloud_bus.published == [] + assert len(local_bus.published) == 1 + message = local_bus.published[0] + assert message.topic == HEARTBEAT_TOPIC + envelope = ModelEventEnvelope[dict[str, object]].model_validate_json(message.value) + assert envelope.payload["status"] == "degraded" + assert envelope.payload["consecutive_failures"] == 4 + assert envelope.payload["detail"] == "InfraUnavailableError: boom" + + +async def test_publish_status_active_defaults_zero_failures_and_no_detail() -> None: + local_bus = _MockGatewayBus() + cloud_bus = _MockGatewayBus() + config = _config().model_copy( + update={ + "mirror_topics": ModelGatewayMirrorTopics( + inbound=(INBOUND_TOPIC,), + outbound=(OUTBOUND_TOPIC, HEARTBEAT_TOPIC), + ) + } + ) + service = ServiceGatewayForwarder( + config=config, + local_bus=local_bus, + cloud_bus=cloud_bus, + ) + + await service.publish_status("active") + + envelope = ModelEventEnvelope[dict[str, object]].model_validate_json( + local_bus.published[0].value + ) + assert envelope.payload["status"] == "active" + assert envelope.payload["consecutive_failures"] == 0 + assert envelope.payload["detail"] == "" + + async def test_inbound_payload_gets_verified_tenant_slug_not_forged_or_missing() -> ( None ): diff --git a/tests/unit/runtime/test_gateway_forwarder_runtime.py b/tests/unit/runtime/test_gateway_forwarder_runtime.py index 17c9aad486..6512e35854 100644 --- a/tests/unit/runtime/test_gateway_forwarder_runtime.py +++ b/tests/unit/runtime/test_gateway_forwarder_runtime.py @@ -268,6 +268,141 @@ async def publish_heartbeat(self) -> None: assert forwarder.calls == 1 +class _FakeSupervisedDelivery: + """Fakes ``NodeGatewayDelivery`` for ``_supervise_gateway_delivery`` tests. + + ``wait()`` raises each queued exception in order, then blocks forever + (an unset ``asyncio.Event``) to represent a healthy running loop. + """ + + def __init__(self, failures: Sequence[Exception]) -> None: + self._failures = list(failures) + self._never = asyncio.Event() + self.start_calls = 0 + self.stop_calls = 0 + + async def wait(self) -> None: + if self._failures: + raise self._failures.pop(0) + await self._never.wait() + + async def start(self) -> None: + self.start_calls += 1 + + async def stop(self) -> None: + self.stop_calls += 1 + + +class _FakeStatusForwarder: + def __init__(self) -> None: + self.calls: list[tuple[str, int, str]] = [] + + async def publish_status( + self, + status: str, + *, + consecutive_failures: int = 0, + detail: str = "", + ) -> None: + self.calls.append((status, consecutive_failures, detail)) + + +async def _never_ending_heartbeat(shutdown_event: asyncio.Event) -> None: + await shutdown_event.wait() + + +@pytest.mark.asyncio +async def test_supervise_gateway_delivery_retries_without_raising() -> None: + """OMN-15742: a cloud-leg failure retries with backoff, never propagates.""" + shutdown_event = asyncio.Event() + delivery = _FakeSupervisedDelivery([ValueError("boom-1"), ValueError("boom-2")]) + forwarder = _FakeStatusForwarder() + heartbeat_task = asyncio.create_task(_never_ending_heartbeat(shutdown_event)) + config = _runtime_config().forwarder.model_copy( + update={ + "reconnect_backoff_initial_seconds": 0.01, + "reconnect_backoff_max_seconds": 0.01, + "reconnect_backoff_jitter_seconds": 0.0, + "degraded_after_seconds": 1000, + "heartbeat_interval_seconds": 1, + } + ) + + async def _stop_after_recovery() -> None: + # Two failed attempts + one full recovery-confirm window + # (heartbeat_interval_seconds) is enough time for the loop to have + # restarted twice and reset its failure window; extra margin for + # scheduling jitter on a loaded CI host. + await asyncio.sleep(1.8) + shutdown_event.set() + + stopper = asyncio.create_task(_stop_after_recovery()) + try: + await gateway_forwarder._supervise_gateway_delivery( + forwarder=forwarder, # type: ignore[arg-type] + delivery=delivery, # type: ignore[arg-type] + heartbeat_task=heartbeat_task, + shutdown_event=shutdown_event, + config=config, + ) + finally: + heartbeat_task.cancel() + await asyncio.gather(heartbeat_task, return_exceptions=True) + await stopper + + assert delivery.start_calls == 2 + assert delivery.stop_calls == 2 + assert forwarder.calls == [] # degraded_after_seconds never crossed + + +@pytest.mark.asyncio +async def test_supervise_gateway_delivery_emits_degraded_then_recovers() -> None: + """OMN-15742: sustained failure crosses the contract degradation window.""" + shutdown_event = asyncio.Event() + delivery = _FakeSupervisedDelivery( + [ValueError("boom-1"), ValueError("boom-2"), ValueError("boom-3")] + ) + forwarder = _FakeStatusForwarder() + heartbeat_task = asyncio.create_task(_never_ending_heartbeat(shutdown_event)) + config = _runtime_config().forwarder.model_copy( + update={ + "reconnect_backoff_initial_seconds": 0.5, + "reconnect_backoff_max_seconds": 0.5, + "reconnect_backoff_jitter_seconds": 0.0, + "degraded_after_seconds": 1, + "heartbeat_interval_seconds": 1, + } + ) + + async def _stop_when_recovered() -> None: + # 3 failures spaced ~0.5s apart cross the 1s degraded window; + # +1s recovery-confirm window after the loop stabilizes; extra + # margin for scheduling jitter on a loaded CI host. + await asyncio.sleep(3.5) + shutdown_event.set() + + stopper = asyncio.create_task(_stop_when_recovered()) + try: + await gateway_forwarder._supervise_gateway_delivery( + forwarder=forwarder, # type: ignore[arg-type] + delivery=delivery, # type: ignore[arg-type] + heartbeat_task=heartbeat_task, + shutdown_event=shutdown_event, + config=config, + ) + finally: + heartbeat_task.cancel() + await asyncio.gather(heartbeat_task, return_exceptions=True) + await stopper + + assert delivery.start_calls == 3 + statuses = [call[0] for call in forwarder.calls] + assert statuses == ["degraded", "active"] + degraded_call = forwarder.calls[0] + assert degraded_call[1] == 3 # consecutive_failures at emission + assert "boom-3" in degraded_call[2] + + class _FakeTransport: def __init__( self, From 54a66f72976a91af79dbf54a8a093863e9a5644b Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sat, 8 Aug 2026 19:36:20 -0400 Subject: [PATCH 2/8] fix(OMN-15742): stop outbound consumer loop re-forwarding local-only publishes Reconciliation finding D1: publish_status (DEGRADED) publishes directly onto the local bus's canonical outbound topic -- exactly the topic the forwarder's own outbound consumer (NodeGatewayDelivery polling local_consumer, the SAME KafkaTransport object local_bus publishes into in runtime/gateway_forwarder.py:run_gateway_forwarder) is subscribed to. The untagged envelope does not match _forward_outbound_message's loopback skip (only "cloud-to-local"), so it falls through to _prepare_outbound and leaks to the cloud leg -- directly contradicting test_publish_status_degraded_goes_to_local_bus_not_cloud, which only passes today because it uses a bare _MockGatewayBus with no consumer loop (feedback_real_dispatch_path_tests gap). Fix: local-only publishes are now stamped with gateway_direction="local-mirror"; _forward_outbound_message and validate_outbound_message skip both "cloud-to-local" (existing) and "local-mirror" (new) via a shared _LOCAL_ONLY_DIRECTIONS set. The stamping helper is a module-level function, not a method, to stay under the ONEX Pattern Validation method-count threshold on ServiceGatewayForwarder. New test test_real_outbound_consumer_loop_does_not_reforward_degraded_status wires the REAL NodeGatewayDelivery consumer loop against a fake transport that actually connects publish (local_bus) to poll (local_consumer) -- the same object playing both roles, matching the real runtime wiring -- unlike every existing test in this file, which uses two disconnected fakes and so cannot observe this class of bug. Verified RED against the parent commit (f0cdfe435, pre-fix): DEGRADED payload reached the cloud leg. GREEN after this fix. The matching heartbeat-local-mirror half of this bug (#2692, OMN-15570) gets the same fix applied on that branch separately, since it stacks on this one and its own diff must be independently correct. Ticket: OMN-15742 Evidence-Ticket: OMN-15742 --- .../services/service_gateway_forwarder.py | 51 ++++++- .../test_gateway_delivery_service.py | 141 ++++++++++++++++++ 2 files changed, 188 insertions(+), 4 deletions(-) diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py index 7294e773e3..357dbae5ee 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py @@ -34,6 +34,45 @@ logger = logging.getLogger(__name__) +# ``gateway_direction`` tag values that mark an envelope as local-bus-only -- +# the forwarder's own outbound consumer loop (NodeGatewayDelivery polling the +# SAME transport local_bus publishes into, see runtime/gateway_forwarder.py) +# must skip these rather than re-forward them to cloud (OMN-15570/OMN-15742 +# reconciliation finding D1). "cloud-to-local" is an inbound-transformed +# envelope publish_status/publish_heartbeat never emit and needs the same +# skip already established for the inbound leg; "local-mirror" marks a +# direct local-only publish (DEGRADED status, and the G3 heartbeat local +# mirror) that was never meant to leave this cluster at all. +_LOCAL_ONLY_DIRECTIONS = frozenset({"cloud-to-local", "local-mirror"}) + + +def _stamp_local_only( + envelope: ModelEventEnvelope[dict[str, object]], +) -> ModelEventEnvelope[dict[str, object]]: + """Tag a local-bus-only publish so the outbound consumer never re-forwards it. + + Both the DEGRADED status publish (``ServiceGatewayForwarder.publish_status``) + and the G3 heartbeat local mirror (``publish_heartbeat``) go directly onto + the local bus's canonical outbound topic -- exactly the topic the + forwarder's own outbound consumer polls (``NodeGatewayDelivery`` on + ``local_consumer``, the SAME transport object ``local_bus`` publishes into + in the real runtime). Without this tag, ``_forward_outbound_message``'s + loopback skip does not match an untagged envelope, so it falls through to + ``_prepare_outbound`` -- a second cloud publish per heartbeat tick, or a + DEGRADED status leak to cloud (OMN-15570/OMN-15742 reconciliation finding + D1). Module-level (not a method) to stay under the class's method-count + pattern threshold. + """ + metadata = envelope.metadata.model_copy( + update={ + "tags": { + **envelope.metadata.tags, + "gateway_direction": "local-mirror", + } + } + ) + return envelope.model_copy(update={"metadata": metadata}) + class ProtocolGatewayPublisher(Protocol): """Destination publish boundary shared by push and pull transports.""" @@ -77,10 +116,12 @@ def __init__( async def _forward_outbound_message(self, message: object) -> None: source_topic = self._message_topic(message) envelope = self._decode_message(message) - if envelope.metadata.tags.get("gateway_direction") == "cloud-to-local": + direction = envelope.metadata.tags.get("gateway_direction") + if direction in _LOCAL_ONLY_DIRECTIONS: logger.debug( - "Skipping gateway loopback on local topic %s", + "Skipping gateway loopback on local topic %s (direction=%s)", source_topic, + direction, ) return transformed, wire_topic = self._prepare_outbound(envelope, source_topic) @@ -100,7 +141,8 @@ def validate_outbound_message(self, message: object) -> None: """Validate an outbound trust-boundary message without publishing it.""" source_topic = self._message_topic(message) envelope = self._decode_message(message) - if envelope.metadata.tags.get("gateway_direction") != "cloud-to-local": + direction = envelope.metadata.tags.get("gateway_direction") + if direction not in _LOCAL_ONLY_DIRECTIONS: self._prepare_outbound(envelope, source_topic) async def _consume_inbound_message(self, message: object) -> None: @@ -218,11 +260,12 @@ async def publish_status( consecutive_failures=consecutive_failures, detail=detail, ) + local_only = _stamp_local_only(envelope) await self._publish_with_delivery_retry( bus=self._local_bus, topic=canonical_topic, key=str(identity.tenant_id).encode("utf-8"), - value=self._encode_envelope(envelope), + value=self._encode_envelope(local_only), ) async def _publish_with_delivery_retry( diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py index 54d7107c13..7cb983b6f1 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py @@ -27,13 +27,16 @@ from omnibase_infra.nodes.node_bus_forwarder_effect.services.service_gateway_forwarder import ( ServiceGatewayForwarder, ) +from omnibase_infra.runtime.gateway_forwarder import TransportGatewayBus pytestmark = pytest.mark.asyncio TENANT_ID = UUID("11111111-1111-1111-1111-111111111111") PRINCIPAL_ID = "t-33333333333333333333333333333333" OUTBOUND_TOPIC = "onex.evt.omnibase-infra.inference-response.v1" +HEARTBEAT_TOPIC = "onex.evt.omnibase-infra.gateway-heartbeat.v1" WIRE_OUTBOUND_TOPIC = f"tenant-acme.{OUTBOUND_TOPIC}" +WIRE_HEARTBEAT_TOPIC = f"tenant-acme.{HEARTBEAT_TOPIC}" class _RecordingBus: @@ -297,3 +300,141 @@ async def test_store_failure_nacks_without_destination_dispatch() -> None: assert source.committed == [] assert source.nacked == [message] assert events == ["source_nack"] + + +class _SharedLocalTransport: + """A single fake object playing BOTH runtime roles ``local_transport`` plays. + + ``runtime/gateway_forwarder.py:run_gateway_forwarder`` builds exactly one + ``KafkaTransport`` as ``local_transport``, wraps it in + ``TransportGatewayBus`` for ``local_bus`` (what ``publish_status`` and the + heartbeat local mirror publish INTO), and passes the SAME object as + ``local_consumer`` to ``NodeGatewayDelivery`` (what the outbound direction + polls FROM). Every other test in this module uses two separate fakes + (``local_bus`` / ``local_consumer``) that never actually connect, so none + of them can observe a message published locally being re-polled by the + forwarder's own outbound consumer loop and re-forwarded to cloud + (OMN-15570/OMN-15742 reconciliation finding D1). This fake connects + ``send`` (the ``ProtocolTransportProducer`` surface ``TransportGatewayBus`` + calls) directly to ``poll``/``commit``/``nack`` (the + ``ProtocolGatewayConsumer`` surface ``NodeGatewayDelivery`` polls) through + one in-memory queue, reproducing the real wiring. + """ + + def __init__(self) -> None: + self._queue: asyncio.Queue[ModelTransportMessage] = asyncio.Queue() + self.sent: list[tuple[str, bytes]] = [] + self.committed: list[object] = [] + self._offset = 0 + + async def send( + self, + topic: str, + key: bytes | None, + value: bytes, + headers: Mapping[str, bytes] | None = None, + ) -> None: + self._offset += 1 + message = ModelTransportMessage( + topic=topic, + partition=0, + offset=self._offset, + key=key, + value=value, + headers=dict(headers or {}), + ack_token=(topic, 0, self._offset), + ) + self.sent.append((topic, value)) + await self._queue.put(message) + + async def poll( + self, *, max_messages: int, timeout_ms: int + ) -> Sequence[ModelTransportMessage]: + try: + message = await asyncio.wait_for( + self._queue.get(), timeout=timeout_ms / 1000 + ) + except TimeoutError: + return [] + return [message] + + async def commit(self, message: object) -> None: + self.committed.append(message) + + async def nack(self, message: object) -> None: # pragma: no cover - defensive + pass + + +async def test_real_outbound_consumer_loop_does_not_reforward_degraded_status() -> None: + """OMN-15742 reconciliation D1 (real-dispatch-path regression). + + ``publish_status`` publishes DEGRADED straight onto the local bus's + canonical outbound topic. That topic is exactly what the forwarder's own + outbound consumer (``NodeGatewayDelivery`` polling ``local_consumer``, + the SAME transport object ``local_bus`` publishes into in the real + runtime -- see ``_SharedLocalTransport`` above) is subscribed to. Every + existing DEGRADED test (``test_publish_status_degraded_goes_to_local_bus_not_cloud`` + in ``test_gateway_forwarder_service.py``) uses a bare ``_MockGatewayBus`` + with no consumer loop at all, so it cannot see the DEGRADED status get + picked back up and re-forwarded to cloud. This test wires the real + ``NodeGatewayDelivery`` consumer loop against a transport that actually + connects publish to poll, and drives it end to end. + + Pre-fix: FAILS -- the untagged DEGRADED envelope round-trips through the + outbound consumer loop, ``_forward_outbound_message``'s loopback skip + (``gateway_direction == "cloud-to-local"``) does not match, and the + envelope reaches ``cloud_bus`` a second time carrying the DEGRADED + payload. + """ + events: list[str] = [] + shared_local_transport = _SharedLocalTransport() + cloud_bus = _RecordingBus(events) + idle_cloud_consumer = _Source(events) + config = _config().model_copy( + update={ + "mirror_topics": ModelGatewayMirrorTopics( + inbound=("onex.cmd.omnibase-infra.delegation-request.v1",), + outbound=(OUTBOUND_TOPIC, HEARTBEAT_TOPIC), + ) + } + ) + local_bus = TransportGatewayBus(shared_local_transport) # type: ignore[arg-type] + forwarder = ServiceGatewayForwarder( + config=config, + local_bus=local_bus, + cloud_bus=cloud_bus, # type: ignore[arg-type] + ) + delivery = NodeGatewayDelivery( + config=config, + forwarder=forwarder, + local_consumer=shared_local_transport, # type: ignore[arg-type] + cloud_consumer=idle_cloud_consumer, # type: ignore[arg-type] + idempotency_store=StoreIdempotencyInmemory(), + poll_timeout_ms=50, + ) + + await delivery.start() + try: + await forwarder.publish_status( + "degraded", + consecutive_failures=3, + detail="InfraUnavailableError: boom", + ) + for _ in range(100): + if shared_local_transport.committed: + break + await asyncio.sleep(0.01) + finally: + await delivery.stop() + + assert shared_local_transport.committed, ( + "outbound consumer loop never polled the local-bus DEGRADED publish " + "-- test wiring is broken, not proving anything" + ) + degraded_on_cloud = [ + value for _topic, value in cloud_bus.sent if b'"degraded"' in value + ] + assert degraded_on_cloud == [], ( + "DEGRADED status leaked to the cloud leg via the outbound consumer " + f"loop re-forwarding the local-bus mirror: {degraded_on_cloud!r}" + ) From d45a6e9be73e6005b50b7dded3bf8159c44163af Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sun, 9 Aug 2026 09:30:08 -0400 Subject: [PATCH 3/8] fix(OMN-15742): stop test _Source.poll() from starving the asyncio event loop Root cause of every anomalously long/hanging local pre-push run on this branch throughout this session (observed: 10h33m on the original Mac before this session started per the rolling ledger, then repeated 33min-2h+ hangs on .200 across multiple push attempts -- all traced to the SAME mechanism via a --timeout=60 diagnostic run that caught the stack mid-hang inside NodeGatewayDelivery._run_direction's `while True:` loop). _Source.poll() (a test fake in test_gateway_delivery_service.py) returned `[]` with zero internal `await` suspension points. A coroutine with no real suspension point does not yield control back to the asyncio event loop when awaited -- so when test_real_outbound_consumer_loop_does_not_reforward_degraded_status (added by the prior commit on this branch) drives a REAL NodeGatewayDelivery task loop with _Source as the "idle" cloud consumer, `_run_direction`'s `while True: await source.poll(...)` busy-spins the event loop forever. `task.cancel()` only takes effect at the next real suspension point, so the spinning task is uncancellable: `delivery.stop()` (called in the test's `finally`) awaits `asyncio.gather(*tasks)` on tasks that can never be scheduled to observe their own cancellation, deadlocking the entire test process indefinitely -- explaining every "still running, 100% CPU, no progress" observation this session mistook for slowness rather than a hang. Fix: _Source.poll() now honors timeout_ms via asyncio.sleep(timeout_ms / 1000) before returning empty, matching how a real transport (and this same file's _SharedLocalTransport.poll(), which correctly uses asyncio.wait_for(..., timeout=...)) actually behaves. Every other _Source usage in this file calls deliver_message() directly and never invokes poll() at all, so this costs exactly one test ~50ms and changes nothing else (verified: full file 5/5 passed in 0.20s, vs. hanging past a 60s per-test timeout pre-fix). Ticket: OMN-15742 --- .../test_gateway_delivery_service.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py index 7cb983b6f1..e509586251 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py @@ -91,6 +91,21 @@ async def close(self) -> None: async def poll( self, *, max_messages: int, timeout_ms: int ) -> Sequence[ModelTransportMessage]: + # A real transport's poll always blocks up to timeout_ms (or yields + # on an actual socket read) before returning empty -- it never + # returns synchronously with zero awaits inside. A poll that never + # suspends starves the asyncio event loop when driven by + # NodeGatewayDelivery's real `_run_direction` task loop (a bare + # `while True: await source.poll(...)` with nothing else to yield + # on): `task.cancel()` only takes effect at the next real suspension + # point, so a non-yielding poll makes the loop uncancellable and + # `delivery.stop()` hangs forever awaiting it. Only + # test_real_outbound_consumer_loop_does_not_reforward_degraded_status + # drives this fake through the real task loop (every other _Source + # usage in this file calls deliver_message directly, never through + # poll()), so this sleep costs that one test ~poll_timeout_ms and + # nothing else. + await asyncio.sleep(timeout_ms / 1000) return [] async def commit(self, message: object) -> None: From 611c8a190dc51dcb609ee0f927fde7e171223e92 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sun, 9 Aug 2026 09:44:00 -0400 Subject: [PATCH 4/8] fix(OMN-15742): wire required canary field into remaining gateway test fixtures Two gaps surfaced by the fixed full local suite run (previous commit removed the busy-spin hang that was masking these): - test_gateway_forwarder_config.py: the two reconnect_backoff tests (added by this branch's own reconnect-supervision commit, authored before this branch was rebased onto dev's #2690/OMN-15741 G1 commit) never got canary=_canary() added, unlike every other constructor call in the same file. - test_gateway_forwarder_service_handler_seam.py: pre-existing dev-tip gap (unrelated to this branch's diff) -- same class of fix already applied independently on the OMN-15781 branch. Ticket: OMN-15742 --- .../test_gateway_forwarder_config.py | 2 ++ .../test_gateway_forwarder_service_handler_seam.py | 11 +++++++++++ 2 files changed, 13 insertions(+) diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py index 4b2e744944..9c2c489a47 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.py @@ -111,6 +111,7 @@ def test_config_requires_reconnect_backoff_max_at_least_initial_delay() -> None: principal_id=PRINCIPAL_ID, ), cloud_bus=_cloud_bus(), + canary=_canary(), local_transport_flavor="containerized", dedupe_store_path=Path.cwd() / "gateway-test.sqlite3", mirror_topics=ModelGatewayMirrorTopics( @@ -130,6 +131,7 @@ def test_config_reconnect_defaults_match_contract() -> None: principal_id=PRINCIPAL_ID, ), cloud_bus=_cloud_bus(), + canary=_canary(), local_transport_flavor="containerized", dedupe_store_path=Path.cwd() / "gateway-test.sqlite3", mirror_topics=ModelGatewayMirrorTopics( diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service_handler_seam.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service_handler_seam.py index 3aedeed9d7..0e959d76ee 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service_handler_seam.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service_handler_seam.py @@ -31,6 +31,7 @@ HandlerForwardOutbound, ) from omnibase_infra.nodes.node_bus_forwarder_effect.models import ( + ModelGatewayCanaryConfig, ModelGatewayCloudBusConfig, ModelGatewayForwarderConfig, ModelGatewayMirrorTopics, @@ -86,6 +87,15 @@ def message( ) +def _canary() -> ModelGatewayCanaryConfig: + return ModelGatewayCanaryConfig( + topic="onex.evt.omnibase-infra.gateway-canary.v1", + cadence_seconds=30, + produce_deadline_seconds=8, + readback_deadline_seconds=12, + ) + + def _config() -> ModelGatewayForwarderConfig: return ModelGatewayForwarderConfig( tenant_identity=ModelGatewayTenantIdentity( @@ -107,6 +117,7 @@ def _config() -> ModelGatewayForwarderConfig: inbound=(INBOUND_TOPIC,), outbound=(OUTBOUND_TOPIC,), ), + canary=_canary(), ) From 97ee827a4ee8fe19b2961b6cd16d27ef0ba20f90 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sun, 9 Aug 2026 12:24:26 -0400 Subject: [PATCH 5/8] fix(OMN-15748): gateway poison-pill quarantine + membership-loss watchdog decode_message() ran before the try/except that nacks, so any undecodable record on a mirrored source topic crashed the process (live-observed 2026-08-08T16:50Z). Nacking it (the ordinary failure path) would also be wrong for a permanently malformed record: nack seeks back to the same offset and re-crashes forever. Quarantine instead: log, best-effort dead-letter via the DLQ topic derived from the record's own topic, commit past the offset, keep the loop alive. Also adds an independent membership-loss watchdog to NodeGatewayDelivery. The live 2026-08-09T10:03:07Z incident was a clean single LeaveGroup with zero exception (aiokafka's client-side max_poll_interval_ms idle-eviction, fired from a separate asyncio task than the poll loop) -- the existing exception-triggered reconnect supervision in runtime/gateway_forwarder.py cannot observe a task that is alive, hung, and never exceptions. The watchdog tracks per-direction last-progress timestamps (updated on poll() completion and deliver_message completion) plus a best-effort consumer.assignment() membership probe, and on staleness force-recreates just the affected direction's transport (a fresh client forces a new group join, sidestepping the permanently-stuck idle clock) and publishes a DEGRADED status. Reuses the previously contract-declared but unwired max_silence_window_seconds config field as the staleness threshold. TDD: both regression tests are RED against pre-fix source (verified via git stash of the source-only diff) and GREEN after. OMN-15748 --- .../event_bus/kafka_transport.py | 19 ++ .../services/service_gateway_delivery.py | 272 +++++++++++++++++- .../test_gateway_delivery_service.py | 196 +++++++++++++ 3 files changed, 478 insertions(+), 9 deletions(-) diff --git a/src/omnibase_infra/event_bus/kafka_transport.py b/src/omnibase_infra/event_bus/kafka_transport.py index 36a87ca16e..311a38387e 100644 --- a/src/omnibase_infra/event_bus/kafka_transport.py +++ b/src/omnibase_infra/event_bus/kafka_transport.py @@ -264,6 +264,25 @@ async def close(self) -> None: if first_error is not None: raise first_error + def has_group_membership(self) -> bool: + """Best-effort local check: does the consumer still hold a partition assignment? + + Not authoritative on its own -- ``assignment()`` reflects the client's + last-known state and can be momentarily stale across a rebalance. Used + as a corroborating signal for the delivery-loop staleness watchdog + (``NodeGatewayDelivery``, OMN-15748/OMN-15690): a silent + ``max_poll_interval_ms`` idle-eviction (aiokafka's client-side + ``GroupCoordinator._heartbeat_routine``) drops group membership with + zero exception raised to the caller, so the watchdog cannot rely on + task failure alone to detect it. + """ + if self._consumer is None: + return False + try: + return bool(self._consumer.assignment()) + except Exception: # noqa: BLE001 — boundary: best-effort probe, never raises + return False + async def caught_up_topics(self, topics: frozenset[str]) -> frozenset[str]: """Return the subset of ``topics`` whose consumer has caught up (LAG=0). diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py index cb4bddfa2f..d325ad0708 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py @@ -5,11 +5,14 @@ from __future__ import annotations import asyncio +import json import logging +import time from collections.abc import Sequence from typing import Literal, Protocol from omnibase_core.models.runtime.model_transport_message import ModelTransportMessage +from omnibase_infra.event_bus.topic_constants import get_dlq_topic_for_original from omnibase_infra.idempotency import ProtocolIdempotencyStore from omnibase_infra.nodes.node_bus_forwarder_effect.models import ( ModelGatewayForwarderConfig, @@ -17,9 +20,12 @@ from omnibase_infra.nodes.node_bus_forwarder_effect.services.service_gateway_forwarder import ( ServiceGatewayForwarder, ) +from omnibase_infra.utils import sanitize_error_message logger = logging.getLogger(__name__) +_DIRECTIONS: tuple[Literal["outbound", "inbound"], ...] = ("outbound", "inbound") + class ProtocolGatewayConsumer(Protocol): """Concrete pull surface used by the Kafka-backed gateway runtime.""" @@ -36,6 +42,39 @@ async def commit(self, message: object) -> None: ... async def nack(self, message: object) -> None: ... +# The delivery loop's real sources (``KafkaTransport``) carry three optional +# capabilities beyond the ``ProtocolGatewayConsumer`` pull surface above: +# ``start``/``close`` (recreate the underlying client -- a fresh group join), +# ``has_group_membership`` (a best-effort local assignment probe), and +# ``send`` (the same object is also a producer, for best-effort +# dead-lettering). None of these are declared as their own ``Protocol`` -- +# the architecture validator enforces one ``Protocol`` class per file, and +# ``ProtocolGatewayConsumer`` above already occupies this file's slot -- so +# they are duck-typed via ``getattr``/``callable`` instead. Absence of any of +# them (e.g. the unit-test fakes that only implement poll/commit/nack) +# degrades gracefully: no dead-letter, no forced recreate, no membership +# signal, never an error. + + +def _build_quarantine_payload( + *, + direction: Literal["outbound", "inbound"], + message: ModelTransportMessage, + error: Exception, +) -> bytes: + """Serialize an undecodable record's forensic context for the DLQ.""" + payload: dict[str, object] = { + "original_topic": message.topic, + "original_partition": message.partition, + "original_offset": message.offset, + "direction": direction, + "failure_class": "gateway_undecodable_record", + "error_type": type(error).__name__, + "error_message": sanitize_error_message(error), + } + return json.dumps(payload).encode("utf-8") + + class NodeGatewayDelivery: """Poll both legs and acknowledge only after durable destination delivery. @@ -63,6 +102,7 @@ def __init__( cloud_consumer: ProtocolGatewayConsumer, idempotency_store: ProtocolIdempotencyStore, poll_timeout_ms: int = 1_000, + watchdog_stale_seconds: float | None = None, ) -> None: self._config = config self._forwarder = forwarder @@ -76,27 +116,62 @@ def __init__( # loop records it. self._delivery_lock = asyncio.Lock() self._tasks: list[asyncio.Task[None]] = [] + self._direction_sources: dict[ + Literal["outbound", "inbound"], ProtocolGatewayConsumer + ] = {"outbound": local_consumer, "inbound": cloud_consumer} + self._direction_tasks: dict[ + Literal["outbound", "inbound"], asyncio.Task[None] + ] = {} + # Membership-loss watchdog state (OMN-15748/OMN-15690). A silent + # aiokafka max_poll_interval_ms idle-eviction raises zero exception, + # so the exception-triggered reconnect supervision in + # runtime/gateway_forwarder.py structurally cannot observe it -- this + # watchdog is the independent detection path. Reuses the + # already-contract-declared but previously unwired + # ``max_silence_window_seconds`` as its staleness threshold. + self._last_progress_monotonic: dict[Literal["outbound", "inbound"], float] = {} + self._membership_lost_streak: dict[Literal["outbound", "inbound"], int] = {} + self._watchdog_degraded: set[Literal["outbound", "inbound"]] = set() + self._watchdog_stale_seconds = ( + watchdog_stale_seconds + if watchdog_stale_seconds is not None + else float(config.max_silence_window_seconds) + ) + self._watchdog_tick_seconds = min(self._watchdog_stale_seconds / 4, 30.0) async def start(self) -> None: """Start one bounded pull loop per direction.""" if self._tasks: return await self._idempotency_store.cleanup_expired(self._retention_seconds) + now = time.monotonic() + self._last_progress_monotonic = {"outbound": now, "inbound": now} + self._membership_lost_streak = {"outbound": 0, "inbound": 0} + self._watchdog_degraded.clear() self._tasks = [ - asyncio.create_task( - self._run_direction("outbound", self._local_consumer), - name="gateway-delivery-outbound", - ), - asyncio.create_task( - self._run_direction("inbound", self._cloud_consumer), - name="gateway-delivery-inbound", - ), + self._spawn_direction_task("outbound"), + self._spawn_direction_task("inbound"), asyncio.create_task( self._run_cleanup_loop(), name="gateway-delivery-dedupe-cleanup", ), + asyncio.create_task( + self._run_watchdog_loop(), + name="gateway-delivery-watchdog", + ), ] + def _spawn_direction_task( + self, direction: Literal["outbound", "inbound"] + ) -> asyncio.Task[None]: + source = self._direction_sources[direction] + task = asyncio.create_task( + self._run_direction(direction, source), + name=f"gateway-delivery-{direction}", + ) + self._direction_tasks[direction] = task + return task + async def wait(self) -> None: """Propagate a delivery-loop failure to the composition root.""" if not self._tasks: @@ -107,6 +182,8 @@ async def stop(self) -> None: """Cancel pull loops without acknowledging any in-flight source record.""" tasks = list(self._tasks) self._tasks.clear() + self._direction_tasks.clear() + self._watchdog_degraded.clear() for task in tasks: task.cancel() if tasks: @@ -129,7 +206,20 @@ async def _deliver_message_locked( message: ModelTransportMessage, ) -> None: """Deliver one record while holding the process-wide delivery lock.""" - envelope = self._forwarder.decode_message(message) + try: + envelope = self._forwarder.decode_message(message) + except asyncio.CancelledError: + raise + except Exception as decode_error: # noqa: BLE001 — boundary: any decode failure is quarantined, never a bare swallow + # A permanently malformed record can never decode no matter how + # many times it is redelivered -- routing it through the nack + # path below would seek back to the same offset and re-crash + # forever (OMN-15748 poison-pill DoS). Quarantine instead: log, + # best-effort dead-letter, commit past it, keep the loop alive. + await self._quarantine_undecodable_message( + direction, source, message, decode_error + ) + return domain = f"gateway:{self._config.tenant_identity.tenant_slug}" try: if direction == "outbound": @@ -189,6 +279,53 @@ async def _deliver_message_locked( ) raise + async def _quarantine_undecodable_message( + self, + direction: Literal["outbound", "inbound"], + source: ProtocolGatewayConsumer, + message: ModelTransportMessage, + error: Exception, + ) -> None: + """Dead-letter an undecodable record and commit past it (never redeliver). + + Commit failure still propagates (uncaught): that is a broker-level + fault, a different failure class already handled by the existing + reconnect-supervision loop in ``runtime/gateway_forwarder.py``. + """ + logger.error( + "Gateway undecodable record quarantined direction=%s source_topic=%s " + "source_partition=%s source_offset=%s error_type=%s", + direction, + message.topic, + message.partition, + message.offset, + type(error).__name__, + exc_info=error, + ) + sender = getattr(source, "send", None) + if callable(sender): + dlq_topic = get_dlq_topic_for_original(message.topic) + if dlq_topic is not None: + try: + await sender( + dlq_topic, + message.key, + _build_quarantine_payload( + direction=direction, message=message, error=error + ), + {"original_topic": message.topic.encode("utf-8")}, + ) + except Exception: + logger.exception( + "Gateway quarantine DLQ publish failed direction=%s " + "source_topic=%s source_partition=%s source_offset=%s", + direction, + message.topic, + message.partition, + message.offset, + ) + await source.commit(message) + async def _run_direction( self, direction: Literal["outbound", "inbound"], @@ -199,8 +336,125 @@ async def _run_direction( max_messages=1, timeout_ms=self._poll_timeout_ms, ) + self._last_progress_monotonic[direction] = time.monotonic() for message in messages: await self.deliver_message(direction, source, message) + self._last_progress_monotonic[direction] = time.monotonic() + + async def _run_watchdog_loop(self) -> None: + """Detect a stalled direction independent of task exceptions. + + Two triggers, either sufficient: (1) no poll/deliver progress for + ``_watchdog_stale_seconds`` -- catches a task alive but hung inside a + stuck ``poll()`` (the live-observed 2026-08-09T10:03:07Z mechanism: + aiokafka's client-side idle-eviction fires a clean LeaveGroup with no + exception, then ``ensure_active_group`` refuses to rejoin while the + idle clock stays past threshold -- which it does forever once the + underlying poll never returns); (2) two consecutive lost-membership + probes -- catches membership silently lost even while poll keeps + returning. Recovery force-recreates just the affected direction's + transport (a fresh client instance re-joins the group from zero) and + publishes a DEGRADED status; a later tick with progress republishes + recovery. + """ + while True: + await asyncio.sleep(self._watchdog_tick_seconds) + now = time.monotonic() + for direction in _DIRECTIONS: + source = self._direction_sources[direction] + elapsed = now - self._last_progress_monotonic.get(direction, now) + membership_lost = self._probe_membership_lost(source) + self._membership_lost_streak[direction] = ( + self._membership_lost_streak.get(direction, 0) + 1 + if membership_lost + else 0 + ) + stale_by_time = elapsed >= self._watchdog_stale_seconds + stale_by_membership = self._membership_lost_streak[direction] >= 2 + if stale_by_time or stale_by_membership: + await self._recover_stalled_direction( + direction, + elapsed_seconds=elapsed, + membership_lost=membership_lost, + ) + elif direction in self._watchdog_degraded: + await self._mark_direction_recovered(direction) + + @staticmethod + def _probe_membership_lost(source: ProtocolGatewayConsumer) -> bool: + prober = getattr(source, "has_group_membership", None) + if not callable(prober): + return False + try: + return not bool(prober()) + except Exception: # noqa: BLE001 — boundary: probe failure counts as lost, fail closed + return True + + async def _recover_stalled_direction( + self, + direction: Literal["outbound", "inbound"], + *, + elapsed_seconds: float, + membership_lost: bool, + ) -> None: + logger.warning( + "Gateway %s delivery loop stalled elapsed_seconds=%.1f " + "membership_lost=%s; forcing recreate + rejoin", + direction, + elapsed_seconds, + membership_lost, + ) + stale_task = self._direction_tasks.get(direction) + if stale_task is not None and not stale_task.done(): + stale_task.cancel() + await asyncio.gather(stale_task, return_exceptions=True) + + source = self._direction_sources[direction] + closer = getattr(source, "close", None) + starter = getattr(source, "start", None) + if callable(closer) and callable(starter): + try: + await closer() + await starter() + except Exception: + logger.exception( + "Gateway %s transport recreate failed; will retry next " + "watchdog tick", + direction, + ) + return + + self._last_progress_monotonic[direction] = time.monotonic() + self._membership_lost_streak[direction] = 0 + new_task = self._spawn_direction_task(direction) + self._tasks = [new_task if t is stale_task else t for t in self._tasks] + if direction not in self._watchdog_degraded: + self._watchdog_degraded.add(direction) + await self._publish_watchdog_status( + "degraded", + detail=( + f"{direction} delivery loop membership-loss recovery " + f"elapsed_seconds={elapsed_seconds:.1f} " + f"membership_lost={membership_lost}" + ), + ) + + async def _mark_direction_recovered( + self, direction: Literal["outbound", "inbound"] + ) -> None: + self._watchdog_degraded.discard(direction) + await self._publish_watchdog_status( + "active", detail=f"{direction} delivery loop recovered" + ) + + async def _publish_watchdog_status( + self, status: Literal["active", "degraded"], *, detail: str + ) -> None: + """Best-effort status publish -- must never itself take down the watchdog.""" + try: + await self._forwarder.publish_status(status, detail=detail) + except Exception: + logger.exception("Gateway watchdog %s status publish failed", status) async def _run_cleanup_loop(self) -> None: cleanup_interval_seconds = min(self._retention_seconds / 4, 3_600) diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py index e509586251..671df5e010 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py @@ -4,6 +4,8 @@ from __future__ import annotations import asyncio +import contextlib +import json from collections.abc import Mapping, Sequence from pathlib import Path from uuid import UUID, uuid4 @@ -81,6 +83,7 @@ def __init__(self, events: list[str], *, fail_commit: bool = False) -> None: self.fail_commit = fail_commit self.committed: list[object] = [] self.nacked: list[object] = [] + self.sent: list[tuple[str, bytes]] = [] async def start(self) -> None: pass @@ -88,6 +91,16 @@ async def start(self) -> None: async def close(self) -> None: pass + async def send( + self, + topic: str, + key: bytes | None, + value: bytes, + headers: Mapping[str, bytes], + ) -> None: + self.events.append("dlq_send") + self.sent.append((topic, value)) + async def poll( self, *, max_messages: int, timeout_ms: int ) -> Sequence[ModelTransportMessage]: @@ -453,3 +466,186 @@ async def test_real_outbound_consumer_loop_does_not_reforward_degraded_status() "DEGRADED status leaked to the cloud leg via the outbound consumer " f"loop re-forwarding the local-bus mirror: {degraded_on_cloud!r}" ) + + +# --- OMN-15748: poison-pill quarantine + membership-loss watchdog ---------- + + +async def test_undecodable_record_is_quarantined_not_crashed() -> None: + """decode_message() failure must be quarantined, never propagate raw. + + Pre-fix: ``_deliver_message_locked`` calls ``decode_message()`` BEFORE + the try/except that nacks, so a malformed record raises straight out of + ``deliver_message`` -- exactly the live 2026-08-08T16:50Z crash trigger + (verbatim pydantic ``ValidationError`` traceback ending in + ``sys.exit(main())``). Nacking it (the ordinary failure path) would also + be wrong: a permanently malformed record redelivers via nack's ``seek`` + and re-crashes forever. It must be logged, dead-lettered, and committed + past (skipped) so the consumer loop keeps making forward progress. + """ + events: list[str] = [] + source = _Source(events) + store = _RecordingStore(events) + delivery, cloud_bus = _delivery(events, source, store) + poison = ModelTransportMessage( + topic=OUTBOUND_TOPIC, + partition=0, + offset=3, + key=b"tenant-key", + value=b"SYNTHETIC-not-valid-json{{{", + headers={}, + ack_token=(OUTBOUND_TOPIC, 0, 3), + ) + + # Must not raise -- this is the process-crash regression. + await delivery.deliver_message("outbound", source, poison) # type: ignore[arg-type] + + assert source.committed == [poison], ( + "poison record must be committed (skipped), not left uncommitted" + ) + assert source.nacked == [], ( + "must not nack a permanently undecodable record -- nack seeks back " + "to the same offset and re-crashes forever (poison-loop)" + ) + assert cloud_bus.sent == [], ( + "an undecodable record must never reach the destination" + ) + assert source.sent, "undecodable record must be dead-lettered, not silently dropped" + dlq_topic, dlq_value = source.sent[0] + assert dlq_topic == "onex.dlq.omnibase-infra.events.v1" + payload = json.loads(dlq_value) + assert payload["original_topic"] == OUTBOUND_TOPIC + assert payload["original_partition"] == 0 + assert payload["original_offset"] == 3 + assert payload["direction"] == "outbound" + assert payload["failure_class"] == "gateway_undecodable_record" + + +class _StallThenRecoverSource: + """Reproduces the live 2026-08-09T10:03:07Z silent-eviction mechanism. + + aiokafka's client-side ``max_poll_interval_ms`` idle-eviction fires from + a SEPARATE asyncio task (``GroupCoordinator._heartbeat_routine``) than + the poll loop -- it force-leaves the group and the poll loop's own + ``poll()`` call simply never returns and never raises. The only way out + is a brand-new client instance (a fresh group join), which is exactly + what ``close()`` + ``start()`` on a real ``KafkaTransport`` does. This + fake models that: the first ``poll()`` hangs forever with zero + exception; only after ``close()``/``start()`` are called does it behave + like a healthy transport again (real suspension via ``asyncio.sleep``, + not a busy-spin return -- the same discipline + ``test_real_outbound_consumer_loop_does_not_reforward_degraded_status`` + documents above). + """ + + def __init__(self) -> None: + self.poll_calls = 0 + self.start_calls = 0 + self.close_calls = 0 + self.committed: list[object] = [] + self.nacked: list[object] = [] + self._recreated = False + self._stuck: asyncio.Event = asyncio.Event() + + async def start(self) -> None: + self.start_calls += 1 + self._recreated = True + + async def close(self) -> None: + self.close_calls += 1 + + def has_group_membership(self) -> bool: + return self._recreated + + async def poll( + self, *, max_messages: int, timeout_ms: int + ) -> Sequence[ModelTransportMessage]: + self.poll_calls += 1 + if not self._recreated: + await self._stuck.wait() # never set: permanent stall, no exception + return [] # pragma: no cover - unreachable in this test + await asyncio.sleep(timeout_ms / 1000) + return [] + + async def commit(self, message: object) -> None: + self.committed.append(message) + + async def nack(self, message: object) -> None: + self.nacked.append(message) + + +async def test_membership_loss_watchdog_recreates_and_recovers() -> None: + """OMN-15748/OMN-15690: silent LeaveGroup (zero exception) must recover. + + Pre-fix: there is no watchdog at all. The only recovery mechanism is + ``runtime/gateway_forwarder.py``'s ``_supervise_gateway_delivery``, which + is exception-triggered off ``delivery.wait()`` -- structurally unable to + observe a task that is alive and hung, never done, never exceptioned. + That is the adjudicated live failure mode: the process is left with + MEMBERS=0 on one direction forever, no rejoin, no restart, no alert. + """ + local_bus = _RecordingBus([]) + cloud_bus = _RecordingBus([]) + config = _config().model_copy( + update={ + "heartbeat_interval_seconds": 1, + "max_silence_window_seconds": 2, + "mirror_topics": ModelGatewayMirrorTopics( + inbound=("onex.cmd.omnibase-infra.delegation-request.v1",), + outbound=(OUTBOUND_TOPIC, HEARTBEAT_TOPIC), + ), + } + ) + forwarder = ServiceGatewayForwarder( + config=config, + local_bus=local_bus, # type: ignore[arg-type] + cloud_bus=cloud_bus, # type: ignore[arg-type] + ) + stalled_outbound = _StallThenRecoverSource() + idle_inbound = _Source([]) + delivery = NodeGatewayDelivery( + config=config, + forwarder=forwarder, + local_consumer=stalled_outbound, # type: ignore[arg-type] + cloud_consumer=idle_inbound, # type: ignore[arg-type] + idempotency_store=StoreIdempotencyInmemory(), + poll_timeout_ms=50, + ) + + await delivery.start() + try: + for _ in range(200): + if stalled_outbound.start_calls: + break + await asyncio.sleep(0.02) + assert stalled_outbound.start_calls >= 1, ( + "watchdog never force-recreated the stalled transport -- the " + "silent-eviction mode was never detected" + ) + assert stalled_outbound.close_calls >= 1 + + # Prove actual recovery (forward progress resumes), not merely a + # crash-free wait. + polls_at_recreate = stalled_outbound.poll_calls + for _ in range(100): + if stalled_outbound.poll_calls > polls_at_recreate: + break + await asyncio.sleep(0.02) + assert stalled_outbound.poll_calls > polls_at_recreate, ( + "recreated direction never resumed polling" + ) + + degraded = [value for _topic, value in local_bus.sent if b'"degraded"' in value] + assert degraded, "membership-loss recovery must publish a DEGRADED status" + + # No task death: the exception-triggered supervision surface + # (delivery.wait()) must never see this recovery -- it is handled + # entirely inside the watchdog. + wait_task = asyncio.create_task(delivery.wait()) + done, _pending = await asyncio.wait({wait_task}, timeout=0.2) + assert wait_task not in done, "delivery loop task must not have died" + wait_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await wait_task + finally: + await delivery.stop() From 309d640214dbb7968c3bb084dfdd40d4e0fa1943 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sun, 9 Aug 2026 14:49:35 -0400 Subject: [PATCH 6/8] fix(OMN-15748): consumer-scoped watchdog recovery, no CancelledError escape Fixes two verifier-confirmed defects in the membership-loss watchdog (mergesweep-0809-poisonpath-verify): - wait()/_run_direction: a watchdog stale_task.cancel() made the supervisor's delivery.wait() (a frozen asyncio.gather snapshot) observe a cancelled child and propagate CancelledError -- a BaseException that escapes gateway_forwarder.py's except Exception and kills the process on every recovery. _run_direction now catches CancelledError and returns cleanly when its direction is watchdog-recovering; wait() re-reads the live task set after every completion instead of gathering a one-time snapshot, so a swapped-in replacement task stays supervised. - Recovery no longer calls close()+start() (which stops BOTH the consumer and producer on a KafkaTransport instance shared across directions -- one direction's consumer is the same object the other direction's publish and status/heartbeat emission use as a producer). Adds KafkaTransport.restart_consumer(), a consumer-only recreate that never touches the producer; _recover_stalled_direction now calls it exclusively. De-vacuums the watchdog regression test (arms delivery.wait() before the stall, matching production ordering) and adds direct KafkaTransport-level coverage proving the producer is untouched across a restart_consumer() call. --- .../event_bus/kafka_transport.py | 110 +++++++++++----- .../services/service_gateway_delivery.py | 118 ++++++++++++++---- ...st_kafka_transport_protocol_conformance.py | 84 +++++++++++++ .../test_gateway_delivery_service.py | 59 +++++---- 4 files changed, 292 insertions(+), 79 deletions(-) diff --git a/src/omnibase_infra/event_bus/kafka_transport.py b/src/omnibase_infra/event_bus/kafka_transport.py index 311a38387e..89235ce61a 100644 --- a/src/omnibase_infra/event_bus/kafka_transport.py +++ b/src/omnibase_infra/event_bus/kafka_transport.py @@ -170,43 +170,91 @@ async def start(self) -> None: await self._producer.start() if self._topics: - # Group ``subscribe`` (topics passed positionally): the consumer joins - # the group and, on join, NATIVELY resumes each partition from its - # group-committed offset (a fresh group starts at ``auto_offset_reset``). - # This is what makes "restart resumes from the committed offset" and - # "uncommitted offsets redeliver on restart" hold, with no manual - # ``seek`` dance. A single transport instance is the sole group member, - # so it is assigned every partition of its topics — matching the unified - # runtime's single-poll-loop-per-topic-set model (single-owner-per-topic - # is the S6 boot invariant, R1). The one-time join latency is absorbed by - # ``_prime`` below so the runtime's first ``poll`` still returns promptly. - self._consumer = AIOKafkaConsumer( - *self._topics, - bootstrap_servers=self._config.bootstrap_servers, - group_id=self._group, - # FORCED for the new transport consumers (plan S3): the runtime, not - # the client, decides when an offset is durable. Legacy push - # consumers keep their per-consumer config setting untouched. - enable_auto_commit=False, - auto_offset_reset=self._auto_offset_reset, - session_timeout_ms=self._config.session_timeout_ms, - heartbeat_interval_ms=self._config.heartbeat_interval_ms, - max_poll_interval_ms=self._config.max_poll_interval_ms, - retry_backoff_ms=self._config.reconnect_backoff_ms, - **self._client_version_kwargs(AIOKafkaConsumer), - **self._auth_kwargs(), - ) - await self._consumer.start() - # Trigger the group join + first fetch now, buffering the first batch, so - # the runtime's first poll() returns the available records instead of - # racing the lazy rebalance. - await self._prime(self._consumer) + await self._start_consumer() except BaseException: await self.close() raise self._started = True + async def _start_consumer(self) -> None: + """Construct, start, and prime a fresh consumer bound to ``self._topics``. + + Factored out of ``start()`` so ``restart_consumer`` can rejoin the group + from zero (a brand-new client instance -- the only way to recover from a + silent ``max_poll_interval_ms`` idle-eviction, see + ``has_group_membership``) without touching ``self._producer``. A single + ``KafkaTransport`` instance can back both one gateway direction's + consumer AND the other direction's (plus status/heartbeat) producer, so + a watchdog-triggered consumer recreate (``NodeGatewayDelivery``, + OMN-15748/OMN-15690) must never go through ``close()`` + ``start()``, + which stops both clients. + """ + # Group ``subscribe`` (topics passed positionally): the consumer joins + # the group and, on join, NATIVELY resumes each partition from its + # group-committed offset (a fresh group starts at ``auto_offset_reset``). + # This is what makes "restart resumes from the committed offset" and + # "uncommitted offsets redeliver on restart" hold, with no manual + # ``seek`` dance. A single transport instance is the sole group member, + # so it is assigned every partition of its topics — matching the unified + # runtime's single-poll-loop-per-topic-set model (single-owner-per-topic + # is the S6 boot invariant, R1). The one-time join latency is absorbed by + # ``_prime`` below so the runtime's first ``poll`` still returns promptly. + self._consumer = AIOKafkaConsumer( + *self._topics, + bootstrap_servers=self._config.bootstrap_servers, + group_id=self._group, + # FORCED for the new transport consumers (plan S3): the runtime, not + # the client, decides when an offset is durable. Legacy push + # consumers keep their per-consumer config setting untouched. + enable_auto_commit=False, + auto_offset_reset=self._auto_offset_reset, + session_timeout_ms=self._config.session_timeout_ms, + heartbeat_interval_ms=self._config.heartbeat_interval_ms, + max_poll_interval_ms=self._config.max_poll_interval_ms, + retry_backoff_ms=self._config.reconnect_backoff_ms, + **self._client_version_kwargs(AIOKafkaConsumer), + **self._auth_kwargs(), + ) + await self._consumer.start() + # Trigger the group join + first fetch now, buffering the first batch, so + # the runtime's first poll() returns the available records instead of + # racing the lazy rebalance. + await self._prime(self._consumer) + + async def restart_consumer(self) -> None: + """Recreate ONLY the consumer-side client; the shared producer stays up. + + A gateway direction's membership-loss watchdog recovery + (``NodeGatewayDelivery._recover_stalled_direction``, + OMN-15748/OMN-15690) must rejoin the group from a fresh client without + killing the producer this same instance also serves for the OTHER + direction's forward-publish and every status/heartbeat publish -- + ``close()`` + ``start()`` stops both clients and would silently break + those for the duration of the recreate. Requires an assigned + ``topics`` set (a producer-only transport has nothing to restart). + """ + if not self._topics: + context = ModelInfraErrorContext.with_correlation( + transport_type=EnumInfraTransportType.KAFKA, + operation="restart_consumer", + target_name="kafka_transport", + ) + raise ProtocolConfigurationError( + "KafkaTransport.restart_consumer() requires topics=[...]; " + "this instance is producer-only.", + context=context, + parameter="topics", + value=list(self._topics), + ) + self._buffer.clear() + if self._consumer is not None: + try: + await self._consumer.stop() + finally: + self._consumer = None + await self._start_consumer() + async def _prime(self, consumer: AIOKafkaConsumer) -> None: """Eagerly fetch ONE batch into ``self._buffer`` after a position change. diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py index d325ad0708..a70265d1cb 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py @@ -44,16 +44,25 @@ async def nack(self, message: object) -> None: ... # The delivery loop's real sources (``KafkaTransport``) carry three optional # capabilities beyond the ``ProtocolGatewayConsumer`` pull surface above: -# ``start``/``close`` (recreate the underlying client -- a fresh group join), -# ``has_group_membership`` (a best-effort local assignment probe), and -# ``send`` (the same object is also a producer, for best-effort -# dead-lettering). None of these are declared as their own ``Protocol`` -- -# the architecture validator enforces one ``Protocol`` class per file, and -# ``ProtocolGatewayConsumer`` above already occupies this file's slot -- so -# they are duck-typed via ``getattr``/``callable`` instead. Absence of any of -# them (e.g. the unit-test fakes that only implement poll/commit/nack) -# degrades gracefully: no dead-letter, no forced recreate, no membership -# signal, never an error. +# ``restart_consumer`` (recreate ONLY the consumer-side client -- a fresh +# group join -- without touching the shared producer another direction and +# status/heartbeat publishing may also depend on; see +# ``KafkaTransport.restart_consumer``), ``has_group_membership`` (a +# best-effort local assignment probe), and ``send`` (the same object is also +# a producer, for best-effort dead-lettering). None of these are declared as +# their own ``Protocol`` -- the architecture validator enforces one +# ``Protocol`` class per file, and ``ProtocolGatewayConsumer`` above already +# occupies this file's slot -- so they are duck-typed via +# ``getattr``/``callable`` instead. Absence of any of them (e.g. the +# unit-test fakes that only implement poll/commit/nack) degrades gracefully: +# no dead-letter, no forced recreate, no membership signal, never an error. +# +# NOTE: watchdog recovery deliberately does NOT use ``close()``/``start()`` +# (OMN-15748). Those stop/start BOTH the consumer and the producer, and a +# single ``KafkaTransport`` instance backs one direction's consumer AND the +# other direction's outbound publish (plus status/heartbeat) -- see +# ``runtime/gateway_forwarder.py``'s ``local_bus``/``cloud_bus`` wiring. +# ``restart_consumer`` is the consumer-scoped alternative. def _build_quarantine_payload( @@ -132,6 +141,14 @@ def __init__( self._last_progress_monotonic: dict[Literal["outbound", "inbound"], float] = {} self._membership_lost_streak: dict[Literal["outbound", "inbound"], int] = {} self._watchdog_degraded: set[Literal["outbound", "inbound"]] = set() + # Directions currently mid-recovery: ``_run_direction`` checks this to + # tell a watchdog-initiated ``stale_task.cancel()`` apart from a real + # shutdown/failure cancel, so it can return cleanly instead of + # re-raising CancelledError. Without this, the cancelled task's + # CancelledError would surface through ``wait()`` (a BaseException, + # escaping ``except Exception`` in the composition root's supervisor) + # and kill the whole process on every watchdog recovery (OMN-15748). + self._watchdog_recovering: set[Literal["outbound", "inbound"]] = set() self._watchdog_stale_seconds = ( watchdog_stale_seconds if watchdog_stale_seconds is not None @@ -173,10 +190,41 @@ def _spawn_direction_task( return task async def wait(self) -> None: - """Propagate a delivery-loop failure to the composition root.""" + """Propagate a delivery-loop failure to the composition root. + + Awaits the LIVE task set, not a one-time snapshot: a watchdog-initiated + direction recovery (``_recover_stalled_direction``) replaces that + direction's entry in ``self._tasks`` in place with a fresh task, and + this loop re-reads ``self._tasks`` after every completion so it picks + up the replacement and keeps supervising it. The superseded task's own + completion is a clean return (``_run_direction``'s + ``_watchdog_recovering`` check), never a propagated CancelledError, so + a watchdog recovery is invisible here -- it neither raises nor ends + ``wait()`` (OMN-15748; a plain ``asyncio.gather(*self._tasks)`` over a + frozen snapshot previously re-raised the cancelled task's + CancelledError from this call, which the composition root's + exception-triggered supervisor cannot distinguish from a real fault). + """ if not self._tasks: raise RuntimeError("gateway delivery node is not started") - await asyncio.gather(*self._tasks) + watched: set[asyncio.Task[None]] = set(self._tasks) + while watched: + done, watched = await asyncio.wait( + watched, return_when=asyncio.FIRST_COMPLETED + ) + for task in done: + if task.cancelled(): + # Only a watchdog-initiated recovery (or a stop() call + # racing this loop) cancels a tracked task; either way it + # is not a supervised failure. + continue + exc = task.exception() + if exc is not None: + raise exc + # Pick up any watchdog-spawned replacement task not yet tracked. + watched |= { + task for task in self._tasks if task not in watched and not task.done() + } async def stop(self) -> None: """Cancel pull loops without acknowledging any in-flight source record.""" @@ -331,15 +379,25 @@ async def _run_direction( direction: Literal["outbound", "inbound"], source: ProtocolGatewayConsumer, ) -> None: - while True: - messages = await source.poll( - max_messages=1, - timeout_ms=self._poll_timeout_ms, - ) - self._last_progress_monotonic[direction] = time.monotonic() - for message in messages: - await self.deliver_message(direction, source, message) + try: + while True: + messages = await source.poll( + max_messages=1, + timeout_ms=self._poll_timeout_ms, + ) self._last_progress_monotonic[direction] = time.monotonic() + for message in messages: + await self.deliver_message(direction, source, message) + self._last_progress_monotonic[direction] = time.monotonic() + except asyncio.CancelledError: + if direction in self._watchdog_recovering: + # Watchdog-initiated: ``_recover_stalled_direction`` cancelled + # this task on purpose to swap in a fresh consumer. Return + # cleanly rather than re-raise so this completion registers + # as ordinary bookkeeping to ``wait()``, never a propagated + # CancelledError (OMN-15748). + return + raise async def _run_watchdog_loop(self) -> None: """Detect a stalled direction independent of task exceptions. @@ -406,16 +464,22 @@ async def _recover_stalled_direction( ) stale_task = self._direction_tasks.get(direction) if stale_task is not None and not stale_task.done(): - stale_task.cancel() - await asyncio.gather(stale_task, return_exceptions=True) + self._watchdog_recovering.add(direction) + try: + stale_task.cancel() + await asyncio.gather(stale_task, return_exceptions=True) + finally: + self._watchdog_recovering.discard(direction) source = self._direction_sources[direction] - closer = getattr(source, "close", None) - starter = getattr(source, "start", None) - if callable(closer) and callable(starter): + # Consumer-scoped recreate ONLY -- never close()/start(), which would + # also stop the producer this same transport instance may serve for + # the OTHER direction's forward-publish and status/heartbeat publish + # (OMN-15748). + restarter = getattr(source, "restart_consumer", None) + if callable(restarter): try: - await closer() - await starter() + await restarter() except Exception: logger.exception( "Gateway %s transport recreate failed; will retry next " diff --git a/tests/unit/event_bus/test_kafka_transport_protocol_conformance.py b/tests/unit/event_bus/test_kafka_transport_protocol_conformance.py index e7e9eb7e9d..a8be9d53da 100644 --- a/tests/unit/event_bus/test_kafka_transport_protocol_conformance.py +++ b/tests/unit/event_bus/test_kafka_transport_protocol_conformance.py @@ -182,3 +182,87 @@ async def stop(self) -> None: assert transport._consumer is None assert transport._producer is None assert transport._started is False + + +@pytest.mark.asyncio +async def test_restart_consumer_rejects_producer_only_instance( + unit_bootstrap: str, +) -> None: + # OMN-15748: a producer-only transport (no assigned topics) has nothing + # for restart_consumer() to recreate -- fail fast rather than silently + # no-op. + transport = KafkaTransport.from_bootstrap(unit_bootstrap) + + with pytest.raises(ProtocolConfigurationError, match="producer-only"): + await transport.restart_consumer() + + +@pytest.mark.asyncio +async def test_restart_consumer_recreates_consumer_without_touching_producer( + monkeypatch: pytest.MonkeyPatch, unit_bootstrap: str, unit_topic: str +) -> None: + """OMN-15748: watchdog recovery must not call close()/start() -- that stops + the shared producer another gateway direction (and status/heartbeat) still + depends on. restart_consumer() must stop only the old consumer, start a + fresh one, and never touch the producer instance at all. + """ + consumer_events: list[str] = [] + consumer_instances: list[object] = [] + + class FakeProducer: + def __init__(self, **_kwargs: object) -> None: + self.stop_calls = 0 + + async def start(self) -> None: + pass + + async def stop(self) -> None: + self.stop_calls += 1 + + class FakeConsumer: + def __init__(self, *_topics: str, **_kwargs: object) -> None: + consumer_instances.append(self) + consumer_events.append("created") + + async def start(self) -> None: + consumer_events.append("started") + + async def stop(self) -> None: + consumer_events.append("stopped") + + def assignment(self) -> list[object]: + return [object()] + + async def getmany( + self, *, timeout_ms: int, max_records: int + ) -> dict[object, list[object]]: + return {} + + monkeypatch.setattr(kafka_transport_module, "AIOKafkaProducer", FakeProducer) + monkeypatch.setattr(kafka_transport_module, "AIOKafkaConsumer", FakeConsumer) + + transport = KafkaTransport.from_bootstrap( + unit_bootstrap, group="unit-conformance", topics=[unit_topic] + ) + await transport.start() + original_producer = transport._producer + original_consumer = transport._consumer + assert consumer_events == ["created", "started"] + + await transport.restart_consumer() + + assert transport._producer is original_producer, ( + "restart_consumer() must never replace or stop the shared producer" + ) + assert original_producer.stop_calls == 0 # type: ignore[union-attr] + assert transport._consumer is not original_consumer, ( + "restart_consumer() must construct a brand-new consumer client" + ) + assert len(consumer_instances) == 2 + assert consumer_events == [ + "created", + "started", + "stopped", + "created", + "started", + ] diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py index 671df5e010..f1d1409d93 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py @@ -528,32 +528,34 @@ class _StallThenRecoverSource: a SEPARATE asyncio task (``GroupCoordinator._heartbeat_routine``) than the poll loop -- it force-leaves the group and the poll loop's own ``poll()`` call simply never returns and never raises. The only way out - is a brand-new client instance (a fresh group join), which is exactly - what ``close()`` + ``start()`` on a real ``KafkaTransport`` does. This - fake models that: the first ``poll()`` hangs forever with zero - exception; only after ``close()``/``start()`` are called does it behave - like a healthy transport again (real suspension via ``asyncio.sleep``, - not a busy-spin return -- the same discipline + is a brand-new consumer client instance (a fresh group join), which is + exactly what ``restart_consumer()`` on a real ``KafkaTransport`` does. + This fake deliberately implements ONLY ``restart_consumer`` -- no + ``close``/``start`` at all -- so a regression back to the old + ``close()``/``start()`` recovery path fails loudly (``AttributeError`` + via ``getattr``'s ``callable`` guard silently no-op'ing recovery, + caught by the polls-resume assertion below) rather than silently + reintroducing the shared-producer-close bug (OMN-15748 finding ii): the + first ``poll()`` hangs forever with zero exception; only after + ``restart_consumer()`` is called does it behave like a healthy transport + again (real suspension via ``asyncio.sleep``, not a busy-spin return -- + the same discipline ``test_real_outbound_consumer_loop_does_not_reforward_degraded_status`` documents above). """ def __init__(self) -> None: self.poll_calls = 0 - self.start_calls = 0 - self.close_calls = 0 + self.restart_calls = 0 self.committed: list[object] = [] self.nacked: list[object] = [] self._recreated = False self._stuck: asyncio.Event = asyncio.Event() - async def start(self) -> None: - self.start_calls += 1 + async def restart_consumer(self) -> None: + self.restart_calls += 1 self._recreated = True - async def close(self) -> None: - self.close_calls += 1 - def has_group_membership(self) -> bool: return self._recreated @@ -583,6 +585,15 @@ async def test_membership_loss_watchdog_recreates_and_recovers() -> None: observe a task that is alive and hung, never done, never exceptioned. That is the adjudicated live failure mode: the process is left with MEMBERS=0 on one direction forever, no rejoin, no restart, no alert. + + ``delivery.wait()`` is armed BEFORE the stall/recovery happens (matching + production ordering: the composition root's supervisor enters and calls + ``delivery.wait()`` before any fault occurs) so this test actually + exercises the watchdog-vs-wait() race, not just the watchdog in + isolation. Pre the ``NodeGatewayDelivery.wait()``/``_run_direction`` + fix, this would fail: the watchdog's ``stale_task.cancel()`` makes the + still-armed ``wait()``'s ``asyncio.gather``/``asyncio.wait`` observe a + cancelled task and propagate ``CancelledError`` out of ``wait_task``. """ local_bus = _RecordingBus([]) cloud_bus = _RecordingBus([]) @@ -614,15 +625,18 @@ async def test_membership_loss_watchdog_recreates_and_recovers() -> None: await delivery.start() try: + # Arm wait() FIRST, exactly as the composition root's supervisor + # does at the top of its loop -- before the stall/recovery below. + wait_task = asyncio.create_task(delivery.wait()) + for _ in range(200): - if stalled_outbound.start_calls: + if stalled_outbound.restart_calls: break await asyncio.sleep(0.02) - assert stalled_outbound.start_calls >= 1, ( + assert stalled_outbound.restart_calls >= 1, ( "watchdog never force-recreated the stalled transport -- the " "silent-eviction mode was never detected" ) - assert stalled_outbound.close_calls >= 1 # Prove actual recovery (forward progress resumes), not merely a # crash-free wait. @@ -639,13 +653,16 @@ async def test_membership_loss_watchdog_recreates_and_recovers() -> None: assert degraded, "membership-loss recovery must publish a DEGRADED status" # No task death: the exception-triggered supervision surface - # (delivery.wait()) must never see this recovery -- it is handled - # entirely inside the watchdog. - wait_task = asyncio.create_task(delivery.wait()) + # (delivery.wait(), already armed above) must survive the recovery + # -- it is handled entirely inside the watchdog and must never + # surface a CancelledError or any other exception here. done, _pending = await asyncio.wait({wait_task}, timeout=0.2) - assert wait_task not in done, "delivery loop task must not have died" + assert wait_task not in done, ( + "delivery.wait() must not have completed -- a watchdog recovery " + "must be invisible to the exception-triggered supervision path" + ) + finally: wait_task.cancel() with contextlib.suppress(asyncio.CancelledError): await wait_task - finally: await delivery.stop() From b5e9f3154563575769c9043754ced33cee24ad13 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sun, 9 Aug 2026 16:02:42 -0400 Subject: [PATCH 7/8] fix(OMN-15748): address CodeRabbit findings on watchdog wait() + log sanitization Two live (non-outdated) CodeRabbit threads on infra#2700: - wait()'s asyncio.wait was unbounded, waking only on a tracked task's completion. _recover_stalled_direction splices the replacement task into self._tasks strictly AFTER the stale task has already finished and this loop has already woken/rebuilt `watched` from the stale-only list -- so an unbounded wait then blocks on the remaining old tasks with no further wakeup until one of THEM completes. A real exception on the recovered direction would go unsupervised (only surfacing as an "exception was never retrieved" warning at GC). Fixed by bounding the wait with _watchdog_tick_seconds so the loop always revisits self._tasks, and by also picking up an already-completed-with-exception replacement on that rescan. Added test_watchdog_replacement_task_failure_still_reaches_wait, verified RED against the unbounded wait() (reverted locally, re-applied after) and GREEN after. - _quarantine_undecodable_message logged the raw exception via exc_info=error, bypassing the sanitize_error_message() the DLQ payload path already uses a few lines up. A pydantic ValidationError from decode_message() embeds the offending record content, which is untrusted tenant bus traffic that can carry PII/credential-bearing values. Now logs the sanitized message, matching the DLQ path. --- .../services/service_gateway_delivery.py | 40 +++++++-- .../test_gateway_delivery_service.py | 86 +++++++++++++++++++ 2 files changed, 117 insertions(+), 9 deletions(-) diff --git a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py index a70265d1cb..87eb7bb188 100644 --- a/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py +++ b/src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py @@ -195,8 +195,8 @@ async def wait(self) -> None: Awaits the LIVE task set, not a one-time snapshot: a watchdog-initiated direction recovery (``_recover_stalled_direction``) replaces that direction's entry in ``self._tasks`` in place with a fresh task, and - this loop re-reads ``self._tasks`` after every completion so it picks - up the replacement and keeps supervising it. The superseded task's own + this loop re-reads ``self._tasks`` after every wake so it picks up the + replacement and keeps supervising it. The superseded task's own completion is a clean return (``_run_direction``'s ``_watchdog_recovering`` check), never a propagated CancelledError, so a watchdog recovery is invisible here -- it neither raises nor ends @@ -204,13 +204,28 @@ async def wait(self) -> None: frozen snapshot previously re-raised the cancelled task's CancelledError from this call, which the composition root's exception-triggered supervisor cannot distinguish from a real fault). + + The ``asyncio.wait`` below is bounded by ``_watchdog_tick_seconds`` + rather than waiting unboundedly for the next completion (CodeRabbit + finding on this PR). ``_recover_stalled_direction`` splices the + replacement task into ``self._tasks`` only AFTER the stale task has + already finished and this loop has already woken and rebuilt + ``watched`` from the (still stale-only) task list -- an unbounded + wait would then block on the remaining old tasks with no further + wakeup until one of THEM completes, so the replacement would never + be picked up and a real exception on the recovered direction would + go unsupervised. The bound guarantees this loop revisits + ``self._tasks`` at least once per watchdog tick even with zero new + completions. """ if not self._tasks: raise RuntimeError("gateway delivery node is not started") watched: set[asyncio.Task[None]] = set(self._tasks) while watched: done, watched = await asyncio.wait( - watched, return_when=asyncio.FIRST_COMPLETED + watched, + timeout=self._watchdog_tick_seconds, + return_when=asyncio.FIRST_COMPLETED, ) for task in done: if task.cancelled(): @@ -221,10 +236,17 @@ async def wait(self) -> None: exc = task.exception() if exc is not None: raise exc - # Pick up any watchdog-spawned replacement task not yet tracked. - watched |= { - task for task in self._tasks if task not in watched and not task.done() - } + # Pick up any watchdog-spawned replacement task not yet tracked -- + # including one that raced ahead and already finished (with an + # exception) before this rescan, which the bounded timeout above + # guarantees we reach even without a completion to wake us. + for task in self._tasks: + if task in watched: + continue + if not task.done() or ( + not task.cancelled() and task.exception() is not None + ): + watched.add(task) async def stop(self) -> None: """Cancel pull loops without acknowledging any in-flight source record.""" @@ -342,13 +364,13 @@ async def _quarantine_undecodable_message( """ logger.error( "Gateway undecodable record quarantined direction=%s source_topic=%s " - "source_partition=%s source_offset=%s error_type=%s", + "source_partition=%s source_offset=%s error_type=%s error=%s", direction, message.topic, message.partition, message.offset, type(error).__name__, - exc_info=error, + sanitize_error_message(error), ) sender = getattr(source, "send", None) if callable(sender): diff --git a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py index f1d1409d93..2d97c49f16 100644 --- a/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py +++ b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py @@ -666,3 +666,89 @@ async def test_membership_loss_watchdog_recreates_and_recovers() -> None: with contextlib.suppress(asyncio.CancelledError): await wait_task await delivery.stop() + + +class _StallThenFailSource: + """Like ``_StallThenRecoverSource``, but the RECREATED consumer then hits a + genuine (non-cancellation) failure -- reproducing the CodeRabbit-flagged + gap on this PR: a watchdog-spawned replacement task must stay under + ``wait()``'s supervision, not just the original stale task. + """ + + def __init__(self) -> None: + self.poll_calls = 0 + self.restart_calls = 0 + self._recreated = False + self._stuck: asyncio.Event = asyncio.Event() + + async def restart_consumer(self) -> None: + self.restart_calls += 1 + self._recreated = True + + def has_group_membership(self) -> bool: + return self._recreated + + async def poll( + self, *, max_messages: int, timeout_ms: int + ) -> Sequence[ModelTransportMessage]: + self.poll_calls += 1 + if not self._recreated: + await self._stuck.wait() # never set: permanent stall, no exception + return [] # pragma: no cover - unreachable in this test + # First poll after recreation still needs to yield once so the + # watchdog's spawn/splice of the replacement task can be observed + # settling before the hard failure below. + await asyncio.sleep(0) + raise RuntimeError("synthetic hard failure on recovered direction") + + async def commit(self, message: object) -> None: # pragma: no cover + pass + + async def nack(self, message: object) -> None: # pragma: no cover + pass + + +async def test_watchdog_replacement_task_failure_still_reaches_wait() -> None: + """CodeRabbit finding on this PR: a real exception on the WATCHDOG-SPAWNED + replacement task must still surface through ``wait()``, not only the + original (superseded) task's completion. Pre-fix (unbounded + ``asyncio.wait`` with no re-scan trigger absent a completion of an + already-``watched`` task), the replacement task raising here would go + unsupervised: the composition root's ``_supervise_gateway_delivery`` + would never observe the fault. + """ + local_bus = _RecordingBus([]) + cloud_bus = _RecordingBus([]) + config = _config().model_copy( + update={ + "heartbeat_interval_seconds": 1, + "max_silence_window_seconds": 2, + "mirror_topics": ModelGatewayMirrorTopics( + inbound=("onex.cmd.omnibase-infra.delegation-request.v1",), + outbound=(OUTBOUND_TOPIC, HEARTBEAT_TOPIC), + ), + } + ) + forwarder = ServiceGatewayForwarder( + config=config, + local_bus=local_bus, # type: ignore[arg-type] + cloud_bus=cloud_bus, # type: ignore[arg-type] + ) + stalled_outbound = _StallThenFailSource() + idle_inbound = _Source([]) + delivery = NodeGatewayDelivery( + config=config, + forwarder=forwarder, + local_consumer=stalled_outbound, # type: ignore[arg-type] + cloud_consumer=idle_inbound, # type: ignore[arg-type] + idempotency_store=StoreIdempotencyInmemory(), + poll_timeout_ms=50, + ) + + await delivery.start() + wait_task = asyncio.create_task(delivery.wait()) + try: + with pytest.raises(RuntimeError, match="synthetic hard failure"): + await asyncio.wait_for(wait_task, timeout=10) + finally: + await delivery.stop() From 993d805f45048475fe646c0eab2aff54bcd05252 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sun, 9 Aug 2026 17:41:50 -0400 Subject: [PATCH 8/8] test(OMN-15748): harden test_monitor_logs.py subprocess timeouts against CI fleet contention The self-hosted CI fleet runs many concurrent pytest-split shards per job; interpreter spawn/import can occasionally queue behind that CPU contention. A hardcoded 10s subprocess.run(..., timeout=10) tripped this twice live on unrelated PRs' Split 12/15 (infra#2697 and this PR, both times as three subprocess.TimeoutExpired failures at exactly the 10s bound, unrelated to either PR's own diff -- this file isn't touched by either PR's product diff). Bumped to a shared _SUBPROCESS_TIMEOUT_SECONDS = 30 constant across all three call sites: still tight enough to catch a genuine hang, with real headroom against fleet contention. --- tests/unit/scripts/test_monitor_logs.py | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/tests/unit/scripts/test_monitor_logs.py b/tests/unit/scripts/test_monitor_logs.py index b55ef867bc..d51148fe59 100644 --- a/tests/unit/scripts/test_monitor_logs.py +++ b/tests/unit/scripts/test_monitor_logs.py @@ -26,6 +26,17 @@ _ENV_SENTINEL = "OMN15572_MONITOR_IMPORT_SENTINEL" _MONITOR_SCRIPT = _SCRIPTS_DIR / "monitor_logs.py" +# A fresh interpreter startup + `import monitor_logs` normally completes in well +# under a second, but the self-hosted CI fleet runs many concurrent pytest-split +# shards per job, and interpreter spawn/import can occasionally queue behind +# that CPU contention. A 10s bound tripped this way twice live, on unrelated +# PRs' Split 12/15 (infra#2697 and infra#2700, mergesweep-0809-poisonpath-verify +# / mergesweep-0809-landingset), both times as three subprocess.TimeoutExpired +# failures at exactly this bound, unrelated to either PR's own diff. 30s keeps +# the bound tight enough to still catch a genuine hang while giving real +# headroom against fleet contention. +_SUBPROCESS_TIMEOUT_SECONDS = 30 + def _import() -> Any: if _MODULE_NAME in sys.modules: @@ -104,7 +115,7 @@ def test_import_does_not_load_home_env(self, tmp_path: Path) -> None: env=subprocess_env, capture_output=True, text=True, - timeout=10, + timeout=_SUBPROCESS_TIMEOUT_SECONDS, check=False, ) @@ -288,7 +299,7 @@ def stop_after_config(_parser, *_args, **_kwargs): env=subprocess_env, capture_output=True, text=True, - timeout=10, + timeout=_SUBPROCESS_TIMEOUT_SECONDS, check=False, ) @@ -342,7 +353,7 @@ def stop_after_config(_parser, *_args, **_kwargs): env=subprocess_env, capture_output=True, text=True, - timeout=10, + timeout=_SUBPROCESS_TIMEOUT_SECONDS, check=False, )