diff --git a/Cargo.lock b/Cargo.lock index 96d6617bb8..7c16585909 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4613,6 +4613,7 @@ dependencies = [ name = "openshell-sdk" version = "0.0.0" dependencies = [ + "async-stream", "async-trait", "futures", "hyper", diff --git a/architecture/gateway.md b/architecture/gateway.md index 742fa5e189..747d328921 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -440,6 +440,82 @@ 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 allocator stamps each with a `cursor`. +Cursor-ordered delivery is guaranteed for the replay phase: on resume the +buffered events from both sources are sorted before emission. Live events 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 an empty cursor. + +#### Cursor spaces + +A sandbox's cursors live in a **cursor space**: a `{epoch, seq}` pair, where the +epoch is a UUID minted on the first publish and `seq` counts from 1. The epoch +is dropped by `TracingLogBus::remove`, so a teardown — or a gateway restart — +retires the space, and the next publish mints a new one. A sequence number alone +cannot distinguish a caught-up client from one holding a cursor out of a space +that no longer exists, because the replacement space reuses the same numbers; +the epoch answers *which counter issued this*, which is the question resume +validation actually has to ask. Behind multiple replicas the same rule makes a +reconnect to a different replica fail loudly rather than return the wrong +events. + +On the wire a cursor is an opaque, fixed-width token. Clients may only compare +two cursors from one stream and keep the greater; the encoding zero-pads `seq` +so that byte-wise comparison matches sequence order, which is what lets every +SDK track a high-water mark without parsing. A stream only ever observes one +epoch — a reset closes both resumable broadcast receivers, ending the stream +rather than switching spaces mid-flight — so that comparison is always well +defined where clients are allowed to use it. The gateway does not rely on it: +server-side ordering runs on the raw `u64` seq carried alongside each event in +`CursoredEvent`, never on the token. + +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. Since + cursors are opaque, the warning is the client's only signal. +- **Unrecoverable gap** — the server sends a snapshot, then terminates with + `OUT_OF_RANGE`. Three cases reach it: the tail was trimmed past the requested + cursor, the cursor's epoch does not match the sandbox's current space, or no + space exists because nothing has been published since teardown. The status + tells the client to restart with an empty cursor; retrying the same token + fails identically. A token the gateway could not have issued is rejected + earlier, as `INVALID_ARGUMENT` on the call itself. + +Both resumable sources draw from one cursor space, so the server merges them by +seq before emitting rather than draining each in turn: on resume it replays only +events after the client's cursor, and without one it replays each bus's retained +tail. Either way the batch leaves in ascending cursor order. The two tails are +bounded independently (`log_tail_lines` and `event_tail`), so merging orders +whatever each bus kept; it does not align their depths. + +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 seq and suppresses live events at or below it, so +each event is delivered once. That mark is per source. The two tails are read at +different instants and bounded independently, so one shared mark would let the +deeper source censor the shallower one — with `event_tail` unset the mark rises +to the newest buffered log while no platform event is replayed at all, and +platform events published during initialization are discarded as duplicates of a +replay that never ran. Subscribing never mints a cursor space, so a +resume against a torn-down sandbox cannot create the space its stale cursor is +then checked against. Clients track the highest observed `cursor` and pass it as +`resume_after_cursor` on reconnect. + +The epoch is validated twice on resume: once before reading the tails and again +once both are in hand, before anything is emitted. The check and each read take +their locks separately, so a teardown plus a republish can retire the validated +space and install a replacement in between; the reads would then apply the old +space's seq to the replacement's buffers, and a trimmed-range check that only +compares numbers would report no gap while skipping the replacement's lower +events. The second look ends the stream with `OUT_OF_RANGE` instead. + ## 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 489db3136f..b6292f5dc9 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -809,6 +809,7 @@ pub async fn sandbox_create( since_time: None, log_sources: vec!["gateway".to_string()], log_min_level: String::new(), + resume_after_cursor: String::new(), }) .await .into_diagnostic()? @@ -3613,6 +3614,7 @@ async fn wait_for_lifecycle_phase( since_time: None, log_sources: Vec::new(), log_min_level: String::new(), + resume_after_cursor: String::new(), }) .await .into_diagnostic()? @@ -5847,6 +5849,7 @@ pub async fn sandbox_logs( .into_diagnostic()?, log_sources: source_filter, log_min_level: level.to_uppercase(), + resume_after_cursor: String::new(), }) .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 f1408493bc..0b638d5694 100644 --- a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs +++ b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs @@ -743,6 +743,7 @@ impl OpenShell for TestOpenShell { let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(provisioning)), + cursor: String::new(), })) .await; if terminal_after_provisional_container_exit @@ -753,6 +754,7 @@ impl OpenShell for TestOpenShell { payload: Some(sandbox_stream_event::Payload::Sandbox( provisional_container_exit, )), + cursor: String::new(), })) .await; provisional_container_exit_sent.notify_waiters(); @@ -764,6 +766,7 @@ impl OpenShell for TestOpenShell { let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(completed)), + cursor: String::new(), })) .await; return; @@ -777,11 +780,13 @@ impl OpenShell for TestOpenShell { message: "Started VM launcher".to_string(), ..PlatformEvent::default() })), + cursor: String::new(), })) .await; let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(error)), + cursor: String::new(), })) .await; tokio::time::sleep(Duration::from_secs(5)).await; @@ -801,12 +806,14 @@ impl OpenShell for TestOpenShell { source: "gateway".to_string(), fields: HashMap::new(), })), + cursor: String::new(), })) .await; } let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(ready)), + cursor: String::new(), })) .await; return; @@ -815,6 +822,7 @@ impl OpenShell for TestOpenShell { let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(completed)), + cursor: String::new(), })) .await; return; @@ -829,6 +837,7 @@ impl OpenShell for TestOpenShell { message: "Preparing rootfs".to_string(), ..PlatformEvent::default() })), + cursor: String::new(), })) .await; tokio::time::sleep(Duration::from_millis(600)).await; @@ -840,12 +849,14 @@ impl OpenShell for TestOpenShell { message: "Formatting root disk".to_string(), ..PlatformEvent::default() })), + cursor: String::new(), })) .await; tokio::time::sleep(Duration::from_millis(600)).await; let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(ready)), + cursor: String::new(), })) .await; return; @@ -857,11 +868,13 @@ impl OpenShell for TestOpenShell { message: "Sandbox scheduled".to_string(), ..PlatformEvent::default() })), + cursor: String::new(), })) .await; let _ = tx .send(Ok(SandboxStreamEvent { payload: Some(sandbox_stream_event::Payload::Sandbox(ready)), + cursor: String::new(), })) .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 b019491d6f..59b42e7997 100644 --- a/crates/openshell-sdk/src/client.rs +++ b/crates/openshell-sdk/src/client.rs @@ -14,13 +14,13 @@ use crate::error::{Result, SdkError}; use crate::pagination::{Page, Pager}; use crate::raw::AuthedGrpcClient; use crate::refresh::{RefreshedToken, TokenSource}; -use crate::transport; use crate::types::{ DeleteOptions, DeletionResult, 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; @@ -662,6 +662,124 @@ 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 recoverable server lag surfaces as + /// [`WatchEvent::Warning`] and the stream continues. + /// + /// [`SdkError::OutOfRange`] is terminal and deliberately not retried: it + /// means the resume point is gone (trimmed from the buffer, or issued by a + /// cursor space the gateway no longer has), so events between it and now + /// are unrecoverable. Auto-restarting from scratch would hide that loss, + /// which is exactly what this API exists to surface. Callers who accept the + /// gap can start a new watch with an empty + /// [`WatchOptions::resume_after_cursor`]; retrying the same cursor fails + /// identically. + 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_name(sandbox.name, "default".to_string(), opts) { + yield event?; + } + ) + } + + /// Shared watch loop over an already-resolved canonical sandbox name. + /// + /// Both [`OpenShellClient::watch_logs`] and + /// [`WorkspaceScopedClient::watch_logs`] resolve and re-confirm a name under + /// their own workspace, then delegate here so the reconnect/resume logic + /// lives in one place. + fn watch_logs_by_name( + &self, + sandbox_name: String, + workspace: 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 { + sandbox: sandbox_name.clone(), + workspace_scope: Some(proto::workspace_selector(&workspace)), + 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.clone(), + ..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 @@ -1118,6 +1236,34 @@ 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 recoverable server lag surfaces as + /// [`WatchEvent::Warning`] and the stream continues. + /// + /// [`SdkError::OutOfRange`] is terminal and deliberately not retried: it + /// means the resume point is gone (trimmed from the buffer, or issued by a + /// cursor space the gateway no longer has), so events between it and now + /// are unrecoverable. Auto-restarting from scratch would hide that loss, + /// which is exactly what this API exists to surface. Callers who accept the + /// gap can start a new watch with an empty + /// [`WatchOptions::resume_after_cursor`]; retrying the same cursor fails + /// identically. + 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_name(sandbox.name, self.workspace.clone(), opts) { + yield event?; + } + ) + } } fn interceptor_from_config(config: &ClientConfig) -> Result { @@ -1275,6 +1421,61 @@ fn map_status(status: tonic::Status) -> SdkError { SdkError::from_status(status) } +/// 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. +/// +/// The comparison is a plain byte-wise string compare on an opaque token. That +/// is the one operation the gateway permits on a cursor, and it is well defined +/// here because both cursors come from the same stream: the encoding is fixed +/// width within a cursor space, and a stream never spans two spaces (a reset +/// ends it). Never parse the token — its layout is not part of the contract. +fn convert_event(event: proto::SandboxStreamEvent, cursor: &mut String) -> Option { + match event.payload? { + proto::sandbox_stream_event::Payload::Log(line) => { + if event.cursor > *cursor { + cursor.clone_from(&event.cursor); + } + Some(WatchEvent::Log { + line: line.into(), + cursor: event.cursor, + }) + } + proto::sandbox_stream_event::Payload::Event(platform) => { + if event.cursor > *cursor { + cursor.clone_from(&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 be61bb00e9..0422f9bf8e 100644 --- a/crates/openshell-sdk/src/error.rs +++ b/crates/openshell-sdk/src/error.rs @@ -104,6 +104,19 @@ pub enum SdkError { /// Original gateway status, including unknown details and metadata. status: Box, }, + + /// 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 { + /// Error message. + message: String, + /// Original gateway status, including details and metadata. + status: Box, + }, } impl SdkError { @@ -155,6 +168,7 @@ impl SdkError { match code { tonic::Code::NotFound => Self::NotFound { message, status }, tonic::Code::AlreadyExists => Self::AlreadyExists { message, status }, + tonic::Code::OutOfRange => Self::OutOfRange { message, status }, tonic::Code::InvalidArgument => Self::InvalidConfig { message, status: Some(status), @@ -178,6 +192,7 @@ impl SdkError { Self::InvalidConfig { status, .. } | Self::Auth { status, .. } => status.as_deref(), Self::NotFound { status, .. } | Self::AlreadyExists { status, .. } + | Self::OutOfRange { status, .. } | Self::Rpc { status, .. } => Some(status), _ => None, } @@ -228,6 +243,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 eac522a187..269104ae2d 100644 --- a/crates/openshell-sdk/src/lib.rs +++ b/crates/openshell-sdk/src/lib.rs @@ -51,8 +51,9 @@ pub use pagination::{Page, Pager}; pub use refresh::{Refresh, RefreshError, RefreshedToken, TokenSource}; pub use types::{ DeleteOptions, DeletionOutcome, DeletionResult, ExecOptions, ExecResult, Health, ListOptions, - SandboxPhase, SandboxRef, SandboxResources, SandboxServiceLevel, SandboxSpec, SandboxStartup, - SandboxTemplateCreateSpec, SandboxTemplateListOptions, SandboxWorkloadConfig, - SandboxWorkloadTemplate, SandboxWorkloadTemplateProvenance, SandboxWorkloadTemplateSpec, - ServiceExposure, ServiceStatus, WorkspaceRef, + LogLine, PlatformEvent, SandboxPhase, SandboxRef, SandboxResources, SandboxServiceLevel, + SandboxSpec, SandboxStartup, SandboxTemplateCreateSpec, SandboxTemplateListOptions, + SandboxWorkloadConfig, SandboxWorkloadTemplate, SandboxWorkloadTemplateProvenance, + SandboxWorkloadTemplateSpec, ServiceExposure, ServiceStatus, WatchEvent, WatchOptions, + WorkspaceRef, }; diff --git a/crates/openshell-sdk/src/types.rs b/crates/openshell-sdk/src/types.rs index a9fd544f14..57bed05e08 100644 --- a/crates/openshell-sdk/src/types.rs +++ b/crates/openshell-sdk/src/types.rs @@ -81,6 +81,61 @@ 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 an opaque resume cursor. + Log { line: LogLine, cursor: String }, + /// A platform event. Carries an opaque resume cursor. + Event { + event: PlatformEvent, + cursor: String, + }, + /// Recoverable loss — the stream continues. No cursor (empty). + 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, + /// Opaque cursor to resume after. Empty starts from the tail. + /// + /// Use a cursor taken from a [`WatchEvent`] of a previous watch on the same + /// sandbox. Do not construct or parse one: the encoding is not part of the + /// gateway's contract. + pub resume_after_cursor: String, + 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 { @@ -92,6 +147,54 @@ 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, + // The wire contract carries `google.protobuf.Timestamp`; these + // curated types stay dependency-light and expose milliseconds, the + // same reduction the CLI applies at its own presentation edge. An + // absent or unrepresentable timestamp reads as 0, which is what + // this field meant before the wire types gained presence. + timestamp_ms: value + .event_time + .as_ref() + .and_then(|time| openshell_core::time::timestamp_to_millis(time).ok()) + .unwrap_or(0), + 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 + .event_time + .as_ref() + .and_then(|time| openshell_core::time::timestamp_to_millis(time).ok()) + .unwrap_or(0), + 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 4051cd5136..1c86ada7dd 100644 --- a/crates/openshell-sdk/tests/client_mock.rs +++ b/crates/openshell-sdk/tests/client_mock.rs @@ -13,7 +13,8 @@ use openshell_core::proto::open_shell_server::{OpenShell, OpenShellServer}; use openshell_sdk::{ AuthConfig, ClientConfig, ExecOptions, ListOptions, OpenShellClient, Refresh, RefreshError, RefreshedToken, SandboxPhase, SandboxSpec, SandboxTemplateCreateSpec, - SandboxTemplateListOptions, ServiceExposure, ServiceStatus as SdkServiceStatus, + SandboxTemplateListOptions, ServiceExposure, ServiceStatus as SdkServiceStatus, WatchEvent, + WatchOptions, }; use std::collections::HashMap; use std::sync::Arc; @@ -21,6 +22,7 @@ use std::sync::atomic::{AtomicU32, Ordering}; use std::time::Duration; use tokio::net::TcpListener; use tokio::sync::Mutex; +use tokio_stream::StreamExt; use tokio_stream::wrappers::TcpListenerStream; use tonic::{Response, Status}; @@ -72,6 +74,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)] @@ -113,6 +134,53 @@ fn sandbox_with_phase_ws( } } +/// Encode the wire cursor for `seq`, spelled out rather than built with the +/// server's encoder. +/// +/// The SDK treats cursors as opaque, so nothing in this crate can produce one. +/// Writing the format by hand also pins it independently: the padding is what +/// makes the SDK's byte-wise high-water comparison agree with sequence order, +/// and a server-side change that dropped it would have to break this literal +/// before it could break a client. +fn test_cursor(seq: u64) -> String { + format!("v1:11111111-1111-4111-8111-111111111111:{seq:020}") +} + +fn log_event(seq: u64, msg: &str) -> proto::SandboxStreamEvent { + proto::SandboxStreamEvent { + payload: Some(proto::sandbox_stream_event::Payload::Log( + proto::SandboxLogLine { + sandbox_id: "id-my-box".into(), + event_time: None, + level: "INFO".into(), + target: "t".into(), + message: msg.into(), + source: "sandbox".into(), + fields: HashMap::new(), + }, + )), + cursor: test_cursor(seq), + } +} + +fn warning_event(msg: &str) -> proto::SandboxStreamEvent { + proto::SandboxStreamEvent { + payload: Some(proto::sandbox_stream_event::Payload::Warning( + proto::SandboxStreamWarning { + message: msg.into(), + }, + )), + cursor: String::new(), + } +} + +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 { @@ -807,9 +875,48 @@ 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; + let req = request.into_inner(); + + // The gateway resolves `sandbox` as a workspace-scoped canonical name, + // so an object ID here is NOT_FOUND rather than a silent no-op stream. + // Mirrored so every watch test fails on ID-addressed requests. + let resolved = self.state.last_get_name.lock().await.clone(); + if let Some(resolved) = resolved + && req.sandbox != resolved + { + return Err(Status::not_found(format!( + "sandbox '{}' not found", + req.sandbox + ))); + } + + self.state.last_watch_requests.lock().await.push(req); + + 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( @@ -1978,3 +2085,287 @@ 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, test_cursor(1)); + assert_eq!(line.message, "a"); + } + e => panic!("expected log, got {e:?}"), + } + + assert!(matches!( + stream.next().await.unwrap().unwrap(), + WatchEvent::Warning { .. } + )); + match stream.next().await.unwrap().unwrap() { + WatchEvent::Log { cursor, .. } => assert_eq!(cursor, test_cursor(2)), + e => panic!("expected log, got {e:?}"), + } + 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 == test_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, ""); + // Resumed from the highest delivered cursor, forwarded verbatim. + assert_eq!(reqs[1].resume_after_cursor, test_cursor(2)); +} + +#[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 == test_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, test_cursor(3)); +} + +#[tokio::test] +async fn watch_logs_resume_cursor_is_padded_high_water() { + // The high-water mark is a byte-wise string comparison on an opaque token, + // so it is only correct while the sequence segment is a fixed width. Seq 9 + // then seq 10 is the case that catches an unpadded encoding: "...:9" sorts + // above "...:10", so the resume point would rewind to 9 and the gateway + // would replay an event the client already has. + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![ + WatchDial { + events: vec![log_event(9, "i"), log_event(10, "j")], + end: DialEnd::Err(tonic::Code::Unavailable), + }, + WatchDial { + events: vec![], + 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 [9u64, 10] { + assert!( + matches!(stream.next().await.unwrap().unwrap(), WatchEvent::Log { cursor, .. } if cursor == test_cursor(want)) + ); + } + assert!(stream.next().await.is_none()); + + let reqs = state.last_watch_requests.lock().await; + assert_eq!(reqs.len(), 2); + assert_eq!(reqs[1].resume_after_cursor, test_cursor(10)); +} + +#[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 { ref cursor, .. } if *cursor == test_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, ""); + assert_eq!(reqs[1].resume_after_cursor, ""); // 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 { ref cursor, .. } if *cursor == test_cursor(1) + )); + let err = stream.next().await.unwrap().unwrap_err(); + assert_eq!(err.code(), "out_of_range"); + assert!(stream.next().await.is_none()); + + // No redial. OUT_OF_RANGE is terminal for every cause the gateway uses it + // for -- a trimmed cursor or one from a retired cursor space -- because + // reconnecting would silently paper over events that are already lost. + assert_eq!(state.last_watch_requests.lock().await.len(), 1); +} + +// `sandbox` addresses a workspace-scoped canonical name, and the mock's +// `id-{name}` ids differ from the names, so sending a resolved object id here +// reaches the gateway as NOT_FOUND and the watch never streams. +#[tokio::test] +async fn watch_logs_addresses_the_sandbox_by_canonical_name() { + for workspace in ["default", "production"] { + let state = Arc::new(MockState { + phase_sequence: vec![proto::SandboxPhase::Ready], + watch_script: vec![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 scoped = client.workspace(workspace); + let mut stream: std::pin::Pin>> = + if workspace == "default" { + Box::pin(client.watch_logs("watched", watch_opts())) + } else { + Box::pin(scoped.watch_logs("watched", watch_opts())) + }; + + assert!( + matches!( + stream.next().await.unwrap().unwrap(), + WatchEvent::Log { ref cursor, .. } if *cursor == test_cursor(1) + ), + "{workspace}: first event must stream" + ); + + let reqs = state.last_watch_requests.lock().await; + assert_eq!(reqs.len(), 1, "{workspace}"); + assert_eq!(reqs[0].sandbox, "watched", "{workspace}: name, not id"); + assert_eq!( + selected_workspace(&reqs[0].workspace_scope), + Some(workspace) + ); + } +} diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 9f32743137..ad98e46247 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -3591,6 +3591,8 @@ impl ComputeRuntime { public_platform_event_from_driver(&event), ), ), + // Placeholder: platform_event_bus.publish() stamps the cursor. + cursor: String::new(), }, ); } @@ -4399,8 +4401,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/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 27fcf946da..4d7c52853d 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -4712,7 +4712,7 @@ pub(super) async fn handle_get_sandbox_logs( .into_iter() .filter_map(|evt| { if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log(log)) = - evt.payload + evt.event.payload { if let Some(since_time) = since_time.as_ref() { let event_time = log.event_time.as_ref()?; diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index b75f464f0b..fd83e008e3 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -18,6 +18,8 @@ use crate::pagination::Pagination; use crate::persistence::{ ObjectLabels, ObjectListQuery, ObjectType, WriteCondition, generate_name, }; +use crate::tracing_bus::CursoredEvent; +use crate::watch_cursor::WatchCursor; use futures::future; use openshell_core::net::set_tcp_nodelay_best_effort; use openshell_core::proto::datamodel::v1::ObjectMeta; @@ -53,7 +55,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}; @@ -81,6 +83,32 @@ const MAX_CREATE_SERVICE_EXPOSURES: usize = 32; #[path = "interactive_exec_tests.rs"] mod interactive_exec_tests; +/// Terminal status for a resume cursor issued by a cursor space that is gone. +/// +/// Retrying the same token fails identically, so the guidance has to be +/// "restart without one" -- otherwise an SDK that reconnects on `OUT_OF_RANGE` +/// spins. The token is not echoed back. +const RESUME_SPACE_GONE: &str = "resume_after_cursor belongs to a cursor space that no longer \ + exists; the gateway restarted or the sandbox's buffers were torn down. Restart the watch \ + with an empty resume_after_cursor. Events published in the meantime are not recoverable."; + +/// Terminal status for a resume cursor ahead of everything its space issued. +const RESUME_CURSOR_AHEAD: &str = "resume_after_cursor is ahead of every cursor this sandbox has \ + issued. Restart the watch with an empty resume_after_cursor."; + +/// Whether `sandbox_id`'s cursor space is still the one that issued `epoch`. +/// +/// A teardown retires the space and the next publish mints a replacement that +/// renumbers from 1, so a surviving epoch is the only proof that a seq validated +/// earlier still addresses the same numbering. Absent counts as changed: there +/// is nothing left for the cursor to point into. +fn cursor_space_is(state: &ServerState, sandbox_id: &str, epoch: uuid::Uuid) -> bool { + state + .tracing_log_bus + .cursor_space(sandbox_id) + .is_some_and(|space| space.epoch == epoch) +} + #[derive(Debug)] pub struct WatchSandboxStream { receiver: ReceiverStream>, @@ -1752,6 +1780,19 @@ pub(super) async fn handle_watch_sandbox( let log_min_level = req.log_min_level; let event_tail = req.event_tail; + // Decode the resume cursor before spawning the producer. A token this + // server could not have issued is pure input validation, in the same class + // as the `id is required` check above, so it fails the RPC rather than + // arriving as the first item of an otherwise-established stream. + let resume_after = if req.resume_after_cursor.is_empty() { + None + } else { + Some( + WatchCursor::parse(&req.resume_after_cursor) + .map_err(|e| Status::invalid_argument(e.to_string()))?, + ) + }; + let (tx, rx) = mpsc::channel::>(256); let state = state.clone(); @@ -1808,6 +1849,8 @@ pub(super) async fn handle_watch_sandbox( sandbox.clone(), ), ), + // Status snapshots are re-read, not resumed by cursor. + cursor: String::new(), })) .await; @@ -1831,13 +1874,142 @@ pub(super) async fn handle_watch_sandbox( } } - // Replay tail logs (best-effort), filtered by log_since_time 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 seq the tail/replay phase already handled, tracked per + // source. 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 its source's mark so each is delivered exactly once. + // + // The two marks must stay separate. Both buses number from one + // shared cursor space, but they are read at different instants and + // bounded independently (`log_tail_lines` vs `event_tail`, which has + // no default and so replays nothing unless the client asks). A + // single shared mark therefore lets the deeper source censor the + // shallower one: with the default `event_tail` of 0 the mark rises + // to the newest buffered log while no platform event was replayed at + // all, and every platform event published in the initialization + // window is dropped as a duplicate of something never sent. Keyed by + // source, an event is suppressed only if its own source's replay + // actually covered it. + // + // No unit test pins this. The only reachable window is between the + // subscribe above and the log tail read below -- an event published + // earlier is replayed rather than live, and one published later + // outranks the mark -- and the producer crosses that window with no + // await a test can wedge open. Reproducing it needs a seam in the + // producer, which is not worth adding to production code. + let resume_seq = resume_after.map_or(0, |resume| resume.seq); + let mut log_cutoff: u64 = resume_seq; + let mut platform_cutoff: u64 = resume_seq; + + if let Some(resume) = resume_after { + // 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 is only a position inside the space that issued it. + // A gateway restart, a bus teardown, or a reconnect landing on + // another replica starts a new space numbered from 1 with no + // record of the cursors the old one handed out. The buses then + // look merely empty, so `tail_after` reports no gap -- treating + // that as "caught up" would pin the cutoff to a stale number + // and silently swallow every live event beneath it. Comparing + // epochs answers "did this cursor come from *this* space?", + // which no numeric bound can. + match state.tracing_log_bus.cursor_space(&sandbox_id) { + None => { + let _ = tx.send(Err(Status::out_of_range(RESUME_SPACE_GONE))).await; + return; + } + Some(space) if space.epoch != resume.epoch => { + let _ = tx.send(Err(Status::out_of_range(RESUME_SPACE_GONE))).await; + return; + } + // Right space, but ahead of anything it issued: only a + // fabricated token gets here. Reject rather than accept a + // cutoff no event can ever exceed. + Some(space) if resume.seq > space.highest_seq => { + let _ = tx + .send(Err(Status::out_of_range(RESUME_CURSOR_AHEAD))) + .await; + return; + } + Some(_) => {} + } + + let log_replay = if follow_logs { + Some(state.tracing_log_bus.tail_after(&sandbox_id, resume.seq)) + } else { + None + }; + + let platform_replay = if follow_events { + Some( + state + .tracing_log_bus + .platform_event_bus + .tail_after(&sandbox_id, resume.seq), + ) + } else { + None + }; + + // Re-check the epoch now that both tails are in hand. The check + // above and each `tail_after` take their locks independently, so + // a teardown plus a republish can retire the validated space and + // install a replacement in between. The reads would then have + // applied the old space's seq to the new space's buffers, and + // `tail_after` -- which only knows numbers -- would report no gap + // while skipping every replacement event at or below it. The + // second look is cheap and runs before anything is emitted, so a + // space that moved under us ends the stream instead of serving a + // truncated replay. + // + // Ordering is unchanged: this takes only the allocator lock and + // releases it, never held across a bus lock. + if !cursor_space_is(&state, &sandbox_id, resume.epoch) { + let _ = tx.send(Err(Status::out_of_range(RESUME_SPACE_GONE))).await; + return; + } + + // 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 seq, then emit ascending. Each + // source's mark comes from its own replay -- `tail_after` + // returns ascending, so its last entry is that source's high + // water. Marking every event the phase examined, not only the + // ones that survived the filters, keeps a filtered event's live + // duplicate suppressed: the live loop does not re-apply + // `log_since_time` and would otherwise let it through. + let mut merged: Vec = Vec::new(); + if let Some(Ok(v)) = log_replay { + if let Some(last) = v.last() { + log_cutoff = log_cutoff.max(last.seq); + } + merged.extend(v); + } + if let Some(Ok(v)) = platform_replay { + if let Some(last) = v.last() { + platform_cutoff = platform_cutoff.max(last.seq); + } + merged.extend(v); + } + + merged.sort_by_key(|c| c.seq); + + for cursored in merged { + if let Some(Payload::Log(ref log)) = cursored.event.payload { if let Some(since_time) = log_since_time.as_ref() { let Some(event_time) = log.event_time.as_ref() else { continue; @@ -1858,27 +2030,88 @@ pub(super) async fn handle_watch_sandbox( continue; } } - if tx.send(Ok(evt)).await.is_err() { + if tx.send(Ok(cursored.event)).await.is_err() { return; } } - } + } else { + // Initial tail, best-effort. Both buses draw from one shared + // cursor space, so draining each in its own pass would order the + // tail by source and only incidentally by cursor: every buffered + // log would precede every buffered platform event regardless of + // which was published first. Collect both windows, sort by the + // shared seq, and emit one ascending run -- the same shape the + // resume path above uses, and the order a client comparing + // cursors expects. + // + // The two windows are truncated independently (log_tail vs + // event_tail), so this merges whatever each bus retained; it + // does not align their depths. + let mut tail: Vec = Vec::new(); + if follow_logs { + let logs = state.tracing_log_bus.tail(&sandbox_id, log_tail as usize); + if let Some(last) = logs.last() { + log_cutoff = log_cutoff.max(last.seq); + } + tail.extend(logs); + } + if follow_events { + let events = state + .tracing_log_bus + .platform_event_bus + .tail(&sandbox_id, event_tail as usize); + if let Some(last) = events.last() { + platform_cutoff = platform_cutoff.max(last.seq); + } + tail.extend(events); + } - // 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() { + tail.sort_by_key(|cursored| cursored.seq); + + for cursored in tail { + // Log filters; platform events carry no log fields and pass. + if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log( + ref log, + )) = cursored.event.payload + { + if let Some(since_time) = log_since_time.as_ref() { + let Some(event_time) = log.event_time.as_ref() else { + continue; + }; + let Ok(ordering) = + openshell_core::time::compare_timestamps(event_time, since_time) + else { + continue; + }; + if ordering == std::cmp::Ordering::Less { + 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(cursored.event)).await.is_err() { return; } } } + // Events drained above the publication watermark. They are held + // back rather than emitted so the client's highest delivered cursor + // is never above an event still queued on the other source. + let mut deferred: Vec = Vec::new(); + loop { - tokio::select! { + // Events withheld by the previous round are already in hand, so + // this round must not block on a new publication: the watermark + // that covers them has already advanced past them, and waiting + // for unrelated traffic would stall their delivery indefinitely. + let first = if deferred.is_empty() { + Some(tokio::select! { () = tx.closed() => { return; } @@ -1893,7 +2126,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: String::new() })).await.is_err() { return; } if stop_on_terminal { @@ -1912,57 +2145,162 @@ 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, + }) + } else { + None + }; + + let mut batch = std::mem::take(&mut deferred); + match first { + None => {} + Some(Ok(evt)) => batch.push(evt), + Some(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; + } + // Carry any withheld events into the next round rather + // than dropping them with this batch. + deferred = batch; + continue; + } + Some(Err(broadcast::error::RecvError::Closed)) => { + let _ = tx.send(Err(Status::cancelled("stream closed"))).await; + return; + } + } + + // Read the publication watermark before draining. Publishers + // assign a sequence and push it onto their broadcast channel + // under one allocator lock, so every event at or below the + // sequence observed here has already reached its channel and + // the drain below is guaranteed to see it. Reading after the + // drain would admit a publish into the gap and defeat this. + // + // `None` means the cursor space is gone (teardown). Nothing + // further can be published into it, so nothing is withheld. + let watermark = state + .tracing_log_bus + .cursor_space(&sandbox_id) + .map_or(u64::MAX, |space| space.highest_seq); + + // Drain what is already queued on both sources. Sorting the + // batch restores cursor order across the two sources without + // waiting on either one. + 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(|cursored| cursored.seq); + + // Withhold anything above the watermark. Such an event was + // published after the drain started, so a lower-cursor event + // from the other source may still be queued behind it. Emitting + // it now would let the client checkpoint above an event it has + // not seen, and resume past it after a disconnect. The next + // round re-reads the watermark, which by then covers these. + let held_from = batch.partition_point(|cursored| cursored.seq <= watermark); + deferred = batch.split_off(held_from); + + // Announce the gap before any event from this batch. A warning + // carries no cursor and is never replayed, so emitting it after + // the events lets a disconnect at the wrong moment strand the + // client past the gap: it would resume from a cursor above the + // skipped events having never learned they were dropped. + if lagged > 0 + && tx + .send(Ok(crate::sandbox_watch::lag_warning_event(lagged))) + .await + .is_err() + { + return; + } + + for cursored in batch { + // Skip events the tail/replay phase already handled, judged + // against the mark for this event's own source. Bus events + // always carry seq >= 1, so no sentinel is needed here: + // non-resumable events never reach this batch. + let is_log = matches!( + cursored.event.payload, + Some(openshell_core::proto::sandbox_stream_event::Payload::Log(_)) + ); + let cutoff = if is_log { log_cutoff } else { platform_cutoff }; + if cursored.seq <= cutoff { + continue; + } + if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log( + ref log, + )) = cursored.event.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(cursored.event)).await.is_err() { + return; + } + } + + if closed { + let _ = tx.send(Err(Status::cancelled("stream closed"))).await; + return; + } } }, request_span, @@ -2882,7 +3220,7 @@ async fn stream_exec_over_relay( )), })) .await; - let _ = proxy_task.await; + finish_interactive_exec_proxy(proxy_task).await; return Ok(()); } } else { @@ -2892,12 +3230,12 @@ async fn stream_exec_over_relay( let exit_code = match exec_result { Ok(code) => code, Err(status) => { - let _ = proxy_task.await; + finish_interactive_exec_proxy(proxy_task).await; return Err(status); } }; - let _ = proxy_task.await; + finish_interactive_exec_proxy(proxy_task).await; let _ = tx .send(Ok(ExecSandboxEvent { @@ -3960,6 +4298,976 @@ 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(), + event_time: openshell_core::time::timestamp_from_millis(i as i64).ok(), + 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 { + event_time: openshell_core::time::timestamp_from_millis(0).ok(), + source: "test".to_string(), + r#type: "Normal".to_string(), + reason: reason.to_string(), + message: reason.to_string(), + metadata: HashMap::new(), + }, + )), + cursor: String::new(), + }, + ); + } + + /// Build the token a client would hold for `seq` in this sandbox's *current* + /// cursor space. Capture it before any teardown to model a real reconnect. + fn cursor_token(state: &ServerState, sandbox_id: &str, seq: u64) -> String { + let space = state + .tracing_log_bus + .cursor_space(sandbox_id) + .expect("cursor space exists; publish before taking a token"); + WatchCursor::new(space.epoch, seq).encode() + } + + /// A well-formed token from an epoch this server never issued. + fn foreign_cursor(seq: u64) -> String { + WatchCursor::new(uuid::Uuid::new_v4(), seq).encode() + } + + /// Sequence number carried by a delivered event. Panics on non-resumable + /// events, so a test that expects a log line cannot silently pass on a + /// snapshot. + fn seq_of(evt: &SandboxStreamEvent) -> u64 { + WatchCursor::parse(&evt.cursor) + .expect("resumable event must carry a valid cursor") + .seq + } + + #[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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: cursor_token(&state, &id, 1), + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + // Snapshot first (status re-read, no cursor). + let snap = stream.next().await.unwrap().unwrap(); + assert!( + snap.cursor.is_empty(), + "first event should be the status snapshot" + ); + + // Then only seqs 2 and 3; seq 1 already seen by the client. + let a = stream.next().await.unwrap().unwrap(); + let b = stream.next().await.unwrap().unwrap(); + assert_eq!(seq_of(&a), 2); + assert_eq!(seq_of(&b), 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(), + event_time: openshell_core::time::timestamp_from_millis(3).ok(), + 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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + resume_after_cursor: cursor_token(&state, &id, 1), + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + // Merged from both buses, ascending by shared seq: 2,3,4. + let mut got = Vec::new(); + for _ in 0..3 { + got.push(seq_of(&stream.next().await.unwrap().unwrap())); + } + assert_eq!(got, vec![2, 3, 4]); + } + + /// The initial tail is the resume path's twin: it draws from the same two + /// buses over the same shared cursor space, so it owes the client the same + /// ascending order. + /// + /// Before the merge, each bus was drained in its own pass and the tail came + /// out grouped by source -- logs 1,3 then platform 2,4 -- so a client + /// tracking the highest cursor saw it go backwards mid-tail. + #[tokio::test] + async fn initial_tail_merges_log_and_platform_events_in_cursor_order() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("tailmerged", 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(), + event_time: openshell_core::time::timestamp_from_millis(3).ok(), + 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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + // event_tail has no default; 0 would replay no platform events. + event_tail: 10, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let mut got = Vec::new(); + for _ in 0..4 { + got.push(seq_of(&stream.next().await.unwrap().unwrap())); + } + assert_eq!(got, vec![1, 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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + // The snapshot itself only proves both subscriptions are live -- the + // replay reads come after it. But on this current-thread runtime the + // producer cannot yield between the two, so by the time the test task + // is scheduled again the producer has run through to the live loop. + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + // 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(seq_of(&stream.next().await.unwrap().unwrap())); + } + assert_eq!(got, (1..=10).collect::>()); + } + + /// A reconnect resumes from the highest cursor the client saw, so live + /// delivery may never emit a cursor while a lower one is still undelivered. + /// The producer drains the log receiver before the platform one: a log line + /// published after that first drain but before the platform drain finishes + /// misses the batch, and its seq is below platform cursors the same batch + /// carries. Emitting them strands the log line -- a disconnect there resumes + /// above it, and no replay ever returns it. The publication watermark holds + /// back anything the drain cannot prove it saw in full. + /// + /// Reaching that interleaving takes a wide drain: the producer is first + /// parked on a full stream channel so a platform backlog accumulates, then + /// both sources are hammered from other worker threads while it walks that + /// backlog. Serialized against the test task the window does not exist -- + /// the drain holds no await a single-threaded test could wedge open. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn live_delivery_never_emits_a_cursor_above_an_undrained_event() { + use tokio_stream::StreamExt as _; + + /// Enough to overrun the 256-slot stream channel and park the producer. + const PARK: usize = 400; + /// Platform events queued while it is parked. The next drain walks all + /// of them, and that walk is the window the hammers below publish into. + /// Kept under the 1024-slot broadcast capacity so nothing is dropped. + const BACKLOG: usize = 600; + const HAMMER: usize = 300; + /// Hitting the window is a race, so repeat it. One round reproduced the + /// unfixed behavior in two runs of three; four caught it in ten of ten. + const ROUNDS: u64 = 4; + /// The single line published to confirm the producer finished + /// initialization before the rounds below start publishing. + const HANDSHAKE: u64 = 1; + const PER_ROUND: u64 = (PARK + BACKLOG + HAMMER * 2) as u64; + const TOTAL: u64 = PER_ROUND * ROUNDS + HANDSHAKE; + + let state = test_server_state().await; + let sandbox = test_sandbox("watermark", 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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + ..Default::default() + }), + ) + .await + .unwrap(); + let mut stream = response.into_inner(); + // The snapshot only proves both subscriptions are live; the producer + // sends it before reading either replay window. Publishing the rest of + // this test against that state is unsound: the log tail is capped at + // 200 by default, so a burst landing before the read is truncated, the + // rest is suppressed as already replayed, and the test stalls. + // + // One line is the handshake. Receiving it with a cursor -- replayed or + // live, either way -- proves the producer is past both tail reads, and + // one line cannot overflow any tail. Everything below is published into + // a producer known to be in the live loop. + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + seed_log_lines(&state, &id, 1); + let handshake = stream.next().await.unwrap().unwrap(); + assert_eq!( + seq_of(&handshake), + HANDSHAKE, + "handshake line should be seq 1" + ); + + let mut highest = HANDSHAKE; + let mut seen = HANDSHAKE; + let mut last_cursor = handshake.cursor; + + for round in 0..ROUNDS { + // Nothing reads during this round's setup, so the producer fills + // the stream channel and blocks part-way through this batch. + seed_log_lines(&state, &id, PARK); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + for i in 0..BACKLOG { + seed_platform_event(&state, &id, &format!("b{round}-{i}")); + } + + let log_hammer = { + let state = Arc::clone(&state); + let id = id.clone(); + tokio::spawn(async move { + for _ in 0..HAMMER { + seed_log_lines(&state, &id, 1); + } + }) + }; + let platform_hammer = { + let state = Arc::clone(&state); + let id = id.clone(); + tokio::spawn(async move { + for i in 0..HAMMER { + seed_platform_event(&state, &id, &format!("h{round}-{i}")); + } + }) + }; + + while seen < PER_ROUND * (round + 1) + HANDSHAKE { + let evt = tokio::time::timeout(std::time::Duration::from_secs(30), stream.next()) + .await + .unwrap_or_else(|_| { + panic!( + "live delivery stalled after {seen}/{TOTAL} events, highest {highest}" + ) + }) + .unwrap() + .unwrap(); + let seq = seq_of(&evt); + // Ascending delivery is what makes a highest-cursor resume safe. + // Without the watermark this trips: a log line published during + // the platform walk follows platform cursors emitted above it. + assert!(seq > highest, "cursor {seq} emitted after {highest}"); + highest = seq; + last_cursor = evt.cursor; + seen += 1; + } + + log_hammer.await.unwrap(); + platform_hammer.await.unwrap(); + } + + // Every seq in the space belongs to one of the two buses, so `TOTAL` + // ascending events ending at `TOTAL` means none was skipped. + assert_eq!(highest, TOTAL, "live delivery skipped a cursor"); + + // Reconnecting from that cursor is consistent with what was delivered: + // the replay resumes at the next event rather than past one. + seed_log_lines(&state, &id, 1); + seed_platform_event(&state, &id, "after-resume"); + drop(stream); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + resume_after_cursor: last_cursor, + ..Default::default() + }), + ) + .await + .unwrap(); + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let a = stream.next().await.unwrap().unwrap(); + let b = stream.next().await.unwrap().unwrap(); + assert_eq!(seq_of(&a), TOTAL + 1); + assert_eq!(seq_of(&b), TOTAL + 2); + } + + /// A lag warning carries no cursor, so it is never replayed on resume. Sent + /// after the surviving events of its own batch, a disconnect in between + /// leaves the client checkpointed past the dropped range having never been + /// told anything was dropped. The warning must lead the batch. + /// + /// Which branch `select!` takes is arbitrary, so the run is repeated: the + /// assertion holds on both paths, but only the platform-first path reaches + /// the drain's lag counter, which is where the ordering used to be wrong. + /// Eight attempts caught the unfixed ordering in three runs of five; forty + /// caught it in six of six. + #[tokio::test] + async fn lag_warning_precedes_the_events_of_its_batch() { + use openshell_core::proto::sandbox_stream_event::Payload; + use tokio_stream::StreamExt as _; + + /// `select!` polls the log receiver before the platform one in three of + /// its four rotations, and only the platform-first path reaches the + /// drain's lag counter. Repeat enough that missing it is negligible. + const ATTEMPTS: usize = 40; + + let state = test_server_state().await; + + for attempt in 0..ATTEMPTS { + let sandbox = test_sandbox(&format!("lagorder{attempt}"), 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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + ..Default::default() + }), + ) + .await + .unwrap(); + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + // One platform event plus a log burst past the 1024-slot broadcast + // capacity, published without an await in between so the producer + // sees both on a single wake-up: a deliverable event and a drop. + seed_platform_event(&state, &id, "e1"); + seed_log_lines(&state, &id, 1100); + + let first = stream.next().await.unwrap().unwrap(); + assert!( + matches!(first.payload, Some(Payload::Warning(_))), + "the lag warning must arrive before any event of the lagged batch, got {:?}", + first.payload + ); + } + } + + #[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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: cursor_token(&state, &id, 3), + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + // 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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + // Seq 2 was trimmed; this is an unrecoverable gap. + resume_after_cursor: cursor_token(&state, &id, 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!(snap.cursor.is_empty()); + + 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()); + } + + /// The guard behind the producer's post-replay epoch re-check. + /// + /// Validation and the two `tail_after` reads take their locks separately, so + /// a teardown plus a republish can swap the space in between and leave the + /// reads applying an old seq to a replacement's buffers -- `tail_after` only + /// compares numbers, so it reports no gap while skipping every replacement + /// event at or below that seq. The producer re-checks the epoch once both + /// tails are in hand and before emitting anything; this pins what that check + /// must answer. + /// + /// The interleaving itself is not reachable from a test: the producer runs + /// validation, both reads, and the re-check with no await in between, so + /// there is nothing to suspend it on. + #[tokio::test] + async fn cursor_space_is_rejects_a_replacement_space() { + let state = test_server_state().await; + let sandbox = test_sandbox("respace", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 1); + let original = state.tracing_log_bus.cursor_space(&id).unwrap().epoch; + assert!(cursor_space_is(&state, &id, original)); + + // Teardown alone leaves no space to point into. + state.tracing_log_bus.remove(&id); + assert!(state.tracing_log_bus.cursor_space(&id).is_none()); + assert!(!cursor_space_is(&state, &id, original)); + + // The republish installs a replacement renumbered from 1. Its seqs + // overlap the retired space's, so only the epoch separates them. + seed_log_lines(&state, &id, 1); + let replacement = state.tracing_log_bus.cursor_space(&id).unwrap(); + assert_ne!(replacement.epoch, original); + assert_eq!(replacement.highest_seq, 1); + assert!(!cursor_space_is(&state, &id, original)); + assert!(cursor_space_is(&state, &id, replacement.epoch)); + } + + #[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(); + + // The reported repro. Before cursors carried an epoch, a bare number + // was all the server had: 2 <= 3 passed the old "is this plausible?" + // bound, the tail replayed only seq 3, and the new space's seqs 1 and 2 + // -- real, unseen events -- were silently swallowed as duplicates. + seed_log_lines(&state, &id, 2); + let retired_cursor = cursor_token(&state, &id, 2); + + // Teardown retires the space; the next publish starts over at seq 1, + // indistinguishable by number from the client's view of a restart. + state.tracing_log_bus.remove(&id); + seed_log_lines(&state, &id, 3); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: retired_cursor, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let item = stream + .next() + .await + .unwrap() + .expect_err("a cursor from a retired space must terminate the stream"); + assert_eq!(item.code(), tonic::Code::OutOfRange, "{item:?}"); + assert!( + item.message().contains("empty resume_after_cursor"), + "status must tell the client to restart without a cursor, not retry: {}", + item.message() + ); + + // Nothing from the new space may be delivered before the error: a + // partial stream would read as "here is everything after your cursor". + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn resume_from_reset_cursor_space_rejected_when_new_space_is_shorter() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("reset-short", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // The shape the old numeric bound did catch (5 > 2), kept so it keeps + // passing -- but now it fails for the reason that generalizes. + seed_log_lines(&state, &id, 5); + let retired_cursor = cursor_token(&state, &id, 5); + state.tracing_log_bus.remove(&id); + seed_log_lines(&state, &id, 2); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: retired_cursor, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + assert!(stream.next().await.unwrap().unwrap().cursor.is_empty()); + let err = stream.next().await.unwrap().expect_err("out of range"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn resume_from_reset_cursor_space_rejected_when_seq_is_within_new_range() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("reset-within", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // Seq 1 sits comfortably inside the new space's 1..=3, so every numeric + // bound accepts it. Only the epoch distinguishes the two spaces. This is + // the assertion the pre-fix design structurally could not make. + seed_log_lines(&state, &id, 3); + let retired_cursor = cursor_token(&state, &id, 1); + state.tracing_log_bus.remove(&id); + seed_log_lines(&state, &id, 3); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: retired_cursor, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + assert!(stream.next().await.unwrap().unwrap().cursor.is_empty()); + let err = stream.next().await.unwrap().expect_err("out of range"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn resume_after_remove_without_republish_terminates_out_of_range() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("reset-empty", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + // No space at all: nothing has been published since teardown. The buses + // look merely empty, so `tail_after` reports no gap -- "caught up" would + // be the wrong reading, because the client's events are gone. + seed_log_lines(&state, &id, 3); + let retired_cursor = cursor_token(&state, &id, 3); + state.tracing_log_bus.remove(&id); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: retired_cursor, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + assert!(stream.next().await.unwrap().unwrap().cursor.is_empty()); + let err = stream.next().await.unwrap().expect_err("out of range"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn resume_with_cursor_from_another_sandbox_terminates_out_of_range() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let a = test_sandbox("epoch-a", Vec::new()); + let b = test_sandbox("epoch-b", Vec::new()); + state.store.put_message(&a).await.unwrap(); + state.store.put_message(&b).await.unwrap(); + let a_id = a.object_id().to_string(); + let b_id = b.object_id().to_string(); + + // Epochs are per sandbox, not per process. A token valid for A must not + // address B's space, even though both are on seq 1..=3 right now. + seed_log_lines(&state, &a_id, 3); + seed_log_lines(&state, &b_id, 3); + let a_cursor = cursor_token(&state, &a_id, 1); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: b.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: a_cursor, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + assert!(stream.next().await.unwrap().unwrap().cursor.is_empty()); + let err = stream.next().await.unwrap().expect_err("out of range"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn resume_with_cursor_ahead_of_the_space_terminates_out_of_range() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("ahead", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 2); + // Right epoch, but a seq this space has never issued: only a fabricated + // token gets here. Accepting it would pin the cutoff above every future + // event and stall the stream silently. + let ahead = cursor_token(&state, &id, 99); + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: ahead, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + assert!(stream.next().await.unwrap().unwrap().cursor.is_empty()); + let err = stream.next().await.unwrap().expect_err("out of range"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn resume_with_malformed_cursor_rejects_invalid_argument() { + let state = test_server_state().await; + let sandbox = test_sandbox("malformed", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 3); + + // Input validation, so it fails the RPC before any stream exists rather + // than arriving as the first item of an apparently-healthy stream. + for raw in ["5", "v1:not-a-uuid:0", &foreign_cursor(1)[..40]] { + let err = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: raw.to_string(), + ..Default::default() + }), + ) + .await + .expect_err("malformed cursor must fail the call"); + assert_eq!(err.code(), tonic::Code::InvalidArgument, "{raw:?}: {err:?}"); + assert!( + !err.message().contains(raw), + "status must not echo the client token: {}", + err.message() + ); + } + } + + #[tokio::test] + async fn resume_with_foreign_epoch_terminates_out_of_range() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("foreign", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 3); + + // Well-formed, so it clears input validation, but from an epoch this + // gateway never minted -- the shape a reconnect to another replica + // takes. + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + resume_after_cursor: foreign_cursor(1), + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + assert!(stream.next().await.unwrap().unwrap().cursor.is_empty()); + let err = stream.next().await.unwrap().expect_err("out of range"); + assert_eq!(err.code(), tonic::Code::OutOfRange, "{err:?}"); + 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 { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + 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(), + event_time: openshell_core::time::timestamp_from_millis(i64::from(i)).ok(), + 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.is_empty() { + cursors.push(seq_of(&evt)); + } + } + + // 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/lib.rs b/crates/openshell-server/src/lib.rs index ee3ef1768e..c0a12facc7 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -48,6 +48,7 @@ mod tls; pub(crate) mod tls_test_utils; pub mod tracing_bus; mod tracing_setup; +mod watch_cursor; mod ws_tunnel; use metrics_exporter_prometheus::PrometheusBuilder; diff --git a/crates/openshell-server/src/sandbox_watch.rs b/crates/openshell-server/src/sandbox_watch.rs index 3a62fce398..679912d352 100644 --- a/crates/openshell-server/src/sandbox_watch.rs +++ b/crates/openshell-server/src/sandbox_watch.rs @@ -7,8 +7,8 @@ use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex}; use std::time::Duration; +use openshell_core::proto::SandboxStreamWarning; use tokio::sync::{broadcast, watch}; -use tonic::Status; use crate::persistence::Store; use openshell_core::proto::Sandbox; @@ -34,6 +34,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 @@ -132,13 +133,24 @@ pub fn spawn_store_poller( }); } -/// 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: String::new(), } } @@ -189,6 +201,49 @@ mod tests { 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"); + } + #[tokio::test] async fn shared_store_poller_notifies_remote_resource_version_change() { let store = Arc::new(crate::persistence::test_store().await); diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index 1b40c62bb1..77fe7937a0 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -118,8 +118,15 @@ mod tests { const STORAGE_V1_SCHEMA_SHA256: &str = "d68401809d8cea445c35233ef32412bbd041cb2ac5acaf368a0d0bf74d2ddf17"; + // Carries both this branch's opaque watch cursor and the well-known time + // types from main. The cursor change adds `WatchSandboxRequest` + // `.resume_after_cursor` and `SandboxStreamEvent.cursor` as strings; both + // fields are unreleased, added on this branch, so no client depends on an + // earlier type. It adds no messages or enums and touches no stored type, + // which is why the durable and overlap fingerprints below are main's + // values unchanged. const PUBLIC_RPC_SCHEMA_SHA256: &str = - "07e889beb43942535d80d635c490f3fb6c6f3e554f3087c101deec8c05502883"; + "fd8a5cad432441be1fe8891332b56f52335c208cc7283dcd93e385e7ae66c6da"; const DURABLE_SCHEMA_SHA256: &str = "9eeaa29dfba187bff69fb7bc4f9a13a0f1d7be3f7049a38c8f0e20ce77ec7d8b"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 91db86c275..fea7abad5c 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; @@ -12,18 +12,174 @@ use tokio::sync::broadcast; use tracing::{Event, Subscriber}; use tracing_subscriber::Layer; use tracing_subscriber::layer::Context; +use uuid::Uuid; + +use crate::watch_cursor::WatchCursor; /// Bus that publishes server log lines keyed by sandbox id. #[derive(Debug, Clone)] 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, +} + +/// A buffered or broadcast stream event paired with its raw sequence number. +/// +/// The wire `SandboxStreamEvent.cursor` is an opaque token; ordering decisions +/// must not depend on its encoding. Carrying the seq alongside keeps every +/// comparison on the watch path numeric, so no consumer parses the token to +/// decide whether an event has already been delivered. +#[derive(Debug, Clone)] +pub(crate) struct CursoredEvent { + pub(crate) seq: u64, + pub(crate) event: SandboxStreamEvent, +} + +#[derive(Debug, Clone)] +struct PerSandbox { + sender: broadcast::Sender, + tail: VecDeque, + /// Highest seq this bus has evicted from `tail`. 0 = nothing trimmed. + /// + /// Under the shared cursor space each bus's tail is non-contiguous in the + /// 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, +} + +/// One sandbox's cursor space: an identity plus its running sequence. +/// +/// The epoch is what makes a cursor verifiable. Sequence numbers restart at 1 +/// whenever a space is recreated, so a bare number from a previous space can +/// look like a valid position in the current one. Minting a fresh epoch with +/// the entry means every reset produces a distinguishable space, and a resume +/// cursor can be checked against the identity that issued it rather than +/// against a plausible range. +#[derive(Debug, Clone)] +struct CursorSpace { + epoch: Uuid, + next: u64, +} + +/// Identity and extent of a sandbox's current cursor space. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct CursorSpaceInfo { + pub(crate) epoch: Uuid, + /// Highest sequence issued so far. `0` means nothing published yet. + pub(crate) highest_seq: 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 `(epoch, seq)` for this sandbox from a locked space. + /// + /// Seq starts at 1 so an empty `resume_after_cursor` means "from the + /// beginning" without skipping event 1. + /// + /// The epoch is minted on the vacant path only — once per cursor space, not + /// once per publish — so generating it under the allocator lock costs + /// nothing on the hot path. + fn next_locked(spaces: &mut HashMap, sandbox_id: &str) -> (Uuid, u64) { + let space = spaces + .entry(sandbox_id.to_string()) + .or_insert_with(|| CursorSpace { + epoch: Uuid::new_v4(), + next: 1, + }); + let seq = space.next; + space.next += 1; + (space.epoch, seq) + } + + /// Identity and extent of this sandbox's space, or `None` when no event has + /// been published into it. + /// + /// A resume cursor is validated against this: a different epoch means the + /// cursor belongs to a space that no longer exists, and `None` means there + /// is no space to resume into at all. An empty `tail_after` result is not + /// on its own proof that a cursor is still valid. + fn space(&self, sandbox_id: &str) -> Option { + self.lock().get(sandbox_id).map(|space| CursorSpaceInfo { + epoch: space.epoch, + highest_seq: space.next.saturating_sub(1), + }) + } +} + +fn tail_after_impl( + tail: &VecDeque, + 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(|cursored| cursored.seq > after_seq) + .cloned() + .collect(); + + Ok(res) } impl Default for TracingLogBus { @@ -35,12 +191,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, } } @@ -51,42 +210,80 @@ impl TracingLogBus { } } - fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { + fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); 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() } - pub fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { + pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { 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. + /// + /// 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. /// - /// This drops the broadcast sender (closing any active receivers with - /// `RecvError::Closed`) and frees the tail buffer. + /// Dropping the allocator entry retires the epoch with it, so the next + /// publish mints a new one. Cursors handed out before this call are + /// therefore rejected on resume rather than mistaken for positions in the + /// replacement space. Both broadcast senders are dropped here too, so a + /// stream that was live across the reset ends rather than silently + /// continuing into a different space. 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 spaces = 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 `spaces`. + self.platform_event_bus.remove(sandbox_id); + spaces.remove(sandbox_id); } - pub fn tail(&self, sandbox_id: &str, max: usize) -> Vec { + pub(crate) 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).cloned().collect::>()) .unwrap_or_default() .into_iter() .rev() - .collect() + .collect::>() + } + + /// Identity and extent of this sandbox's current cursor space. + /// + /// `None` means nothing has been published for the sandbox, so there is no + /// space to resume into. Takes only the allocator lock, never the bus map, + /// so it cannot invert the documented allocator -> bus map order. + pub(crate) fn cursor_space(&self, sandbox_id: &str) -> Option { + self.seq.space(sandbox_id) + } + + pub(crate) 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 +296,9 @@ impl TracingLogBus { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( log.clone(), )), + // Placeholder: publish() stamps the real cursor from the sandbox + // cursor space. + cursor: String::new(), }; self.publish(&log.sandbox_id, evt, Self::DEFAULT_TAIL); } @@ -106,15 +306,27 @@ 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 spaces = self.seq.lock(); + let (epoch, seq) = SeqAllocator::next_locked(&mut spaces, sandbox_id); + event.cursor = WatchCursor::new(epoch, seq).encode(); 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 cursored = CursoredEvent { seq, event }; + let _ = per.sender.send(cursored.clone()); + per.tail.push_back(cursored); + while per.tail.len() > tail_cap { + if let Some(trimmed) = per.tail.pop_front() { + per.last_trimmed_seq = trimmed.seq; + } } } } @@ -155,6 +367,9 @@ where payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( log, )), + // Placeholder: publish() stamps the real cursor from the sandbox + // cursor space. + cursor: String::new(), }; self.bus.publish(&sandbox_id, evt, self.default_tail); } @@ -208,6 +423,163 @@ mod tests { } } + /// Fixed epoch for hand-built events, so encoded cursors are deterministic. + fn test_epoch() -> Uuid { + Uuid::parse_str("11111111-1111-4111-8111-111111111111").expect("valid uuid") + } + + /// 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: WatchCursor::new(test_epoch(), seq).encode(), + } + } + + /// Build a buffered event stamped at `seq`. + fn cursored(seq: u64) -> CursoredEvent { + CursoredEvent { + seq, + event: stream_event(seq), + } + } + + /// Build a contiguous tail with seqs `lo..=hi`. + fn tail_of(lo: u64, hi: u64) -> VecDeque { + (lo..=hi).map(cursored).collect() + } + + /// Extract seqs from a run of buffered events, in order. + fn cursors(events: &[CursoredEvent]) -> Vec { + events.iter().map(|c| c.seq).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!(tail_after_impl(&tail, 0, 0).unwrap().is_empty()); + assert!(tail_after_impl(&tail, 0, 42).unwrap().is_empty()); + } + + #[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!(tail_after_impl(&tail, 0, 5).expect("ok").is_empty()); + } + + #[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!(tail_after_impl(&tail, 0, 99).expect("ok").is_empty()); + } + + #[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 = [cursored(2), cursored(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!(bus.tail_after("nope", 5).unwrap().is_empty()); + } + + #[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 +599,132 @@ mod tests { assert!(bus.tail(sandbox_id, 10).is_empty()); } + #[test] + fn cursor_space_is_absent_until_first_publish() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-epoch-lazy"; + + assert_eq!(bus.cursor_space(sandbox_id), None); + + // Subscribing must not mint a space. `sender_for` touches only the bus + // map, never the allocator -- and the watch handler subscribes before + // it validates the resume cursor. If a subscription could manufacture + // an epoch, a client resuming against a torn-down sandbox would create + // the very space its stale cursor is then checked against. + let _log_rx = bus.subscribe(sandbox_id); + let _platform_rx = bus.platform_event_bus.subscribe(sandbox_id); + assert_eq!(bus.cursor_space(sandbox_id), None); + + bus.publish_external(make_log_event(sandbox_id, "first")); + let space = bus.cursor_space(sandbox_id).expect("space after publish"); + assert_eq!(space.highest_seq, 1); + } + + #[test] + fn publish_after_remove_starts_a_new_epoch() { + // The defect this whole mechanism exists for: teardown restarts seqs at + // 1, so a cursor from before the reset is numerically indistinguishable + // from a position in the new space. A fresh epoch makes it distinct. + let bus = TracingLogBus::new(); + let sandbox_id = "sb-epoch-reset"; + + bus.publish_external(make_log_event(sandbox_id, "a")); + bus.publish_external(make_log_event(sandbox_id, "b")); + let before = bus.cursor_space(sandbox_id).expect("space exists"); + assert_eq!(before.highest_seq, 2); + + bus.remove(sandbox_id); + assert_eq!(bus.cursor_space(sandbox_id), None); + + bus.publish_external(make_log_event(sandbox_id, "c")); + let after = bus.cursor_space(sandbox_id).expect("space recreated"); + + assert_ne!(before.epoch, after.epoch, "teardown must retire the epoch"); + // Seq alone cannot tell the spaces apart -- that is the point. + assert_eq!(after.highest_seq, 1); + } + + #[test] + fn log_and_platform_publishes_share_one_epoch() { + // Both buses draw from one allocator, so a cursor observed on either + // resumes both. Separate epochs would make a merged resume impossible. + let bus = TracingLogBus::new(); + let sandbox_id = "sb-epoch-shared"; + + bus.publish_external(make_log_event(sandbox_id, "a")); + let after_log = bus.cursor_space(sandbox_id).expect("space exists"); + + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); + let after_platform = bus.cursor_space(sandbox_id).expect("space exists"); + + assert_eq!(after_log.epoch, after_platform.epoch); + assert_eq!(after_platform.highest_seq, 2); + + let log_cursor = &bus.tail(sandbox_id, 10)[0].event.cursor; + let platform_cursor = &bus.platform_event_bus.tail(sandbox_id, 10)[0].event.cursor; + assert_eq!( + WatchCursor::parse(log_cursor).expect("valid").epoch, + WatchCursor::parse(platform_cursor).expect("valid").epoch, + ); + } + + #[test] + fn concurrent_publish_and_remove_keeps_cursors_in_one_ascending_space() { + // Teardown retires the cursor space 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, or worse, one that mixes + // two epochs. + // + // Both halves matter. Resume validation trusts that every cursor in a + // tail belongs to the epoch `cursor_space` reports, so a mixed tail + // would let a cursor pass the epoch check and still address the wrong + // events. + 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 observed: Vec = bus + .tail(sandbox_id, usize::MAX) + .iter() + .map(|c| WatchCursor::parse(&c.event.cursor).expect("bus stamps valid cursors")) + .collect(); + assert!( + observed + .windows(2) + .all(|w| w[0].epoch == w[1].epoch && w[0].seq < w[1].seq), + "tail must stay in one epoch with strictly ascending seqs, got {observed:?}" + ); + } + + 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(); @@ -243,7 +741,7 @@ mod tests { // New publish should reach the new subscriber bus.publish_external(make_log_event(sandbox_id, "new message")); let evt = rx.try_recv().expect("should receive new event"); - assert!(evt.payload.is_some()); + assert!(evt.event.payload.is_some()); } #[test] @@ -278,13 +776,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: String::new(), + }; bus.publish(sandbox_id, evt); assert!(rx.try_recv().is_ok()); @@ -300,7 +801,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 +809,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: String::new(), + }; 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 +828,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 +842,7 @@ mod tests { message: format!("Message {i}"), metadata: HashMap::new(), })), + cursor: String::new(), }; bus.publish(sandbox_id, evt); } @@ -347,8 +852,8 @@ mod tests { assert_eq!(events.len(), 5); // Verify order (oldest first) - for (i, evt) in events.iter().enumerate() { - if let Some(sandbox_stream_event::Payload::Event(ref e)) = evt.payload { + for (i, cursored) in events.iter().enumerate() { + if let Some(sandbox_stream_event::Payload::Event(ref e)) = cursored.event.payload { assert_eq!(e.reason, format!("Event{i}")); } else { panic!("expected Event payload"); @@ -358,27 +863,30 @@ mod tests { // Tail with smaller max should return most recent events let events = bus.tail(sandbox_id, 2); assert_eq!(events.len(), 2); - if let Some(sandbox_stream_event::Payload::Event(ref e)) = events[0].payload { + if let Some(sandbox_stream_event::Payload::Event(ref e)) = events[0].event.payload { assert_eq!(e.reason, "Event3"); } - if let Some(sandbox_stream_event::Payload::Event(ref e)) = events[1].payload { + if let Some(sandbox_stream_event::Payload::Event(ref e)) = events[1].event.payload { assert_eq!(e.reason, "Event4"); } } #[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: String::new(), + }; bus.publish(sandbox_id, evt); assert_eq!(bus.tail(sandbox_id, 10).len(), 1); @@ -392,13 +900,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,63 +909,86 @@ 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 { + 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() } - pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { + pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { 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 spaces = self.seq.lock(); + let (epoch, seq) = SeqAllocator::next_locked(&mut spaces, sandbox_id); + event.cursor = WatchCursor::new(epoch, seq).encode(); 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 cursored = CursoredEvent { seq, event }; + let _ = per.sender.send(cursored.clone()); + per.tail.push_back(cursored); + while per.tail.len() > Self::DEFAULT_TAIL { + if let Some(trimmed) = per.tail.pop_front() { + per.last_trimmed_seq = trimmed.seq; + } } } /// Return buffered platform events for replay to late subscribers. - pub(crate) fn tail(&self, sandbox_id: &str, max: usize) -> Vec { + 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).cloned().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/crates/openshell-server/src/watch_cursor.rs b/crates/openshell-server/src/watch_cursor.rs new file mode 100644 index 0000000000..894b52aa18 --- /dev/null +++ b/crates/openshell-server/src/watch_cursor.rs @@ -0,0 +1,206 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Opaque, epoch-bound cursors for the `WatchSandbox` stream. +//! +//! A cursor is only meaningful inside the cursor space that issued it. A +//! gateway restart, or teardown of a sandbox's buses, starts a new space whose +//! sequence numbers restart at 1, so a bare number cannot tell "caught up" +//! apart from "belongs to a space that no longer exists". Binding the sequence +//! to the space's epoch makes that distinction explicit: a cursor from a dead +//! space is rejected instead of silently suppressing live events beneath it. +//! +//! # Wire contract +//! +//! `v1::<20-digit zero-padded seq>` +//! +//! The token is **opaque to clients**. The only operation a client may perform +//! is comparing two cursors from the same stream and keeping the greater one as +//! its resume point. That comparison is byte-wise: the version prefix and uuid +//! are fixed width, and the sequence is zero-padded, so lexicographic order +//! equals sequence order within one epoch. Every stream observes exactly one +//! epoch (a reset closes the stream), so the comparison is always well defined +//! where clients are allowed to use it. +//! +//! The server does **not** rely on that property. Ordering decisions on the +//! watch path run on the raw `u64` sequence carried alongside each event in +//! [`crate::tracing_bus::CursoredEvent`], so no server-side correctness +//! decision depends on the encoding. + +use std::fmt; + +use uuid::Uuid; + +/// Current cursor encoding version. +const VERSION: &str = "v1"; + +/// Zero-padded width of the sequence segment. `u64::MAX` is 20 digits. +const SEQ_WIDTH: usize = 20; + +/// Exact encoded length: `"v1"` + `':'` + hyphenated uuid + `':'` + seq. +const ENCODED_LEN: usize = VERSION.len() + 1 + 36 + 1 + SEQ_WIDTH; + +/// A resume cursor: a sequence number bound to the cursor space that issued it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct WatchCursor { + pub(crate) epoch: Uuid, + pub(crate) seq: u64, +} + +/// The client supplied a cursor this server could not have issued. +/// +/// Deliberately carries no detail from the input: the token is echoed back to +/// nobody, and a single opaque reason keeps malformed input from becoming a +/// reflection vector. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CursorParseError; + +impl fmt::Display for CursorParseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("resume_after_cursor is not a valid watch cursor") + } +} + +impl std::error::Error for CursorParseError {} + +impl WatchCursor { + pub(crate) const fn new(epoch: Uuid, seq: u64) -> Self { + Self { epoch, seq } + } + + /// Encode as the opaque wire token. + pub(crate) fn encode(&self) -> String { + format!( + "{VERSION}:{}:{:0SEQ_WIDTH$}", + self.epoch.as_hyphenated(), + self.seq + ) + } + + /// Parse a client-supplied token. + /// + /// Strict by design. Anything this server would not have produced is + /// rejected, so a client cannot hand back a hand-built or truncated cursor + /// and have it silently treated as a position in the current space. + pub(crate) fn parse(raw: &str) -> Result { + // Fixed-width encoding, so one length check also bounds the work done + // on hostile input. + if raw.len() != ENCODED_LEN { + return Err(CursorParseError); + } + + let mut parts = raw.split(':'); + let (Some(version), Some(epoch), Some(seq), None) = + (parts.next(), parts.next(), parts.next(), parts.next()) + else { + return Err(CursorParseError); + }; + + if version != VERSION { + return Err(CursorParseError); + } + + // Re-encode and compare so only the canonical lowercase hyphenated form + // is accepted. `Uuid::parse_str` also takes braced and simple forms, + // which would give one epoch several spellings and break the + // lexicographic ordering clients rely on. + let parsed_epoch = Uuid::parse_str(epoch).map_err(|_| CursorParseError)?; + if parsed_epoch.as_hyphenated().to_string() != epoch { + return Err(CursorParseError); + } + + if seq.len() != SEQ_WIDTH || !seq.bytes().all(|b| b.is_ascii_digit()) { + return Err(CursorParseError); + } + // 20 digits can exceed u64::MAX, so this also rejects overflow. + let seq: u64 = seq.parse().map_err(|_| CursorParseError)?; + + // Sequences start at 1; an empty cursor is the only "from the + // beginning" signal. + if seq == 0 { + return Err(CursorParseError); + } + + Ok(Self { + epoch: parsed_epoch, + seq, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn epoch() -> Uuid { + Uuid::parse_str("3f2a9c14-7b6e-4d81-9a02-1c5d8e4f7b30").expect("valid uuid") + } + + #[test] + fn encode_parse_roundtrip() { + for seq in [1, 2, 9, 10, 999, u64::MAX] { + let cursor = WatchCursor::new(epoch(), seq); + let encoded = cursor.encode(); + assert_eq!(encoded.len(), ENCODED_LEN); + assert_eq!(WatchCursor::parse(&encoded).expect("roundtrip"), cursor); + } + } + + #[test] + fn encoding_is_lexicographically_ordered_by_seq() { + // Clients are told they may compare two cursors from one stream + // byte-wise and keep the greater. That only holds because the sequence + // is zero-padded to a fixed width; dropping the padding would make + // "...:9" sort above "...:10" and silently rewind every reconnect. + let encoded: Vec = [1_u64, 2, 9, 10, 99, 100, u64::MAX] + .into_iter() + .map(|seq| WatchCursor::new(epoch(), seq).encode()) + .collect(); + + let mut sorted = encoded.clone(); + sorted.sort(); + assert_eq!(sorted, encoded, "lexicographic order must match seq order"); + } + + #[test] + fn parse_rejects_malformed() { + let e = epoch().as_hyphenated().to_string(); + let cases = [ + ("empty", String::new()), + ("bare number", "5".to_string()), + ("unpadded seq", format!("v1:{e}:5")), + ("wrong version", format!("v2:{e}:{:020}", 1)), + ( + "not a uuid", + format!("v1:not-a-uuid-not-a-uuid-not-a-uuid-x:{:020}", 1), + ), + ( + "uppercase uuid", + format!("v1:{}:{:020}", e.to_uppercase(), 1), + ), + ( + "simple uuid", + format!("v1:{}:{:020}", epoch().as_simple(), 1), + ), + ("four segments", format!("v1:{e}:{:020}:x", 1)), + ("zero seq", format!("v1:{e}:{:020}", 0)), + ("non-digit seq", format!("v1:{e}:0000000000000000000x")), + ("seq overflows u64", format!("v1:{e}:99999999999999999999")), + ("over-long input", "x".repeat(200)), + ]; + + for (name, raw) in cases { + assert_eq!( + WatchCursor::parse(&raw), + Err(CursorParseError), + "expected rejection for {name}" + ); + } + } + + #[test] + fn parse_error_does_not_echo_input() { + let rendered = CursorParseError.to_string(); + assert!(!rendered.contains("v1:"), "error must not echo the token"); + } +} diff --git a/docs/observability/accessing-logs.mdx b/docs/observability/accessing-logs.mdx index 4b755f74cc..e5981c9583 100644 --- a/docs/observability/accessing-logs.mdx +++ b/docs/observability/accessing-logs.mdx @@ -39,6 +39,25 @@ 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 an opaque `cursor` token. Status snapshots and warnings carry an empty cursor. + +Do not parse a cursor. The only supported operation is comparing two cursors observed on the same stream and keeping the greater one as the resume point. That comparison is a plain string comparison in any language. + +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. The warning is the signal that events were skipped. +- **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. The client should restart observation and, if it needs the missing lines, read them from the log files inside the sandbox. + +A cursor is bound to the cursor space that issued it. A gateway restart, teardown of the sandbox's buffers, or a reconnect that lands on a different gateway replica starts a new space, and cursors from the previous one no longer refer to anything. The gateway rejects them with `OUT_OF_RANGE` instead of treating them as caught up — which would silently suppress the new space's events. A malformed cursor is rejected with `INVALID_ARGUMENT`. + +`OUT_OF_RANGE` is terminal for that cursor. Restart the watch with an empty resume cursor; retrying the same one fails identically. + +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 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. + +The gateway merges the log and platform event sources before emitting, so events normally arrive in ascending cursor order. That applies to the buffered tail you receive when the stream opens, to a replay after a resume, and to live delivery. 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. Track the highest cursor seen as the resume point rather than the last one received. + ## 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 0623943f87..dd05f0fd2b 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -2010,6 +2010,25 @@ 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. Empty means no cursor resume: the + // server falls back to tail-limited replay controlled by log_tail_lines and + // event_tail. Otherwise 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). + // + // A cursor is bound to the cursor space that issued it. A gateway restart, + // teardown of the sandbox's buffers, or a reconnect to a different gateway + // replica starts a new space, and cursors from the previous one are rejected + // with OUT_OF_RANGE rather than silently suppressing live events beneath + // them. OUT_OF_RANGE is terminal for that cursor: restart the watch with an + // empty resume_after_cursor, because retrying the same token fails + // identically. A cursor this server could not have issued is rejected with + // INVALID_ARGUMENT. + string resume_after_cursor = 12; } // One event in a sandbox watch stream. @@ -2021,11 +2040,25 @@ 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 the warning itself. SandboxStreamWarning warning = 4; // Draft policy update notification. DraftPolicyUpdate draft_policy_update = 5; } + // Opaque position in this sandbox's cursor space, shared across the resumable + // log and platform event sources. Empty for non-resumable events (status + // snapshots, warnings). + // + // Do not parse this token; its encoding is not part of the contract. The only + // supported operation is comparing two non-empty cursors observed on the same + // stream and keeping the greater one, then passing it as + // WatchSandboxRequest.resume_after_cursor to resume without loss or + // duplication. That comparison is a plain byte-wise string comparison. It is + // well defined only within one stream: a stream never spans two cursor + // spaces, because a reset ends it. + string cursor = 6; } // Log line correlated to a sandbox. @@ -2044,6 +2077,11 @@ 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. Cursors are opaque, so this message is the only signal that events +// were skipped. Unrecoverable loss (a trimmed or foreign resume cursor) is +// reported as an OUT_OF_RANGE stream status, not this message. message SandboxStreamWarning { string message = 1; } diff --git a/sdk/go/proto/openshellv1/openshell.pb.go b/sdk/go/proto/openshellv1/openshell.pb.go index 8b3438eff2..89e10bd26c 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -7467,9 +7467,27 @@ type WatchSandboxRequest struct { // Filter by log source (e.g. "gateway", "sandbox"). Empty means all sources. LogSources []string `protobuf:"bytes,9,rep,name=log_sources,json=logSources,proto3" json:"log_sources,omitempty"` // Minimum log level to include (e.g. "INFO", "WARN", "ERROR"). Empty means all levels. - LogMinLevel string `protobuf:"bytes,10,opt,name=log_min_level,json=logMinLevel,proto3" json:"log_min_level,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + LogMinLevel string `protobuf:"bytes,10,opt,name=log_min_level,json=logMinLevel,proto3" json:"log_min_level,omitempty"` + // Resume streaming after this cursor. Empty means no cursor resume: the + // server falls back to tail-limited replay controlled by log_tail_lines and + // event_tail. Otherwise 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). + // + // A cursor is bound to the cursor space that issued it. A gateway restart, + // teardown of the sandbox's buffers, or a reconnect to a different gateway + // replica starts a new space, and cursors from the previous one are rejected + // with OUT_OF_RANGE rather than silently suppressing live events beneath + // them. OUT_OF_RANGE is terminal for that cursor: restart the watch with an + // empty resume_after_cursor, because retrying the same token fails + // identically. A cursor this server could not have issued is rejected with + // INVALID_ARGUMENT. + ResumeAfterCursor string `protobuf:"bytes,12,opt,name=resume_after_cursor,json=resumeAfterCursor,proto3" json:"resume_after_cursor,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *WatchSandboxRequest) Reset() { @@ -7579,6 +7597,13 @@ func (x *WatchSandboxRequest) GetLogMinLevel() string { return "" } +func (x *WatchSandboxRequest) GetResumeAfterCursor() string { + if x != nil { + return x.ResumeAfterCursor + } + return "" +} + // One event in a sandbox watch stream. type SandboxStreamEvent struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -7589,7 +7614,19 @@ type SandboxStreamEvent struct { // *SandboxStreamEvent_Event // *SandboxStreamEvent_Warning // *SandboxStreamEvent_DraftPolicyUpdate - Payload isSandboxStreamEvent_Payload `protobuf_oneof:"payload"` + Payload isSandboxStreamEvent_Payload `protobuf_oneof:"payload"` + // Opaque position in this sandbox's cursor space, shared across the resumable + // log and platform event sources. Empty for non-resumable events (status + // snapshots, warnings). + // + // Do not parse this token; its encoding is not part of the contract. The only + // supported operation is comparing two non-empty cursors observed on the same + // stream and keeping the greater one, then passing it as + // WatchSandboxRequest.resume_after_cursor to resume without loss or + // duplication. That comparison is a plain byte-wise string comparison. It is + // well defined only within one stream: a stream never spans two cursor + // spaces, because a reset ends it. + Cursor string `protobuf:"bytes,6,opt,name=cursor,proto3" json:"cursor,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -7676,6 +7713,13 @@ func (x *SandboxStreamEvent) GetDraftPolicyUpdate() *DraftPolicyUpdate { return nil } +func (x *SandboxStreamEvent) GetCursor() string { + if x != nil { + return x.Cursor + } + return "" +} + type isSandboxStreamEvent_Payload interface { isSandboxStreamEvent_Payload() } @@ -7696,7 +7740,9 @@ type SandboxStreamEvent_Event struct { } type SandboxStreamEvent_Warning struct { - // 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 the warning itself. Warning *SandboxStreamWarning `protobuf:"bytes,4,opt,name=warning,proto3,oneof"` } @@ -7811,6 +7857,11 @@ func (x *SandboxLogLine) GetFields() map[string]string { return nil } +// Recoverable loss notification on a watch stream. Emitted when the server +// skips ahead after a broadcast lag instead of terminating; the stream keeps +// running. Cursors are opaque, so this message is the only signal that events +// were skipped. Unrecoverable loss (a trimmed or foreign resume cursor) is +// reported as an OUT_OF_RANGE stream status, not this message. type SandboxStreamWarning struct { state protoimpl.MessageState `protogen:"open.v1"` Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"` @@ -17976,7 +18027,7 @@ const file_openshell_proto_rawDesc = "" + "sandbox_id\x18\x02 \x01(\tR\tsandboxId\x12\x1a\n" + "\x05token\x18\x03 \x01(\tB\x04\x88\xb5\x18\x01R\x05token\x12C\n" + "\x0fexpiration_time\x18h \x01(\v2\x1a.google.protobuf.TimestampR\x0eexpirationTime\x12\x18\n" + - "\arevoked\x18\x05 \x01(\bR\arevokedJ\x04\b\x04\x10\x05R\rexpires_at_ms\"\xf1\x03\n" + + "\arevoked\x18\x05 \x01(\bR\arevokedJ\x04\b\x04\x10\x05R\rexpires_at_ms\"\xa1\x04\n" + "\x13WatchSandboxRequest\x12R\n" + "\x0fworkspace_scope\x18\v \x01(\v2).openshell.datamodel.v1.WorkspaceSelectorR\x0eworkspaceScope\x12\x18\n" + "\asandbox\x18\x01 \x01(\tR\asandbox\x12#\n" + @@ -17993,13 +18044,15 @@ const file_openshell_proto_rawDesc = "" + "\vlog_sources\x18\t \x03(\tR\n" + "logSources\x12\"\n" + "\rlog_min_level\x18\n" + - " \x01(\tR\vlogMinLevelJ\x04\b\b\x10\tR\flog_since_ms\"\xcc\x02\n" + + " \x01(\tR\vlogMinLevel\x12.\n" + + "\x13resume_after_cursor\x18\f \x01(\tR\x11resumeAfterCursorJ\x04\b\b\x10\tR\flog_since_ms\"\xe4\x02\n" + "\x12SandboxStreamEvent\x121\n" + "\asandbox\x18\x01 \x01(\v2\x15.openshell.v1.SandboxH\x00R\asandbox\x120\n" + "\x03log\x18\x02 \x01(\v2\x1c.openshell.v1.SandboxLogLineH\x00R\x03log\x123\n" + "\x05event\x18\x03 \x01(\v2\x1b.openshell.v1.PlatformEventH\x00R\x05event\x12>\n" + "\awarning\x18\x04 \x01(\v2\".openshell.v1.SandboxStreamWarningH\x00R\awarning\x12Q\n" + - "\x13draft_policy_update\x18\x05 \x01(\v2\x1f.openshell.v1.DraftPolicyUpdateH\x00R\x11draftPolicyUpdateB\t\n" + + "\x13draft_policy_update\x18\x05 \x01(\v2\x1f.openshell.v1.DraftPolicyUpdateH\x00R\x11draftPolicyUpdate\x12\x16\n" + + "\x06cursor\x18\x06 \x01(\tR\x06cursorB\t\n" + "\apayload\"\xdb\x02\n" + "\x0eSandboxLogLine\x12\x1d\n" + "\n" +