diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index a6768040e6..cd14dc28de 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -6573,6 +6573,12 @@ impl ComputeRuntime { sandbox_id: &str, ) -> Result { let _guard = self.lock_sandbox_for_lifecycle(delete_guard).await; + // A missing sandbox row means the sandbox is gone. Several readers make + // that inference with their own error policy -- this teardown, the + // push-log handler's store recheck (`ensure_log_stream_sandbox_exists` + // in grpc/policy.rs), and the `WatchSandbox` producer in + // grpc/sandbox.rs -- so a change to deletion semantics (soft delete, + // migration windows, record recreation) must update every one of them. let record = self .store .get(Sandbox::object_type(), sandbox_id) @@ -6591,9 +6597,10 @@ 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); + // `tracing_log_bus.close` also clears the platform event bus and leaves + // a terminal tombstone in the shared cursor allocator, so late + // publishers cannot re-create the buffers this frees (see its docs). + self.tracing_log_bus.close(sandbox_id); self.sandbox_watch_bus.remove(sandbox_id); } @@ -15519,6 +15526,105 @@ mod tests { )); } + #[tokio::test] + async fn delete_tombstones_sandbox_log_and_event_buses() { + // #5402: cleanup frees the sandbox's log and platform-event buffers. + // Resetting the cursor space (`remove`) lets a late publisher re-create + // both, and nothing ever frees them again. The delete path must leave a + // terminal tombstone in the shared allocator instead. + let runtime = test_runtime(ControlledDriver::new()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let log_line = |message: &str| openshell_core::proto::SandboxLogLine { + sandbox_id: "sb-1".to_string(), + event_time: openshell_core::time::timestamp_from_millis(1_000).ok(), + level: "INFO".to_string(), + target: "delete-tombstone".to_string(), + message: message.to_string(), + source: "sandbox".to_string(), + fields: HashMap::new(), + }; + let platform_event = |message: &str| openshell_core::proto::SandboxStreamEvent { + payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Event( + PlatformEvent { + source: "test".to_string(), + reason: "Delete".to_string(), + message: message.to_string(), + ..Default::default() + }, + )), + cursor: String::new(), + }; + + // Both buses hold buffers before the delete, so cleanup has memory to + // reclaim and the assertions below observe a real transition. + runtime + .tracing_log_bus + .publish_external(log_line("before delete")); + runtime + .tracing_log_bus + .platform_event_bus + .publish("sb-1", platform_event("before delete")); + assert_eq!(runtime.tracing_log_bus.tail("sb-1", 10).len(), 1); + assert_eq!( + runtime + .tracing_log_bus + .platform_event_bus + .tail("sb-1", 10) + .len(), + 1 + ); + + // The real delete: request, driver acknowledgement, then the Deleted + // watch event that owns gateway-side cleanup. + runtime + .delete_sandbox("default", "sandbox-a") + .await + .expect("the delete request is accepted"); + runtime + .apply_deleted("sb-1") + .await + .expect("the deleted watch event completes cleanup"); + + assert!( + runtime.tracing_log_bus.is_closed("sb-1"), + "delete cleanup must tombstone the sandbox's cursor space, not reset it" + ); + assert!(runtime.tracing_log_bus.tail("sb-1", 10).is_empty()); + assert!( + runtime + .tracing_log_bus + .platform_event_bus + .tail("sb-1", 10) + .is_empty() + ); + + // Late publishers: an open PushSandboxLogs stream, a driver event + // queued behind the Deleted event, and the gateway tracing layer. + runtime + .tracing_log_bus + .publish_external(log_line("after delete")); + runtime + .tracing_log_bus + .platform_event_bus + .publish("sb-1", platform_event("after delete")); + + assert!( + runtime.tracing_log_bus.tail("sb-1", 10).is_empty(), + "a late log publish must not re-create the buffers cleanup freed" + ); + assert!( + runtime + .tracing_log_bus + .platform_event_bus + .tail("sb-1", 10) + .is_empty(), + "a late platform event must not re-create the buffers cleanup freed" + ); + assert!(runtime.tracing_log_bus.is_closed("sb-1")); + } + #[tokio::test] async fn apply_deleted_releases_driver_resources_for_out_of_band_removal() { // Regression test for #2352: a container removed without going diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index cc83789417..d5455b44c2 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -5455,6 +5455,35 @@ pub(super) async fn handle_get_sandbox_logs( Ok(Response::new(GetSandboxLogsResponse { logs, buffer_total })) } +/// How often an open push stream re-reads its sandbox record to detect that +/// the sandbox was deleted. +/// +/// Batches flush every 500ms, so this costs roughly one primary-key read per +/// stream per second, and only when a batch arrives. +const LOG_STREAM_EXISTENCE_RECHECK_PERIOD: std::time::Duration = std::time::Duration::from_secs(1); + +/// Mutable state for one `PushSandboxLogs` stream. +#[derive(Debug, Default)] +struct LogStreamState { + /// Sandbox id the first batch validated as belonging to the caller. + validated_sandbox_id: Option, + /// When the sandbox record was last confirmed to exist. + last_existence_check: Option, +} + +impl LogStreamState { + /// True when the cached existence proof is stale enough to re-read. + fn should_recheck(&self, now: std::time::Instant) -> bool { + self.last_existence_check.is_none_or(|checked_at| { + now.duration_since(checked_at) >= LOG_STREAM_EXISTENCE_RECHECK_PERIOD + }) + } + + fn mark_checked(&mut self, now: std::time::Instant) { + self.last_existence_check = Some(now); + } +} + pub(super) async fn handle_push_sandbox_logs( state: &Arc, request: Request>, @@ -5467,7 +5496,7 @@ pub(super) async fn handle_push_sandbox_logs( .cloned() .ok_or_else(|| Status::unauthenticated("missing principal"))?; let mut stream = request.into_inner(); - let mut validated_sandbox_id = None; + let mut stream_state = LogStreamState::default(); while let Some(batch) = stream .message() @@ -5478,11 +5507,15 @@ pub(super) async fn handle_push_sandbox_logs( continue; } - ensure_log_stream_sandbox_scope( + // Auth first: a cross-sandbox push must keep the audited + // PermissionDenied rather than learn about existence. + ensure_log_stream_sandbox_scope(state, &principal, &batch.sandbox_id, &mut stream_state) + .await?; + ensure_log_stream_sandbox_exists( state, - &principal, + &mut stream_state, &batch.sandbox_id, - &mut validated_sandbox_id, + std::time::Instant::now(), ) .await?; @@ -5511,6 +5544,55 @@ pub(super) async fn handle_push_sandbox_logs( Ok(Response::new(PushSandboxLogsResponse {})) } +/// End an open push stream once the sandbox behind it is gone. +/// +/// A tombstone from [`crate::tracing_bus::TracingLogBus::close`] is checked on +/// every batch and costs no store read. It ends the stream on the replica that +/// ran delete cleanup. The store is re-read at most once per +/// [`LOG_STREAM_EXISTENCE_RECHECK_PERIOD`], which ends a stream whose record +/// disappears while the stream is still open, as a best effort on the other +/// replicas. Freeing another replica's buffers is a separate concern (#5401); +/// only the replica that removed the record arms a permanent tombstone. +/// +/// Only a missing record ends the stream. A transient store failure is not +/// proof of absence, so it keeps the stream open and the next cadence retries. +/// +/// A missing sandbox row means the sandbox is gone. Several readers make that +/// inference with their own error policy -- delete cleanup, this recheck, and +/// the `WatchSandbox` producer in grpc/sandbox.rs -- so a change to deletion +/// semantics (soft delete, migration windows, record recreation) must update +/// every one of them. +async fn ensure_log_stream_sandbox_exists( + state: &Arc, + stream_state: &mut LogStreamState, + sandbox_id: &str, + now: std::time::Instant, +) -> Result<(), Status> { + if state.tracing_log_bus.is_closed(sandbox_id) { + return Err(Status::not_found("sandbox not found")); + } + if !stream_state.should_recheck(now) { + return Ok(()); + } + + match state.store.get(Sandbox::object_type(), sandbox_id).await { + Ok(Some(_)) => { + stream_state.mark_checked(now); + Ok(()) + } + Ok(None) => Err(Status::not_found("sandbox not found")), + Err(error) => { + warn!( + sandbox_id, + %error, + "Sandbox existence recheck failed; keeping the log stream open" + ); + stream_state.mark_checked(now); + Ok(()) + } + } +} + fn validate_sandbox_log_timestamps(logs: &[SandboxLogLine]) -> Result<(), Status> { for (index, log) in logs.iter().enumerate() { if let Some(event_time) = log.event_time.as_ref() { @@ -5522,16 +5604,21 @@ fn validate_sandbox_log_timestamps(logs: &[SandboxLogLine]) -> Result<(), Status Ok(()) } +/// Validate one batch's sandbox scope and, on the first batch, its existence. +/// +/// On the first batch this reads the sandbox record and marks the +/// existence-recheck clock, so the already-paid read also seeds the cadence +/// instead of the caller issuing a second one on the next batch. async fn ensure_log_stream_sandbox_scope( state: &Arc, principal: &Principal, sandbox_id: &str, - validated_sandbox_id: &mut Option, + stream_state: &mut LogStreamState, ) -> Result<(), Status> { // Guard every batch, so switching to another sandbox mid-stream is // recorded like any other cross-sandbox attempt. crate::auth::guard::ensure_sandbox_scope(principal, sandbox_id)?; - if let Some(validated) = validated_sandbox_id.as_deref() { + if let Some(validated) = stream_state.validated_sandbox_id.as_deref() { if sandbox_id != validated { return Err(Status::permission_denied( "log stream sandbox_id changed after validation", @@ -5546,7 +5633,8 @@ async fn ensure_log_stream_sandbox_scope( .await .map_err(|e| Status::internal(format!("fetch sandbox failed: {e}")))? .ok_or_else(|| Status::not_found("sandbox not found"))?; - *validated_sandbox_id = Some(sandbox_id.to_string()); + stream_state.validated_sandbox_id = Some(sandbox_id.to_string()); + stream_state.mark_checked(std::time::Instant::now()); Ok(()) } @@ -8468,6 +8556,322 @@ mod tests { server.abort(); } + #[test] + fn log_stream_state_rechecks_only_after_the_period() { + let mut stream_state = LogStreamState::default(); + let now = std::time::Instant::now(); + assert!( + stream_state.should_recheck(now), + "a stream that never read the record must check" + ); + + stream_state.mark_checked(now); + assert!(!stream_state.should_recheck(now)); + let just_inside = now + + LOG_STREAM_EXISTENCE_RECHECK_PERIOD + .checked_sub(std::time::Duration::from_millis(1)) + .expect("the recheck period must be at least a millisecond"); + assert!(!stream_state.should_recheck(just_inside)); + assert!(stream_state.should_recheck(now + LOG_STREAM_EXISTENCE_RECHECK_PERIOD)); + } + + #[tokio::test] + async fn log_stream_tombstone_ends_stream_without_a_store_read() { + // The record is present, so only the tombstone can end the stream: a + // path that consulted the store first would report it alive. + let state = test_server_state().await; + let sandbox_id = "sb-recheck-tombstone"; + state + .store + .put_message(&test_sandbox( + sandbox_id, + "recheck-tombstone", + ProtoSandboxPolicy::default(), + vec![], + )) + .await + .unwrap(); + let mut stream_state = LogStreamState::default(); + let now = std::time::Instant::now(); + stream_state.mark_checked(now); + state.tracing_log_bus.close(sandbox_id); + + let error = ensure_log_stream_sandbox_exists(&state, &mut stream_state, sandbox_id, now) + .await + .expect_err("a tombstoned sandbox must end the stream on the next batch"); + + assert_eq!(error.code(), Code::NotFound); + } + + #[tokio::test] + async fn log_stream_recheck_ends_stream_when_the_record_is_gone() { + let state = test_server_state().await; + let sandbox_id = "sb-recheck-gone"; + let mut stream_state = LogStreamState::default(); + let checked_at = std::time::Instant::now(); + stream_state.mark_checked(checked_at); + + let error = ensure_log_stream_sandbox_exists( + &state, + &mut stream_state, + sandbox_id, + checked_at + LOG_STREAM_EXISTENCE_RECHECK_PERIOD, + ) + .await + .expect_err("a missing record must end the stream"); + + assert_eq!(error.code(), Code::NotFound); + assert!( + !state.tracing_log_bus.is_closed(sandbox_id), + "one absent sample must not arm a permanent tombstone" + ); + } + + #[tokio::test] + async fn log_stream_recheck_keeps_stream_when_the_record_still_exists() { + let state = test_server_state().await; + let sandbox_id = "sb-recheck-alive"; + state + .store + .put_message(&test_sandbox( + sandbox_id, + "recheck-alive", + ProtoSandboxPolicy::default(), + vec![], + )) + .await + .unwrap(); + let mut stream_state = LogStreamState::default(); + let checked_at = std::time::Instant::now(); + stream_state.mark_checked(checked_at); + let now = checked_at + LOG_STREAM_EXISTENCE_RECHECK_PERIOD; + + ensure_log_stream_sandbox_exists(&state, &mut stream_state, sandbox_id, now) + .await + .expect("a present record keeps the stream open"); + + assert!(!state.tracing_log_bus.is_closed(sandbox_id)); + assert!( + !stream_state.should_recheck(now), + "the confirmed read restarts the cadence" + ); + } + + #[tokio::test] + async fn log_stream_recheck_survives_a_transient_store_failure() { + let state = test_server_state().await; + let sandbox_id = "sb-recheck-transient"; + state + .store + .put_message(&test_sandbox( + sandbox_id, + "recheck-transient", + ProtoSandboxPolicy::default(), + vec![], + )) + .await + .unwrap(); + let mut stream_state = LogStreamState::default(); + let checked_at = std::time::Instant::now(); + stream_state.mark_checked(checked_at); + let now = checked_at + LOG_STREAM_EXISTENCE_RECHECK_PERIOD; + + // A failed read is not proof of absence: every later read fails too. + state.store.close().await; + + ensure_log_stream_sandbox_exists(&state, &mut stream_state, sandbox_id, now) + .await + .expect("a store failure must not end the stream"); + assert!(!state.tracing_log_bus.is_closed(sandbox_id)); + + ensure_log_stream_sandbox_exists( + &state, + &mut stream_state, + sandbox_id, + now + LOG_STREAM_EXISTENCE_RECHECK_PERIOD, + ) + .await + .expect("the stream must survive to the next batch"); + } + + #[tokio::test] + async fn push_sandbox_logs_stream_ends_not_found_after_sandbox_delete() { + use openshell_core::proto::open_shell_client::OpenShellClient; + use openshell_core::proto::open_shell_server::OpenShellServer; + use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream}; + + let state = test_server_state().await; + let sandbox_id = "sb-log-deleted"; + state + .store + .put_message(&test_sandbox( + sandbox_id, + "log-deleted", + ProtoSandboxPolicy::default(), + vec![], + )) + .await + .unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let principal = crate::auth::principal::test_support::sandbox(sandbox_id); + let service = OpenShellServer::with_interceptor( + crate::grpc::OpenShellService::new(state.clone()), + move |mut request: Request<()>| { + request.extensions_mut().insert(principal.clone()); + Ok(request) + }, + ); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(service) + .serve_with_incoming(TcpListenerStream::new(listener)), + ); + let mut client = OpenShellClient::connect(format!("http://{address}")) + .await + .unwrap(); + + let (tx, rx) = tokio::sync::mpsc::channel(4); + let call = + tokio::spawn(async move { client.push_sandbox_logs(ReceiverStream::new(rx)).await }); + + let batch = |message: &str, millis: i64| PushSandboxLogsRequest { + sandbox_id: sandbox_id.to_string(), + logs: vec![SandboxLogLine { + sandbox_id: sandbox_id.to_string(), + event_time: openshell_core::time::timestamp_from_millis(millis).ok(), + level: "INFO".to_string(), + target: "delete-test".to_string(), + message: message.to_string(), + source: "sandbox".to_string(), + fields: HashMap::new(), + }], + }; + + tx.send(batch("before delete", 1)).await.unwrap(); + // Wait for the first batch to be ingested, so the second batch is the + // one that meets the tombstone. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while state.tracing_log_bus.tail(sandbox_id, 10).is_empty() { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("the first batch must be ingested"); + + // What `cleanup_sandbox_state` does on the replica that serves the + // delete. + state.tracing_log_bus.close(sandbox_id); + + tx.send(batch("after delete", 2)).await.unwrap(); + + let error = call + .await + .expect("client task must not panic") + .expect_err("a deleted sandbox must end its push stream"); + assert_eq!(error.code(), Code::NotFound); + server.abort(); + } + + #[tokio::test] + async fn push_sandbox_logs_probes_a_sandbox_record_once_per_validated_batch() { + use crate::otel_tracing::test_exporter; + use openshell_core::proto::open_shell_client::OpenShellClient; + use openshell_core::proto::open_shell_server::OpenShellServer; + use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream}; + + let state = test_server_state().await; + let sandbox_id = "sb-log-seeded-recheck"; + state + .store + .put_message(&test_sandbox( + sandbox_id, + "log-seeded-recheck", + ProtoSandboxPolicy::default(), + vec![], + )) + .await + .unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let principal = crate::auth::principal::test_support::sandbox(sandbox_id); + let service = OpenShellServer::with_interceptor( + crate::grpc::OpenShellService::new(state.clone()), + move |mut request: Request<()>| { + request.extensions_mut().insert(principal.clone()); + Ok(request) + }, + ); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(service) + .serve_with_incoming(TcpListenerStream::new(listener)), + ); + let mut client = OpenShellClient::connect(format!("http://{address}")) + .await + .unwrap(); + let (tx, rx) = tokio::sync::mpsc::channel(4); + let call = + tokio::spawn(async move { client.push_sandbox_logs(ReceiverStream::new(rx)).await }); + + // Count the sandbox-record reads the handler makes. The first batch's + // scope check already fetched the record, so the recheck clock must be + // seeded from it: a second read in the same batch would double what + // every open stream pays at startup. + let traced = test_exporter::install_traced(); + + tx.send(PushSandboxLogsRequest { + sandbox_id: sandbox_id.to_string(), + logs: vec![SandboxLogLine { + sandbox_id: sandbox_id.to_string(), + event_time: openshell_core::time::timestamp_from_millis(1).ok(), + level: "INFO".to_string(), + target: "seeded-recheck".to_string(), + message: "first batch".to_string(), + source: "sandbox".to_string(), + fields: HashMap::new(), + }], + }) + .await + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while state.tracing_log_bus.tail(sandbox_id, 10).is_empty() { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("the first batch must be ingested"); + + let record_reads: Vec<_> = traced + .wait_for_spans(|spans| { + spans.iter().any(|span| { + span.name == "store.get" + && test_exporter::attribute(span, "object.id").as_deref() + == Some(sandbox_id) + }) + }) + .await + .into_iter() + .filter(|span| { + span.name == "store.get" + && test_exporter::attribute(span, "object.id").as_deref() == Some(sandbox_id) + }) + .collect(); + assert_eq!( + record_reads.len(), + 1, + "a validated batch must read the sandbox record once, not twice: the \ + scope check proves existence, so only the recheck-cadence seed prevents \ + a duplicate read per open stream" + ); + + drop(tx); + call.await + .expect("client task must not panic") + .expect("closing the client stream ends the push cleanly"); + server.abort(); + } + #[test] fn sandbox_log_timestamp_validation_rejects_malformed_batches_atomically() { let logs = vec![ @@ -11990,15 +12394,15 @@ mod tests { } let req = with_sandbox(Request::new(()), "sb-a"); let principal = req.extensions().get::().unwrap().clone(); - let mut validated = None; + let mut stream_state = LogStreamState::default(); let capture = openshell_ocsf::test_support::OcsfCapture::default(); let _guard = capture.install(); - ensure_log_stream_sandbox_scope(&state, &principal, "sb-a", &mut validated) + ensure_log_stream_sandbox_scope(&state, &principal, "sb-a", &mut stream_state) .await .expect("first frame should validate"); assert!(capture.events().is_empty(), "own sandbox is not a denial"); - let err = ensure_log_stream_sandbox_scope(&state, &principal, "sb-b", &mut validated) + let err = ensure_log_stream_sandbox_scope(&state, &principal, "sb-b", &mut stream_state) .await .expect_err("later frame must not switch sandbox ids"); @@ -12014,9 +12418,9 @@ mod tests { let state = test_server_state().await; let req = with_sandbox(Request::new(()), "sb-a"); let principal = req.extensions().get::().unwrap().clone(); - let mut validated = None; + let mut stream_state = LogStreamState::default(); - let err = ensure_log_stream_sandbox_scope(&state, &principal, "sb-a", &mut validated) + let err = ensure_log_stream_sandbox_scope(&state, &principal, "sb-a", &mut stream_state) .await .expect_err("missing sandbox must not validate"); diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 92d30f5571..908685bb34 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -84,6 +84,22 @@ pub struct ResumeGap { struct CursorSpace { epoch: Uuid, next: u64, + /// Terminal tombstone written by [`TracingLogBus::close`]. A closed space + /// never allocates another cursor, so late publishers cannot re-create the + /// buffers deletion just freed. + closed: bool, +} + +impl CursorSpace { + /// A bare tombstone for an id that never published, so a subscribe-only + /// sandbox can be closed too. No epoch is ever handed out for it. + fn tombstone() -> Self { + Self { + epoch: Uuid::nil(), + next: 1, + closed: true, + } + } } /// Identity and extent of a sandbox's current cursor space. @@ -108,8 +124,9 @@ struct SeqAllocator { 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 + /// Publication, subscription, and teardown each hold this guard across + /// their bus-map mutation, which is what keeps cursors monotonic and makes + /// the closed check atomic with the entry it guards. 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. @@ -128,33 +145,65 @@ impl SeqAllocator { /// 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) { + /// + /// Returns `None` for a closed space: the timeline is over and the caller + /// must drop the event rather than re-create the buffers [`TracingLogBus::close`] + /// freed. + fn next_locked( + spaces: &mut HashMap, + sandbox_id: &str, + ) -> Option<(Uuid, u64)> { let space = spaces .entry(sandbox_id.to_string()) .or_insert_with(|| CursorSpace { epoch: Uuid::new_v4(), next: 1, + closed: false, }); + if space.closed { + return None; + } let seq = space.next; space.next += 1; - (space.epoch, seq) + Some((space.epoch, seq)) } /// Identity and extent of this sandbox's space, or `None` when no event has - /// been published into it. + /// been published into it or the space is closed. /// /// 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), - }) + self.lock() + .get(sandbox_id) + .filter(|space| !space.closed) + .map(|space| CursorSpaceInfo { + epoch: space.epoch, + highest_seq: space.next.saturating_sub(1), + }) + } + + /// True when `sandbox_id` carries a tombstone in the locked space map. + /// + /// Shared by `subscribe` (which must test it atomically with the bus-map + /// insert) and `is_closed`, so the predicate has one definition. + fn is_closed_locked(spaces: &HashMap, sandbox_id: &str) -> bool { + spaces.get(sandbox_id).is_some_and(|space| space.closed) } } +/// A receiver that is already closed, for a tombstoned id. +/// +/// Nothing is ever sent for a closed id, so the sender is dropped immediately +/// and the receiver reports `Closed` on first use. Capacity is irrelevant. +fn closed_receiver() -> broadcast::Receiver { + let (sender, receiver) = broadcast::channel(1); + drop(sender); + receiver +} + fn tail_after_impl( tail: &VecDeque, last_trimmed_seq: u64, @@ -221,9 +270,73 @@ impl TracingLogBus { } pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { + // Hold the allocator across the bus-map insert so the closed check is + // atomic with a concurrent [`Self::close`], mirroring `publish`. A + // closed id yields a receiver that is already closed instead of a new + // entry: re-creating the entry would undo the memory `close` reclaimed. + let spaces = self.seq.lock(); + if SeqAllocator::is_closed_locked(&spaces, sandbox_id) { + return closed_receiver(); + } self.sender_for(sandbox_id).subscribe() } + /// Terminally close a sandbox's log and platform event streams. + /// + /// This is the deletion path: it frees both buses' buffers and leaves a + /// tombstone in the shared cursor space, so a late publisher — an open + /// `PushSandboxLogs` stream, a driver event queued after `Deleted`, the + /// gateway tracing layer including OCSF audit events — drops its event + /// instead of re-creating the ~1.7 MiB of channel and tail that nothing + /// would ever free again. Idempotent. + /// + /// Lifecycle: a sandbox's space stays open until [`Self::remove`] retires + /// it or this call closes it. `remove` is the reset path (re-publishing + /// mints a new epoch, as the resume tests pin). `closed` is terminal: + /// `publish`, `subscribe`, and the platform bus all refuse the id, the + /// cursor-space lookup reports no space, and `remove` leaves the tombstone + /// in place. + /// + /// Tombstones are unbounded on purpose. Each costs roughly a hundred + /// bytes, against the ~1.7 MiB of buffers it replaces; sandbox ids are + /// fresh UUIDs that are never reused, and all bus state is cleared on + /// gateway restart. + /// + /// The whole sequence runs under the cursor-space lock, so it is atomic + /// against publication and subscription on either bus. The platform bus + /// removal only takes its own map lock, never the allocator, so the + /// documented allocator -> bus map order holds. + pub fn close(&self, sandbox_id: &str) { + let mut spaces = self.seq.lock(); + self.drop_bus_entries_locked(sandbox_id); + spaces + .entry(sandbox_id.to_string()) + .and_modify(|space| space.closed = true) + .or_insert_with(CursorSpace::tombstone); + } + + /// Drop both buses' per-sandbox entries. + /// + /// The caller holds the cursor-space lock. The platform bus removal takes + /// only its own map lock and never reaches for the allocator, so this + /// preserves the documented allocator -> bus map order. + fn drop_bus_entries_locked(&self, sandbox_id: &str) { + { + let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + inner.per_id.remove(sandbox_id); + } + self.platform_event_bus.remove(sandbox_id); + } + + /// True once [`Self::close`] has run for this sandbox. + /// + /// The push-stream handler uses this as a cheap, lock-only existence + /// check: a tombstoned id can never gain a buffer again, so it needs no + /// store read. + pub fn is_closed(&self, sandbox_id: &str) -> bool { + SeqAllocator::is_closed_locked(&self.seq.lock(), sandbox_id) + } + /// Remove all bus entries for the given sandbox id, including the platform /// event bus that shares this bus's cursor allocator. /// @@ -242,15 +355,17 @@ impl TracingLogBus { /// 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. + /// + /// [`Self::close`] is the terminal counterpart: a removed space can be + /// republished into, a closed one cannot. On a closed id this is a no-op on + /// the allocator, because dropping the tombstone there would let a late + /// publisher re-create the buffers deletion just freed. pub fn remove(&self, sandbox_id: &str) { let mut spaces = self.seq.lock(); - { - let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); - inner.per_id.remove(sandbox_id); + self.drop_bus_entries_locked(sandbox_id); + if spaces.get(sandbox_id).is_none_or(|space| !space.closed) { + spaces.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(crate) fn tail(&self, sandbox_id: &str, max: usize) -> Vec { @@ -309,9 +424,13 @@ impl TracingLogBus { 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`. + // allocator -> bus map, matching `remove` and `close`. let mut spaces = self.seq.lock(); - let (epoch, seq) = SeqAllocator::next_locked(&mut spaces, sandbox_id); + // A closed space is over: drop the event instead of allocating a + // cursor and re-creating the buffers `close` freed. + let Some((epoch, seq)) = SeqAllocator::next_locked(&mut spaces, sandbox_id) else { + return; + }; event.cursor = WatchCursor::new(epoch, seq).encode(); let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); @@ -420,6 +539,20 @@ fn display_level(target: &str, level: &str) -> String { mod tests { use super::*; + /// Assert that neither bus holds a per-sandbox entry. + #[track_caller] + fn assert_bus_maps_empty(bus: &TracingLogBus) { + assert!(bus.inner.lock().unwrap().per_id.is_empty()); + assert!( + bus.platform_event_bus + .inner + .lock() + .unwrap() + .per_id + .is_empty() + ); + } + #[test] fn cross_sandbox_denials_remain_gateway_wide() { use crate::auth::{audit, principal::test_support}; @@ -705,11 +838,12 @@ mod tests { 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. + // Subscribing must not mint a space. `subscribe` takes the allocator + // lock to check the tombstone, but only inserts into the bus map -- 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); @@ -867,6 +1001,222 @@ mod tests { bus.remove("nonexistent"); } + #[test] + fn publish_after_close_drops_without_recreating_buffers() { + // The issue's leak: late publishers used to re-create the buffers a + // deleted sandbox's cleanup had just freed, and nothing ever freed them + // again. + let bus = TracingLogBus::new(); + let sandbox_id = "sb-closed-publish"; + bus.publish_external(make_log_event(sandbox_id, "before")); + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); + bus.close(sandbox_id); + + bus.publish_external(make_log_event(sandbox_id, "after")); + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); + + assert!( + bus.tail(sandbox_id, 10).is_empty(), + "a closed sandbox must drop log lines" + ); + assert!( + bus.platform_event_bus.tail(sandbox_id, 10).is_empty(), + "a closed sandbox must drop platform events" + ); + assert_bus_maps_empty(&bus); + assert_eq!(bus.cursor_space(sandbox_id), None); + assert!(bus.is_closed(sandbox_id)); + } + + #[test] + fn tracing_log_bus_close_closes_active_receivers() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-close-receivers"; + let mut log_rx = bus.subscribe(sandbox_id); + let mut platform_rx = bus.platform_event_bus.subscribe(sandbox_id); + + bus.publish_external(make_log_event(sandbox_id, "hello")); + assert!(log_rx.try_recv().is_ok()); + + bus.close(sandbox_id); + + match log_rx.try_recv() { + Err(broadcast::error::TryRecvError::Closed) => {} // expected + other => panic!("expected Closed, got {other:?}"), + } + match platform_rx.try_recv() { + Err(broadcast::error::TryRecvError::Closed) => {} // expected + other => panic!("expected Closed, got {other:?}"), + } + // A close that tombstoned but forgot the map removal would still pass + // the receiver assertions above; the map lengths are what prove the + // buffers were actually freed. + assert_bus_maps_empty(&bus); + } + + #[test] + fn subscribe_after_close_yields_closed_receiver_and_creates_no_entry() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-sub-after-close"; + bus.close(sandbox_id); + + let mut log_rx = bus.subscribe(sandbox_id); + assert!(matches!( + log_rx.try_recv(), + Err(broadcast::error::TryRecvError::Closed) + )); + let mut platform_rx = bus.platform_event_bus.subscribe(sandbox_id); + assert!(matches!( + platform_rx.try_recv(), + Err(broadcast::error::TryRecvError::Closed) + )); + + assert_bus_maps_empty(&bus); + } + + #[test] + fn close_is_idempotent_and_remove_keeps_the_tombstone() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-close-idempotent"; + + bus.publish_external(make_log_event(sandbox_id, "line")); + bus.close(sandbox_id); + bus.close(sandbox_id); + assert!(bus.is_closed(sandbox_id)); + assert_eq!(bus.seq.lock().len(), 1); + + // `remove` is the reset path, but a closed id is terminal: the + // tombstone must survive so a late publisher still cannot re-create the + // buffers. + bus.remove(sandbox_id); + assert!(bus.is_closed(sandbox_id)); + bus.publish_external(make_log_event(sandbox_id, "late")); + assert!(bus.tail(sandbox_id, 10).is_empty()); + assert_bus_maps_empty(&bus); + } + + #[test] + fn close_without_prior_publish_inserts_a_tombstone() { + // A subscribe-only sandbox has no cursor space to close, but deleting + // it must still refuse later publishes. + let bus = TracingLogBus::new(); + let sandbox_id = "sb-subscribe-only"; + let _rx = bus.subscribe(sandbox_id); + bus.close(sandbox_id); + + assert!(bus.is_closed(sandbox_id)); + bus.publish_external(make_log_event(sandbox_id, "late")); + assert!(bus.tail(sandbox_id, 10).is_empty()); + assert_bus_maps_empty(&bus); + } + + #[test] + fn cursor_space_reports_none_for_a_closed_sandbox() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-closed-space"; + bus.publish_external(make_log_event(sandbox_id, "line")); + assert!(bus.cursor_space(sandbox_id).is_some()); + + bus.close(sandbox_id); + + assert_eq!( + bus.cursor_space(sandbox_id), + None, + "a tombstone must not look like a live space" + ); + } + + #[test] + fn tracing_layer_after_close_allocates_nothing() { + // The gateway tracing layer -- including the OCSF audit event emitted + // while serving the delete -- publishes for any event carrying a + // `sandbox_id`. Pin that route against the tombstone too. + use tracing_subscriber::prelude::*; + + let bus = TracingLogBus::new(); + let sandbox_id = "sb-layer-closed"; + bus.publish_external(make_log_event(sandbox_id, "before")); + bus.close(sandbox_id); + + let subscriber = tracing_subscriber::registry().with(bus.layer()); + tracing::subscriber::with_default(subscriber, || { + tracing::info!(sandbox_id = "sb-layer-closed", "late gateway event"); + }); + + assert!(bus.tail(sandbox_id, 10).is_empty()); + // The layer must not re-create a closed entry. + assert_bus_maps_empty(&bus); + let spaces = bus.seq.lock(); + assert_eq!(spaces.len(), 1, "only the tombstone may remain"); + assert!(spaces[sandbox_id].closed); + } + + #[test] + fn concurrent_publish_and_close_never_recreates_buffers() { + // A publish that allocated before `close` must not land in a re-created + // entry. The tombstone check and the bus-map insert share the allocator + // lock, so once `close` returns for an id nothing can refill its tail. + for round in 0..64 { + let bus = TracingLogBus::new(); + let sandbox_id = format!("sb-close-race-{round}"); + let barrier = Arc::new(std::sync::Barrier::new(5)); + + let publishers: Vec<_> = (0..4) + .map(|_| { + let bus = bus.clone(); + let barrier = Arc::clone(&barrier); + let sandbox_id = sandbox_id.clone(); + std::thread::spawn(move || { + barrier.wait(); + for _ in 0..64 { + bus.publish_external(make_log_event(&sandbox_id, "x")); + bus.platform_event_bus.publish(&sandbox_id, stream_event(0)); + } + }) + }) + .collect(); + + barrier.wait(); + bus.close(&sandbox_id); + + for publisher in publishers { + publisher.join().expect("publisher thread panicked"); + } + + assert!(bus.is_closed(&sandbox_id)); + assert!( + bus.tail(&sandbox_id, usize::MAX).is_empty(), + "a closed sandbox must never accumulate a tail again" + ); + assert!( + bus.platform_event_bus + .tail(&sandbox_id, usize::MAX) + .is_empty() + ); + assert_bus_maps_empty(&bus); + assert_eq!(bus.seq.lock().len(), 1, "only the tombstone remains"); + } + } + + #[test] + fn repeated_create_delete_cycles_keep_bus_maps_bounded() { + // The RSS proxy: buffers are freed every cycle and only the small + // tombstone accumulates. + let bus = TracingLogBus::new(); + for index in 0..1_000 { + let sandbox_id = format!("sb-cycle-{index}"); + bus.publish_external(make_log_event(&sandbox_id, "line")); + bus.platform_event_bus.publish(&sandbox_id, stream_event(0)); + bus.close(&sandbox_id); + assert_bus_maps_empty(&bus); + } + assert_eq!( + bus.seq.lock().len(), + 1_000, + "one tombstone per deleted sandbox" + ); + } + #[test] fn display_level_maps_ocsf_target_to_ocsf() { assert_eq!(display_level(OCSF_TARGET, "INFO"), "OCSF"); @@ -1030,6 +1380,13 @@ impl PlatformEventBus { } pub(crate) fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { + // Same guard as `TracingLogBus::subscribe`: take the allocator lock so + // the closed check is atomic with `TracingLogBus::close`, and return an + // already-closed receiver for a tombstoned id. + let spaces = self.seq.lock(); + if SeqAllocator::is_closed_locked(&spaces, sandbox_id) { + return closed_receiver(); + } self.sender_for(sandbox_id).subscribe() } @@ -1038,7 +1395,9 @@ impl PlatformEventBus { // 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); + let Some((epoch, seq)) = SeqAllocator::next_locked(&mut spaces, sandbox_id) else { + return; + }; event.cursor = WatchCursor::new(epoch, seq).encode(); let mut inner = self.inner.lock().expect("platform event bus lock poisoned");