From cc477c5678fd724c2da6078cf3b01c38c09bebd0 Mon Sep 17 00:00:00 2001 From: divesh Date: Mon, 9 Nov 2026 09:58:18 -0800 Subject: [PATCH] fix(server): allocate sandbox stream buffers lazily and share events (#5399) Create the per-sandbox broadcast channel on first subscribe instead of first publish, and store events as Arc shared by the channel ring and the replay tail. An unwatched sandbox no longer holds two pre-allocated 1024-slot rings sized for the full Sandbox payload. Signed-off-by: divesh --- crates/openshell-server/src/grpc/sandbox.rs | 4 +- crates/openshell-server/src/tracing_bus.rs | 141 ++++++++++++-------- 2 files changed, 89 insertions(+), 56 deletions(-) diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 86ff12b170..7d922034b4 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -2566,7 +2566,7 @@ pub(super) async fn handle_watch_sandbox( let mut batch = std::mem::take(&mut deferred); match first { None => {} - Some(Ok(evt)) => batch.push(evt), + Some(Ok(evt)) => batch.push(Arc::unwrap_or_clone(evt)), Some(Err(broadcast::error::RecvError::Lagged(n))) => { // Lag is recoverable: surface a warning and keep streaming. if tx @@ -2612,7 +2612,7 @@ pub(super) async fn handle_watch_sandbox( { loop { match rx.try_recv() { - Ok(evt) => batch.push(evt), + Ok(evt) => batch.push(Arc::unwrap_or_clone(evt)), Err(broadcast::error::TryRecvError::Empty) => break, // Keep draining: the receiver is usable after a skip. Err(broadcast::error::TryRecvError::Lagged(n)) => lagged += n, diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 92d30f5571..1df3034006 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -41,10 +41,17 @@ pub(crate) struct CursoredEvent { pub(crate) event: SandboxStreamEvent, } +/// Slot capacity of a per-sandbox broadcast channel. +const BROADCAST_CAPACITY: usize = 1024; + #[derive(Debug, Clone)] struct PerSandbox { - sender: broadcast::Sender, - tail: VecDeque, + /// Created on first subscribe. Every slot is allocated with the channel, + /// so a sandbox nobody watches must not pay for one. + sender: Option>>, + /// Shares each event's allocation with the channel; an inline + /// `CursoredEvent` is sized for the full `Sandbox` payload variant. + tail: VecDeque>, /// Highest seq this bus has evicted from `tail`. 0 = nothing trimmed. /// /// Under the shared cursor space each bus's tail is non-contiguous in the @@ -55,13 +62,40 @@ struct PerSandbox { impl PerSandbox { fn new() -> Self { - let (tx, _rx) = broadcast::channel(1024); Self { - sender: tx, + sender: None, tail: VecDeque::new(), last_trimmed_seq: 0, } } + + fn subscribe(&mut self) -> broadcast::Receiver> { + self.sender + .get_or_insert_with(|| broadcast::channel(BROADCAST_CAPACITY).0) + .subscribe() + } + + fn push(&mut self, cursored: CursoredEvent, tail_cap: usize) { + let cursored = Arc::new(cursored); + if let Some(sender) = &self.sender { + let _ = sender.send(Arc::clone(&cursored)); + } + self.tail.push_back(cursored); + while self.tail.len() > tail_cap { + if let Some(trimmed) = self.tail.pop_front() { + self.last_trimmed_seq = trimmed.seq; + } + } + } + + fn tail(&self, max: usize) -> Vec { + let skip = self.tail.len().saturating_sub(max); + self.tail + .iter() + .skip(skip) + .map(|cursored| (**cursored).clone()) + .collect() + } } /// The requested resume cursor is older than the oldest buffered event; @@ -156,7 +190,7 @@ impl SeqAllocator { } fn tail_after_impl( - tail: &VecDeque, + tail: &VecDeque>, last_trimmed_seq: u64, after_seq: u64, ) -> Result, ResumeGap> { @@ -176,7 +210,7 @@ fn tail_after_impl( let res: Vec = tail .iter() .filter(|cursored| cursored.seq > after_seq) - .cloned() + .map(|cursored| (**cursored).clone()) .collect(); Ok(res) @@ -210,18 +244,13 @@ impl TracingLogBus { } } - fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { + pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver> { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); inner .per_id .entry(sandbox_id.to_string()) .or_insert_with(PerSandbox::new) - .sender - .clone() - } - - pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { - self.sender_for(sandbox_id).subscribe() + .subscribe() } /// Remove all bus entries for the given sandbox id, including the platform @@ -258,11 +287,8 @@ impl TracingLogBus { inner .per_id .get(sandbox_id) - .map(|d| d.tail.iter().rev().take(max).cloned().collect::>()) + .map(|per| per.tail(max)) .unwrap_or_default() - .into_iter() - .rev() - .collect::>() } /// Identity and extent of this sandbox's current cursor space. @@ -315,19 +341,11 @@ impl TracingLogBus { event.cursor = WatchCursor::new(epoch, seq).encode(); let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); - let per = inner + inner .per_id .entry(sandbox_id.to_string()) - .or_insert_with(PerSandbox::new); - - let cursored = CursoredEvent { seq, event }; - let _ = per.sender.send(cursored.clone()); - per.tail.push_back(cursored); - while per.tail.len() > tail_cap { - if let Some(trimmed) = per.tail.pop_front() { - per.last_trimmed_seq = trimmed.seq; - } - } + .or_insert_with(PerSandbox::new) + .push(CursoredEvent { seq, event }, tail_cap); } } @@ -496,7 +514,7 @@ mod tests { .try_recv() .expect("sandbox audit event must reach live stream"); let Some(openshell_core::proto::sandbox_stream_event::Payload::Log(log)) = - live.event.payload + &live.event.payload else { panic!("expected log payload"); }; @@ -546,8 +564,8 @@ mod tests { } /// Build a contiguous tail with seqs `lo..=hi`. - fn tail_of(lo: u64, hi: u64) -> VecDeque { - (lo..=hi).map(cursored).collect() + fn tail_of(lo: u64, hi: u64) -> VecDeque> { + (lo..=hi).map(|seq| Arc::new(cursored(seq))).collect() } /// Extract seqs from a run of buffered events, in order. @@ -621,7 +639,10 @@ mod tests { // Simulate the shared cursor space: this bus only owns seqs 2 and 4 // (the other bus owns 1 and 3), and never trimmed. Resuming from 0 must // not report a gap just because seq 1 is absent here. - let tail: VecDeque = [cursored(2), cursored(4)].into_iter().collect(); + let tail: VecDeque> = [cursored(2), cursored(4)] + .into_iter() + .map(Arc::new) + .collect(); let events = tail_after_impl(&tail, 0, 0).expect("no gap"); assert_eq!(cursors(&events), vec![2, 4]); } @@ -679,6 +700,34 @@ mod tests { assert_eq!(events, vec![2]); } + #[test] + fn broadcast_channel_is_created_only_on_subscribe() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-lazy"; + let has_sender = |bus: &TracingLogBus| { + bus.inner.lock().unwrap().per_id[sandbox_id] + .sender + .is_some() + && bus.platform_event_bus.inner.lock().unwrap().per_id[sandbox_id] + .sender + .is_some() + }; + + bus.publish_external(make_log_event(sandbox_id, "unwatched")); + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); + assert!(!has_sender(&bus), "publishing must not allocate a channel"); + assert_eq!(bus.tail(sandbox_id, 10).len(), 1); + + let mut log_rx = bus.subscribe(sandbox_id); + let mut platform_rx = bus.platform_event_bus.subscribe(sandbox_id); + assert!(has_sender(&bus)); + + bus.publish_external(make_log_event(sandbox_id, "watched")); + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); + assert!(log_rx.try_recv().is_ok()); + assert!(platform_rx.try_recv().is_ok()); + } + #[test] fn tracing_log_bus_remove_cleans_up_all_maps() { let bus = TracingLogBus::new(); @@ -1019,18 +1068,13 @@ impl PlatformEventBus { } } - fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { + pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver> { let mut inner = self.inner.lock().expect("platform event bus lock poisoned"); inner .per_id .entry(sandbox_id.to_string()) .or_insert_with(PerSandbox::new) - .sender - .clone() - } - - pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { - self.sender_for(sandbox_id).subscribe() + .subscribe() } pub(crate) fn publish(&self, sandbox_id: &str, mut event: SandboxStreamEvent) { @@ -1042,19 +1086,11 @@ impl PlatformEventBus { event.cursor = WatchCursor::new(epoch, seq).encode(); let mut inner = self.inner.lock().expect("platform event bus lock poisoned"); - let per = inner + inner .per_id .entry(sandbox_id.to_string()) - .or_insert_with(PerSandbox::new); - - let cursored = CursoredEvent { seq, event }; - let _ = per.sender.send(cursored.clone()); - per.tail.push_back(cursored); - while per.tail.len() > Self::DEFAULT_TAIL { - if let Some(trimmed) = per.tail.pop_front() { - per.last_trimmed_seq = trimmed.seq; - } - } + .or_insert_with(PerSandbox::new) + .push(CursoredEvent { seq, event }, Self::DEFAULT_TAIL); } /// Return buffered platform events for replay to late subscribers. @@ -1063,11 +1099,8 @@ impl PlatformEventBus { inner .per_id .get(sandbox_id) - .map(|d| d.tail.iter().rev().take(max).cloned().collect::>()) + .map(|per| per.tail(max)) .unwrap_or_default() - .into_iter() - .rev() - .collect() } pub(crate) fn tail_after(