From 37b72a93b765e315be515967df32b2f3e22e2672 Mon Sep 17 00:00:00 2001 From: Yuriy Butenko Date: Sat, 29 Aug 2026 17:57:34 +0300 Subject: [PATCH 1/4] fix(sidecar): report a control-stream EOF instead of racing it as a shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The only `shutdown_tx` lives in the response/control reader task, so the shutdown channel closes both on a real host shutdown and when that reader dies on an abnormal EOF or decode error. The `biased` select polls the shutdown arm first, so whichever the cause, `None` meant `break 'protocol` and a 0 exit — and whether the sidecar instead surfaced the reader's `write_error_rx` report (exit 1) came down to whether the protocol task got polled in the window between the reader's `try_send` and its drop. The reader always enqueues its failure before exiting, so a queued transport error is authoritative: consult it before treating the closed shutdown channel as a clean stop. `closing_either_required_ingress_stream_is_terminal` was passing on that race, not on the invariant it asserts. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01VK2HMiUAJryC8KcZCHgUJp --- crates/native-sidecar/src/stdio.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/crates/native-sidecar/src/stdio.rs b/crates/native-sidecar/src/stdio.rs index 913b2eb022..29e28f0f30 100644 --- a/crates/native-sidecar/src/stdio.rs +++ b/crates/native-sidecar/src/stdio.rs @@ -1069,6 +1069,17 @@ async fn run_async( biased; maybe_shutdown = shutdown_rx.recv() => { let Some(control) = maybe_shutdown else { + // The only shutdown sender lives in the response/control + // reader, so this channel also closes when that reader dies + // on an abnormal EOF or decode error — a transport failure, + // not a graceful stop. The reader reports the failure on + // `write_error_rx` *before* it exits, so a queued error here + // is authoritative; without this check, whether the sidecar + // exits 0 or 1 on a control-stream EOF is a race between + // this arm and the `write_error_rx` arm below. + if let Ok(error) = write_error_rx.try_recv() { + return Err(io::Error::new(io::ErrorKind::BrokenPipe, error).into()); + } break 'protocol; }; match control.payload { From a7a5b636c573d94786bbe40d65f40764dfcf46f1 Mon Sep 17 00:00:00 2001 From: Yuriy Butenko Date: Sat, 29 Aug 2026 17:24:51 +0300 Subject: [PATCH 2/4] fix(sidecar): flush queued VM events before admitting new stdin requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit odw-43s, event-pump half. The `'protocol` loop's `tokio::select!` is `biased`, so it polls arms in declaration order and takes the first ready one. `stdin_rx` sat above both event-pump arms, so a host that pipelines requests keeps that arm permanently ready and NO VM's queued output is ever flushed: one tenant's request stream starves every other tenant's events indefinitely. Move `event_ready_rx` and `process_event_notify` above `stdin_rx`, below the control lane (cancels and permission replies must still outrank routine event flushing). Both arms are bounded and cannot starve stdin in return: the drain only empties what is already queued and never produces, and the pump is capped by `runtime.fairness.vm_quantum_operations` plus the pending-process-event capacity. The other half — a long dispatch is still awaited INLINE holding `&mut sidecar`, so an ACP prompt blocks every other VM — needs shared-state access that `dispatch_wire`, `poll_event_wire` and `pump_process_events` cannot give while they are all `&mut self`. DESIGN-43s.md specifies that restructure (per-VM `Arc>`, `&self` dispatch, actor per VM in the transport) and the regression test it unlocks. --- DESIGN-43s.md | 129 +++++++++++++++++++++++++++++ crates/native-sidecar/src/stdio.rs | 91 +++++++++++--------- 2 files changed, 180 insertions(+), 40 deletions(-) create mode 100644 DESIGN-43s.md diff --git a/DESIGN-43s.md b/DESIGN-43s.md new file mode 100644 index 0000000000..d59a65c9b3 --- /dev/null +++ b/DESIGN-43s.md @@ -0,0 +1,129 @@ +# DESIGN-43s — request concurrency for the stdio protocol loop + +Scope: `crates/native-sidecar/src/stdio.rs`, beads `odw-43s`. + +## What shipped + +The `biased` `tokio::select!` in `run_async` (`stdio.rs`, the `'protocol` loop) +polls arms in declaration order and services the first ready one. `stdin_rx` +sat **above** both event-pump arms (`event_ready_rx`, `process_event_notify`), +so a host that pipelines requests keeps `stdin_rx` permanently ready and no +VM's queued output is ever flushed — one tenant's request stream starves every +other tenant's events indefinitely. + +Fix: the two event-pump arms now sit above `stdin_rx` and below the control +lane. Both are bounded, so they cannot starve stdin in return: + +- the drain arm only empties what is already queued (it calls + `poll_event_wire`, never `pump_process_events`, so the queue can only shrink); +- the pump arm is capped by `runtime.fairness.vm_quantum_operations` per VM and + by the pending-process-event capacity. + +Ordering is now: shutdown → control lane (cancels, permission replies) → event +drain → process-event pump → new stdin requests → limit warnings → write errors. + +## What did NOT ship, and why + +A long dispatch is still awaited **inline** with `&mut sidecar` held: + +``` +'protocol loop → handle_protocol_frame(&mut sidecar, …).await + → dispatch_with_prompt_interrupt(&mut sidecar, …) + → sidecar.dispatch_wire(request).await +``` + +`NativeSidecar::dispatch_wire`, `NativeSidecar::poll_event_wire` +(`service.rs:1601`, `service.rs:1610`) and +`NativeSidecar::pump_process_events` (`execution/process_events.rs:476`) are all +`&mut self`. While the dispatch future is alive it holds the only mutable +borrow, so nothing else in the loop — no pump, no unrelated VM's request — can +touch the sidecar. `dispatch_with_prompt_interrupt` (`stdio.rs:1286`) already +shows the ceiling of what is reachable without restructuring: it can select on +`stdin_rx` beside the pinned dispatch, but every frame it reads is either a +same-request ACP interrupt or gets parked in the single-slot `pending_frame` and +waits for the dispatch to resolve. + +So an ACP prompt (minutes long) still blocks every other VM's requests *and* +their process-event pumping. Removing that needs the restructure below; it is +not a select-ordering change. + +## Design for full request concurrency + +### 1. Split the sidecar into shared state + per-VM ownership + +`NativeSidecar` (`crates/native-sidecar/src/state.rs`, `service.rs`) is one +struct holding `vms: BTreeMap` plus process-global pieces +(config, metrics, extension registry, `process_event_receiver`, +`SharedSidecarRequestClient`, `SharedEventSink`). Today every method takes +`&mut self`, so VM-local work and process-global work are indistinguishable to +the borrow checker. + +Change: + +- Move `VmState` behind `Arc>` (tokio `Mutex`; these paths await) + and keep `vms: BTreeMap>>` behind an `RwLock` so + creating/removing a VM does not block dispatches on other VMs. +- Take `&self` on `NativeSidecar` for anything that resolves a VM and then + operates under that VM's lock. `dispatch`, `dispatch_wire`, + `pump_process_events`, `poll_event`, `poll_event_wire` become `&self`. +- Keep genuinely process-global mutable pieces (`process_event_receiver`, the + disposed-session signal) in their own small mutexes rather than the outer one, + so the pump does not serialize behind a dispatch. + +Files: `state.rs` (the `VmState`/`NativeSidecar` field layout and every +`self.vms.get_mut(..)` caller), `service.rs`, `vm.rs`, +`execution/process_events.rs`, `language_execution.rs`, `filesystem.rs`, +`extension.rs` (`ExtensionHost::poll_event` and friends take `&mut self` today +and would follow). + +### 2. Actor per VM in the transport + +With `&self` dispatch, `stdio.rs` stops awaiting inline: + +- `run_async` holds `sidecar: Arc>`. +- Route each inbound `RequestFrame` by `request.ownership` to a per-VM actor: + `BTreeMap>`, one bounded channel + each (backpressure per VM, not per process). Connection- and session-scoped + frames keep a single shared lane, since they mutate the connection/session + registries. +- Each actor is a `tokio::spawn`ed task owning that VM's serial ordering: recv a + frame, `sidecar.dispatch_wire(frame).await`, write the response through the + existing `ProtocolFrameWriter` (already `Clone` + `Send`, already the + synchronization point for egress ordering). +- The `'protocol` loop keeps only: shutdown, control lane, the two event-pump + arms, routing, limit warnings, write errors. No arm awaits a dispatch, so no + arm can starve another. + +`pending_frame` disappears: the single-slot park exists only because one loop +owns both the dispatch and the stdin reader. `dispatch_with_prompt_interrupt` +shrinks to "the VM actor also selects on its own interrupt channel"; the control +lane feeds interrupts into the owning VM's actor instead of into a shared slot. + +### 3. Ordering and shutdown invariants to preserve + +- Per-VM request ordering must stay FIFO — one actor task per VM, never a task + per request. +- Events for a VM must not overtake the response to the request that produced + them; both go through `ProtocolFrameWriter`, so keep the existing + ordinary/control lane split and emit the response before returning from the + actor turn. +- `cleanup_connections` / `untrack_disposed_sessions` must join or cancel a VM's + actor before `remove_connection`, otherwise a dispatch races VM teardown. +- `active_sessions` is read by the event pump and written by + `track_session_state`; it moves into the shared lane or behind its own lock. + +### 4. Test plan for the restructure + +`crates/native-sidecar/tests/stdio_binary.rs` already spawns the real sidecar +binary over stdio and creates VMs; it is the right harness. The regression test +is: create VM A and VM B, start a blocking extension request (an ACP prompt, or +the existing `TEST_EXTENSION_NAMESPACE` blocking request) on A, then assert B's +`execute` gets an `execution_accepted` response and B's `execution_output` event +frames arrive while A's request is still outstanding. That test cannot be +written today — it deadlocks on the inline `.await` — which is precisely the +gap this design closes. + +The shipped ordering fix is not covered by a new test: it is a reorder of +`select!` arms with no new logic, and reproducing the starvation deterministically +needs a stdin flood that races the reader. Covered by the existing suite for +regressions. diff --git a/crates/native-sidecar/src/stdio.rs b/crates/native-sidecar/src/stdio.rs index 29e28f0f30..b72a1e1bb3 100644 --- a/crates/native-sidecar/src/stdio.rs +++ b/crates/native-sidecar/src/stdio.rs @@ -1110,6 +1110,57 @@ async fn run_async( } } } + // Fairness (odw-43s): the two event-pump arms sit ABOVE `stdin_rx`. + // `biased` polls arms in order and takes the first ready one, so + // with stdin first a host that pipelines requests keeps that arm + // permanently ready and NO VM's queued output is ever flushed — + // one tenant's request stream starves every other tenant's events. + // Both arms are bounded (the drain only empties what is already + // queued and never produces; the pump is capped by + // `runtime.fairness.vm_quantum_operations` and the pending-event + // capacity), so they cannot starve stdin in return. The control + // lane stays above them: cancels and permission replies must still + // outrank routine event flushing. + maybe_ready = event_ready_rx.recv() => { + let Some(()) = maybe_ready else { + break; + }; + loop { + let mut emitted_frame = false; + for session in active_sessions.iter().cloned().collect::>() { + if let Some(frame) = sidecar + .poll_event_wire(&session.ownership_scope(), Duration::ZERO) + .await? + { + send_output_frame(&frame_writer, ProtocolFrame::EventFrame(frame))?; + emitted_frame = true; + } + } + + if !emitted_frame { + break; + } + } + flush_sidecar_requests(&mut sidecar, &frame_writer)?; + } + _ = process_event_notify.notified() => { + for session in active_sessions.iter().cloned().collect::>() { + if sidecar.pump_process_events(&session.compat_ownership_scope()).await? { + match event_ready_tx.try_send(()) { + Ok(()) + | Err(tokio::sync::mpsc::error::TrySendError::Full(())) => {} + Err(tokio::sync::mpsc::error::TrySendError::Closed(())) => { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "event-ready wake receiver closed", + ) + .into()); + } + } + } + } + flush_sidecar_requests(&mut sidecar, &frame_writer)?; + } maybe_frame = stdin_rx.recv(), if !stdin_closed => { match maybe_frame { Some(frame) => { @@ -1168,46 +1219,6 @@ async fn run_async( } } } - maybe_ready = event_ready_rx.recv() => { - let Some(()) = maybe_ready else { - break; - }; - loop { - let mut emitted_frame = false; - for session in active_sessions.iter().cloned().collect::>() { - if let Some(frame) = sidecar - .poll_event_wire(&session.ownership_scope(), Duration::ZERO) - .await? - { - send_output_frame(&frame_writer, ProtocolFrame::EventFrame(frame))?; - emitted_frame = true; - } - } - - if !emitted_frame { - break; - } - } - flush_sidecar_requests(&mut sidecar, &frame_writer)?; - } - _ = process_event_notify.notified() => { - for session in active_sessions.iter().cloned().collect::>() { - if sidecar.pump_process_events(&session.compat_ownership_scope()).await? { - match event_ready_tx.try_send(()) { - Ok(()) - | Err(tokio::sync::mpsc::error::TrySendError::Full(())) => {} - Err(tokio::sync::mpsc::error::TrySendError::Closed(())) => { - return Err(io::Error::new( - io::ErrorKind::BrokenPipe, - "event-ready wake receiver closed", - ) - .into()); - } - } - } - } - flush_sidecar_requests(&mut sidecar, &frame_writer)?; - } maybe_write_error = write_error_rx.recv() => { if let Some(error) = maybe_write_error { return Err(io::Error::new(io::ErrorKind::BrokenPipe, error).into()); From 19ed09128d797246f1b09a2edad59f1d5cab4f98 Mon Sep 17 00:00:00 2001 From: Yuriy Butenko Date: Sat, 29 Aug 2026 17:58:32 +0300 Subject: [PATCH 3/4] fix(sidecar): round-robin stdin and the process-event pump, not rank them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 27a1199 moved `event_ready_rx` and `process_event_notify` above `stdin_rx` in the `biased` select. That inverts the starvation rather than removing it, and the inverted form is the worse one: neither event arm is self-limiting under a guest emitting sustained stdout. `queue_pending_execution_event` notifies for every queued event (execution/process.rs:243, :374) and `pump_process_events` re-arms the notify whenever a VM burns its `vm_quantum_operations` quantum (execution/process_events.rs:531), so a permit is stored at essentially all times; and the drain is no pure drain either — a zero-timeout `poll_event` still pulls `process_event_receiver` into `pending_process_events` (service.rs:1694-1717), so the guest refills it as fast as the loop empties it. Above stdin, those arms alternate forever and `stdin_rx.recv()` is never polled. `stdin_rx` is also the cancel lane, contrary to the comment that shipped: `route_decoded_combined_frame` sends every `RequestFrame` — including `RequestPayload::CancelExecution` — to `stdin_tx`, while the control lane admits only `SidecarResponseFrame` and shutdown. So one untrusted guest could stall every other tenant's requests on a shared sidecar and make its own flood uncancellable. Restore the arm order and make the two lanes alternate instead: `service_process_events` does one bounded round (pump every active session, then emit at most `runtime.fairness.vm_quantum_operations` frames, re-arming `event_ready_tx` for whatever the cap leaves), called once per `'protocol` turn at the top of the loop. Every frame-handling path — the `pending_frame` park, the stdin arm, the control lane — returns through it, so a pipelining host can no longer keep queued VM output unflushed, and the cap keeps the converse true. The wake arms now only consume the edge that woke the loop. DESIGN-43s.md §"What shipped" is corrected on both counts. A new unit test pins the routing fact the ordering rests on: cancels are admitted on the stdin lane, not the control lane. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01VK2HMiUAJryC8KcZCHgUJp --- DESIGN-43s.md | 75 +++++++--- crates/native-sidecar/src/stdio.rs | 215 ++++++++++++++++++++++------- 2 files changed, 219 insertions(+), 71 deletions(-) diff --git a/DESIGN-43s.md b/DESIGN-43s.md index d59a65c9b3..e02def75f0 100644 --- a/DESIGN-43s.md +++ b/DESIGN-43s.md @@ -5,22 +5,52 @@ Scope: `crates/native-sidecar/src/stdio.rs`, beads `odw-43s`. ## What shipped The `biased` `tokio::select!` in `run_async` (`stdio.rs`, the `'protocol` loop) -polls arms in declaration order and services the first ready one. `stdin_rx` -sat **above** both event-pump arms (`event_ready_rx`, `process_event_notify`), -so a host that pipelines requests keeps `stdin_rx` permanently ready and no -VM's queued output is ever flushed — one tenant's request stream starves every -other tenant's events indefinitely. - -Fix: the two event-pump arms now sit above `stdin_rx` and below the control -lane. Both are bounded, so they cannot starve stdin in return: - -- the drain arm only empties what is already queued (it calls - `poll_event_wire`, never `pump_process_events`, so the queue can only shrink); -- the pump arm is capped by `runtime.fairness.vm_quantum_operations` per VM and - by the pending-process-event capacity. - -Ordering is now: shutdown → control lane (cancels, permission replies) → event -drain → process-event pump → new stdin requests → limit warnings → write errors. +polls arms in declaration order and services the first ready one. With +`stdin_rx` **above** both event-pump arms (`event_ready_rx`, +`process_event_notify`), a host that pipelines requests keeps `stdin_rx` +permanently ready and no VM's queued output is ever flushed — one tenant's +request stream starves every other tenant's events indefinitely. + +Reordering the arms does not fix that; it inverts which lane starves, and the +inverted form is the worse of the two. Neither event arm is self-limiting under +load: + +- `ActiveProcess::queue_pending_execution_event` calls + `process_event_notify.notify_one()` for *every* queued execution event + (`execution/process.rs:243`, `:374`), and `pump_process_events` re-arms the + notify itself whenever a VM burns its `runtime.fairness.vm_quantum_operations` + quantum (`execution/process_events.rs:531`). A guest emitting sustained stdout + keeps a `Notify` permit stored at essentially all times. +- The drain is not a pure drain either: `poll_event` with a zero timeout still + pulls `process_event_receiver` into `pending_process_events` + (`service.rs:1694-1717`), so the guest refills it as fast as the loop empties + it. + +And `stdin_rx` is the cancel lane, not merely the new-work lane: +`route_decoded_combined_frame` (`stdio.rs`) sends every +`ProtocolFrame::RequestFrame` — `cancel_execution` included +(`language_execution.rs:1637`) — to `stdin_tx`, while +`route_decoded_control_frame` admits only `SidecarResponseFrame` and shutdown +`ControlFrame`s to `stdin_control_tx`. Put above stdin, the event arms let one +untrusted guest stall the whole multi-tenant process's request lane *and* make +its own flood uncancellable. + +The fix is therefore a round-robin, not a priority swap: + +- `service_process_events` (`stdio.rs`) performs one **bounded** round — pump + every active session, then emit at most + `runtime.fairness.vm_quantum_operations` event frames. Work left over by the + cap re-arms `event_ready_tx` so the next turn resumes it. +- The `'protocol` loop calls it once per turn, at the top, before taking on more + work. Every path that handles a frame (the `pending_frame` park, the stdin + arm, the control lane) comes back through there, so a pipelining host can no + longer keep queued output unflushed. +- Arm order is unchanged from before this work — both event-pump arms stay below + `stdin_rx` — and their bodies now only consume the edge that woke the loop. + +Ordering is: shutdown → control lane (sidecar responses, permission replies) → +stdin requests (cancels included) → event-pump wakes → limit warnings → write +errors, with one bounded event flush per turn regardless of which arm fired. ## What did NOT ship, and why @@ -123,7 +153,12 @@ frames arrive while A's request is still outstanding. That test cannot be written today — it deadlocks on the inline `.await` — which is precisely the gap this design closes. -The shipped ordering fix is not covered by a new test: it is a reorder of -`select!` arms with no new logic, and reproducing the starvation deterministically -needs a stdin flood that races the reader. Covered by the existing suite for -regressions. +The shipped fairness round is covered indirectly. Reproducing either starvation +deterministically needs a real guest flooding stdout racing a pipelining host — +`stdio_binary.rs` can host that test, but only once the restructure above makes +the outcome deterministic rather than timing-dependent. What *is* pinned today +is the fact the arm order rests on: +`cancel_execution_is_admitted_on_the_stdin_lane_not_the_control_lane` +(`stdio.rs` unit tests) fails if cancels ever move off `stdin_tx`, which is the +only condition under which demoting stdin below the event arms would become +defensible. diff --git a/crates/native-sidecar/src/stdio.rs b/crates/native-sidecar/src/stdio.rs index b72a1e1bb3..4849e26eef 100644 --- a/crates/native-sidecar/src/stdio.rs +++ b/crates/native-sidecar/src/stdio.rs @@ -1047,6 +1047,22 @@ async fn run_async( let mut limit_warning_closed = false; let mut stdin_closed = false; 'protocol: loop { + // Fairness (odw-43s): service queued VM events once per turn, before + // taking on more work. Doing it here rather than only in the wake arms + // below is what makes the two lanes round-robin: every path that + // handles a frame — the `pending_frame` park, the stdin arm, the + // control lane — comes back through here, so a host that pipelines + // requests can no longer keep queued VM output unflushed. The call is + // bounded (see `service_process_events`), so the reverse cannot happen + // either. + service_process_events( + &mut sidecar, + &frame_writer, + &active_sessions, + &event_ready_tx, + ) + .await?; + if let Some(frame) = pending_frame.take() { handle_protocol_frame( frame, @@ -1110,57 +1126,6 @@ async fn run_async( } } } - // Fairness (odw-43s): the two event-pump arms sit ABOVE `stdin_rx`. - // `biased` polls arms in order and takes the first ready one, so - // with stdin first a host that pipelines requests keeps that arm - // permanently ready and NO VM's queued output is ever flushed — - // one tenant's request stream starves every other tenant's events. - // Both arms are bounded (the drain only empties what is already - // queued and never produces; the pump is capped by - // `runtime.fairness.vm_quantum_operations` and the pending-event - // capacity), so they cannot starve stdin in return. The control - // lane stays above them: cancels and permission replies must still - // outrank routine event flushing. - maybe_ready = event_ready_rx.recv() => { - let Some(()) = maybe_ready else { - break; - }; - loop { - let mut emitted_frame = false; - for session in active_sessions.iter().cloned().collect::>() { - if let Some(frame) = sidecar - .poll_event_wire(&session.ownership_scope(), Duration::ZERO) - .await? - { - send_output_frame(&frame_writer, ProtocolFrame::EventFrame(frame))?; - emitted_frame = true; - } - } - - if !emitted_frame { - break; - } - } - flush_sidecar_requests(&mut sidecar, &frame_writer)?; - } - _ = process_event_notify.notified() => { - for session in active_sessions.iter().cloned().collect::>() { - if sidecar.pump_process_events(&session.compat_ownership_scope()).await? { - match event_ready_tx.try_send(()) { - Ok(()) - | Err(tokio::sync::mpsc::error::TrySendError::Full(())) => {} - Err(tokio::sync::mpsc::error::TrySendError::Closed(())) => { - return Err(io::Error::new( - io::ErrorKind::BrokenPipe, - "event-ready wake receiver closed", - ) - .into()); - } - } - } - } - flush_sidecar_requests(&mut sidecar, &frame_writer)?; - } maybe_frame = stdin_rx.recv(), if !stdin_closed => { match maybe_frame { Some(frame) => { @@ -1181,6 +1146,21 @@ async fn run_async( None => stdin_closed = true, } } + // Event-pump wakes. These stay BELOW `stdin_rx`: a guest emitting + // sustained stdout keeps `process_event_notify` armed at essentially + // all times (every queued execution event notifies, and + // `pump_process_events` re-arms whenever a VM burns its quantum), so + // above stdin they would win the `biased` select forever and stall + // the request lane — which is also the cancel lane, the only way to + // stop that guest. The arms carry no work: the flush happens once + // per turn at the top of the `'protocol` loop, so consuming the edge + // that woke the loop is all they have to do here. + maybe_ready = event_ready_rx.recv() => { + let Some(()) = maybe_ready else { + break; + }; + } + _ = process_event_notify.notified() => {} maybe_warning = limit_warning_rx.recv(), if !limit_warning_closed => { match maybe_warning { Some(warning) => { @@ -1808,6 +1788,81 @@ fn enqueue_stdin_frame( }) } +/// One bounded round of process-event servicing: pump every active session's +/// queued execution events into the durable sidecar queues, then emit at most +/// `runtime.fairness.vm_quantum_operations` event frames. +/// +/// Fairness (odw-43s). The `'protocol` loop calls this once per turn, so a host +/// that pipelines stdin frames can no longer keep queued VM output unflushed. +/// The frame cap is what keeps the converse true, and it is load-bearing rather +/// than defensive: neither half of this work is self-limiting under a guest +/// emitting sustained stdout. `ActiveProcess::queue_pending_execution_event` +/// notifies for *every* queued event and `pump_process_events` re-arms the +/// notify itself whenever a VM burns its `vm_quantum_operations` quantum, so +/// the wake edge is essentially always armed; and a zero-timeout +/// `poll_event_wire` is not a pure drain either — `poll_event` still pulls +/// `process_event_receiver` into `pending_process_events`, so the guest can +/// refill it as fast as this loop empties it. Draining to exhaustion here would +/// therefore hold the loop off `stdin_rx` indefinitely, and `stdin_rx` is the +/// lane that carries cancels: `route_decoded_combined_frame` sends every +/// `RequestFrame` — `cancel_execution` included — to `stdin_tx`, while the +/// control lane admits only sidecar responses and shutdown. Whatever the cap +/// leaves behind re-arms `event_ready_tx`, so the next turn resumes it after +/// stdin has had its own. +async fn service_process_events( + sidecar: &mut NativeSidecar, + writer: &ProtocolFrameWriter, + active_sessions: &BTreeSet, + event_ready_tx: &Sender<()>, +) -> Result<(), Box> { + let sessions = active_sessions.iter().cloned().collect::>(); + for session in &sessions { + sidecar + .pump_process_events(&session.compat_ownership_scope()) + .await?; + } + + let frame_budget = sidecar.config.runtime.fairness.vm_quantum_operations.max(1); + let mut emitted = 0usize; + let mut more_pending = false; + 'drain: loop { + let mut emitted_frame = false; + for session in &sessions { + if let Some(frame) = sidecar + .poll_event_wire(&session.ownership_scope(), Duration::ZERO) + .await? + { + send_output_frame(writer, ProtocolFrame::EventFrame(frame))?; + emitted_frame = true; + emitted += 1; + if emitted >= frame_budget { + more_pending = true; + break 'drain; + } + } + } + + if !emitted_frame { + break; + } + } + + if more_pending { + match event_ready_tx.try_send(()) { + Ok(()) | Err(tokio::sync::mpsc::error::TrySendError::Full(())) => {} + Err(tokio::sync::mpsc::error::TrySendError::Closed(())) => { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "event-ready wake receiver closed", + ) + .into()); + } + } + } + + flush_sidecar_requests(sidecar, writer) +} + fn flush_sidecar_requests( sidecar: &mut NativeSidecar, writer: &ProtocolFrameWriter, @@ -2067,6 +2122,64 @@ mod tests { ); } + /// The `'protocol` loop's arm order (odw-43s) rests on this: the event-pump + /// wakes sit BELOW `stdin_rx` because `stdin_rx` is the lane a cancel + /// arrives on. Demoting stdin under the event arms would leave a guest + /// flooding stdout uncancellable, since its own flood keeps those arms + /// ready. If cancels ever move to the control lane, this test fails and the + /// ordering argument can be revisited. + #[test] + fn cancel_execution_is_admitted_on_the_stdin_lane_not_the_control_lane() { + let transport = test_callback_transport(FrameSidecarRequestLimits { + max_pending_responses: 4, + max_pending_response_bytes: 4096, + max_frame_bytes: 4096, + }); + let ingress_budget = test_protocol_budget(4, 4096, "test ordinary ingress"); + let control_budget = test_protocol_budget(4, 4096, "test control ingress"); + let (ordinary_tx, mut ordinary_rx) = + channel::, String>>(4); + let (control_tx, mut control_rx) = channel::(4); + let (shutdown_tx, _shutdown_rx) = channel::(1); + let (overload_tx, _overload_rx) = test_frame_writer(4); + + let cancel = ProtocolFrame::RequestFrame(request_frame( + 7, + vm_ownership("conn-1", "session-1", "vm-1"), + RequestPayload::CancelExecutionRequest(wire::CancelExecutionRequest { + execution_id: String::from("operation-1-1"), + }), + )); + assert_eq!( + route_decoded_combined_frame( + test_decoded_frame(cancel), + &ordinary_tx, + &transport, + &control_tx, + &shutdown_tx, + &overload_tx, + &ingress_budget, + &control_budget, + ), + StdinReaderFlow::Continue, + ); + + assert!( + control_rx.try_recv().is_err(), + "cancel_execution must not be admitted on the response/control lane" + ); + let Ok(Ok(Some(routed))) = ordinary_rx.try_recv() else { + panic!("cancel_execution must be admitted on the stdin request lane"); + }; + let ProtocolFrame::RequestFrame(request) = routed.frame else { + panic!("the stdin lane must carry the request frame itself"); + }; + assert!( + matches!(request.payload, RequestPayload::CancelExecutionRequest(_)), + "the stdin lane must carry the cancel, not a substitute frame" + ); + } + #[test] fn stdio_work_queues_are_bounded() { let capacity = agentos_runtime::DEFAULT_PROTOCOL_MAX_INGRESS_FRAMES; From 2350d635c672aa746e647b3a3186dd8629c724c8 Mon Sep 17 00:00:00 2001 From: Yuriy Butenko Date: Sat, 29 Aug 2026 20:27:21 +0300 Subject: [PATCH 4/4] fix(sidecar): rotate the event round's starting session so the frame cap cannot pin one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `service_process_events` caps a round at `runtime.fairness.vm_quantum_operations` frames (default 64, crates/runtime/src/lib.rs:83) and the cap is global across sessions, not per session. The emit pass walks the sessions one frame each, so they share a round fairly, but a round that hits the cap breaks partway down the list — and the list was rebuilt in `BTreeSet` order every round, so the break always fell in the same place. With more than 64 sessions emitting, every round served the same leading 64 and the tail was never reached: the fairness fix introduced its own starvation, just one lane over from the one it removed. `rotated_sessions` advances the round's starting session by one per call, so the cut moves instead of pinning. Keeping the cap global (rather than per session) is deliberate: a per-session cap would make one round cost `sessions * quantum` frames and put the cancel lane back behind a wait that grows with tenant count. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01VK2HMiUAJryC8KcZCHgUJp --- DESIGN-43s.md | 6 +++ crates/native-sidecar/src/stdio.rs | 59 +++++++++++++++++++++++++++++- 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/DESIGN-43s.md b/DESIGN-43s.md index e02def75f0..97edf3df94 100644 --- a/DESIGN-43s.md +++ b/DESIGN-43s.md @@ -41,6 +41,12 @@ The fix is therefore a round-robin, not a priority swap: every active session, then emit at most `runtime.fairness.vm_quantum_operations` event frames. Work left over by the cap re-arms `event_ready_tx` so the next turn resumes it. +- That cap is global across sessions, not per session, so one round stays cheap + however many tenants are attached. The emit pass itself walks the sessions one + frame at a time, so they share a round fairly; `rotated_sessions` then advances + the round's starting session by one per call so the point where the cap cuts + the list moves. Without that rotation, more than `vm_quantum_operations` + emitting sessions would leave everything past the cap permanently unserved. - The `'protocol` loop calls it once per turn, at the top, before taking on more work. Every path that handles a frame (the `pending_frame` park, the stdin arm, the control lane) comes back through there, so a pipelining host can no diff --git a/crates/native-sidecar/src/stdio.rs b/crates/native-sidecar/src/stdio.rs index 4849e26eef..7aaa07f1a9 100644 --- a/crates/native-sidecar/src/stdio.rs +++ b/crates/native-sidecar/src/stdio.rs @@ -1046,6 +1046,7 @@ async fn run_async( let mut pending_frame: Option = None; let mut limit_warning_closed = false; let mut stdin_closed = false; + let mut event_round_rotation = 0usize; 'protocol: loop { // Fairness (odw-43s): service queued VM events once per turn, before // taking on more work. Doing it here rather than only in the wake arms @@ -1060,6 +1061,7 @@ async fn run_async( &frame_writer, &active_sessions, &event_ready_tx, + &mut event_round_rotation, ) .await?; @@ -1788,6 +1790,23 @@ fn enqueue_stdin_frame( }) } +/// The emit round in `service_process_events` walks the sessions one frame at a +/// time, so sessions are served fairly *within* a round — but the frame budget is +/// global, so a round that hits the cap stops partway down the list. Always +/// starting at the same end would therefore starve every session past the cap +/// outright once more than `vm_quantum_operations` sessions (default 64, +/// `crates/runtime/src/lib.rs`) are emitting: every round would serve the same +/// leading 64 and never reach the rest. Advancing the starting point by one per +/// round moves the cut instead of pinning it. +fn rotated_sessions(sessions: &BTreeSet, rotation: usize) -> Vec { + let mut sessions = sessions.iter().cloned().collect::>(); + let len = sessions.len(); + if len > 0 { + sessions.rotate_left(rotation % len); + } + sessions +} + /// One bounded round of process-event servicing: pump every active session's /// queued execution events into the durable sidecar queues, then emit at most /// `runtime.fairness.vm_quantum_operations` event frames. @@ -1809,13 +1828,20 @@ fn enqueue_stdin_frame( /// control lane admits only sidecar responses and shutdown. Whatever the cap /// leaves behind re-arms `event_ready_tx`, so the next turn resumes it after /// stdin has had its own. +/// +/// The cap is global across sessions rather than per session — a per-session cap +/// would make one round cost `sessions * quantum` frames and put the cancel lane +/// back behind a wait that grows with tenant count. `rotated_sessions` is what +/// keeps the global cap fair across sessions; see its comment. async fn service_process_events( sidecar: &mut NativeSidecar, writer: &ProtocolFrameWriter, active_sessions: &BTreeSet, event_ready_tx: &Sender<()>, + rotation: &mut usize, ) -> Result<(), Box> { - let sessions = active_sessions.iter().cloned().collect::>(); + let sessions = rotated_sessions(active_sessions, *rotation); + *rotation = rotation.wrapping_add(1); for session in &sessions { sidecar .pump_process_events(&session.compat_ownership_scope()) @@ -2180,6 +2206,37 @@ mod tests { ); } + /// The emit budget in `service_process_events` is global, so the round's + /// starting session must advance — otherwise every session past the cap + /// starves permanently once more than `vm_quantum_operations` sessions are + /// emitting. + #[test] + fn event_emission_rounds_start_at_a_rotating_session() { + let sessions = (0..3) + .map(|index| SessionScope { + connection_id: String::from("conn-1"), + session_id: format!("session-{index}"), + }) + .collect::>(); + + let starts = (0..4) + .map(|rotation| rotated_sessions(&sessions, rotation)[0].session_id.clone()) + .collect::>(); + assert_eq!(starts, ["session-0", "session-1", "session-2", "session-0"]); + + // Every round still offers every session; only the order shifts. + for rotation in 0..4 { + assert_eq!( + rotated_sessions(&sessions, rotation) + .into_iter() + .collect::>(), + sessions, + ); + } + + assert!(rotated_sessions(&BTreeSet::new(), 7).is_empty()); + } + #[test] fn stdio_work_queues_are_bounded() { let capacity = agentos_runtime::DEFAULT_PROTOCOL_MAX_INGRESS_FRAMES;