fix(server): allocate sandbox stream buffers lazily and share events

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,

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<CursoredEvent>,
-    tail: VecDeque<CursoredEvent>,
+    /// Created on first subscribe. Every slot is allocated with the channel,
+    /// so a sandbox nobody watches must not pay for one.
+    sender: Option<broadcast::Sender<Arc<CursoredEvent>>>,
+    /// Shares each event's allocation with the channel; an inline
+    /// `CursoredEvent` is sized for the full `Sandbox` payload variant.
+    tail: VecDeque<Arc<CursoredEvent>>,
     /// 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<Arc<CursoredEvent>> {
+        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<CursoredEvent> {
+        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<CursoredEvent>,
+    tail: &VecDeque<Arc<CursoredEvent>>,
     last_trimmed_seq: u64,
     after_seq: u64,
 ) -> Result<Vec<CursoredEvent>, ResumeGap> {
@@ -176,7 +210,7 @@ fn tail_after_impl(
     let res: Vec<CursoredEvent> = 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<CursoredEvent> {
+    pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver<Arc<CursoredEvent>> {
         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<CursoredEvent> {
-        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::<Vec<_>>())
+            .map(|per| per.tail(max))
             .unwrap_or_default()
-            .into_iter()
-            .rev()
-            .collect::<Vec<CursoredEvent>>()
     }
 
     /// 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<CursoredEvent> {
-        (lo..=hi).map(cursored).collect()
+    fn tail_of(lo: u64, hi: u64) -> VecDeque<Arc<CursoredEvent>> {
+        (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<CursoredEvent> = [cursored(2), cursored(4)].into_iter().collect();
+        let tail: VecDeque<Arc<CursoredEvent>> = [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<CursoredEvent> {
+    pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver<Arc<CursoredEvent>> {
         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<CursoredEvent> {
-        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::<Vec<_>>())
+            .map(|per| per.tail(max))
             .unwrap_or_default()
-            .into_iter()
-            .rev()
-            .collect()
     }
 
     pub(crate) fn tail_after(