Repository navigation
fix(OMN-15748): gateway poison-pill quarantine + membership-loss watchdog - #2700
Merged
jonahgabriel merged 9 commits intoAug 9, 2026
Merged
jonahgabriel merged 9 commits into
jonahgabriel merged 9 commits into
Conversation
…ateway 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)
…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 (f0cdfe4, 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
…ent 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
…t 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
…hdog 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
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 41 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughThe gateway forwarder now supports bounded reconnect retries, degraded and recovered status events, stalled-consumer recovery, and poison-record quarantine. Kafka consumer recreation preserves the producer. Configuration and tests cover the new liveness behavior. ChangesGateway liveness and recovery
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant GatewayForwarder
participant NodeGatewayDelivery
participant KafkaTransport
participant ServiceGatewayForwarder
GatewayForwarder->>NodeGatewayDelivery: supervise delivery
NodeGatewayDelivery->>KafkaTransport: restart stalled consumer
NodeGatewayDelivery->>ServiceGatewayForwarder: publish degraded status
GatewayForwarder->>NodeGatewayDelivery: retry after bounded backoff
GatewayForwarder->>ServiceGatewayForwarder: publish active status after recovery
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
Base automatically changed from
jonah/omn-15742-gateway-reconnect-supervision
to
dev
August 9, 2026 18:27
…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.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (5)
src/omnibase_infra/runtime/gateway_forwarder.py (1)
399-415: 🩺 Stability & Availability | 🔵 TrivialTwo uncoordinated status emitters now publish to the same local topic.
This helper and
NodeGatewayDelivery._publish_watchdog_status(src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.pyLine 514-521) both callforwarder.publish_statuswithactive/degraded, and each tracks its own degraded state. The runtime supervisor tracksdegraded_emitted; the watchdog tracks_watchdog_degradedper direction.An operator consuming the heartbeat topic can therefore see
activepublished by the watchdog while the supervisor still considers the gateway degraded, and the reverse. The payload carries no emitter identity, so the two streams are indistinguishable downstream.Consider adding an emitter or scope field to the status payload, or centralizing degraded state in one owner. Neither change is required for this PR to be correct.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/omnibase_infra/runtime/gateway_forwarder.py` around lines 399 - 415, The gateway status payload has two uncoordinated emitters: _publish_gateway_status and NodeGatewayDelivery._publish_watchdog_status, each maintaining independent active/degraded state. Coordinate these status streams by either adding an emitter or scope identifier to publish_status payloads or centralizing degraded-state ownership, ensuring downstream consumers can distinguish or receive a single authoritative status.src/omnibase_infra/event_bus/kafka_transport.py (1)
250-256: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueConsider tolerating a failing
stop()so the replacement consumer is still created.If
self._consumer.stop()raises, the exception propagates before_start_consumer()runs. The transport is then left withself._consumer = None.NodeGatewayDelivery._recover_stalled_directioncatches this, logs, and returns without spawning the direction task, so that direction stays down until the next watchdog tick. Recovery is not lost, but it is delayed by one tick. Suppressing the stop error keeps the recreate atomic.♻️ Optional: log and continue on stop failure
self._buffer.clear() if self._consumer is not None: try: await self._consumer.stop() + except Exception: # noqa: BLE001 — the old client is being discarded anyway. + logger.warning( + "KafkaTransport consumer stop failed during restart_consumer; " + "discarding the client and recreating", + exc_info=True, + ) finally: self._consumer = None await self._start_consumer()This requires a module-level
loggerin this file; confirm one exists before applying.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/omnibase_infra/event_bus/kafka_transport.py` around lines 250 - 256, Update the consumer replacement flow around _consumer.stop() so a stop failure is caught, optionally logged through the existing module-level logger, and does not prevent _start_consumer() from running. Ensure self._consumer is still cleared and the replacement consumer is created even when stopping the previous consumer raises.src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py (2)
68-84: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider adding a generated correlation ID to the quarantine payload.
The record is undecodable, so no envelope correlation ID exists. A generated
uuid4()would let an operator join the DLQ record to theGateway undecodable record quarantinedlog line. Today only (topic, partition, offset) links them, and that tuple is not logged as a single field.♻️ Optional: emit and log a quarantine correlation ID
def _build_quarantine_payload( *, direction: Literal["outbound", "inbound"], message: ModelTransportMessage, error: Exception, + correlation_id: str, ) -> bytes: """Serialize an undecodable record's forensic context for the DLQ.""" payload: dict[str, object] = { + "correlation_id": correlation_id, "original_topic": message.topic,Pass the same value to the
logger.errorcall in_quarantine_undecodable_message.Based on learnings, this repository accepts plain forensic payloads; treat this as optional. As per coding guidelines "Propagate incoming correlation IDs, generate one with
uuid4()when absent, and include it in error context."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py` around lines 68 - 84, The quarantine payload currently lacks a correlation identifier for undecodable records. Generate one with uuid4() in _quarantine_undecodable_message, pass the same value into _build_quarantine_payload, include it in the serialized payload, and add it to the corresponding logger.error context for “Gateway undecodable record quarantined.”Source: Coding guidelines
430-439: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winWatchdog recovery has no confirmation window, unlike the runtime supervisor.
On the tick after a successful recreate,
elapsedis small and membership is regained, so neither staleness condition holds and_mark_direction_recoveredpublishesactiveimmediately. The gap between thedegradedandactiveevents can be as short as one_watchdog_tick_seconds.
_supervise_gateway_deliveryinsrc/omnibase_infra/runtime/gateway_forwarder.pydeliberately takes the opposite approach: it requires a fullheartbeat_interval_secondswithout a new failure before it publishesactive(see the docstring at Line 266-271 of that file). If a direction stalls repeatedly, this watchdog emits an alternatingdegraded/activestream on the local bus while the sibling supervisor suppresses the same flapping.Consider requiring N consecutive healthy ticks, or one full
heartbeat_interval_secondsof progress, before declaring recovery.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py` around lines 430 - 439, Update the recovery branch in the watchdog flow around _recover_stalled_direction and _mark_direction_recovered to require a confirmation window before publishing active. Track either the configured number of consecutive healthy ticks or one full heartbeat_interval_seconds without renewed staleness, reset that progress when the direction degrades again, and call _mark_direction_recovered only after the threshold is satisfied.tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py (1)
559-560: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThis test exercises only the membership-streak trigger, not the staleness timer.
has_group_membershipreturnsself._recreated, which isFalseuntil recovery. The watchdog therefore records a lost-membership probe on every tick. Withmax_silence_window_seconds=2,_watchdog_tick_secondsis0.5, so_membership_lost_streakreaches 2 at about 1.0s. Thestale_by_timethreshold of 2.0s is never reached.Trigger (1) in the
_run_watchdog_loopdocstring — no poll progress for_watchdog_stale_seconds— is the mechanism described as the live 2026-08-09 failure, and it currently has no coverage. A second fake that implementsrestart_consumerbut omitshas_group_membershipwould isolate the time-based path.Do you want me to generate that test?
Also applies to: 600-609
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py` around lines 559 - 560, Add a separate watchdog test fake near the existing membership fake that implements restart_consumer but omits has_group_membership, then use it to exercise _run_watchdog_loop’s time-based stale_by_time path. Configure max_silence_window_seconds=2 and verify the watchdog triggers after approximately the 2-second no-progress threshold, without relying on _membership_lost_streak.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@src/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_forwarder_config.py`:
- Around line 45-48: Update _compute_reconnect_delay_seconds to cap the retry
exponent before calculating reconnect_backoff_initial_seconds * 2 ** exponent,
using the configured reconnect_backoff_max_seconds so the intermediate
multiplication cannot overflow. Preserve the existing maximum-delay cap, jitter
behavior, and retry progression for normal attempt values.
In
`@src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py`:
- Around line 343-352: The gateway undecodable-record log in the delivery
service exposes unsanitized exception content through exc_info. Replace the raw
exception logging with sanitize_error_message(error), matching the existing DLQ
sanitization path, and remove traceback logging that could include untrusted
record data while preserving the existing quarantine context fields.
- Around line 210-227: Update the task supervision loop around the local watched
set so it periodically yields or uses a bounded wait, ensuring self._tasks is
re-read after watchdog recovery can splice in replacement tasks. Preserve
cancellation handling and exception propagation, and ensure newly added live
tasks are included in watched before supervision blocks indefinitely.
In
`@src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.py`:
- Around line 242-269: Sanitize the raw detail text in
service_gateway_forwarder.py:242-269 within publish_status before passing it to
_build_status_envelope, using the repository’s sanitization utility or a safe
error classification; preserve status publication behavior while ensuring
credentials are never serialized. Add a test in
tests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service.py:281-294
that supplies credential-bearing diagnostic text and asserts the published
payload is redacted.
- Around line 264-269: Update publish_status around the
_publish_with_delivery_retry call to avoid unbounded retries for status events.
Publish the local status once or through a short bounded timeout, allowing
RuntimeHostError to propagate or be logged so reconnect supervision and watchdog
recovery can continue.
In `@tests/unit/runtime/test_gateway_forwarder_runtime.py`:
- Around line 367-375: Increase reconnect_backoff_initial_seconds and
reconnect_backoff_max_seconds in the test configuration to add timing headroom
before the degraded threshold is evaluated. Update the _stop_when_recovered
sleep duration and its accompanying comment to reflect the new backoff spacing,
while preserving the existing degraded-threshold assertion behavior.
---
Nitpick comments:
In `@src/omnibase_infra/event_bus/kafka_transport.py`:
- Around line 250-256: Update the consumer replacement flow around
_consumer.stop() so a stop failure is caught, optionally logged through the
existing module-level logger, and does not prevent _start_consumer() from
running. Ensure self._consumer is still cleared and the replacement consumer is
created even when stopping the previous consumer raises.
In
`@src/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.py`:
- Around line 68-84: The quarantine payload currently lacks a correlation
identifier for undecodable records. Generate one with uuid4() in
_quarantine_undecodable_message, pass the same value into
_build_quarantine_payload, include it in the serialized payload, and add it to
the corresponding logger.error context for “Gateway undecodable record
quarantined.”
- Around line 430-439: Update the recovery branch in the watchdog flow around
_recover_stalled_direction and _mark_direction_recovered to require a
confirmation window before publishing active. Track either the configured number
of consecutive healthy ticks or one full heartbeat_interval_seconds without
renewed staleness, reset that progress when the direction degrades again, and
call _mark_direction_recovered only after the threshold is satisfied.
In `@src/omnibase_infra/runtime/gateway_forwarder.py`:
- Around line 399-415: The gateway status payload has two uncoordinated
emitters: _publish_gateway_status and
NodeGatewayDelivery._publish_watchdog_status, each maintaining independent
active/degraded state. Coordinate these status streams by either adding an
emitter or scope identifier to publish_status payloads or centralizing
degraded-state ownership, ensuring downstream consumers can distinguish or
receive a single authoritative status.
In `@tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py`:
- Around line 559-560: Add a separate watchdog test fake near the existing
membership fake that implements restart_consumer but omits has_group_membership,
then use it to exercise _run_watchdog_loop’s time-based stale_by_time path.
Configure max_silence_window_seconds=2 and verify the watchdog triggers after
approximately the 2-second no-progress threshold, without relying on
_membership_lost_streak.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 48ce90c3-b87f-4627-9201-bfc32fda1997
📒 Files selected for processing (13)
src/omnibase_infra/event_bus/kafka_transport.pysrc/omnibase_infra/nodes/node_bus_forwarder_effect/contract.yamlsrc/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_forwarder_config.pysrc/omnibase_infra/nodes/node_bus_forwarder_effect/models/model_gateway_heartbeat.pysrc/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_delivery.pysrc/omnibase_infra/nodes/node_bus_forwarder_effect/services/service_gateway_forwarder.pysrc/omnibase_infra/runtime/gateway_forwarder.pytests/unit/event_bus/test_kafka_transport_protocol_conformance.pytests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.pytests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_config.pytests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service.pytests/unit/nodes/node_bus_forwarder_effect/test_gateway_forwarder_service_handler_seam.pytests/unit/runtime/test_gateway_forwarder_runtime.py
…y-poison-quarantine-membership-recovery # Conflicts: # tests/unit/nodes/node_bus_forwarder_effect/test_gateway_delivery_service.py
Contributor
✅ Hostile Reviewer — PASSEDBlocking findings (critical): 0 Gate semantics (pilot phase)
Powered by omniintelligence.review_pairing.cli_review — node-based adversarial review via HandlerLlmCliSubprocess (OMN-8468/OMN-8524) |
…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.
Collaborator
Author
|
Addressed both live CodeRabbit findings on
Both fixes in b5e9f31. |
jonahgabriel
added a commit
that referenced
this pull request
Aug 9, 2026
… flake OMN-15749
Deploy Agent Tests failed twice on TestRepoDigestsOnContainerIsRealDockerFailureMode
(docker create / docker rm -f timing out at 30s). Root-caused, not transient
hand-waving: this is the pre-existing, already-ticketed OMN-15749 ("deploy-agent
digest-verification fixture tests time out on real docker create/docker rm -f --
30s too short under runner contention", Backlog, unrelated to this PR's diff --
this PR touches zero deploy-agent files). Confirmed non-systemic: sibling PRs
#2701/#2702/#2700 all passed the identical test on concurrent runs at the same
time, so this is isolated runner-instance docker contention, not a fleet-wide
or code-caused failure. Retrying past the known flake rather than blocking an
unrelated CI-workflow-trigger fix on it.
jonahgabriel
added a commit
to OmniNode-ai/onex_change_control
that referenced
this pull request
Aug 9, 2026
…ibase_infra#2700 (#6265) * evidence(OMN-15748): author OCC companion for OmniNode-ai/omnibase_infra#2700 OCC companion by node_pr_lifecycle_fix_effect (OMN-13317 F1 / OMN-13990 / OMN-14285). Product PR head 97ee827a4ee8fe19b2961b6cd16d27ef0ba20f90. * evidence(OMN-15748): self-bind OCC#6265 + rebind contract_sha256 * evidence(OMN-15748): rebind contract to PR#2700 head b5e9f3154 (post-retarget) The original OMN-15748 autobind (commit_sha 97ee827a4) predated the base retarget to dev and three follow-up commits: 309d64021 (consumer-scoped watchdog recovery + no-CancelledError-escape fix), f4c241d93 (merge dev, resolving the retarget conflict), and b5e9f3154 (CodeRabbit findings on the watchdog wait() bounded-rescan gap and log sanitization). Per OCC append-only convention, adds a new rebind evidence entry and updates the content-bound probe's ref to the current head instead of mutating history. * evidence(OMN-15748): rebind receipts to PR#2700 head b5e9f3154 (post-retarget) Companion to the contract rebind (previous commit). Re-verified the content-bound probe live at the new head (green: has_group_membership present, exit 0; red at merge-base 611c8a190, exit 1, unchanged) and updated its receipt with the fresh commit_sha/contract_entry_sha256/probe evidence. Added the receipt for the new occ-rebind-post-retarget-b5e9f315 evidence entry, confirming PR #2700's live head matches this binding. Bumped the legacy whole-file contract_sha256 on the three unaffected receipts for consistency (their contract_entry_sha256 is unchanged and remains authoritative per OMN-13888 -- their own content did not change). Local verification: uv run pytest tests/test_evidence_admissibility.py -q -> 76 passed. uv run validate-yaml contracts/OMN-15748.yaml -> OK. scripts/validation/check_receipt_hardening.py -> exit 0, no findings. --------- Co-authored-by: omnimarket-bot <bot@omninode.ai> Co-authored-by: Jonah Gray <jonah@omninode.ai>
…nst 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.
jonahgabriel
deleted the
jonah/omn-15748-gateway-poison-quarantine-membership-recovery
branch
August 9, 2026 22:59
jonahgabriel
added a commit
that referenced
this pull request
Aug 10, 2026
…d PR events (#2703) * fix(OMN-14241): occ-preflight eval-path callers must listen for `edited` PR events ci.yml (implicit default types, no `edited`) and hostile-reviewer.yml (`types: [opened, synchronize, reopened]`) each declare a top-level `occ-preflight` job calling the canonical reusable occ-preflight.yml, but neither retriggers when a PR body is edited without a new commit. The reusable workflow already live-fetches the current PR body via `gh pr view` on every run it does execute -- the defect was never a frozen-body read, it was that no new run fired at all when a body-only edit (typically stamping `Evidence-Source: OCC#<n>` after the companion OCC PR merges) landed. The last commit-triggered run -- correctly FAILURE before the stamp existed -- then sits as the permanent `occ-preflight / eligibility` status, and CI Summary fails closed on it forever without a manual empty-commit retrigger. Live-root-caused on infra#2696 and #2694 (2026-08-09): both PRs stamped Evidence-Source via a body edit; CI's and Hostile Reviewer's occ-preflight copies stayed FAILURE from before the edit while call-reject-skip.yml's copy (which already lists `edited`) correctly re-ran and passed. Adds tests/ci/test_occ_eval_path_trigger_coverage.py (RED before this change on both workflows, GREEN after), sibling to the existing test_occ_born_path_trigger_coverage.py (OMN-14987) guard for the autobind/companion-effect minting workflows. OMN-14241 (adopted existing Backlog ticket; relatedTo OMN-15725/OMN-15214 lineage). No debounce/diff guard added -- today's actual behavior already pays a full-CI-run cost per stamp fix via manual empty-commit retriggers, so automating the same retrigger via `edited` is net-neutral CI cost, not new cost; it removes the manual-toil failure mode currently blocking both PRs. * chore(OMN-14241): [mergesweep-0809-infraunblock] retrigger CI Deploy Agent Tests (OMN-15378) failed once with a docker-teardown flake unrelated to this PR's diff (subprocess.TimeoutExpired on `docker rm -f` in test_executor_digest_verification_image_id_omn15181.py, a file this PR never touches). First-strike retry per two-strike allowance. * chore(OMN-14241): [mergesweep-0809-infraunblock] retrigger CodeRabbit gate gate/CodeRabbit Thread Check's prior run (31333667026) was CANCELLED by a concurrent same-SHA run collision, not a real failure (log confirmed "Unresolved CodeRabbit threads: 0"). An issue_comment retrigger attempt resolved to the wrong head SHA (dev tip, not this PR's head) due to how cr-thread-gate-caller.yml's issue_comment path checks out ref -- a genuine workflow quirk, not applicable to this PR's actual commit. Using a synchronize-triggering empty commit instead, which correctly targets this PR's head per the workflow's pull_request trigger. * chore(OMN-14241): [mergesweep-0809-infraunblock] retrigger CI Summary poll CI Summary FAILED (run 31334333041) purely on poller-deadline timing: its own log shows CodeQL still listed as "missing/pending" at the 20:37:31Z deadline, but CodeQL's own run (31334296327) completed SUCCESS at 20:36:49Z -- 42 seconds earlier. A fresh poll will observe CodeQL as already-passed. Not a repeat of the same failure class (prior strikes were Deploy Agent Tests docker-teardown timeout and a CodeRabbit gate wrong-SHA issue) -- this is a first-instance poller/job-completion race. * chore(OMN-14241): [mergesweep-0809-infraunblock] retrigger past known flake OMN-15749 Deploy Agent Tests failed twice on TestRepoDigestsOnContainerIsRealDockerFailureMode (docker create / docker rm -f timing out at 30s). Root-caused, not transient hand-waving: this is the pre-existing, already-ticketed OMN-15749 ("deploy-agent digest-verification fixture tests time out on real docker create/docker rm -f -- 30s too short under runner contention", Backlog, unrelated to this PR's diff -- this PR touches zero deploy-agent files). Confirmed non-systemic: sibling PRs #2701/#2702/#2700 all passed the identical test on concurrent runs at the same time, so this is isolated runner-instance docker contention, not a fleet-wide or code-caused failure. Retrying past the known flake rather than blocking an unrelated CI-workflow-trigger fix on it. * chore(OMN-14241): [mergesweep-0809-infraunblock] retrigger past transient pin-reachability timeout Tests (Split 1/15) failed on tests/integration/ci/test_pin_reachability_live_omn15538.py with "transport error: The read operation timed out" resolving the pre-existing core-ref pin in ci.yml/hostile-reviewer.yml (unrelated to this PR's diff -- neither line is touched here; the stale pin itself is OMN-15725's scope, PR #2680 still open). Confirmed transient, not a durable unreachable-pin state: sibling PR #2688/#2697 passed this same check hours ago; sibling PR #2701 failed the identical check in the same ~30-40min window as this PR, consistent with a shared transient GitHub API rate-limit/timeout window (cf. OMN-14352) rather than a code-caused or permanently-broken pin.
jonahgabriel
added a commit
that referenced
this pull request
Aug 10, 2026
… with measured evidence (#2714) The 10s->30s subprocess timeout bump in test_monitor_logs.py landed via commit 993d805 (PR #2700, OMN-15748) with a speculative, unmeasured justification comment claiming import monitor_logs "normally completes in well under a second." That claim is false and was never measured. Root-caused with python -X importtime on this repo's own .venv/bin/python3 (the same interpreter sys.executable resolves to inside the test's own subprocess): monitor_logs.py's module-scope _load_monitor_alert_topic() call imports omnibase_infra.models.contracts.model_monitor_alert_contract, which pays the transitive cost of omnibase_infra/models/__init__.py's eager-loading chain (node_architecture_validator, node_registration_orchestrator.dispatchers, omnibase_core dispatch/ reducer/common/validation models). Measured cumulative import cost: 3,464,293us (~3.46s), of which 3,415,000us (~3.42s) is that one subtree. 10 serial cold-subprocess runs: p50 4.452s, max 5.515s -- multi-second even completely uncontended. Reproduced the live CI failure mode directly by running N copies of the same import concurrently via ThreadPoolExecutor (24-core box): N=24 (1x core count) -> max 11.425s, already past the old 10s bound; N=48 (2x oversubscription) -> max 18.906s. 30s keeps ~1.6x margin over the worst measured case and ~2.6x over 1x-oversubscription -- a real, quantified margin that did not exist when the constant was chosen. Scope is deliberately narrow: the heavy transitive import in omnibase_infra/models/__init__.py is a known architecture cost, not touched here (would be an unsolicited architecture change affecting every consumer of the models package). This PR only corrects the comment with real evidence and confirms 30s is justified by that evidence; the timeout value itself is unchanged. No dedicated ticket previously tracked this fix -- it was piggybacked onto the unrelated OMN-15748 gateway-forwarder ticket. OMN-15836 now tracks it directly. Ticket: OMN-15836
jonahgabriel
added a commit
that referenced
this pull request
Aug 10, 2026
Advances the merge base to origin/dev @ 303e008 to unblock PR #2681 (OMN-15700) on the required CI Summary external-context check. The PR previously pinned merge SHA a90460f replayed over the Aug-6 base 4bf964a, which contained neither: - PR #2700 (dev squash 6b6cdc1): monitor_logs test timeout bump 10s -> 30s - PR #2714 (303e008): measured-comment fix for the same test Merging origin/dev in (not rebasing) so 0e21e0c stays in ancestry, keeping the prior hostile-review approval at that exact head citable against an unrewritten commit. No conflicts: git merge --no-commit --no-ff reported "Automatic merge went well" with a single auto-merged file (src/omnibase_infra/runtime/service_kernel.py, the savings-estimator seam file) and zero UU entries / conflict markers. Refs OMN-15700
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Two independent recovery mechanisms for
NodeGatewayDelivery, both from the2026-08-09 mergesweep-0809-poisonpath adjudication:
1. Poison-pill quarantine (OMN-15748).
_deliver_message_lockedcalledself._forwarder.decode_message(message)before thetry/exceptthatnacks. Any undecodable record on a mirrored source topic crashed the process
(live-observed 2026-08-08T16:50Z, verbatim pydantic
ValidationErrortraceback ending in
sys.exit(main()),RestartCount0→1). Moving decodeinside
tryand routing failures to nack (the existing behavior formid-pipeline errors) would be wrong for THIS class of failure: a permanently
malformed record can never decode no matter how many times it's redelivered,
so nack's
seek-back-and-redeliver just re-crashes forever. Fix: decodefailures are logged, best-effort dead-lettered (via
get_dlq_topic_for_original, the record's own topic — never a hardcodedliteral), and committed past (skipped), so the consumer loop keeps making
forward progress.
2. Membership-loss watchdog (OMN-15748/OMN-15690). The live
2026-08-09T10:03:07Z incident was a clean single
LeaveGroupwith zeroexception — aiokafka's client-side
max_poll_interval_msidle-eviction,fired from
GroupCoordinator._heartbeat_routine, a separate asyncio taskfrom the poll loop. The existing exception-triggered reconnect supervision
(
_supervise_gateway_deliveryinruntime/gateway_forwarder.py, landed in#2688) gates off
delivery.wait()'s gather — structurally unable to observea task that is alive, hung inside a stuck
poll(), and never exceptions.Once evicted,
ensure_active_grouprefuses to rejoin while the client'sidle clock stays past threshold, and that clock never resets on a
permanently stuck poll, so the outage is permanent with no exit event.
Fix: an independent watchdog task inside
NodeGatewayDeliverytracksper-direction last-progress timestamps (updated on
poll()completion anddeliver_messagecompletion) plus a best-effortKafkaTransport.has_group_membership()(new: wrapsconsumer.assignment()) probe. On staleness (either signal), itforce-cancels and recreates just the affected direction's task and
transport (
close()+start()— a brand-newAIOKafkaConsumerforces afresh group join, sidestepping the permanently-stuck idle clock) and
publishes a
DEGRADEDstatus via the existingpublish_statusmechanism(#2688). Reuses the previously contract-declared but completely unwired
max_silence_window_secondsconfig field as the staleness threshold — itexisted in
ModelGatewayForwarderConfigwith no reader anywhere in thecodebase before this PR.
Both optional capabilities (
close/start,has_group_membership,send)are duck-typed via
getattr/callablerather than declared as additionalProtocolclasses — the architecture validator enforces oneProtocolperfile and this file already owns
ProtocolGatewayConsumer.Deferred (flagged, not fixed here)
The adjudication also recommended a bounded-timeout wrap
(
asyncio.wait_for) aroundself._delivery_lockacquisition asdefense-in-depth against cross-direction lock starvation. Not included:
inbound was NOT stuck in the live incident (it kept flowing through the same
lock the entire time), so this isn't the root cause here, and it has no
dedicated RED test in this pass. Left as a follow-up per the adjudication's
own precedent of deferring a full lock-split audit.
TDD
Both regression tests verified RED against pre-fix source (
git stashofthe source-only diff, tests kept) and GREEN after:
test_undecodable_record_is_quarantined_not_crashed— feeds the exactlive poison shape through the real delivery path; asserts logged, DLQ'd,
offset committed (not nacked), no exception escapes.
test_membership_loss_watchdog_recreates_and_recovers— a fake sourcewhose first
poll()hangs forever with zero exception (reproducing thelive silent-eviction mechanism); asserts the watchdog force-recreates,
polling resumes, DEGRADED is published, and
delivery.wait()'sexception-triggered gather never fires (proving recovery bypasses the
broken supervision path entirely, not merely surviving it).
Verification
uv run mypy --stricton both touched source files: clean.uv run ruff format/ruff check: clean.node_bus_forwarder_effect/+runtime/test_gateway_forwarder_runtime.pyevent_bus/, 664 tests): all pass, both hosts (.200 and this Mac).the first draft (3 auxiliary
Protocolclasses) — fixed viagetattr/callableduck-typing, re-verified clean.unit suite (touches
kafka_transport.py, a shared module) — 23188passed, 40 skipped, one pre-existing, host-dependent, unrelated
failure on
.200(test_omninode_ci_required_context_probe.py:: test_reporter_degrades_visibly_when_the_probe_is_missing), confirmed bybash --versiondiff:.200runs macOS system bash 3.2, this testrequires bash 4+ (OMN-15788, previously flagged/disclosed). Re-ran the
full suite on this Mac (bash 5.3): 23188 passed, 0 failed, push succeeded
from here as the disclosed OMN-15788 exception.
Coordination / landing order
Stacked on
jonah/omn-15742-gateway-reconnect-supervision(infra#2688,head
611c8a190) rather thandevor a new branch off#2697— thiswork's
_recover_stalled_directionpublishesDEGRADED/activethroughServiceGatewayForwarder.publish_status, the mechanism #2688 introduced, sostacking avoids re-deriving that seam. Does not touch
#2688's or#2697's branch heads directly — new commits only, on a third branch. Perthe runner-recovery lane owning reruns on #2688/#2697: this PR should not
be merged ahead of #2688 landing on
dev(its base branch is#2688'sbranch, not
dev) — land#2688first, then rebase/retarget this PR'sbase to
dev(or #2697's branch, if that lands first) before merge.OMN-15748
Evidence-Ticket: OMN-15748
Evidence-Source: OCC#6265
Stacked-Parent: #2688
Summary by CodeRabbit
New Features
Bug Fixes