diff --git a/DESIGN-43s.md b/DESIGN-43s.md new file mode 100644 index 0000000000..97edf3df94 --- /dev/null +++ b/DESIGN-43s.md @@ -0,0 +1,170 @@ +# 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. 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. +- 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 + 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 + +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 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 913b2eb022..7aaa07f1a9 100644 --- a/crates/native-sidecar/src/stdio.rs +++ b/crates/native-sidecar/src/stdio.rs @@ -1046,7 +1046,25 @@ 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 + // 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, + &mut event_round_rotation, + ) + .await?; + if let Some(frame) = pending_frame.take() { handle_protocol_frame( frame, @@ -1069,6 +1087,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 { @@ -1119,6 +1148,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) => { @@ -1157,46 +1201,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()); @@ -1786,6 +1790,105 @@ 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. +/// +/// 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. +/// +/// 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 = rotated_sessions(active_sessions, *rotation); + *rotation = rotation.wrapping_add(1); + 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, @@ -2045,6 +2148,95 @@ 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" + ); + } + + /// 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;