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..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 @@ -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 @@ -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: @@ -140,16 +182,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 +221,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 +239,35 @@ 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, + ) + 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(local_only), + ) + 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_delivery_service.py b/tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py index 54d7107c13..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 @@ -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: @@ -88,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: @@ -297,3 +315,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}" + ) 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..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 @@ -102,6 +102,49 @@ 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(), + canary=_canary(), + 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(), + canary=_canary(), + 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/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(), ) 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,