diff --git a/Cargo.lock b/Cargo.lock index edc940ea68..a3a5dbc11d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -127,7 +127,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -138,7 +138,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -228,6 +228,28 @@ dependencies = [ "serde_json", ] +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -1822,7 +1844,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3564,7 +3586,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3651,7 +3673,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http 1.4.0", @@ -4300,6 +4322,7 @@ dependencies = [ name = "openshell-sdk" version = "0.0.0" dependencies = [ + "async-stream", "async-trait", "futures", "hyper", @@ -5942,7 +5965,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6010,7 +6033,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6540,7 +6563,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7038,7 +7061,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7074,7 +7097,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -8061,7 +8084,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/architecture/gateway.md b/architecture/gateway.md index 769f57f6a0..cecf5e6e76 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -284,6 +284,37 @@ 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`. 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: + +- **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. 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 The gateway persistence layer is a protobuf object store. Domain services store diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index 78a794aa30..f01d05155a 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -745,6 +745,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()? @@ -3085,6 +3086,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()? @@ -5525,6 +5527,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 7be771c442..01b65ad710 100644 --- a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs +++ b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs @@ -607,6 +607,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 { @@ -618,11 +619,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; @@ -642,12 +645,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; @@ -656,6 +661,7 @@ impl OpenShell for TestOpenShell { let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(completed)), + cursor: 0, })) .await; return; @@ -670,6 +676,7 @@ impl OpenShell for TestOpenShell { message: "Preparing rootfs".to_string(), ..PlatformEvent::default() })), + cursor: 0, })) .await; tokio::time::sleep(Duration::from_millis(600)).await; @@ -681,12 +688,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; @@ -698,11 +707,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-sdk/Cargo.toml b/crates/openshell-sdk/Cargo.toml index 8d80beaa74..abbac2a9de 100644 --- a/crates/openshell-sdk/Cargo.toml +++ b/crates/openshell-sdk/Cargo.toml @@ -28,6 +28,7 @@ tokio-tungstenite = { workspace = true } tonic = { workspace = true, features = ["tls-native-roots"] } tower = { workspace = true } tracing = { workspace = true } +async-stream = "0.3.6" [dev-dependencies] serde_json = { workspace = true } diff --git a/crates/openshell-sdk/src/client.rs b/crates/openshell-sdk/src/client.rs index b5486812f7..8b5f64eef3 100644 --- a/crates/openshell-sdk/src/client.rs +++ b/crates/openshell-sdk/src/client.rs @@ -13,12 +13,12 @@ use crate::config::{AuthConfig, ClientConfig}; use crate::error::{Result, SdkError}; use crate::raw::{AuthedGrpcClient, AuthedInferenceClient}; use crate::refresh::{RefreshedToken, TokenSource}; -use crate::transport; use crate::types::{ ExecOptions, ExecResult, Health, ListOptions, SandboxPhase, SandboxRef, SandboxSpec, SandboxTemplateCreateSpec, SandboxTemplateListOptions, SandboxWorkloadTemplate, WorkspaceRef, }; -use futures::StreamExt; +use crate::{WatchEvent, WatchOptions, transport}; +use futures::{Stream, StreamExt}; use openshell_core::proto; use std::collections::HashMap; use std::future::Future; @@ -532,6 +532,114 @@ impl OpenShellClient { }) } + /// Watch a sandbox's logs and platform events with loss-aware resume. + /// + /// Reconnects transparently on transient stream errors, resuming from the + /// highest cursor already delivered. A trimmed resume cursor ends the stream + /// with [`SdkError::OutOfRange`]; a recoverable server lag surfaces as + /// [`WatchEvent::Warning`] and the stream continues. + pub fn watch_logs( + &self, + name: &str, + opts: WatchOptions, + ) -> impl Stream> + '_ { + let name = name.to_string(); + async_stream::try_stream!( + let sandbox = self.get_sandbox(&name).await?; + for await event in self.watch_logs_by_id(sandbox.id, opts) { + yield event?; + } + ) + } + + /// Shared watch loop over an already-resolved sandbox id. + /// + /// Both [`OpenShellClient::watch_logs`] and + /// [`WorkspaceScopedClient::watch_logs`] resolve a name to an id under their + /// own workspace, then delegate here so the reconnect/resume logic lives in + /// one place. + fn watch_logs_by_id( + &self, + sandbox_id: String, + opts: WatchOptions, + ) -> impl Stream> + '_ { + async_stream::try_stream!( + let mut cursor = opts.resume_after_cursor; + let mut backoff = Duration::from_millis(100); + loop { + let request = proto::WatchSandboxRequest { + id: sandbox_id.clone(), + follow_status: false, + follow_logs: opts.follow_logs, + follow_events: opts.follow_events, + log_tail_lines: opts.log_tail_lines, + event_tail: opts.event_tail, + log_sources: opts.log_sources.clone(), + log_min_level: opts.log_min_level.clone().unwrap_or_default(), + resume_after_cursor: cursor, + ..Default::default() + }; + // Apply the same reconnect policy to the initial dial: `unary` + // only retries `Unauthenticated`, so a pre-stream transient + // (e.g. `Unavailable`) would otherwise exit without resuming. + let mut stream = match self + .unary(|mut grpc| { + let req = request.clone(); + async move { grpc.watch_sandbox(req).await } + }) + .await + { + Ok(stream) => stream, + // Transient — back off and redial from `cursor`. + Err(err) if is_retryable_stream(&err) => { + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(Duration::from_secs(2)); + continue; + } + // Trimmed cursor or any other error — terminal. + Err(err) => Err(err)?, + }; + let mut clean_eof = true; + while let Some(item) = stream.next().await { + match item { + Ok(event) => { + // A delivered event means the connection is healthy + // again; reset the reconnect backoff so a later drop + // retries promptly instead of at the capped delay. + backoff = Duration::from_millis(100); + if let Some(ev) = convert_event(event, &mut cursor) { + yield ev; + } + } + Err(status) => { + clean_eof = false; + let err = map_status(status); + match err { + // Terminal gap — never silently restart. + SdkError::OutOfRange { .. } => { + Err(err)?; + } + // Transient — back off and redial from `cursor`. + _ if is_retryable_stream(&err) => { + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(Duration::from_secs(2)); + } + // Anything else is terminal. + _ => { + Err(err)?; + } + } + break; + } + } + } + if clean_eof { + break; + } + } + ) + } + /// Run a unary RPC with OIDC-aware auth: refresh proactively before the /// call (if the token is near expiry) and, on an `Unauthenticated` /// response, force a refresh and retry exactly once. No-op auth behaves @@ -924,6 +1032,26 @@ impl WorkspaceScopedClient { stderr, }) } + + /// Watch a sandbox's logs and platform events with loss-aware resume. + /// + /// Reconnects transparently on transient stream errors, resuming from the + /// highest cursor already delivered. A trimmed resume cursor ends the stream + /// with [`SdkError::OutOfRange`]; a recoverable server lag surfaces as + /// [`WatchEvent::Warning`] and the stream continues. + pub fn watch_logs( + &self, + name: &str, + opts: WatchOptions, + ) -> impl Stream> + '_ { + let name = name.to_string(); + async_stream::try_stream!( + let sandbox = self.get_sandbox(&name).await?; + for await event in self.client.watch_logs_by_id(sandbox.id, opts) { + yield event?; + } + ) + } } fn interceptor_from_config(config: &ClientConfig) -> Result { @@ -1059,6 +1187,7 @@ fn map_status(status: tonic::Status) -> SdkError { tonic::Code::NotFound => SdkError::NotFound { message }, tonic::Code::AlreadyExists => SdkError::AlreadyExists { message }, tonic::Code::InvalidArgument => SdkError::invalid_config(message), + tonic::Code::OutOfRange => SdkError::OutOfRange { message }, tonic::Code::Unauthenticated | tonic::Code::PermissionDenied => SdkError::auth(message), _ => SdkError::Rpc { code: status.code() as i32, @@ -1067,6 +1196,51 @@ fn map_status(status: tonic::Status) -> SdkError { } } +/// Convert a wire watch event into the curated [`WatchEvent`], advancing +/// `cursor` for resumable payloads. +/// +/// Log and platform events carry the shared per-sandbox cursor and update it. +/// Warnings are recoverable loss notices with no cursor, so they never advance +/// it. Status snapshots and draft-policy updates are not part of the log/event +/// stream and are dropped (`None`). +/// +/// `cursor` is a high-water mark, not the last cursor seen. The gateway reads +/// the log and platform sources independently during live delivery, so arrival +/// order can differ from cursor order. Taking the max keeps the resume point +/// monotonic; assigning directly would let a later lower-cursor event rewind it +/// and replay already-delivered events after a reconnect. +fn convert_event(event: proto::SandboxStreamEvent, cursor: &mut u64) -> Option { + match event.payload? { + proto::sandbox_stream_event::Payload::Log(line) => { + *cursor = (*cursor).max(event.cursor); + Some(WatchEvent::Log { + line: line.into(), + cursor: event.cursor, + }) + } + proto::sandbox_stream_event::Payload::Event(platform) => { + *cursor = (*cursor).max(event.cursor); + Some(WatchEvent::Event { + event: platform.into(), + cursor: event.cursor, + }) + } + proto::sandbox_stream_event::Payload::Warning(warning) => Some(WatchEvent::Warning { + message: warning.message, + }), + proto::sandbox_stream_event::Payload::Sandbox(_) + | proto::sandbox_stream_event::Payload::DraftPolicyUpdate(_) => None, + } +} + +/// Whether a mid-stream error is a transient condition worth reconnecting on. +/// +/// Only `Unavailable` (connection drop, gateway restart) is retryable; every +/// other status is terminal so the caller surfaces it instead of looping. +fn is_retryable_stream(err: &SdkError) -> bool { + matches!(err, SdkError::Rpc { code, .. } if *code == tonic::Code::Unavailable as i32) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/openshell-sdk/src/error.rs b/crates/openshell-sdk/src/error.rs index 343d66fd05..0869285815 100644 --- a/crates/openshell-sdk/src/error.rs +++ b/crates/openshell-sdk/src/error.rs @@ -93,6 +93,14 @@ pub enum SdkError { /// Error message. message: String, }, + + /// Gateway could not honor a resume cursor because the requested position + /// was already trimmed from its buffer (gRPC `OutOfRange`). The stream is + /// terminated; restart observation and, if needed, read missing lines from + /// the sandbox log files. + #[error("out of range: {message}")] + #[diagnostic(code(openshell::sdk::out_of_range))] + OutOfRange { message: String }, } impl SdkError { @@ -133,6 +141,13 @@ impl SdkError { } } + /// Create an `OutOfrange` error. + pub fn out_of_range(message: impl Into) -> Self { + Self::OutOfRange { + message: message.into(), + } + } + /// Stable string code for cross-language binding consumers. /// /// Returns one of: `invalid_config`, `tls`, `connect`, `auth`, `io`, @@ -165,6 +180,7 @@ impl SdkError { Self::NotFound { .. } => "not_found", Self::AlreadyExists { .. } => "already_exists", Self::Rpc { .. } => "rpc", + Self::OutOfRange { .. } => "out_of_range", } } } diff --git a/crates/openshell-sdk/src/lib.rs b/crates/openshell-sdk/src/lib.rs index 985c7ecc05..160d1ee0c8 100644 --- a/crates/openshell-sdk/src/lib.rs +++ b/crates/openshell-sdk/src/lib.rs @@ -47,8 +47,9 @@ pub use config::{AuthConfig, ClientConfig}; pub use error::SdkError; pub use refresh::{Refresh, RefreshError, RefreshedToken, TokenSource}; pub use types::{ - ExecOptions, ExecResult, Health, ListOptions, SandboxPhase, SandboxRef, SandboxResources, - SandboxServiceLevel, SandboxSpec, SandboxStartup, SandboxTemplateCreateSpec, + ExecOptions, ExecResult, Health, ListOptions, LogLine, PlatformEvent, SandboxPhase, SandboxRef, + SandboxResources, SandboxServiceLevel, SandboxSpec, SandboxStartup, SandboxTemplateCreateSpec, SandboxTemplateListOptions, SandboxWorkloadConfig, SandboxWorkloadTemplate, - SandboxWorkloadTemplateProvenance, SandboxWorkloadTemplateSpec, ServiceStatus, WorkspaceRef, + SandboxWorkloadTemplateProvenance, SandboxWorkloadTemplateSpec, ServiceStatus, WatchEvent, + WatchOptions, WorkspaceRef, }; diff --git a/crates/openshell-sdk/src/types.rs b/crates/openshell-sdk/src/types.rs index db2944474b..837ae8bde0 100644 --- a/crates/openshell-sdk/src/types.rs +++ b/crates/openshell-sdk/src/types.rs @@ -35,6 +35,53 @@ pub enum ServiceStatus { Unhealthy, } +/// One item from a reusable sandbox stream. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum WatchEvent { + /// A server/supervisor log line. Carries a resume cursor. + Log { line: LogLine, cursor: u64 }, + /// A platform event. Carries a resume cursor. + Event { event: PlatformEvent, cursor: u64 }, + /// Recoverable loss — the stream continues. No cursor (0). + Warning { message: String }, +} + +/// Options for [`crate::client::OpenShellClient::watch_logs`]. +#[derive(Debug, Clone, Default)] +pub struct WatchOptions { + pub follow_logs: bool, + pub follow_events: bool, + pub log_sources: Vec, + pub log_min_level: Option, + pub resume_after_cursor: u64, + pub log_tail_lines: u32, + pub event_tail: u32, +} + +#[derive(Debug, Clone)] +#[non_exhaustive] +pub struct LogLine { + pub sandbox_id: String, + pub timestamp_ms: i64, + pub level: String, + pub target: String, + pub message: String, + pub source: String, + pub fields: HashMap, +} + +#[derive(Debug, Clone)] +#[non_exhaustive] +pub struct PlatformEvent { + pub timestamp_ms: i64, + pub source: String, + pub r#type: String, + pub reason: String, + pub message: String, + pub metadata: HashMap, +} + impl From for ServiceStatus { fn from(value: proto::ServiceStatus) -> Self { match value { @@ -46,6 +93,41 @@ impl From for ServiceStatus { } } +impl From for LogLine { + fn from(value: proto::SandboxLogLine) -> Self { + // The wire contract treats an empty source as "gateway" for backward + // compatibility with pre-`source` producers. Normalize here so callers + // never have to special-case the empty string. + let source = if value.source.is_empty() { + "gateway".to_string() + } else { + value.source + }; + Self { + sandbox_id: value.sandbox_id, + timestamp_ms: value.timestamp_ms, + level: value.level, + target: value.target, + message: value.message, + source, + fields: value.fields, + } + } +} + +impl From for PlatformEvent { + fn from(value: proto::PlatformEvent) -> Self { + Self { + timestamp_ms: value.timestamp_ms, + source: value.source, + r#type: value.r#type, + reason: value.reason, + message: value.message, + metadata: value.metadata, + } + } +} + impl From for ServiceStatus { fn from(value: i32) -> Self { proto::ServiceStatus::try_from(value).map_or(Self::Unspecified, Self::from) diff --git a/crates/openshell-sdk/tests/client_mock.rs b/crates/openshell-sdk/tests/client_mock.rs index 58633ceb17..957234d8e4 100644 --- a/crates/openshell-sdk/tests/client_mock.rs +++ b/crates/openshell-sdk/tests/client_mock.rs @@ -13,13 +13,14 @@ use openshell_core::proto::open_shell_server::{OpenShell, OpenShellServer}; use openshell_sdk::{ AuthConfig, ClientConfig, ExecOptions, ListOptions, OpenShellClient, Refresh, RefreshError, RefreshedToken, SandboxPhase, SandboxSpec, SandboxTemplateCreateSpec, - SandboxTemplateListOptions, ServiceStatus as SdkServiceStatus, + SandboxTemplateListOptions, ServiceStatus as SdkServiceStatus, WatchEvent, WatchOptions, }; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; use tokio::net::TcpListener; use tokio::sync::Mutex; +use tokio_stream::StreamExt; use tokio_stream::wrappers::TcpListenerStream; use tonic::{Response, Status}; @@ -50,6 +51,25 @@ struct MockState { require_bearer: Option, /// Count of requests rejected by the `require_bearer` gate. unauth_hits: AtomicU32, + last_watch_requests: Mutex>, + watch_calls: AtomicU32, + // Per-dial script. Outer Vec index = dial number. Inner = events to send, + // then how to end that dial. + watch_script: Vec, +} + +#[derive(Debug, Clone)] +struct WatchDial { + events: Vec, + end: DialEnd, +} + +#[derive(Debug, Clone)] +enum DialEnd { + Clean, + Err(tonic::Code), + /// The dial itself fails before any stream opens (pre-stream RPC error). + FailDial(tonic::Code), } #[derive(Clone)] @@ -91,6 +111,41 @@ fn sandbox_with_phase_ws( } } +fn log_event(cursor: u64, msg: &str) -> proto::SandboxStreamEvent { + proto::SandboxStreamEvent { + payload: Some(proto::sandbox_stream_event::Payload::Log( + proto::SandboxLogLine { + sandbox_id: "id-my-box".into(), + timestamp_ms: 0, + level: "INFO".into(), + target: "t".into(), + message: msg.into(), + source: "sandbox".into(), + fields: HashMap::new(), + }, + )), + cursor, + } +} + +fn warning_event(msg: &str) -> proto::SandboxStreamEvent { + proto::SandboxStreamEvent { + payload: Some(proto::sandbox_stream_event::Payload::Warning( + proto::SandboxStreamWarning { + message: msg.into(), + }, + )), + cursor: 0, + } +} + +fn watch_opts() -> WatchOptions { + WatchOptions { + follow_logs: true, + ..Default::default() + } +} + fn workspace_proto(name: &str, phase: proto::datamodel::v1::WorkspacePhase) -> proto::Workspace { proto::Workspace { metadata: Some(proto::datamodel::v1::ObjectMeta { @@ -666,9 +721,37 @@ impl OpenShell for TestOpenShell { async fn watch_sandbox( &self, - _: tonic::Request, + request: tonic::Request, ) -> Result, Status> { - Err(Status::unimplemented("unused")) + let dial = self.state.watch_calls.fetch_add(1, Ordering::SeqCst) as usize; + self.state + .last_watch_requests + .lock() + .await + .push(request.into_inner()); + + let script = self.state.watch_script.get(dial).cloned(); + if let Some(WatchDial { + end: DialEnd::FailDial(code), + .. + }) = script + { + return Err(Status::new(code, "scripted dial failure")); + } + let (tx, rx) = tokio::sync::mpsc::channel(8); + tokio::spawn(async move { + let Some(dial) = script else { return }; + for ev in dial.events { + let _ = tx.send(Ok(ev)).await; + } + if let DialEnd::Err(code) = dial.end { + let _ = tx.send(Err(Status::new(code, "scripted"))).await; + } + // tx dropped here → stream ends + }); + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } async fn submit_policy_analysis( @@ -1595,3 +1678,202 @@ async fn raw_grpc_fresh_refreshes_before_raw_call() { "raw_grpc_fresh must refresh the near-expiry token exactly once" ); } + +#[tokio::test] +async fn watch_logs_forwards_logs_and_warnings() { + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![WatchDial { + events: vec![ + log_event(1, "a"), + warning_event("lagged"), + log_event(2, "b"), + ], + end: DialEnd::Clean, + }], + ..Default::default() + }); + let endpoint = start_mock(state.clone()).await; + let client = connect(&endpoint).await; + + let stream = client.watch_logs("my-box", watch_opts()); + tokio::pin!(stream); + + match stream.next().await.unwrap().unwrap() { + WatchEvent::Log { line, cursor } => { + assert_eq!(cursor, 1); + assert_eq!(line.message, "a"); + } + e => panic!("expected log, got {e:?}"), + } + + assert!(matches!( + stream.next().await.unwrap().unwrap(), + WatchEvent::Warning { .. } + )); + assert!(matches!( + stream.next().await.unwrap().unwrap(), + WatchEvent::Log { cursor: 2, .. } + )); + assert!(stream.next().await.is_none()); +} + +#[tokio::test] +async fn watch_logs_resumes_after_reconnect() { + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![ + WatchDial { + events: vec![log_event(1, "a"), log_event(2, "b")], + end: DialEnd::Err(tonic::Code::Unavailable), + }, + WatchDial { + events: vec![log_event(3, "c")], + end: DialEnd::Clean, + }, + ], + ..Default::default() + }); + let endpoint = start_mock(state.clone()).await; + let client = connect(&endpoint).await; + + let stream = client.watch_logs("my-box", watch_opts()); + tokio::pin!(stream); + for want in [1u64, 2, 3] { + assert!( + matches!(stream.next().await.unwrap().unwrap(), WatchEvent::Log { cursor, .. } if cursor == want) + ); + } + assert!(stream.next().await.is_none()); + + let reqs = state.last_watch_requests.lock().await; + assert_eq!(reqs.len(), 2); + assert_eq!(reqs[0].resume_after_cursor, 0); + assert_eq!(reqs[1].resume_after_cursor, 2); // resumed from highest delivered +} + +#[tokio::test] +async fn watch_logs_resumes_from_highest_cursor_when_arrival_is_unordered() { + // The gateway reads the log and platform sources independently during live + // delivery, so arrival order can differ from cursor order. The resume point + // must be the highest cursor seen, not the last one. + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![ + WatchDial { + events: vec![log_event(3, "c"), log_event(1, "a")], + end: DialEnd::Err(tonic::Code::Unavailable), + }, + WatchDial { + events: vec![log_event(4, "d")], + end: DialEnd::Clean, + }, + ], + ..Default::default() + }); + let endpoint = start_mock(state.clone()).await; + let client = connect(&endpoint).await; + + let stream = client.watch_logs("my-box", watch_opts()); + tokio::pin!(stream); + for want in [3u64, 1, 4] { + assert!( + matches!(stream.next().await.unwrap().unwrap(), WatchEvent::Log { cursor, .. } if cursor == want) + ); + } + assert!(stream.next().await.is_none()); + + let reqs = state.last_watch_requests.lock().await; + assert_eq!(reqs.len(), 2); + // Cursor 1 arrived last but must not rewind the resume point to 1, which + // would make the gateway replay cursors 2 and 3 all over again. + assert_eq!(reqs[1].resume_after_cursor, 3); +} + +#[tokio::test] +async fn watch_logs_normalizes_empty_log_source_to_gateway() { + let mut event = log_event(1, "a"); + if let Some(proto::sandbox_stream_event::Payload::Log(ref mut line)) = event.payload { + line.source = String::new(); + } + + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![WatchDial { + events: vec![event], + end: DialEnd::Clean, + }], + ..Default::default() + }); + let endpoint = start_mock(state.clone()).await; + let client = connect(&endpoint).await; + + let stream = client.watch_logs("my-box", watch_opts()); + tokio::pin!(stream); + let WatchEvent::Log { line, .. } = stream.next().await.unwrap().unwrap() else { + panic!("expected a log event"); + }; + // The wire contract treats an omitted source as "gateway". + assert_eq!(line.source, "gateway"); +} + +#[tokio::test] +async fn watch_logs_retries_initial_dial_failure() { + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![ + // First dial fails before the stream opens. + WatchDial { + events: vec![], + end: DialEnd::FailDial(tonic::Code::Unavailable), + }, + // Second dial succeeds and delivers. + WatchDial { + events: vec![log_event(1, "a")], + end: DialEnd::Clean, + }, + ], + ..Default::default() + }); + let endpoint = start_mock(state.clone()).await; + let client = connect(&endpoint).await; + + let stream = client.watch_logs("my-box", watch_opts()); + tokio::pin!(stream); + assert!(matches!( + stream.next().await.unwrap().unwrap(), + WatchEvent::Log { cursor: 1, .. } + )); + assert!(stream.next().await.is_none()); + + let reqs = state.last_watch_requests.lock().await; + assert_eq!(reqs.len(), 2); // dialed twice: failed, then reconnected + assert_eq!(reqs[0].resume_after_cursor, 0); + assert_eq!(reqs[1].resume_after_cursor, 0); // nothing delivered yet on retry +} + +#[tokio::test] +async fn watch_logs_gap_terminates_out_of_range() { + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![WatchDial { + events: vec![log_event(1, "a")], + end: DialEnd::Err(tonic::Code::OutOfRange), + }], + ..Default::default() + }); + let endpoint = start_mock(state.clone()).await; + let client = connect(&endpoint).await; + + let stream = client.watch_logs("my-box", watch_opts()); + tokio::pin!(stream); + assert!(matches!( + stream.next().await.unwrap().unwrap(), + WatchEvent::Log { cursor: 1, .. } + )); + let err = stream.next().await.unwrap().unwrap_err(); + assert_eq!(err.code(), "out_of_range"); + assert!(stream.next().await.is_none()); + + assert_eq!(state.last_watch_requests.lock().await.len(), 1); // no redial +} diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 5f7524f838..0a6d60be34 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -2818,6 +2818,8 @@ impl ComputeRuntime { public_platform_event_from_driver(&event), ), ), + // Placeholder: platform_event_bus.publish() stamps the cursor. + cursor: 0, }, ); } @@ -3498,8 +3500,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 64fd40cee2..d1463b9f7c 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -46,7 +46,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}; @@ -1448,6 +1448,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(); @@ -1505,6 +1506,8 @@ pub(super) async fn handle_watch_sandbox( sandbox.clone(), ), ), + // Status snapshots are re-read, not resumed by cursor. + cursor: 0, })) .await; @@ -1528,13 +1531,88 @@ 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 - { + // 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 + // unrecoverable gap -> terminate with a documented status. + use openshell_core::proto::sandbox_stream_event::Payload; + + // A cursor above everything this space has issued cannot be + // resumed: a gateway restart or bus teardown restarts the + // allocator at 1 with no record of the cursors it already gave + // out. The buses look merely empty, so `tail_after` reports no + // gap -- treating that as "caught up" would pin the cutoff to a + // stale cursor and silently swallow every live event beneath it. + let highest_cursor = state.tracing_log_bus.highest_cursor(&sandbox_id); + if resume_after_cursor > highest_cursor { + let _ = tx + .send(Err(Status::out_of_range(format!( + "resume cursor {resume_after_cursor} is no longer available; earliest resumable cursor is {}", + highest_cursor + 1 + )))) + .await; + return; + } + + 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); + + // 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 { continue; } @@ -1549,23 +1627,49 @@ 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; + } + } + replay_cutoff = replay_cutoff.max(evt.cursor); + 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) + { + replay_cutoff = replay_cutoff.max(evt.cursor); + if tx.send(Ok(evt)).await.is_err() { + return; + } } } } loop { - tokio::select! { + let first = tokio::select! { () = tx.closed() => { return; } @@ -1580,7 +1684,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 { @@ -1599,57 +1703,121 @@ 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; } } + // Status snapshots carry cursor 0 and are outside the + // resumable cursor space, so they never join a batch. + continue; } + // Both resumable sources feed one cursor space, so neither + // can be emitted on its own: `select!` picks an arbitrary + // ready branch, which would emit a higher cursor ahead of a + // lower one waiting on the other source. Take whichever woke + // us as the start of a batch and merge below. res = async { match log_rx.as_mut() { Some(rx) => rx.recv().await, None => future::pending().await, } - } => { - match res { - Ok(evt) => { - 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; - } - if !level_matches(&log.level, &log_min_level) { - continue; - } - } - if tx.send(Ok(evt)).await.is_err() { - return; - } - } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; - return; - } - } - } + } => res, res = async { match platform_rx.as_mut() { Some(rx) => rx.recv().await, None => future::pending().await, } - } => { - match res { - Ok(evt) => { - if tx.send(Ok(evt)).await.is_err() { - return; - } - } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; - return; + } => res, + }; + + let mut batch = Vec::new(); + match first { + Ok(evt) => batch.push(evt), + 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; + } + continue; + } + Err(broadcast::error::RecvError::Closed) => { + let _ = tx.send(Err(Status::cancelled("stream closed"))).await; + return; + } + } + + // Drain what is already queued on both sources. Anything ready + // now was published before the event we just took, so sorting + // the batch restores cursor order without waiting on either + // source. Events published after this drain are not held back: + // strict global ordering would mean delaying every event to see + // whether a lower cursor still arrives. + let mut lagged = 0u64; + let mut closed = false; + for rx in [log_rx.as_mut(), platform_rx.as_mut()] + .into_iter() + .flatten() + { + loop { + match rx.try_recv() { + Ok(evt) => batch.push(evt), + Err(broadcast::error::TryRecvError::Empty) => break, + // Keep draining: the receiver is usable after a skip. + Err(broadcast::error::TryRecvError::Lagged(n)) => lagged += n, + Err(broadcast::error::TryRecvError::Closed) => { + closed = true; + break; } } } } + + batch.sort_by_key(|evt| evt.cursor); + + for evt in batch { + // 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; + } + if !level_matches(&log.level, &log_min_level) { + continue; + } + } + if tx.send(Ok(evt)).await.is_err() { + return; + } + } + + if lagged > 0 + && tx + .send(Ok(crate::sandbox_watch::lag_warning_event(lagged))) + .await + .is_err() + { + return; + } + if closed { + let _ = tx.send(Err(Status::cancelled("stream closed"))).await; + return; + } } }, request_span, @@ -3537,6 +3705,364 @@ 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 live_delivery_orders_events_across_sources_by_cursor() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("liveorder", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + id: id.clone(), + follow_logs: true, + follow_events: true, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + // Draining the snapshot proves the producer reached the live loop, so + // it is subscribed to both buses before anything below is published. + let snap = stream.next().await.unwrap().unwrap(); + assert_eq!(snap.cursor, 0); + + // Publish without awaiting in between. On the current-thread runtime + // the producer cannot interleave, so both channels hold ready events + // when it next polls -- the state where `select!` picks arbitrarily and + // would otherwise emit a log cursor ahead of a lower platform cursor. + for i in 0..5 { + seed_log_lines(&state, &id, 1); // odd cursors + seed_platform_event(&state, &id, &format!("e{i}")); // even cursors + } + + let mut got = Vec::new(); + for _ in 0..10 { + got.push(stream.next().await.unwrap().unwrap().cursor); + } + assert_eq!(got, (1..=10).collect::>()); + } + + #[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 resume_from_reset_cursor_space_terminates_out_of_range() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("reset", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 5); + // Teardown resets the shared allocator, so the next publish starts over + // at cursor 1 -- the same as a gateway restart from the client's view. + state.tracing_log_bus.remove(&id); + seed_log_lines(&state, &id, 2); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + id: id.clone(), + follow_logs: true, + // Valid in the previous cursor space, unreachable in this one. + resume_after_cursor: 5, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert_eq!(snap.cursor, 0); + + let err = stream + .next() + .await + .unwrap() + .expect_err("a cursor from a reset space must terminate the stream"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + + // The events published after the reset must never be silently dropped + // as "already delivered" duplicates. + 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/sandbox_watch.rs b/crates/openshell-server/src/sandbox_watch.rs index ac38eba8db..bff687538c 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,24 @@ 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))), + // Warnings are not part of the resumable log/platform sequence. + cursor: 0, } } @@ -114,4 +126,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"); + } } diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index a91a5fd877..9ebf6b773e 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -4,7 +4,7 @@ //! Capture openshell-server tracing logs for streaming over gRPC. use std::collections::{HashMap, VecDeque}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, MutexGuard}; use openshell_core::proto::{SandboxLogLine, SandboxStreamEvent}; use openshell_ocsf::OCSF_TARGET; @@ -18,12 +18,120 @@ use tracing_subscriber::layer::Context; pub struct TracingLogBus { inner: Arc>, pub(crate) platform_event_bus: PlatformEventBus, + seq: SeqAllocator, } -#[derive(Debug)] +#[derive(Debug, Clone)] struct Inner { - per_id: HashMap>, - tails: HashMap>, + per_id: HashMap, +} + +#[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 { + fn new() -> Self { + let (tx, _rx) = broadcast::channel(1024); + 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 +/// 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 { + /// Lock the cursor space. + /// + /// Publication and teardown each hold this guard across their bus-map + /// mutation, which is what keeps cursors monotonic. Allocating and then + /// releasing would let a teardown reset the counter in between, so the + /// in-flight event lands in a freshly recreated entry carrying a cursor from + /// the old space while the next publish restarts at 1. + /// + /// The lock order is always allocator -> bus map. No path takes a bus map + /// lock and then reaches for the allocator, so the nesting cannot deadlock. + fn lock(&self) -> MutexGuard<'_, HashMap> { + self.inner.lock().expect("seq allocator lock poisoned") + } + + /// Take the next sequence number for this sandbox from a locked space. + /// + /// Seq starts at 1 so the proto default `resume_after_cursor` (0) means + /// "from the beginning" without skipping event 1. + fn next_locked(counters: &mut HashMap, sandbox_id: &str) -> u64 { + let counter = counters.entry(sandbox_id.to_string()).or_insert(1); + let seq = *counter; + *counter += 1; + seq + } + + /// Highest cursor handed out for this sandbox, or `0` when none is. + /// + /// Bounds the current cursor space. A resume cursor above this belongs to a + /// previous space (gateway restart, or the sandbox's buses were removed and + /// recreated), because the counter restarts at 1 with no memory of the + /// cursors it already issued. + fn highest_allocated(&self, sandbox_id: &str) -> u64 { + self.lock() + .get(sandbox_id) + .map_or(0, |next| next.saturating_sub(1)) + } +} + +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 { @@ -35,12 +143,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(), - tails: HashMap::new(), })), - platform_event_bus: PlatformEventBus::new(), + platform_event_bus: PlatformEventBus::new(seq.clone()), + seq, } } @@ -56,10 +167,8 @@ impl TracingLogBus { inner .per_id .entry(sandbox_id.to_string()) - .or_insert_with(|| { - let (tx, _rx) = broadcast::channel(1024); - tx - }) + .or_insert_with(PerSandbox::new) + .sender .clone() } @@ -67,26 +176,67 @@ 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 senders (closing any active receivers with + /// `RecvError::Closed`) and frees the tail buffers. /// - /// This drops the broadcast sender (closing any active receivers with - /// `RecvError::Closed`) and frees the tail buffer. + /// The whole sequence runs under the cursor-space lock, so it is atomic + /// against publication on either bus. Clearing the maps first is not enough + /// on its own: a publisher that had already allocated a cursor would insert + /// it into a recreated entry after the maps were cleared, and the next + /// publisher would restart at 1 behind 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); - inner.tails.remove(sandbox_id); + let mut counters = self.seq.lock(); + { + let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + inner.per_id.remove(sandbox_id); + } + // Takes only the platform bus map lock; never reaches for `counters`. + self.platform_event_bus.remove(sandbox_id); + counters.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::>() + } + + /// Highest cursor issued in the current cursor space for this sandbox. + /// + /// `0` means nothing has been published yet. Callers resuming from a client + /// cursor use this to tell "caught up" apart from "cursor belongs to a + /// cursor space that no longer exists": an empty `tail_after` result is not + /// on its own proof that the cursor is still valid. + pub fn highest_cursor(&self, sandbox_id: &str) -> u64 { + self.seq.highest_allocated(sandbox_id) + } + + 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). @@ -99,6 +249,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); } @@ -106,15 +258,26 @@ 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) { + // Hold the cursor space across the tail insert so a teardown cannot + // reset the counter between allocation and insertion. Lock order is + // allocator -> bus map, matching `remove`. + let mut counters = self.seq.lock(); + let seq = SeqAllocator::next_locked(&mut counters, sandbox_id); + event.cursor = seq; 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 = 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() > tail_cap { + if let Some((trimmed, _)) = per.tail.pop_front() { + per.last_trimmed_seq = trimmed; + } } } } @@ -155,6 +318,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); } @@ -208,6 +373,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(); @@ -227,6 +539,55 @@ mod tests { assert!(bus.tail(sandbox_id, 10).is_empty()); } + #[test] + fn concurrent_publish_and_remove_keeps_cursors_monotonic() { + // Teardown resets the shared allocator while both buses can still + // accept a publish. Unless the whole sequence is atomic against + // publication, a publisher that allocated before the reset inserts its + // old cursor into a recreated entry, and the next publisher restarts at + // 1 behind it -- leaving a tail whose cursors go backwards. + let bus = TracingLogBus::new(); + let sandbox_id = "sb-race"; + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let barrier = Arc::new(std::sync::Barrier::new(5)); + + let publishers: Vec<_> = (0..4) + .map(|_| { + let bus = bus.clone(); + let stop = Arc::clone(&stop); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + barrier.wait(); + while !stop.load(std::sync::atomic::Ordering::Relaxed) { + bus.publish_external(make_log_event(sandbox_id, "x")); + } + }) + }) + .collect(); + + barrier.wait(); + // Interleave teardown with in-flight publication. Each observation is a + // sample of the tail mid-race; a non-monotonic one means an event was + // stamped from a cursor space that no longer existed when it landed. + for _ in 0..20_000 { + bus.remove(sandbox_id); + let cursors: Vec = bus + .tail(sandbox_id, usize::MAX) + .iter() + .map(|e| e.cursor) + .collect(); + assert!( + cursors.windows(2).all(|w| w[0] < w[1]), + "tail cursors must stay strictly ascending, got {cursors:?}" + ); + } + + stop.store(true, std::sync::atomic::Ordering::Relaxed); + for publisher in publishers { + publisher.join().expect("publisher thread panicked"); + } + } + #[test] fn tracing_log_bus_subscribe_after_remove_creates_fresh_channel() { let bus = TracingLogBus::new(); @@ -278,13 +639,16 @@ 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); // 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()); @@ -300,7 +664,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); @@ -308,14 +672,17 @@ 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()); } #[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"); } @@ -324,7 +691,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 @@ -338,6 +705,7 @@ mod tests { message: format!("Message {i}"), metadata: HashMap::new(), })), + cursor: 0, }; bus.publish(sandbox_id, evt); } @@ -368,17 +736,20 @@ 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 { payload: None }; + let evt = SandboxStreamEvent { + payload: None, + cursor: 0, + }; bus.publish(sandbox_id, evt); assert_eq!(bus.tail(sandbox_id, 10).len(), 1); @@ -392,13 +763,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 { @@ -406,24 +772,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() } @@ -431,15 +797,26 @@ 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) { + // Hold the cursor space across the tail insert (same allocator -> map + // lock order as `TracingLogBus::publish`), so teardown cannot reset the + // counter underneath an in-flight publish. + let mut counters = self.seq.lock(); + let seq = SeqAllocator::next_locked(&mut counters, 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 { + if let Some((trimmed, _)) = per.tail.pop_front() { + per.last_trimmed_seq = trimmed; + } } } @@ -447,22 +824,40 @@ 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() .collect() } + 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, /// 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); } } diff --git a/docs/observability/accessing-logs.mdx b/docs/observability/accessing-logs.mdx index 4b755f74cc..67333ebb92 100644 --- a/docs/observability/accessing-logs.mdx +++ b/docs/observability/accessing-logs.mdx @@ -39,6 +39,21 @@ 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. + +Cursors are only meaningful within one cursor space. A gateway restart begins a new space numbered from 1, so a cursor held across the restart no longer refers to anything. The gateway rejects it with `OUT_OF_RANGE` instead of treating it as caught up. + +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. The handoff from replay to live delivery is exact: an event buffered while the stream was reopening is delivered once, never twice. It is not a guarantee that nothing was lost — a warning event or an `OUT_OF_RANGE` status still reports loss, both before and after a reconnect. + +Replay is emitted in cursor order. Live delivery merges the log and platform event sources by cursor before emitting, so events normally arrive in ascending cursor order. The gateway does not delay an event to wait for a lower cursor that has not been published yet, so a cursor can still arrive late under concurrent publication. Treat `cursor` as the ordering key rather than arrival order, and track the highest cursor seen as the resume point. + ## Direct Filesystem Access Start an independent shell with `sandbox exec` to read log files directly: diff --git a/examples/supervisor-middleware-content-guard/Cargo.lock b/examples/supervisor-middleware-content-guard/Cargo.lock index f31d5be9b5..8ce16c0100 100644 --- a/examples/supervisor-middleware-content-guard/Cargo.lock +++ b/examples/supervisor-middleware-content-guard/Cargo.lock @@ -854,6 +854,8 @@ dependencies = [ "prost-types", "protoc-bin-vendored", "rustix", + "rustls", + "rustls-pemfile", "serde", "serde_json", "thiserror", @@ -1246,6 +1248,15 @@ dependencies = [ "security-framework", ] +[[package]] +name = "rustls-pemfile" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "rustls-pki-types" version = "1.14.1" diff --git a/proto/openshell.proto b/proto/openshell.proto index e5e47b44e8..fdfdc1b9b4 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1536,6 +1536,21 @@ 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 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). + // + // Cursors are only meaningful within one cursor space. A gateway restart, or + // teardown of the sandbox's buffers, starts a new space numbered from 1. A + // cursor above everything the current space has issued is rejected with + // OUT_OF_RANGE rather than silently suppressing live events beneath it. + uint64 resume_after_cursor = 11; } // One event in a sandbox watch stream. @@ -1547,11 +1562,18 @@ 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-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; } // Log line correlated to a sandbox. @@ -1568,6 +1590,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; }