fix(router): never await channel delivery in the shared inbound router - #618
fix(router): never await channel delivery in the shared inbound router#618jamiepine wants to merge 1 commit into
Conversation
The main select loop forwarded messages with an awaited send into each channel's bounded queue. One channel that stopped draining (long tool call, wedged LLM stream) filled its 64-slot queue and blocked the router on the await, stalling inbound delivery for every conversation on every agent. Each active channel now gets an InboundRelay: the router enqueues with a non-blocking send and a per-channel relay task owns the awaited forward into the channel's bounded queue. Per-channel FIFO order is unchanged, send-failure semantics still mean the channel is gone, and relay depth is logged at 128 and each doubling so a stalled channel is visible.
WalkthroughAdded ChangesInbound relay delivery
Estimated code review effort: 4 (Complex) | ~45 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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/agent/inbound_relay.rs`:
- Around line 96-102: Replace the unbounded channel in the relay ingress setup
with a bounded channel and use non-blocking try_send for message admission.
Define the overload behavior explicitly, including the queued-counter update and
pressure logging, so rejected messages are handled without blocking the router
or retaining unlimited InboundMessage values; update RelayState integration
accordingly.
- Around line 59-74: Update the enqueue path in the relay sender method
containing tx.send and RelayState::queued so queued is incremented before
tx.send can wake the relay task. If sending fails and returns the message,
decrement queued to roll back the reservation before propagating RelaySendError;
preserve the warning-threshold logic using the reserved depth. Add a
multi-threaded test that races relay consumption with enqueue accounting and
verifies queued never underflows or corrupts backpressure tracking.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: ea6fe120-0044-47ce-b22a-13f71b3588f2
📒 Files selected for processing (3)
src/agent.rssrc/agent/inbound_relay.rssrc/main.rs
| self.tx | ||
| .send(message) | ||
| .map_err(|error| RelaySendError(Box::new(error.0)))?; | ||
|
|
||
| let depth = self.state.queued.fetch_add(1, Ordering::Relaxed) + 1; | ||
| let warn_at = self.state.warn_at.load(Ordering::Relaxed); | ||
| if depth >= warn_at | ||
| && self | ||
| .state | ||
| .warn_at | ||
| .compare_exchange( | ||
| warn_at, | ||
| warn_at.saturating_mul(2), | ||
| Ordering::Relaxed, | ||
| Ordering::Relaxed, | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
fd -a 'inbound_relay.rs|RUST_STYLE_GUIDE.md|Cargo.toml|Cargo.lock' . | sed 's#^\./##'
echo "== relay outline =="
ast-grep outline src/agent/inbound_relay.rs --view expanded || true
echo "== relay relevant lines =="
nl -ba src/agent/inbound_relay.rs | sed -n '1,160p'
echo "== tests / silent results mentions =="
rg -n "let _ =|Result|fetch_add|queued|send\\(|UnboundedSender|tokio::spawn|handle" src/agent/inbound_relay.rsRepository: spacedriveapp/spacebot
Length of output: 1623
🌐 Web query:
Tokio UnboundedSender send multiple threads can wake receiver before later increment atomic
💡 Result:
In Tokio's mpsc unbounded channel, the send operation and the wakeup mechanism are correctly synchronized to ensure the receiver is properly notified, preventing race conditions where a receiver might miss a message or fail to wake despite a message being sent. The UnboundedSender::send method performs two primary steps: it updates the channel state (incrementing the message count via an atomic read-modify-write operation) and then pushes the actual data into the internal channel structure [1][2][3]. The critical synchronization points are: 1. Atomic State Update: The sender uses compare_exchange with AcqRel (Acquire-Release) ordering on the semaphore/channel state atomic to increment the message count [1][2][3]. This ensures that the message count increment is visible and establishes the necessary happens-before relationship between the send and any subsequent receiver activity. 2. Wakeup Mechanism: After successfully incrementing the count and pushing the message, the channel triggers the receiver's waker [4]. The underlying AtomicWaker uses atomic operations (specifically AcqRel or Release orderings) to safely manage the registration and consumption of the waker [5][6]. When a receiver polls the channel and receives Poll::Pending, it registers a waker [7]. Any subsequent call to send will see this registered waker (or the result of the atomic synchronization) and trigger the wakeup [5][4]. Because the incrementing of the atomic state occurs before the message is pushed and the waker is triggered, the receiver will not miss the message. Even if multiple threads are sending concurrently, the atomic RMW (read-modify-write) operations on the channel state ensure a consistent total order, and the waker logic ensures that the receiver is notified once, even if multiple messages arrive in quick succession [5][8][4]. The "spurious wakeup" cases are handled by the receiver's poll loop, which is designed to check for messages after being woken, maintaining correctness even if wakeups occur in ways that don't perfectly align with every individual message send [7].
Citations:
- 1: https://docs.rs/tokio/latest/src/tokio/sync/mpsc/unbounded.rs.html
- 2: https://github.com/tokio-rs/tokio/blob/master/tokio/src/sync/mpsc/unbounded.rs
- 3: https://github.com/tokio-rs/tokio/blob/5030b300/tokio/src/sync/mpsc/unbounded.rs
- 4: https://github.com/tokio-rs/tokio/blob/efe3ab679a05f3da3fcc511a44120239830254f2/tokio/src/sync/mpsc/chan.rs
- 5: https://github.com/tokio-rs/tokio/blob/master/tokio/src/sync/task/atomic_waker.rs
- 6: sync: improve AtomicWaker::wake performance tokio-rs/tokio#7450
- 7: https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.UnboundedReceiver.html
- 8: sync: Replace AtomicWaker in sync::mpsc with an algorithm derived from spmc-waker tokio-rs/tokio#8256
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== relevant inbound_relay.rs sections =="
awk 'BEGIN{FS=1} {printf "%6d %s\n", NR, $0}' src/agent/inbound_relay.rs | sed -n '50,130p'
awk 'BEGIN{FS=1} {printf "%6d %s\n", NR, $0}' src/agent/inbound_relay.rs | sed -n '160,220p'
echo "== atomic counter uses =="
awk 'BEGIN{FS=1} {printf "%6d %s\n", NR, $0}' src/agent/inbound_relay.rs | sed -n '1,160p' | rg -n "queued|fetch_add|fetch_sub|compare_exchange|send\("
echo "== deterministic probe: current queued underflow simulation =="
python3 - <<'PY'
from dataclasses import dataclass
import sys
`@dataclass`
class State:
queued: int
depth: int
def current_enqueue_before_send(s: State):
try:
s.queued = s.queued + 1
return None
except:
pass
def current_decrement(s: State):
old = s.queued
s.queued = s.queued - 1
return (old, s.queued)
state = State(queued=0, depth=0)
current_enqueue_before_send(state)
before, after = current_decrement(state)
print(f"current_decrement_from_zero: before={before}, after={after}")
print("Python cannot directly demonstrate Rust usize underflow wrap/assert here, but Python wraps negative modulo.", sys.version)
def relay_order_enqueue_after_send(decrement_order, enqueue_order):
# Model task receives/wakes before enqueue order increments local.
state = State(queued=0, depth=0)
if decrement_order == 1:
old = state.queued
state.queued = state.queued - 1
enqueue_order(state)
return False, state.queued
enqueue_order(state)
if decrement_order == 2:
old = state.queued
state.queued = state.queued - 1
return True, state.queued
return True, state.queued
valid, stale_after = relay_order_enqueue_after_send(2, lambda s: s.queued + 0)
print(f"task_decrements_after_successful_rx_without_enqueue: stale={stale_after}, actual_in_flight={1}")
PYRepository: spacedriveapp/spacebot
Length of output: 6824
Defer relay-depth accounting until the message is actually enqueued.
tx.send(message) can wake the relay task before RelayState::queued.fetch_add(...) increments. The task can then decrement the depth at RelayState::queued.fetch_sub(...) while the queue counter is still zero, which underflows in debug builds and corrupts backpressure tracking.
Increment RelayState::queued before calling tx.send(), and roll the increment back if tx.send() returns the message. Include a multi-threaded test that races the relay task against this enqueue accounting.
🤖 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/agent/inbound_relay.rs` around lines 59 - 74, Update the enqueue path in
the relay sender method containing tx.send and RelayState::queued so queued is
incremented before tx.send can wake the relay task. If sending fails and returns
the message, decrement queued to roll back the reservation before propagating
RelaySendError; preserve the warning-threshold logic using the reserved depth.
Add a multi-threaded test that races relay consumption with enqueue accounting
and verifies queued never underflows or corrupts backpressure tracking.
| let (tx, mut rx) = mpsc::unbounded_channel::<InboundMessage>(); | ||
| let state = Arc::new(RelayState { | ||
| conversation_id: conversation_id.to_string(), | ||
| agent_id: agent_id.to_string(), | ||
| queued: AtomicUsize::new(0), | ||
| warn_at: AtomicUsize::new(PRESSURE_WARN_DEPTH), | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)src/agent/inbound_relay\.rs$|InboundMessage|RelayState|PRESSURE_WARN_DEPTH' || true
echo "== outline =="
if [ -f src/agent/inbound_relay.rs ]; then
ast-grep outline src/agent/inbound_relay.rs || true
echo "== relevant lines =="
sed -n '1,220p' src/agent/inbound_relay.rs | nl -ba
fi
echo "== searches =="
rg -n "unbounded_channel|try_send|send\\(|InboundMessage|RelayState|PRESSURE_WARN_DEPTH|pressure|relay" src/agent src -S || trueRepository: spacedriveapp/spacebot
Length of output: 906
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== read inbound_relay.rs =="
sed -n '1,240p' src/agent/inbound_relay.rs
echo "== search relay symbols =="
rg -n "InboundRelay::spawn|send\\(|InboundMessage|RelaySendError|unbounded_channel|PRESSURE_WARN_DEPTH" . -SRepository: spacedriveapp/spacebot
Length of output: 48362
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== read RUST_STYLE_GUIDE around logging and send/error handling =="
sed -n '280,390p' RUST_STYLE_GUIDE.md
sed -n '740,765p' RUST_STYLE_GUIDE.md
sed -n '790,830p' RUST_STYLE_GUIDE.md
echo "== read main relay send sites =="
sed -n '2430,2545p' src/main.rsRepository: spacedriveapp/spacebot
Length of output: 11175
Bound relay ingress memory and define overload handling.
mpsc::unbounded_channel has no capacity limit, and RelayState.queued is incremented before any drop-path can occur. A stalled channel under sustained traffic can retain unlimited InboundMessages and exhaust process memory. Pressure logs do not limit relay buffering. Use bounded ingress with non-blocking try_send, then define an explicit overload policy that preserves the router non-blocking guarantee.
🤖 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/agent/inbound_relay.rs` around lines 96 - 102, Replace the unbounded
channel in the relay ingress setup with a bounded channel and use non-blocking
try_send for message admission. Define the overload behavior explicitly,
including the queued-counter update and pressure logging, so rejected messages
are handled without blocking the router or retaining unlimited InboundMessage
values; update RelayState integration accordingly.
The main select loop forwarded inbound messages with an awaited
sendinto each channel's bounded (64) queue — three sites: the inbound forward, deferred injection replay, and cross-agent injection. All three sit in the single shared router task, so one channel that stops draining (long tool call, wedged LLM stream) fills its queue and blocks the router on the await. From that moment no conversation on any agent receives messages until the stuck channel drains. Classic head-of-line blocking at the fan-in.The fix: each active channel gets an
InboundRelay(src/agent/inbound_relay.rs). The router enqueues with a synchronous, never-blocking send; a per-channel relay task owns the awaited forward into the channel's bounded queue. A saturated channel now stalls only its own relay task.Semantics preserved:
active_channelseviction, deferred-injection requeue) work unchanged, they just no longer awaitChannelitself, cron's ephemeral channels, and coalescing are untouched — the 64-slot consumer queue still bounds what the channel event loop seesOverflow beyond the bounded queue buffers in the relay with depth tracking: a warning at depth 128 and each doubling after (O(log n) log lines, not O(n)), plus a drained notice — so a stalling channel is loud instead of invisibly freezing the fleet.
Tests cover: enqueue never blocks against a full, unconsumed channel queue; FIFO holds through pressure; sends fail (and hand the message back) once the channel receiver is gone.
Note
This PR introduces an
InboundRelayto decouple the shared router task from individual channel delivery, preventing head-of-line blocking when a channel's queue saturates. The router now uses non-blocking sends; relay tasks handle bounded queue delivery asynchronously. Pressure is tracked and logged per-channel (O(log n) warnings on overflow), and all existing recovery semantics are preserved.Written by Tembo for commit 676e0add. This will update automatically on new commits.