feat(realtime): fix head-of-line blocking in realtime fanout with bounded concurrency and proactive slow-consumer quarantine (#379)

[s6pa1rta3n-lab]

Overview

Resolves #379.

This pull request addresses the head-of-line blocking vulnerability in fanout_message by decoupling message delivery from overflow and eviction handling, enforcing bounded concurrency with per-connection isolation, and introducing a proactive slow-consumer detection and quarantine policy in policy.rs.

Key Changes

1. Bounded-Concurrency Fanout (backend/src/realtime/fanout.rs):

   • Converted fanout to bounded-concurrency execution via buffer_unordered (configurable concurrency limit, default 64).
   • Removed sequential await loops and in-line handle_overflow(&id).await execution from the hot fanout path.
   • Offloaded connection overflow handling and registry eviction to detached asynchronous tasks.
   • Guaranteed per-connection FIFO message ordering.

2. Proactive Slow-Consumer Detection & Quarantine (backend/src/realtime/policy.rs):

   • Implemented ConnectionPolicyTracker with queue depth trend analysis (Stable, Growing, Draining, Stalled).
   • Added ConnectionHealthState lifecycle transitions: Healthy -> Degraded -> Quarantined -> Evicted.
   • Integrated a leaky-bucket burst tolerance model to avoid false-positive evictions during transient network blips while rapidly isolating sustained stalled consumers.
   • Added automatic quarantine recovery once consumers drain their backlog after a cooldown period.

3. Isolated Connection Accounting (backend/src/realtime/connection.rs):

   • Enhanced Connection with atomic queue depth tracking and non-blocking policy evaluation.
   • Introduced ConnectionReceiver for automated message consumption accounting, drain rate metrics, and queue depth synchronization.

4. Comprehensive Test Suite & Benchmarks (backend/tests/slow_consumer_test.rs):

   • Added load test verifying p99 fanout latency across 50 healthy connections stays flat and sub-millisecond even when a slow/stalled connection has a full buffer.
   • Added strict per-connection FIFO message sequence ordering verification across 10 concurrent consumers (100 messages each).
   • Added bounded concurrency validation under high connection counts (128 clients, concurrency limit 16).
   • Added proactive quarantine lifecycle and cooldown recovery tests.
   • Validated backwards-compatible ConnectionRegistry::add interface.

Acceptance Criteria Checklist

• A stalled/slow single connection cannot measurably delay delivery to any other connection (demonstrated by a test/benchmark, not just code review).
• Per-connection message ordering is preserved.
• Concurrency is bounded (no unbounded task/future spawning per fanout call).
• policy.rs gains a documented, tested slow-consumer detection policy beyond "channel send failed."
• Existing cargo test suite stays green.

Verification

• cargo test --lib: 36 unit tests passed.
• cargo test --tests: All integration test suites passed (connected_trace_test, discrepancy_alert_test, double_replay_test, dual_network_ingestion_test, mixed_version_replay_test, slow_consumer_test).
• Contract tests in contracts/ passed.

Payout Routing

• EVM (Base/Arbitrum/Polygon/ETH): 0xF46C9F6d70C50BF81ef3588AB523a90a594a2F89
• Stellar: GCL6OXAMLD75BMTINA6EMRUDWK5THQUSHMYNLSNBCJAPZJHNYJTUNIBC