diff --git a/src/omnibase_infra/event_bus/kafka_transport.py b/src/omnibase_infra/event_bus/kafka_transport.py index 36a87ca16e..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. @@ -264,6 +312,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..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 @@ -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,48 @@ 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: +# ``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( + *, + 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 +111,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,37 +125,135 @@ 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() + # 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 + 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.""" + """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 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 + ``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). + + 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") - await asyncio.gather(*self._tasks) + watched: set[asyncio.Task[None]] = set(self._tasks) + while watched: + done, watched = await asyncio.wait( + watched, + timeout=self._watchdog_tick_seconds, + 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 -- + # 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.""" 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 +276,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,18 +349,198 @@ 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 error=%s", + direction, + message.topic, + message.partition, + message.offset, + type(error).__name__, + sanitize_error_message(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"], source: ProtocolGatewayConsumer, ) -> None: + 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. + + 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: - messages = await source.poll( - max_messages=1, - timeout_ms=self._poll_timeout_ms, + 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(): + 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] + # 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 restarter() + 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}" + ), ) - for message in messages: - await self.deliver_message(direction, source, message) + + 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/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 e509586251..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 @@ -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,289 @@ 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 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.restart_calls = 0 + self.committed: list[object] = [] + self.nacked: list[object] = [] + 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 + 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. + + ``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([]) + 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: + # 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.restart_calls: + break + await asyncio.sleep(0.02) + assert stalled_outbound.restart_calls >= 1, ( + "watchdog never force-recreated the stalled transport -- the " + "silent-eviction mode was never detected" + ) + + # 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(), 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.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 + 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() 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, )