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(