From 12383ea351271072c6afe7f7b4dad4dac105d2df Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Thu, 3 Sep 2026 12:08:19 +0100 Subject: [PATCH 1/9] fix(api): emit warning on WatchSandbox broadcast lag instead of terminating Broadcast lag on the status, log, and platform receivers was converted to a RESOURCE_EXHAUSTED status that terminated the whole watch stream. Lag is recoverable: the receiver resumes at the oldest surviving message. Emit a SandboxStreamWarning and continue streaming instead; keep terminating on Closed. Add helpers and unit tests covering the warning payload and receiver recovery after lag. Partially addresses #3055 (cursor/resume follow up separately). Signed-off-by: Artem Lytvyn --- crates/openshell-server/src/grpc/sandbox.rs | 32 +++++++-- crates/openshell-server/src/sandbox_watch.rs | 69 +++++++++++++++++--- 2 files changed, 86 insertions(+), 15 deletions(-) diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index a8d198bc80..2597f7f0e8 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -41,7 +41,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Poll}; use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{broadcast, mpsc, oneshot}; use tokio_stream::wrappers::ReceiverStream; use tonic::{Request, Response, Status}; use tracing::{debug, info, warn}; @@ -1125,8 +1125,14 @@ pub(super) async fn handle_watch_sandbox( } } } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; + Err(broadcast::error::RecvError::Lagged(n)) => { + // Lag is recoverable: surface a warning and keep streaming. + if tx.send(Ok(crate::sandbox_watch::lag_warning_event(n))).await.is_err() { + return; + } + } + Err(broadcast::error::RecvError::Closed) => { + let _ = tx.send(Err(Status::cancelled("stream closed"))).await; return; } } @@ -1151,8 +1157,14 @@ pub(super) async fn handle_watch_sandbox( return; } } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; + Err(broadcast::error::RecvError::Lagged(n)) => { + // Lag is recoverable: surface a warning and keep streaming. + if tx.send(Ok(crate::sandbox_watch::lag_warning_event(n))).await.is_err() { + return; + } + }, + Err(broadcast::error::RecvError::Closed) => { + let _ = tx.send(Err(Status::cancelled("stream closed"))).await; return; } } @@ -1169,8 +1181,14 @@ pub(super) async fn handle_watch_sandbox( return; } } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; + Err(broadcast::error::RecvError::Lagged(n)) => { + // Lag is recoverable: surface a warning and keep streaming. + if tx.send(Ok(crate::sandbox_watch::lag_warning_event(n))).await.is_err() { + return; + } + }, + Err(broadcast::error::RecvError::Closed) => { + let _ = tx.send(Err(Status::cancelled("stream closed"))).await; return; } } diff --git a/crates/openshell-server/src/sandbox_watch.rs b/crates/openshell-server/src/sandbox_watch.rs index ac38eba8db..a80d21d653 100644 --- a/crates/openshell-server/src/sandbox_watch.rs +++ b/crates/openshell-server/src/sandbox_watch.rs @@ -6,8 +6,8 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; +use openshell_core::proto::SandboxStreamWarning; use tokio::sync::broadcast; -use tonic::Status; /// Broadcast bus of sandbox updates keyed by sandbox id. /// @@ -26,6 +26,7 @@ impl SandboxWatchBus { } } + /// Private method to register sandbox in the `SandboxWatchBus` registry if it does not exist. fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender<()> { let mut inner = self.inner.lock().expect("sandbox watch bus lock poisoned"); inner @@ -59,13 +60,22 @@ impl SandboxWatchBus { } } -/// Helper to translate broadcast lag into a gRPC status. -pub fn broadcast_to_status(err: broadcast::error::RecvError) -> Status { - match err { - broadcast::error::RecvError::Closed => Status::cancelled("stream closed"), - broadcast::error::RecvError::Lagged(n) => { - Status::resource_exhausted(format!("watch stream lagged; dropped {n} messages")) - } +/// Build the warning payload emitted when a watch broadcast receiver lags. +/// +/// Broadcast lag is recoverable: the receiver skips ahead to the oldest +/// surviving message, so the stream continues after surfacing this warning +/// instead of terminating. +pub fn lag_warning(n: u64) -> SandboxStreamWarning { + SandboxStreamWarning { + message: format!("watch stream lagged; dropped {n} messages"), + } +} + +/// Wrap [`lag_warning`] in a `SandboxStreamEvent` ready to send on the stream. +pub fn lag_warning_event(n: u64) -> openshell_core::proto::SandboxStreamEvent { + use openshell_core::proto::sandbox_stream_event::Payload; + openshell_core::proto::SandboxStreamEvent { + payload: Some(Payload::Warning(lag_warning(n))), } } @@ -114,4 +124,47 @@ mod tests { // Should not panic bus.remove("nonexistent"); } + + #[test] + fn lag_warning_reports_dropped_count() { + let warning = lag_warning(7); + assert!( + warning.message.contains('7'), + "message: {}", + warning.message + ); + assert!( + warning.message.contains("lagged"), + "message: {}", + warning.message + ); + } + + #[test] + fn lag_warning_event_wraps_warning_payload() { + use openshell_core::proto::sandbox_stream_event::Payload; + let evt = lag_warning_event(3); + match evt.payload { + Some(Payload::Warning(w)) => assert!(w.message.contains('3')), + other => panic!("expected Warning payload, got {other:?}"), + } + } + + // Broadcast lag is recoverable at the tokio layer: after `Lagged`, the same + // receiver keeps yielding the oldest surviving messages instead of closing. + #[tokio::test] + async fn lagged_receiver_recovers_after_lag() { + const N: usize = 4; + let (tx, mut rx) = broadcast::channel(N); + for _ in 0..=N { + let _ = tx.send(()); + } + + let err = rx.recv().await.expect_err("expected Lagged"); + assert!(matches!(err, broadcast::error::RecvError::Lagged(_))); + + // The receiver is still usable: after lag it resumes at the oldest + // surviving message instead of closing. + assert!(rx.recv().await.is_ok(), "receiver should recover after lag"); + } } From 7209f30844b9008b0efcc1694d79989aa8e131d9 Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Thu, 3 Sep 2026 21:13:58 +0100 Subject: [PATCH 2/9] refactor(server): group per-sandbox log bus state and stamp sequence numbers Signed-off-by: Artem Lytvyn --- crates/openshell-server/src/tracing_bus.rs | 49 ++++++++++++++++------ 1 file changed, 36 insertions(+), 13 deletions(-) diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index a91a5fd877..2fb865f1d8 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -22,8 +22,14 @@ pub struct TracingLogBus { #[derive(Debug)] struct Inner { - per_id: HashMap>, - tails: HashMap>, + per_id: HashMap, +} + +#[derive(Debug)] +struct PerSandbox { + sender: broadcast::Sender, + tail: VecDeque<(u64, SandboxStreamEvent)>, + next_seq: u64, } impl Default for TracingLogBus { @@ -37,8 +43,7 @@ impl TracingLogBus { pub fn new() -> Self { Self { inner: Arc::new(Mutex::new(Inner { - per_id: HashMap::new(), - tails: HashMap::new(), + per_id: HashMap::::new(), })), platform_event_bus: PlatformEventBus::new(), } @@ -58,8 +63,15 @@ impl TracingLogBus { .entry(sandbox_id.to_string()) .or_insert_with(|| { let (tx, _rx) = broadcast::channel(1024); - tx + PerSandbox { + sender: tx, + tail: VecDeque::new(), + // Seq starts at 1 so the proto default resume_after_cursor + // (0) means "from the beginning" without skipping event 1. + next_seq: 1, + } }) + .sender .clone() } @@ -74,19 +86,25 @@ impl TracingLogBus { pub fn remove(&self, sandbox_id: &str) { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); inner.per_id.remove(sandbox_id); - inner.tails.remove(sandbox_id); } pub fn tail(&self, sandbox_id: &str, max: usize) -> Vec { let inner = self.inner.lock().expect("tracing bus lock poisoned"); inner - .tails + .per_id .get(sandbox_id) - .map(|d| d.iter().rev().take(max).cloned().collect::>()) + .map(|d| { + d.tail + .iter() + .rev() + .take(max) + .map(|(_seq, event)| event.clone()) + .collect::>() + }) .unwrap_or_default() .into_iter() .rev() - .collect() + .collect::>() } /// Publish a log line from an external source (e.g., sandbox push). @@ -111,10 +129,15 @@ impl TracingLogBus { let _ = tx.send(event.clone()); let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); - let deque = inner.tails.entry(sandbox_id.to_string()).or_default(); - deque.push_back(event); - while deque.len() > tail_cap { - deque.pop_front(); + let per_sandbox = inner + .per_id + .get_mut(sandbox_id) + .expect("sender_for inserted the entry above"); + let seq = per_sandbox.next_seq; + per_sandbox.next_seq += 1; + per_sandbox.tail.push_back((seq, event)); + while per_sandbox.tail.len() > tail_cap { + per_sandbox.tail.pop_front(); } } } From 345ca7c2975a07b5055bddfca03b9d665d6c412d Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Fri, 4 Sep 2026 11:33:21 +0100 Subject: [PATCH 3/9] feat(proto): add resume cursor fields to sandbox watch API Signed-off-by: Artem Lytvyn --- crates/openshell-cli/src/run.rs | 3 +++ .../sandbox_create_lifecycle_integration.rs | 11 ++++++++++ crates/openshell-server/src/compute/mod.rs | 2 ++ crates/openshell-server/src/grpc/sandbox.rs | 4 +++- crates/openshell-server/src/sandbox_watch.rs | 2 ++ crates/openshell-server/src/tracing_bus.rs | 20 ++++++++++++++++--- proto/openshell.proto | 5 +++++ 7 files changed, 43 insertions(+), 4 deletions(-) diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index bead4b4319..be9f7f4574 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -708,6 +708,7 @@ pub async fn sandbox_create( log_since_ms: 0, log_sources: vec!["gateway".to_string()], log_min_level: String::new(), + resume_after_cursor: 0, }) .await .into_diagnostic()? @@ -2607,6 +2608,7 @@ async fn wait_for_lifecycle_phase( log_since_ms: 0, log_sources: Vec::new(), log_min_level: String::new(), + resume_after_cursor: 0, }) .await .into_diagnostic()? @@ -7072,6 +7074,7 @@ pub async fn sandbox_logs( log_since_ms: since_ms, log_sources: source_filter, log_min_level: level.to_uppercase(), + resume_after_cursor: 0, }) .await .into_diagnostic()? diff --git a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs index 118daff902..b358922cc2 100644 --- a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs +++ b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs @@ -489,6 +489,7 @@ impl OpenShell for TestOpenShell { let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(provisioning)), + cursor: 0, })) .await; if vm_error_after_started { @@ -500,11 +501,13 @@ impl OpenShell for TestOpenShell { message: "Started VM launcher".to_string(), ..PlatformEvent::default() })), + cursor: 0, })) .await; let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(error)), + cursor: 0, })) .await; tokio::time::sleep(Duration::from_secs(5)).await; @@ -524,12 +527,14 @@ impl OpenShell for TestOpenShell { source: "gateway".to_string(), fields: HashMap::new(), })), + cursor: 0, })) .await; } let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(ready)), + cursor: 0, })) .await; return; @@ -538,6 +543,7 @@ impl OpenShell for TestOpenShell { let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(completed)), + cursor: 0, })) .await; return; @@ -552,6 +558,7 @@ impl OpenShell for TestOpenShell { message: "Preparing rootfs".to_string(), ..PlatformEvent::default() })), + cursor: 0, })) .await; tokio::time::sleep(Duration::from_millis(600)).await; @@ -563,12 +570,14 @@ impl OpenShell for TestOpenShell { message: "Formatting root disk".to_string(), ..PlatformEvent::default() })), + cursor: 0, })) .await; tokio::time::sleep(Duration::from_millis(600)).await; let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(ready)), + cursor: 0, })) .await; return; @@ -580,11 +589,13 @@ impl OpenShell for TestOpenShell { message: "Sandbox scheduled".to_string(), ..PlatformEvent::default() })), + cursor: 0, })) .await; let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(ready)), + cursor: 0, })) .await; }); diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2dd1f7cd2e..3f5b34db61 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -2890,6 +2890,8 @@ impl ComputeRuntime { public_platform_event_from_driver(&event), ), ), + // Placeholder: platform_event_bus.publish() stamps the cursor. + cursor: 0, }, ); } diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 2597f7f0e8..b4dce2a0e8 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -1031,6 +1031,8 @@ pub(super) async fn handle_watch_sandbox( sandbox.clone(), ), ), + // Status snapshots are re-read, not resumed by cursor. + cursor: 0, })) .await; @@ -1106,7 +1108,7 @@ pub(super) async fn handle_watch_sandbox( match state.store.get_message::(&sandbox_id).await { Ok(Some(sandbox)) => { state.sandbox_index.update_from_sandbox(&sandbox); - if tx.send(Ok(SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Sandbox(sandbox.clone()))})).await.is_err() { + if tx.send(Ok(SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Sandbox(sandbox.clone())), cursor: 0 })).await.is_err() { return; } if stop_on_terminal { diff --git a/crates/openshell-server/src/sandbox_watch.rs b/crates/openshell-server/src/sandbox_watch.rs index a80d21d653..bff687538c 100644 --- a/crates/openshell-server/src/sandbox_watch.rs +++ b/crates/openshell-server/src/sandbox_watch.rs @@ -76,6 +76,8 @@ pub fn lag_warning_event(n: u64) -> openshell_core::proto::SandboxStreamEvent { use openshell_core::proto::sandbox_stream_event::Payload; openshell_core::proto::SandboxStreamEvent { payload: Some(Payload::Warning(lag_warning(n))), + // Warnings are not part of the resumable log/platform sequence. + cursor: 0, } } diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 2fb865f1d8..c56e313f8f 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -117,6 +117,8 @@ impl TracingLogBus { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( log.clone(), )), + // Placeholder: publish() stamps the real cursor from next_seq. + cursor: 0, }; self.publish(&log.sandbox_id, evt, Self::DEFAULT_TAIL); } @@ -178,6 +180,8 @@ where payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( log, )), + // Placeholder: publish() stamps the real cursor from next_seq. + cursor: 0, }; self.bus.publish(&sandbox_id, evt, self.default_tail); } @@ -307,7 +311,10 @@ mod tests { let mut rx = bus.subscribe(sandbox_id); // Publish an event - let evt = SandboxStreamEvent { payload: None }; + let evt = SandboxStreamEvent { + payload: None, + cursor: 0, + }; bus.publish(sandbox_id, evt); assert!(rx.try_recv().is_ok()); @@ -331,7 +338,10 @@ mod tests { // New subscription should work let mut new_rx = bus.subscribe(sandbox_id); - let evt = SandboxStreamEvent { payload: None }; + let evt = SandboxStreamEvent { + payload: None, + cursor: 0, + }; bus.publish(sandbox_id, evt); assert!(new_rx.try_recv().is_ok()); } @@ -361,6 +371,7 @@ mod tests { message: format!("Message {i}"), metadata: HashMap::new(), })), + cursor: 0, }; bus.publish(sandbox_id, evt); } @@ -401,7 +412,10 @@ mod tests { let bus = PlatformEventBus::new(); let sandbox_id = "sb-7"; - let evt = SandboxStreamEvent { payload: None }; + let evt = SandboxStreamEvent { + payload: None, + cursor: 0, + }; bus.publish(sandbox_id, evt); assert_eq!(bus.tail(sandbox_id, 10).len(), 1); diff --git a/proto/openshell.proto b/proto/openshell.proto index 42e635927f..0480602273 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1390,6 +1390,9 @@ message WatchSandboxRequest { // Minimum log level to include (e.g. "INFO", "WARN", "ERROR"). Empty means all levels. string log_min_level = 10; + + // Resume streaming after this cursor (0 = from the beginning). + uint64 resume_after_cursor = 11; } // One event in a sandbox watch stream. @@ -1406,6 +1409,8 @@ message SandboxStreamEvent { // Draft policy update notification. DraftPolicyUpdate draft_policy_update = 5; } + // Monotonic per-source position for resuming after a cursor. + uint64 cursor = 6; } // Log line correlated to a sandbox. From 8880cf900bbc83649fc980474589da9f133fa424 Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Fri, 4 Sep 2026 14:18:28 +0100 Subject: [PATCH 4/9] feat(server): stamp watch cursors from a shared per-sandbox sequence Allocate cursors from a single SeqAllocator shared by the log and platform event buses, so a sandbox's merged watch stream carries unique, strictly increasing cursors. A single resume_after_cursor can then unambiguously locate a client's position across both sources. Rewrite both publish paths to allocate the sequence, stamp event.cursor, send, and append to the tail under one lock. This removes the previous get_mut().expect() TOCTOU race where a concurrent remove() between the two lock sections could panic. Signed-off-by: Artem Lytvyn --- crates/openshell-server/src/tracing_bus.rs | 164 ++++++++++++++------- 1 file changed, 107 insertions(+), 57 deletions(-) diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index c56e313f8f..9aa8d9d64b 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -18,6 +18,7 @@ use tracing_subscriber::layer::Context; pub struct TracingLogBus { inner: Arc>, pub(crate) platform_event_bus: PlatformEventBus, + seq: SeqAllocator, } #[derive(Debug)] @@ -29,7 +30,49 @@ struct Inner { struct PerSandbox { sender: broadcast::Sender, tail: VecDeque<(u64, SandboxStreamEvent)>, - next_seq: u64, +} + +impl PerSandbox { + fn new() -> Self { + let (tx, _rx) = broadcast::channel(1024); + Self { + sender: tx, + tail: VecDeque::new(), + } + } +} + +/// Per-sandbox monotonic sequence allocator. +/// +/// Shared across the resumable buses (`TracingLogBus`, `PlatformEventBus`) so +/// cursors are unique and strictly ordered within a single sandbox's merged +/// stream. Stamping at publish time keeps tail cursors stable across client +/// reconnects, which is what a single `resume_after_cursor` needs. +#[derive(Debug, Clone, Default)] +struct SeqAllocator { + inner: Arc>>, +} + +impl SeqAllocator { + /// Return the next sequence number for this sandbox. + /// + /// Seq starts at 1 so the proto default `resume_after_cursor` (0) means + /// "from the beginning" without skipping event 1. + fn next(&self, sandbox_id: &str) -> u64 { + let mut counters = self.inner.lock().expect("seq allocator lock poisoned"); + let counter = counters.entry(sandbox_id.to_string()).or_insert(1); + let seq = *counter; + *counter += 1; + seq + } + + /// Drop the counter for a sandbox once its buses are torn down. + fn remove(&self, sandbox_id: &str) { + self.inner + .lock() + .expect("seq allocator lock poisoned") + .remove(sandbox_id); + } } impl Default for TracingLogBus { @@ -41,11 +84,15 @@ impl Default for TracingLogBus { impl TracingLogBus { #[must_use] pub fn new() -> Self { + // One allocator, shared with the platform event bus so both draw from + // a single per-sandbox cursor space. + let seq = SeqAllocator::default(); Self { inner: Arc::new(Mutex::new(Inner { - per_id: HashMap::::new(), + per_id: HashMap::new(), })), - platform_event_bus: PlatformEventBus::new(), + platform_event_bus: PlatformEventBus::new(seq.clone()), + seq, } } @@ -61,16 +108,7 @@ impl TracingLogBus { inner .per_id .entry(sandbox_id.to_string()) - .or_insert_with(|| { - let (tx, _rx) = broadcast::channel(1024); - PerSandbox { - sender: tx, - tail: VecDeque::new(), - // Seq starts at 1 so the proto default resume_after_cursor - // (0) means "from the beginning" without skipping event 1. - next_seq: 1, - } - }) + .or_insert_with(PerSandbox::new) .sender .clone() } @@ -86,6 +124,8 @@ impl TracingLogBus { pub fn remove(&self, sandbox_id: &str) { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); inner.per_id.remove(sandbox_id); + drop(inner); + self.seq.remove(sandbox_id); } pub fn tail(&self, sandbox_id: &str, max: usize) -> Vec { @@ -126,20 +166,22 @@ impl TracingLogBus { /// Default tail buffer capacity (lines per sandbox). const DEFAULT_TAIL: usize = 2000; - fn publish(&self, sandbox_id: &str, event: SandboxStreamEvent, tail_cap: usize) { - let tx = self.sender_for(sandbox_id); - let _ = tx.send(event.clone()); + fn publish(&self, sandbox_id: &str, mut event: SandboxStreamEvent, tail_cap: usize) { + // Allocate the cursor first; next() takes and releases its own lock + // before we lock `inner`, so the two locks are never nested. + let seq = self.seq.next(sandbox_id); + event.cursor = seq; let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); - let per_sandbox = inner + let per = inner .per_id - .get_mut(sandbox_id) - .expect("sender_for inserted the entry above"); - let seq = per_sandbox.next_seq; - per_sandbox.next_seq += 1; - per_sandbox.tail.push_back((seq, event)); - while per_sandbox.tail.len() > tail_cap { - per_sandbox.tail.pop_front(); + .entry(sandbox_id.to_string()) + .or_insert_with(PerSandbox::new); + + let _ = per.sender.send(event.clone()); + per.tail.push_back((seq, event)); + while per.tail.len() > tail_cap { + per.tail.pop_front(); } } } @@ -305,7 +347,7 @@ mod tests { #[test] fn platform_event_bus_remove_cleans_up() { - let bus = PlatformEventBus::new(); + let bus = PlatformEventBus::new(SeqAllocator::default()); let sandbox_id = "sb-4"; let mut rx = bus.subscribe(sandbox_id); @@ -330,7 +372,7 @@ mod tests { #[test] fn platform_event_bus_subscribe_after_remove_creates_fresh_channel() { - let bus = PlatformEventBus::new(); + let bus = PlatformEventBus::new(SeqAllocator::default()); let sandbox_id = "sb-5"; let _old_rx = bus.subscribe(sandbox_id); @@ -348,7 +390,7 @@ mod tests { #[test] fn platform_event_bus_remove_nonexistent_is_noop() { - let bus = PlatformEventBus::new(); + let bus = PlatformEventBus::new(SeqAllocator::default()); // Should not panic bus.remove("nonexistent"); } @@ -357,7 +399,7 @@ mod tests { fn platform_event_bus_tail_returns_buffered_events() { use openshell_core::proto::{PlatformEvent, sandbox_stream_event}; - let bus = PlatformEventBus::new(); + let bus = PlatformEventBus::new(SeqAllocator::default()); let sandbox_id = "sb-6"; // Publish some events @@ -402,14 +444,14 @@ mod tests { #[test] fn platform_event_bus_tail_empty_sandbox() { - let bus = PlatformEventBus::new(); + let bus = PlatformEventBus::new(SeqAllocator::default()); let events = bus.tail("nonexistent", 10); assert!(events.is_empty()); } #[test] fn platform_event_bus_remove_clears_tail() { - let bus = PlatformEventBus::new(); + let bus = PlatformEventBus::new(SeqAllocator::default()); let sandbox_id = "sb-7"; let evt = SandboxStreamEvent { @@ -429,13 +471,8 @@ mod tests { /// This keeps platform events isolated from tracing capture. #[derive(Debug, Clone)] pub(crate) struct PlatformEventBus { - inner: Arc>, -} - -#[derive(Debug)] -struct PlatformEventBusInner { - senders: HashMap>, - tails: HashMap>, + inner: Arc>, + seq: SeqAllocator, } impl PlatformEventBus { @@ -443,24 +480,24 @@ impl PlatformEventBus { /// Platform events are infrequent (typically 5-10 per sandbox lifecycle). const DEFAULT_TAIL: usize = 50; - fn new() -> Self { + /// Build a platform event bus sharing `seq` with its owning `TracingLogBus` + /// so both stamp cursors from the same per-sandbox sequence. + fn new(seq: SeqAllocator) -> Self { Self { - inner: Arc::new(Mutex::new(PlatformEventBusInner { - senders: HashMap::new(), - tails: HashMap::new(), + inner: Arc::new(Mutex::new(Inner { + per_id: HashMap::new(), })), + seq, } } fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { let mut inner = self.inner.lock().expect("platform event bus lock poisoned"); inner - .senders + .per_id .entry(sandbox_id.to_string()) - .or_insert_with(|| { - let (tx, _rx) = broadcast::channel(1024); - tx - }) + .or_insert_with(PerSandbox::new) + .sender .clone() } @@ -468,15 +505,22 @@ impl PlatformEventBus { self.sender_for(sandbox_id).subscribe() } - pub(crate) fn publish(&self, sandbox_id: &str, event: SandboxStreamEvent) { - let tx = self.sender_for(sandbox_id); - let _ = tx.send(event.clone()); + pub(crate) fn publish(&self, sandbox_id: &str, mut event: SandboxStreamEvent) { + // Allocate before locking `inner` (same non-nested lock order as + // TracingLogBus::publish). + let seq = self.seq.next(sandbox_id); + event.cursor = seq; let mut inner = self.inner.lock().expect("platform event bus lock poisoned"); - let deque = inner.tails.entry(sandbox_id.to_string()).or_default(); - deque.push_back(event); - while deque.len() > Self::DEFAULT_TAIL { - deque.pop_front(); + let per = inner + .per_id + .entry(sandbox_id.to_string()) + .or_insert_with(PerSandbox::new); + + let _ = per.sender.send(event.clone()); + per.tail.push_back((seq, event)); + while per.tail.len() > Self::DEFAULT_TAIL { + per.tail.pop_front(); } } @@ -484,9 +528,16 @@ impl PlatformEventBus { pub(crate) fn tail(&self, sandbox_id: &str, max: usize) -> Vec { let inner = self.inner.lock().expect("platform event bus lock poisoned"); inner - .tails + .per_id .get(sandbox_id) - .map(|d| d.iter().rev().take(max).cloned().collect::>()) + .map(|d| { + d.tail + .iter() + .rev() + .take(max) + .map(|(_seq, event)| event.clone()) + .collect::>() + }) .unwrap_or_default() .into_iter() .rev() @@ -499,7 +550,6 @@ impl PlatformEventBus { /// and frees the tail buffer. pub(crate) fn remove(&self, sandbox_id: &str) { let mut inner = self.inner.lock().expect("platform event bus lock poisoned"); - inner.senders.remove(sandbox_id); - inner.tails.remove(sandbox_id); + inner.per_id.remove(sandbox_id); } } From 5281e9e962d794d4a1c14cde87dd3ed1b813607f Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Fri, 4 Sep 2026 16:50:49 +0100 Subject: [PATCH 5/9] feat(server): serve WatchSandbox resume from cursor with gap detection Add tail_after() to the log and platform event buses, returning every buffered event newer than a client's resume cursor. Each PerSandbox now tracks last_trimmed_seq (the highest seq it has evicted) so a resume is reported as an unrecoverable ResumeGap only when this bus dropped an event the client still needs. Judging gaps by evictions, not by the tail's oldest seq, is required under the shared cursor space: each bus's tail is non-contiguous in the global sequence because the other bus owns the missing seqs, so comparing against tail.front() would flag false gaps. Signed-off-by: Artem Lytvyn --- crates/openshell-server/src/tracing_bus.rs | 226 ++++++++++++++++++++- 1 file changed, 222 insertions(+), 4 deletions(-) diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 9aa8d9d64b..53d6e92502 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -21,15 +21,21 @@ pub struct TracingLogBus { seq: SeqAllocator, } -#[derive(Debug)] +#[derive(Debug, Clone)] struct Inner { per_id: HashMap, } -#[derive(Debug)] +#[derive(Debug, Clone)] struct PerSandbox { sender: broadcast::Sender, tail: VecDeque<(u64, SandboxStreamEvent)>, + /// 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 + /// global seq (the other bus owns the missing seqs), so a resume gap can + /// only be judged by what *this* bus actually dropped. + last_trimmed_seq: u64, } impl PerSandbox { @@ -38,10 +44,19 @@ impl PerSandbox { Self { sender: tx, tail: VecDeque::new(), + last_trimmed_seq: 0, } } } +/// The requested resume cursor is older than the oldest buffered event; +/// the events between them were trimmed and cannot be replayed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ResumeGap { + pub requested_after: u64, + pub oldest_available: u64, +} + /// Per-sandbox monotonic sequence allocator. /// /// Shared across the resumable buses (`TracingLogBus`, `PlatformEventBus`) so @@ -75,6 +90,33 @@ impl SeqAllocator { } } +fn tail_after_impl( + tail: &VecDeque<(u64, SandboxStreamEvent)>, + last_trimmed_seq: u64, + after_seq: u64, +) -> Result, ResumeGap> { + // Gap iff this bus dropped an event the client still needs, i.e. the + // highest seq we evicted is newer than the client's position. Judged only + // on this bus's own evictions — the other bus owns the seqs missing here. + if after_seq < last_trimmed_seq { + return Err(ResumeGap { + requested_after: after_seq, + oldest_available: last_trimmed_seq + 1, + }); + } + + // Skippable events (seq <= after_seq) are the oldest, at the front, so a + // take-while would stop before reaching the wanted ones. Filter the whole + // tail instead; order is preserved and caught-up yields an empty vec. + let res: Vec = tail + .iter() + .filter(|(seq, _)| *seq > after_seq) + .map(|(_, event)| event.clone()) + .collect(); + + Ok(res) +} + impl Default for TracingLogBus { fn default() -> Self { Self::new() @@ -147,6 +189,18 @@ impl TracingLogBus { .collect::>() } + pub fn tail_after( + &self, + sandbox_id: &str, + after_seq: u64, + ) -> Result, ResumeGap> { + let inner = self.inner.lock().expect("tracing bus lock poisoned"); + inner.per_id.get(sandbox_id).map_or_else( + || Ok(Vec::new()), + |per| tail_after_impl(&per.tail, per.last_trimmed_seq, after_seq), + ) + } + /// Publish a log line from an external source (e.g., sandbox push). /// /// Injects the line into the same broadcast channel and tail buffer @@ -181,7 +235,9 @@ impl TracingLogBus { let _ = per.sender.send(event.clone()); per.tail.push_back((seq, event)); while per.tail.len() > tail_cap { - per.tail.pop_front(); + if let Some((trimmed, _)) = per.tail.pop_front() { + per.last_trimmed_seq = trimmed; + } } } } @@ -277,6 +333,153 @@ mod tests { } } + /// Build a stream event carrying `seq` in its cursor for assertion. + fn stream_event(seq: u64) -> SandboxStreamEvent { + SandboxStreamEvent { + payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( + make_log_event("sb", &seq.to_string()), + )), + cursor: seq, + } + } + + /// Build a contiguous tail with seqs `lo..=hi`. + fn tail_of(lo: u64, hi: u64) -> VecDeque<(u64, SandboxStreamEvent)> { + (lo..=hi).map(|s| (s, stream_event(s))).collect() + } + + /// Extract cursors from a run of events, in order. + fn cursors(events: &[SandboxStreamEvent]) -> Vec { + events.iter().map(|e| e.cursor).collect() + } + + #[test] + fn tail_after_impl_empty_tail_returns_empty() { + let tail = VecDeque::new(); + // Nothing trimmed (last_trimmed_seq = 0): any cursor is serviceable. + assert_eq!(tail_after_impl(&tail, 0, 0).unwrap(), Vec::new()); + assert_eq!(tail_after_impl(&tail, 0, 42).unwrap(), Vec::new()); + } + + #[test] + fn tail_after_impl_from_zero_returns_all() { + let tail = tail_of(1, 5); + let events = tail_after_impl(&tail, 0, 0).expect("serviceable"); + assert_eq!(cursors(&events), vec![1, 2, 3, 4, 5]); + } + + #[test] + fn tail_after_impl_mid_range_returns_newer_in_order() { + let tail = tail_of(1, 5); + let events = tail_after_impl(&tail, 0, 3).expect("serviceable"); + assert_eq!(cursors(&events), vec![4, 5]); + } + + #[test] + fn tail_after_impl_caught_up_returns_empty() { + let tail = tail_of(1, 5); + // Cursor at the newest seq: nothing newer, but not a gap. + assert_eq!(tail_after_impl(&tail, 0, 5).expect("ok"), Vec::new()); + } + + #[test] + fn tail_after_impl_future_cursor_returns_empty() { + let tail = tail_of(1, 5); + // Cursor beyond newest (client claims to have seen more than exists): + // still serviceable, just nothing to send. + assert_eq!(tail_after_impl(&tail, 0, 99).expect("ok"), Vec::new()); + } + + #[test] + fn tail_after_impl_boundary_at_last_trimmed_is_serviceable() { + // Bus trimmed up to seq 2, retains 3..=5. Client saw exactly 2, so + // nothing they still need was dropped. + let tail = tail_of(3, 5); + let events = tail_after_impl(&tail, 2, 2).expect("serviceable"); + assert_eq!(cursors(&events), vec![3, 4, 5]); + } + + #[test] + fn tail_after_impl_gap_returns_err() { + // Bus trimmed up to seq 2, retains 3..=5. Client wants everything after + // 1, but seq 2 was evicted and cannot be replayed. + let tail = tail_of(3, 5); + let err = tail_after_impl(&tail, 2, 1).expect_err("gap"); + assert_eq!( + err, + ResumeGap { + requested_after: 1, + oldest_available: 3, + } + ); + } + + #[test] + fn tail_after_impl_non_contiguous_tail_no_false_gap() { + // 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<(u64, SandboxStreamEvent)> = + [(2, stream_event(2)), (4, stream_event(4))] + .into_iter() + .collect(); + let events = tail_after_impl(&tail, 0, 0).expect("no gap"); + assert_eq!(cursors(&events), vec![2, 4]); + } + + #[test] + fn tracing_log_bus_tail_after_serviceable_and_missing() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-ta"; + for _ in 0..3 { + bus.publish_external(make_log_event(sandbox_id, "line")); + } + // Cursors start at 1, so three publishes are seqs 1,2,3. + assert_eq!( + cursors(&bus.tail_after(sandbox_id, 0).unwrap()), + vec![1, 2, 3] + ); + assert_eq!(cursors(&bus.tail_after(sandbox_id, 2).unwrap()), vec![3]); + // Unknown sandbox: no entry, nothing buffered, no gap. + assert_eq!(bus.tail_after("nope", 5).unwrap(), Vec::new()); + } + + #[test] + fn platform_event_bus_tail_after_serviceable() { + let bus = TracingLogBus::new(); + let platform = &bus.platform_event_bus; + let sandbox_id = "sb-pe"; + for _ in 0..3 { + platform.publish(sandbox_id, stream_event(0)); + } + // Shared allocator, but only the platform bus published here, so its + // seqs are 1,2,3. + assert_eq!( + cursors(&platform.tail_after(sandbox_id, 0).unwrap()), + vec![1, 2, 3] + ); + assert_eq!( + cursors(&platform.tail_after(sandbox_id, 1).unwrap()), + vec![2, 3] + ); + } + + #[test] + fn shared_allocator_interleaves_cursors_across_buses() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-mix"; + // Interleave log and platform publishes; the shared allocator gives + // each a unique, increasing cursor in one merged space. + bus.publish_external(make_log_event(sandbox_id, "a")); // seq 1 + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); // seq 2 + bus.publish_external(make_log_event(sandbox_id, "b")); // seq 3 + + let logs = cursors(&bus.tail_after(sandbox_id, 0).unwrap()); + let events = cursors(&bus.platform_event_bus.tail_after(sandbox_id, 0).unwrap()); + assert_eq!(logs, vec![1, 3]); + assert_eq!(events, vec![2]); + } + #[test] fn tracing_log_bus_remove_cleans_up_all_maps() { let bus = TracingLogBus::new(); @@ -520,7 +723,9 @@ impl PlatformEventBus { let _ = per.sender.send(event.clone()); per.tail.push_back((seq, event)); while per.tail.len() > Self::DEFAULT_TAIL { - per.tail.pop_front(); + if let Some((trimmed, _)) = per.tail.pop_front() { + per.last_trimmed_seq = trimmed; + } } } @@ -544,6 +749,19 @@ impl PlatformEventBus { .collect() } + #[allow(dead_code)] + pub(crate) fn tail_after( + &self, + sandbox_id: &str, + after_seq: u64, + ) -> Result, ResumeGap> { + let inner = self.inner.lock().expect("platform event bus lock poisoned"); + inner.per_id.get(sandbox_id).map_or_else( + || Ok(Vec::new()), + |per| tail_after_impl(&per.tail, per.last_trimmed_seq, after_seq), + ) + } + /// Remove the bus entry for the given sandbox id. /// /// This drops the broadcast sender, closing any active receivers, From d5c12f88f5e427e34856ddbcd0900ad41ba261c3 Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Sat, 5 Sep 2026 21:07:10 +0100 Subject: [PATCH 6/9] feat(server): resume WatchSandbox from cursor across log and platform buses Wire resume_after_cursor into the watch producer. On a non-zero cursor, replay events strictly after it from both the log and platform buses, merge by shared cursor, and emit in order before entering the live loop. A trimmed range on either bus is an unrecoverable gap and terminates the stream with OUT_OF_RANGE carrying the requested and earliest-available cursors, distinct from recoverable lag which warns and continues. Signed-off-by: Artem Lytvyn --- crates/openshell-server/src/grpc/sandbox.rs | 104 ++++++++++++++++---- crates/openshell-server/src/tracing_bus.rs | 1 - 2 files changed, 87 insertions(+), 18 deletions(-) diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index b4dce2a0e8..50ae1aa515 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -974,6 +974,7 @@ pub(super) async fn handle_watch_sandbox( let log_sources = req.log_sources; let log_min_level = req.log_min_level; let event_tail = req.event_tail; + let resume_after_cursor = req.resume_after_cursor; let (tx, rx) = mpsc::channel::>(256); let state = state.clone(); @@ -1056,13 +1057,58 @@ pub(super) async fn handle_watch_sandbox( } } - // Replay tail logs (best-effort), filtered by log_since_ms and log_sources. - if follow_logs { - for evt in state.tracing_log_bus.tail(&sandbox_id, log_tail as usize) { - if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log( - ref log, - )) = evt.payload - { + if resume_after_cursor > 0 { + // Resume: replay events strictly after the client's cursor from both + // resumable buses. Either bus reporting a trimmed range is an + // unrecoverable gap -> terminate with a documented status. + use openshell_core::proto::sandbox_stream_event::Payload; + + let log_replay = if follow_logs { + Some( + state + .tracing_log_bus + .tail_after(&sandbox_id, resume_after_cursor), + ) + } else { + None + }; + + let platform_replay = if follow_events { + Some( + state + .tracing_log_bus + .platform_event_bus + .tail_after(&sandbox_id, resume_after_cursor), + ) + } else { + None + }; + + // Gap check FIRST (borrows), before the merge moves the vecs. + for replay in [&log_replay, &platform_replay] { + if let Some(Err(gap)) = replay { + let _ = tx.send(Err(Status::out_of_range(format!( + "resume cursor {} is no longer available; earliest resumable cursor is {}", + gap.requested_after, gap.oldest_available + )))) + .await; + return; + } + } + + // Merge both buses by shared cursor, then emit ascending. + let mut merged: Vec = Vec::new(); + if let Some(Ok(v)) = log_replay { + merged.extend(v); + } + if let Some(Ok(v)) = platform_replay { + merged.extend(v); + } + + merged.sort_by_key(|e| e.cursor); + + for evt in merged { + if let Some(Payload::Log(ref log)) = evt.payload { if log_since_ms > 0 && log.timestamp_ms < log_since_ms { continue; } @@ -1077,17 +1123,41 @@ pub(super) async fn handle_watch_sandbox( return; } } - } + } else { + // Replay tail logs (best-effort), filtered by log_since_ms and log_sources. + if follow_logs { + for evt in state.tracing_log_bus.tail(&sandbox_id, log_tail as usize) { + if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log( + ref log, + )) = evt.payload + { + if log_since_ms > 0 && log.timestamp_ms < log_since_ms { + continue; + } + if !log_sources.is_empty() && !source_matches(&log.source, &log_sources) + { + continue; + } + if !level_matches(&log.level, &log_min_level) { + continue; + } + } + if tx.send(Ok(evt)).await.is_err() { + return; + } + } + } - // Replay buffered platform events. - if follow_events { - for evt in state - .tracing_log_bus - .platform_event_bus - .tail(&sandbox_id, event_tail as usize) - { - if tx.send(Ok(evt)).await.is_err() { - return; + // Replay buffered platform events. + if follow_events { + for evt in state + .tracing_log_bus + .platform_event_bus + .tail(&sandbox_id, event_tail as usize) + { + if tx.send(Ok(evt)).await.is_err() { + return; + } } } } diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 53d6e92502..9187cf9682 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -749,7 +749,6 @@ impl PlatformEventBus { .collect() } - #[allow(dead_code)] pub(crate) fn tail_after( &self, sandbox_id: &str, From 2fd2c7599631e2621b1cc3f7303d73e4e2aaf061 Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Sat, 5 Sep 2026 21:20:21 +0100 Subject: [PATCH 7/9] test(server): cover WatchSandbox cursor resume paths Add handler-level tests for the resumable watch stream: replay strictly after the client cursor, merge log and platform events in shared-cursor order, suppress duplicates when resuming at the latest cursor, and terminate with OUT_OF_RANGE when the requested cursor has been trimmed. Signed-off-by: Artem Lytvyn --- crates/openshell-server/src/grpc/sandbox.rs | 204 ++++++++++++++++++++ 1 file changed, 204 insertions(+) diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 50ae1aa515..b2d46c8c8b 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -3056,6 +3056,210 @@ mod tests { ); } + /// Seed `n` log lines onto the log bus; cursors run 1..=n. + fn seed_log_lines(state: &ServerState, sandbox_id: &str, n: usize) { + for i in 0..n { + state + .tracing_log_bus + .publish_external(openshell_core::proto::SandboxLogLine { + sandbox_id: sandbox_id.to_string(), + timestamp_ms: i as i64, + level: "INFO".to_string(), + target: "test".to_string(), + message: format!("line {i}"), + source: "gateway".to_string(), + ..Default::default() + }); + } + } + + fn seed_platform_event(state: &ServerState, sandbox_id: &str, reason: &str) { + state.tracing_log_bus.platform_event_bus.publish( + sandbox_id, + SandboxStreamEvent { + payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Event( + openshell_core::proto::PlatformEvent { + timestamp_ms: 0, + source: "test".to_string(), + r#type: "Normal".to_string(), + reason: reason.to_string(), + message: reason.to_string(), + metadata: HashMap::new(), + }, + )), + cursor: 0, + }, + ); + } + + #[tokio::test] + async fn resume_replays_only_events_after_cursor() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("resumed", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // Cursors 1,2,3. + seed_log_lines(&state, &id, 3); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + id: id.clone(), + follow_logs: true, + resume_after_cursor: 1, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + // Snapshot first (status re-read, cursor 0). + let snap = stream.next().await.unwrap().unwrap(); + assert_eq!(snap.cursor, 0, "first event should be the status snapshot"); + + // Then only cursors 2 and 3; cursor 1 already seen by the client. + let a = stream.next().await.unwrap().unwrap(); + let b = stream.next().await.unwrap().unwrap(); + assert_eq!(a.cursor, 2); + assert_eq!(b.cursor, 3); + } + + #[tokio::test] + async fn resume_merges_log_and_platform_events_in_cursor_order() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("merged", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // Interleave across the shared allocator: log=1, platform=2, log=3, platform=4. + seed_log_lines(&state, &id, 1); // cursor 1 + seed_platform_event(&state, &id, "e2"); // cursor 2 + state + .tracing_log_bus + .publish_external(openshell_core::proto::SandboxLogLine { + sandbox_id: id.clone(), + timestamp_ms: 3, + level: "INFO".to_string(), + target: "test".to_string(), + message: "line 3".to_string(), + source: "gateway".to_string(), + ..Default::default() + }); // cursor 3 + seed_platform_event(&state, &id, "e4"); // cursor 4 + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + id: id.clone(), + follow_logs: true, + follow_events: true, + resume_after_cursor: 1, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert_eq!(snap.cursor, 0); + + // Merged from both buses, ascending by shared cursor: 2,3,4. + let mut got = Vec::new(); + for _ in 0..3 { + got.push(stream.next().await.unwrap().unwrap().cursor); + } + assert_eq!(got, vec![2, 3, 4]); + } + + #[tokio::test] + async fn resume_at_latest_cursor_suppresses_duplicates() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("nodup", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // Cursors 1,2,3; client already saw through 3. + seed_log_lines(&state, &id, 3); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + id: id.clone(), + follow_logs: true, + resume_after_cursor: 3, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert_eq!(snap.cursor, 0); + + // No resumable events remain; the live loop yields nothing promptly. + let next = tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await; + assert!( + next.is_err(), + "expected no further events after resume at latest cursor, got {next:?}" + ); + } + + #[tokio::test] + async fn resume_from_trimmed_cursor_terminates_out_of_range() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("gap", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // Exceed the 2000-line tail so the earliest cursors are trimmed. + seed_log_lines(&state, &id, 2005); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + id: id.clone(), + follow_logs: true, + // Cursor 2 was trimmed; this is an unrecoverable gap. + resume_after_cursor: 2, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + // Snapshot still arrives first (fresh state), then the terminal gap status. + let snap = stream.next().await.unwrap().unwrap(); + assert_eq!(snap.cursor, 0); + + let err = stream + .next() + .await + .unwrap() + .expect_err("trimmed cursor must terminate the stream"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + assert!( + err.message().contains('2'), + "gap status should report the requested cursor: {}", + err.message() + ); + + // Stream ends after the terminal status. + assert!(stream.next().await.is_none()); + } + #[tokio::test] async fn delete_handler_ends_telemetry_for_the_resolved_sandbox_id() { let state = test_server_state().await; From e501015a668e73d37d85a453f2f46b8e7735c537 Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Sat, 5 Sep 2026 21:26:19 +0100 Subject: [PATCH 8/9] docs(api): document WatchSandbox loss-awareness and resume Signed-off-by: Artem Lytvyn --- architecture/gateway.md | 24 ++++++++++++++++++++++++ docs/observability/accessing-logs.mdx | 11 +++++++++++ proto/openshell.proto | 21 ++++++++++++++++++--- 3 files changed, 53 insertions(+), 3 deletions(-) diff --git a/architecture/gateway.md b/architecture/gateway.md index 0430d95159..9fae7434d4 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -284,6 +284,30 @@ Domain objects use shared metadata: stable server-generated IDs, human-readable names, creation timestamps, and labels. Crate-level details live in `crates/openshell-core/README.md`. +### Watch streams + +`WatchSandbox` merges three per-sandbox sources into one client stream: status +snapshots, server/sandbox logs, and platform events. Logs and platform events +are resumable; a shared per-sandbox counter stamps each with a monotonic +`cursor` so the merged stream is linearly ordered across both sources. Status +snapshots and warnings are re-read on demand and carry `cursor = 0`. + +The gateway holds a bounded in-memory tail per sandbox. Loss is reported with +two distinct, documented behaviors: + +- **Recoverable lag** — a broadcast receiver falls behind and the server skips + ahead. The stream emits a `SandboxStreamWarning` event and continues; the + client sees the gap as a cursor discontinuity. +- **Unrecoverable gap** — a reconnect requests `resume_after_cursor` below the + oldest buffered cursor (the tail has been trimmed past it). The server sends a + snapshot, then terminates the stream with `OUT_OF_RANGE` carrying the + requested and earliest-available cursors so the client can restart cleanly. + +On resume the server replays only events after the client's cursor from both +resumable sources, merged in cursor order, before entering live delivery — no +loss and no duplication. Clients track the highest observed `cursor` and pass it +as `resume_after_cursor` on reconnect. + ## Persistence The gateway persistence layer is a protobuf object store. Domain services store diff --git a/docs/observability/accessing-logs.mdx b/docs/observability/accessing-logs.mdx index 4b755f74cc..7ac30791ed 100644 --- a/docs/observability/accessing-logs.mdx +++ b/docs/observability/accessing-logs.mdx @@ -39,6 +39,17 @@ The sandbox pushes logs to the gateway over gRPC in real time. The gateway store For durable log storage, use the log files inside the sandbox or enable [OCSF JSON export](/observability/ocsf-json-export) and ship the JSONL files to an external log aggregator. +## Loss Awareness and Resume + +The watch stream behind `openshell logs` is loss-aware. Each resumable event (log line or platform event) carries a monotonic `cursor`. Status snapshots and warnings carry cursor `0`. + +The gateway distinguishes recoverable from unrecoverable loss: + +- **Recoverable lag.** When a consumer falls behind and the gateway skips ahead in its buffer, the stream emits a warning event and keeps running. Clients see the gap as a jump in cursor values. +- **Unrecoverable gap.** When a client reconnects and asks to resume after a cursor the gateway has already trimmed from its buffer, the stream ends with an `OUT_OF_RANGE` status that reports the requested and earliest-available cursors. The client should restart observation and, if it needs the missing lines, read them from the log files inside the sandbox. + +On reconnect, a client passes the highest cursor it processed as the resume point. The gateway replays only events after that cursor — logs and platform events merged in cursor order — then resumes live delivery, so no events are lost or duplicated across the reconnect. + ## Direct Filesystem Access Start an independent shell with `sandbox exec` to read log files directly: diff --git a/proto/openshell.proto b/proto/openshell.proto index 0480602273..3aec339735 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1391,7 +1391,13 @@ message WatchSandboxRequest { // Minimum log level to include (e.g. "INFO", "WARN", "ERROR"). Empty means all levels. string log_min_level = 10; - // Resume streaming after this cursor (0 = from the beginning). + // Resume streaming after this cursor (0 = from the beginning). On reconnect, + // set this to the highest `SandboxStreamEvent.cursor` already processed; the + // server replays only log and platform events after it, merged in cursor + // order, before resuming live delivery. If the requested cursor has already + // been trimmed from the server's buffer, the resume is unrecoverable and the + // stream terminates with OUT_OF_RANGE (see SandboxStreamWarning for the + // recoverable case). uint64 resume_after_cursor = 11; } @@ -1404,12 +1410,17 @@ message SandboxStreamEvent { SandboxLogLine log = 2; // One platform event. PlatformEvent event = 3; - // Warning from the server (e.g. missed messages due to lag). + // Recoverable warning from the server, e.g. messages dropped because a + // broadcast receiver lagged. The stream continues after this warning; the + // client can detect the gap from cursor discontinuity. SandboxStreamWarning warning = 4; // Draft policy update notification. DraftPolicyUpdate draft_policy_update = 5; } - // Monotonic per-source position for resuming after a cursor. + // Monotonic per-sandbox position shared across the resumable log and platform + // event sources. Pass the highest observed value as + // WatchSandboxRequest.resume_after_cursor to resume without loss or + // duplication. 0 for non-resumable events (status snapshots, warnings). uint64 cursor = 6; } @@ -1427,6 +1438,10 @@ message SandboxLogLine { map fields = 7; } +// Recoverable loss notification on a watch stream. Emitted when the server +// skips ahead after a broadcast lag instead of terminating; the stream keeps +// running. Unrecoverable loss (a trimmed resume cursor) is reported as an +// OUT_OF_RANGE stream status, not this message. message SandboxStreamWarning { string message = 1; } From c3527e5879a4687468ab80db04113384f57598dd Mon Sep 17 00:00:00 2001 From: Artem Lytvyn Date: Sun, 6 Sep 2026 21:04:49 +0100 Subject: [PATCH 9/9] fix(server): deliver watch events once and harden cursor teardown Signed-off-by: Artem Lytvyn --- architecture/gateway.md | 17 ++-- crates/openshell-server/src/compute/mod.rs | 3 +- crates/openshell-server/src/grpc/sandbox.rs | 90 +++++++++++++++++++++ crates/openshell-server/src/tracing_bus.rs | 18 +++-- docs/observability/accessing-logs.mdx | 2 + proto/openshell.proto | 15 ++-- 6 files changed, 126 insertions(+), 19 deletions(-) diff --git a/architecture/gateway.md b/architecture/gateway.md index 9fae7434d4..863f375312 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -289,8 +289,12 @@ names, creation timestamps, and labels. Crate-level details live in `WatchSandbox` merges three per-sandbox sources into one client stream: status snapshots, server/sandbox logs, and platform events. Logs and platform events are resumable; a shared per-sandbox counter stamps each with a monotonic -`cursor` so the merged stream is linearly ordered across both sources. Status -snapshots and warnings are re-read on demand and carry `cursor = 0`. +`cursor`. Cursor-ordered delivery is guaranteed for the replay phase: on +resume the buffered events from both sources are sorted by cursor before +emission. Live events carry cursors and are monotonic within each source, but +the two sources are read independently, so a client should order across sources +by `cursor` rather than by arrival. Status snapshots and warnings are re-read on +demand and carry `cursor = 0`. The gateway holds a bounded in-memory tail per sandbox. Loss is reported with two distinct, documented behaviors: @@ -304,9 +308,12 @@ two distinct, documented behaviors: requested and earliest-available cursors so the client can restart cleanly. On resume the server replays only events after the client's cursor from both -resumable sources, merged in cursor order, before entering live delivery — no -loss and no duplication. Clients track the highest observed `cursor` and pass it -as `resume_after_cursor` on reconnect. +resumable sources, merged in cursor order, before entering live delivery. The +broadcast receivers are subscribed before replay, so an event buffered during +initialization could appear in both replay and the live receiver; the producer +tracks the highest replayed cursor and suppresses live events at or below it, so +each event is delivered once. Clients track the highest observed `cursor` and +pass it as `resume_after_cursor` on reconnect. ## Persistence diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 3f5b34db61..0296d44899 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -3388,8 +3388,9 @@ impl ComputeRuntime { } fn cleanup_sandbox_state(&self, sandbox_id: &str) { + // `tracing_log_bus.remove` also clears the platform event bus and resets + // the shared cursor allocator last (see its docs). self.tracing_log_bus.remove(sandbox_id); - self.tracing_log_bus.platform_event_bus.remove(sandbox_id); self.sandbox_watch_bus.remove(sandbox_id); } diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index b2d46c8c8b..f7bafab058 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -1057,6 +1057,13 @@ pub(super) async fn handle_watch_sandbox( } } + // Highest resumable cursor already handled by the tail/replay phase. + // The broadcast receivers were subscribed before replay ran, so an + // event published during initialization can sit in both the replay + // buffer and a live receiver. The live loop suppresses events at or + // below this cutoff so each is delivered exactly once. + let mut replay_cutoff: u64 = resume_after_cursor; + if resume_after_cursor > 0 { // Resume: replay events strictly after the client's cursor from both // resumable buses. Either bus reporting a trimmed range is an @@ -1107,6 +1114,12 @@ pub(super) async fn handle_watch_sandbox( merged.sort_by_key(|e| e.cursor); + // Everything through the highest replayed cursor is now handled; + // suppress its live duplicate below. + if let Some(last) = merged.last() { + replay_cutoff = replay_cutoff.max(last.cursor); + } + for evt in merged { if let Some(Payload::Log(ref log)) = evt.payload { if log_since_ms > 0 && log.timestamp_ms < log_since_ms { @@ -1142,6 +1155,7 @@ pub(super) async fn handle_watch_sandbox( continue; } } + replay_cutoff = replay_cutoff.max(evt.cursor); if tx.send(Ok(evt)).await.is_err() { return; } @@ -1155,6 +1169,7 @@ pub(super) async fn handle_watch_sandbox( .platform_event_bus .tail(&sandbox_id, event_tail as usize) { + replay_cutoff = replay_cutoff.max(evt.cursor); if tx.send(Ok(evt)).await.is_err() { return; } @@ -1217,6 +1232,10 @@ pub(super) async fn handle_watch_sandbox( } => { match res { Ok(evt) => { + // Skip events already delivered by the tail/replay phase. + if evt.cursor != 0 && evt.cursor <= replay_cutoff { + continue; + } if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log(ref log)) = evt.payload { if !log_sources.is_empty() && !source_matches(&log.source, &log_sources) { continue; @@ -1249,6 +1268,10 @@ pub(super) async fn handle_watch_sandbox( } => { match res { Ok(evt) => { + // Skip events already delivered by the tail/replay phase. + if evt.cursor != 0 && evt.cursor <= replay_cutoff { + continue; + } if tx.send(Ok(evt)).await.is_err() { return; } @@ -3260,6 +3283,73 @@ mod tests { assert!(stream.next().await.is_none()); } + #[tokio::test] + async fn watch_delivers_each_event_once_during_init_race() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("race", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // Seed events that land in the tail before the watch subscribes. + seed_log_lines(&state, &id, 5); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + id: id.clone(), + follow_logs: true, + ..Default::default() + }), + ) + .await + .unwrap(); + + // Publish more concurrently with producer initialization. Some of these + // can land after the broadcast subscription but before the tail read, + // putting them in both replay and the live receiver. + for i in 5..15 { + state + .tracing_log_bus + .publish_external(openshell_core::proto::SandboxLogLine { + sandbox_id: id.clone(), + timestamp_ms: i64::from(i), + level: "INFO".to_string(), + target: "test".to_string(), + message: format!("line {i}"), + source: "gateway".to_string(), + ..Default::default() + }); + } + + let mut stream = response.into_inner(); + let mut cursors = Vec::new(); + while let Ok(Some(item)) = + tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await + { + let evt = item.unwrap(); + if evt.cursor != 0 { + cursors.push(evt.cursor); + } + } + + // Every delivered cursor is unique (no double delivery) and monotonically + // increasing (replay ordered, then live in cursor order for one source). + let mut sorted = cursors.clone(); + sorted.sort_unstable(); + sorted.dedup(); + assert_eq!( + sorted.len(), + cursors.len(), + "duplicate cursors delivered: {cursors:?}" + ); + assert_eq!( + cursors, sorted, + "cursors not delivered in order: {cursors:?}" + ); + } + #[tokio::test] async fn delete_handler_ends_telemetry_for_the_resolved_sandbox_id() { let state = test_server_state().await; diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 9187cf9682..262f581a76 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -159,14 +159,20 @@ impl TracingLogBus { self.sender_for(sandbox_id).subscribe() } - /// Remove all bus entries for the given sandbox id. + /// Remove all bus entries for the given sandbox id, including the platform + /// event bus that shares this bus's cursor allocator. /// - /// This drops the broadcast sender (closing any active receivers with - /// `RecvError::Closed`) and frees the tail buffer. + /// This drops the broadcast senders (closing any active receivers with + /// `RecvError::Closed`) and frees the tail buffers. Both per-sandbox maps + /// are cleared before the shared `SeqAllocator` entry is reset, so the + /// allocator is never reset while either map can still accept a publish that + /// references it. pub fn remove(&self, sandbox_id: &str) { - let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); - inner.per_id.remove(sandbox_id); - drop(inner); + { + let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + inner.per_id.remove(sandbox_id); + } + self.platform_event_bus.remove(sandbox_id); self.seq.remove(sandbox_id); } diff --git a/docs/observability/accessing-logs.mdx b/docs/observability/accessing-logs.mdx index 7ac30791ed..bf9f7c794b 100644 --- a/docs/observability/accessing-logs.mdx +++ b/docs/observability/accessing-logs.mdx @@ -50,6 +50,8 @@ The gateway distinguishes recoverable from unrecoverable loss: On reconnect, a client passes the highest cursor it processed as the resume point. The gateway replays only events after that cursor — logs and platform events merged in cursor order — then resumes live delivery, so no events are lost or duplicated across the reconnect. +Replay is emitted in cursor order. During live delivery the log and platform event sources are read independently, so events from different sources can interleave; order across sources by `cursor` rather than by arrival. + ## Direct Filesystem Access Start an independent shell with `sandbox exec` to read log files directly: diff --git a/proto/openshell.proto b/proto/openshell.proto index 3aec339735..c8b6235e98 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1391,13 +1391,14 @@ message WatchSandboxRequest { // Minimum log level to include (e.g. "INFO", "WARN", "ERROR"). Empty means all levels. string log_min_level = 10; - // Resume streaming after this cursor (0 = from the beginning). On reconnect, - // set this to the highest `SandboxStreamEvent.cursor` already processed; the - // server replays only log and platform events after it, merged in cursor - // order, before resuming live delivery. If the requested cursor has already - // been trimmed from the server's buffer, the resume is unrecoverable and the - // stream terminates with OUT_OF_RANGE (see SandboxStreamWarning for the - // recoverable case). + // Resume streaming after this cursor. 0 means no cursor resume: the server + // falls back to tail-limited replay controlled by log_tail_lines and + // event_tail. When greater than zero, set it to the highest + // `SandboxStreamEvent.cursor` already processed; the server replays only log + // and platform events after it, merged in cursor order, before resuming live + // delivery. If the requested cursor has already been trimmed from the + // server's buffer, the resume is unrecoverable and the stream terminates with + // OUT_OF_RANGE (see SandboxStreamWarning for the recoverable case). uint64 resume_after_cursor = 11; }