From bb9c594ed72f793d78926eb3358d888b6b011ab1 Mon Sep 17 00:00:00 2001 From: divesh Date: Tue, 10 Nov 2026 11:08:59 -0800 Subject: [PATCH 1/2] fix(server): free deleted sandboxes' buffers on every replica (#5401) Delete cleanup runs only on the replica that served the delete or holds the reconciler lease, so other replicas kept a deleted sandbox's log and event buffers forever. Each multi-replica gateway now sweeps its own idle buffers every 30s, checks which sandboxes still exist with a batched primary-key lookup, and closes the ones missing on two consecutive sweeps. Buffers with live receivers or activity in the last two minutes are skipped, so the cost scales with a replica's idle local state, not fleet size or replica count. Signed-off-by: divesh --- crates/openshell-server/src/bus_sweeper.rs | 211 ++++++++++++++++++ crates/openshell-server/src/lib.rs | 10 +- .../openshell-server/src/persistence/mod.rs | 29 ++- .../src/persistence/postgres.rs | 31 ++- .../src/persistence/sqlite.rs | 31 ++- .../openshell-server/src/persistence/tests.rs | 34 +++ crates/openshell-server/src/tracing_bus.rs | 40 ++++ 7 files changed, 381 insertions(+), 5 deletions(-) create mode 100644 crates/openshell-server/src/bus_sweeper.rs diff --git a/crates/openshell-server/src/bus_sweeper.rs b/crates/openshell-server/src/bus_sweeper.rs new file mode 100644 index 0000000000..445d826408 --- /dev/null +++ b/crates/openshell-server/src/bus_sweeper.rs @@ -0,0 +1,211 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Frees the log and event buffers a replica holds for deleted sandboxes. +//! +//! Delete cleanup runs only on the replica that served the delete or ran the +//! driver watch. Any other replica that buffered the sandbox's logs or events +//! keeps them until this sweeper sees the record is gone. Each replica checks +//! only its own idle buffers, in batched primary-key lookups, so the cost +//! scales with local state rather than with fleet size or replica count. + +use std::collections::HashSet; +use std::sync::Arc; +use std::time::Duration; + +use openshell_core::proto::Sandbox; +use tokio::sync::watch; +use tracing::{info, warn}; + +use crate::persistence::{ObjectType, PersistenceResult, Store}; +use crate::tracing_bus::TracingLogBus; + +/// How often each replica sweeps its buffers. +pub const SWEEP_INTERVAL: Duration = Duration::from_secs(30); + +/// Buffers active more recently than this are not checked. A sandbox that is +/// still publishing or watched almost certainly still exists. +pub const MIN_IDLE: Duration = Duration::from_mins(2); + +/// Spawn the sweeper. Only multi-replica deployments need it: a single replica +/// runs every delete cleanup itself. +pub fn spawn(store: Arc, bus: TracingLogBus, mut shutdown_rx: watch::Receiver) { + tokio::spawn(async move { + let mut suspects = HashSet::new(); + let mut timer = tokio::time::interval(SWEEP_INTERVAL); + timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + tokio::select! { + changed = shutdown_rx.changed() => { + if changed.is_err() || *shutdown_rx.borrow() { + break; + } + } + _ = timer.tick() => { + match sweep_once(&store, &bus, MIN_IDLE, &mut suspects).await { + Ok(0) => {} + Ok(closed) => info!(closed, "freed log and event buffers of deleted sandboxes"), + Err(err) => warn!(error = %err, "bus sweeper: failed to check sandbox records"), + } + } + } + } + }); +} + +/// Run one sweep and return how many sandboxes were closed. +/// +/// A sandbox is closed only when its record is missing on two sweeps in a +/// row; `suspects` carries the ids missing on the previous one. A store error +/// closes nothing and leaves `suspects` unchanged. +async fn sweep_once( + store: &Store, + bus: &TracingLogBus, + min_idle: Duration, + suspects: &mut HashSet, +) -> PersistenceResult { + let idle = bus.idle_sandbox_ids(min_idle); + let existing = store.existing_ids(Sandbox::object_type(), &idle).await?; + + let mut closed = 0; + let mut missing = HashSet::new(); + for sandbox_id in idle { + if existing.contains(&sandbox_id) { + continue; + } + if suspects.contains(&sandbox_id) { + bus.close(&sandbox_id); + closed += 1; + } else { + missing.insert(sandbox_id); + } + } + *suspects = missing; + Ok(closed) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::persistence::test_postgres::TestSchema; + use openshell_core::proto::{SandboxLogLine, SandboxStreamEvent}; + + async fn put_sandbox(store: &Store, sandbox_id: &str) { + store + .put( + Sandbox::object_type(), + sandbox_id, + sandbox_id, + "default", + b"payload", + None, + ) + .await + .unwrap(); + } + + fn publish_log(bus: &TracingLogBus, sandbox_id: &str) { + bus.publish_external(SandboxLogLine { + sandbox_id: sandbox_id.to_string(), + message: "line".to_string(), + ..Default::default() + }); + } + + fn publish_event(bus: &TracingLogBus, sandbox_id: &str) { + bus.platform_event_bus + .publish(sandbox_id, SandboxStreamEvent::default()); + } + + #[tokio::test] + async fn deleted_sandbox_is_freed_on_another_replica() { + let store = crate::persistence::test_store().await; + deleted_sandbox_is_freed_on_another_replica_on(&store, &store).await; + } + + #[tokio::test] + #[ignore = "requires a disposable PostgreSQL database in OPENSHELL_TEST_POSTGRES_URL; run mise run test:rust:postgres"] + async fn postgres_deleted_sandbox_is_freed_on_another_replica() { + let schema = TestSchema::create("bus_sweeper").await; + let owner = schema.connect_store().await; + let other = schema.connect_store().await; + deleted_sandbox_is_freed_on_another_replica_on(&owner, &other).await; + schema.drop_schema().await; + } + + /// `owner` serves the delete; `other` is a replica that buffered the + /// sandbox but never runs its delete cleanup. + async fn deleted_sandbox_is_freed_on_another_replica_on(owner: &Store, other: &Store) { + put_sandbox(owner, "sb-live").await; + put_sandbox(owner, "sb-gone").await; + let bus = TracingLogBus::new(); + publish_log(&bus, "sb-live"); + publish_log(&bus, "sb-gone"); + publish_event(&bus, "sb-gone"); + + owner + .delete(Sandbox::object_type(), "sb-gone") + .await + .unwrap(); + + let mut suspects = HashSet::new(); + let first = sweep_once(other, &bus, Duration::ZERO, &mut suspects).await; + assert_eq!(first.unwrap(), 0, "first miss only marks a suspect"); + assert!(!bus.is_closed("sb-gone")); + + let second = sweep_once(other, &bus, Duration::ZERO, &mut suspects).await; + assert_eq!(second.unwrap(), 1); + assert!(bus.is_closed("sb-gone")); + assert!(bus.tail("sb-gone", 10).is_empty()); + assert!(bus.platform_event_bus.tail("sb-gone", 10).is_empty()); + + assert!(!bus.is_closed("sb-live")); + assert_eq!(bus.tail("sb-live", 10).len(), 1); + } + + #[tokio::test] + async fn active_buffers_are_not_checked() { + let store = crate::persistence::test_store().await; + let bus = TracingLogBus::new(); + publish_log(&bus, "sb-gone"); + let mut suspects = HashSet::new(); + + // Recently active: skipped until it has been idle for `min_idle`. + for _ in 0..2 { + let swept = sweep_once(&store, &bus, Duration::from_hours(1), &mut suspects).await; + assert_eq!(swept.unwrap(), 0); + } + + // A live receiver on either bus keeps the sandbox out of the sweep. + let platform_rx = bus.platform_event_bus.subscribe("sb-gone"); + for _ in 0..2 { + let swept = sweep_once(&store, &bus, Duration::ZERO, &mut suspects).await; + assert_eq!(swept.unwrap(), 0); + } + assert!(!bus.is_closed("sb-gone")); + + drop(platform_rx); + for expected in [0, 1] { + let swept = sweep_once(&store, &bus, Duration::ZERO, &mut suspects).await; + assert_eq!(swept.unwrap(), expected); + } + assert!(bus.is_closed("sb-gone")); + } + + #[tokio::test] + async fn store_failure_frees_nothing() { + let store = crate::persistence::test_store().await; + let bus = TracingLogBus::new(); + publish_log(&bus, "sb-gone"); + let mut suspects = HashSet::from(["sb-gone".to_string()]); + + store.close().await; + let swept = sweep_once(&store, &bus, Duration::ZERO, &mut suspects).await; + + assert!(swept.is_err()); + assert!(!bus.is_closed("sb-gone")); + assert_eq!(bus.tail("sb-gone", 10).len(), 1); + assert!(suspects.contains("sb-gone")); + } +} diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 0c9a352b26..37862e6e7a 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -14,6 +14,7 @@ //! external endpoint; it does not switch on driver names. mod auth; +mod bus_sweeper; pub mod certgen; pub mod cli; mod compute; @@ -1282,8 +1283,8 @@ pub(crate) async fn run_server( } startup_tx.send_replace(true); - // The poller exists to observe writes made by other replicas. Single- - // replica backends have none, so it would only add load. + // The poller and sweeper exist to observe writes made by other replicas. + // Single-replica backends have none, so they would only add load. if !store.is_single_replica() { sandbox_watch::spawn_store_poller( store.clone(), @@ -1291,6 +1292,11 @@ pub(crate) async fn run_server( sandbox_watch::DEFAULT_STORE_POLL_INTERVAL, shutdown_rx.clone(), ); + bus_sweeper::spawn( + store.clone(), + state.tracing_log_bus.clone(), + shutdown_rx.clone(), + ); } let membership_worker = gateway_members::spawn_membership_worker( state.clone(), diff --git a/crates/openshell-server/src/persistence/mod.rs b/crates/openshell-server/src/persistence/mod.rs index d61451604d..0f5f001e40 100644 --- a/crates/openshell-server/src/persistence/mod.rs +++ b/crates/openshell-server/src/persistence/mod.rs @@ -20,7 +20,7 @@ pub use crate::storage_proto::{ use openshell_core::{Error as CoreError, Result as CoreResult}; use prost::Message; use rand::Rng; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use thiserror::Error; pub use mutation_lock::{LockMode, MutationLockKey, MutationLockSet}; @@ -40,6 +40,11 @@ pub type PersistenceResult = Result; /// requests into independently retryable, bounded write statements. pub const DELETE_MANY_BATCH_SIZE: usize = 128; +/// Maximum number of object ids sent in one existence lookup. +/// +/// Keep this well below `SQLite`'s bind-variable limit. +pub const EXISTING_IDS_BATCH_SIZE: usize = 500; + /// Persistence-layer error type. #[derive(Debug, Error, Clone)] pub enum PersistenceError { @@ -830,6 +835,28 @@ impl Store { store_dispatch_traced!(self.delete_many(object_type, ids)) } + /// Return the subset of `ids` that exist for an object type, using bounded + /// primary-key lookups. + #[tracing::instrument( + name = "store", + level = "debug", + skip_all, + fields( + otel.name = "store.existing_ids", + otel.status_code = tracing::field::Empty, + object_type = %object_type, + object_count = ids.len(), + batch_count = ids.len().div_ceil(EXISTING_IDS_BATCH_SIZE), + ) + )] + pub async fn existing_ids( + &self, + object_type: &str, + ids: &[String], + ) -> PersistenceResult> { + store_dispatch_traced!(self.existing_ids(object_type, ids)) + } + /// Count objects of a given type within a workspace. #[tracing::instrument( name = "store", diff --git a/crates/openshell-server/src/persistence/postgres.rs b/crates/openshell-server/src/persistence/postgres.rs index 45511726fd..8e1d7f9fd4 100644 --- a/crates/openshell-server/src/persistence/postgres.rs +++ b/crates/openshell-server/src/persistence/postgres.rs @@ -36,7 +36,10 @@ pub(super) fn embedded_migration_sql(version: i64) -> Option<&'static str> { .map(|migration| migration.sql.as_ref()) } -use super::{DELETE_MANY_BATCH_SIZE, DRAFT_CHUNK_OBJECT_TYPE, POLICY_OBJECT_TYPE}; +use super::{ + DELETE_MANY_BATCH_SIZE, DRAFT_CHUNK_OBJECT_TYPE, EXISTING_IDS_BATCH_SIZE, POLICY_OBJECT_TYPE, +}; +use std::collections::HashSet; #[derive(Debug, Clone)] pub struct PostgresStore { @@ -1150,6 +1153,32 @@ WHERE object_type = $1 AND workspace = $2 AND name = $3 Ok(deleted) } + pub async fn existing_ids( + &self, + object_type: &str, + ids: &[String], + ) -> PersistenceResult> { + let mut existing = HashSet::with_capacity(ids.len()); + for ids in ids.chunks(EXISTING_IDS_BATCH_SIZE) { + let mut query = + QueryBuilder::::new("SELECT id FROM objects WHERE object_type = "); + query.push_bind(object_type).push(" AND id IN ("); + let mut separated = query.separated(", "); + for id in ids { + separated.push_bind(id); + } + separated.push_unseparated(")"); + + let rows = query + .build() + .fetch_all(&self.pool) + .await + .map_err(|e| map_db_error(&e))?; + existing.extend(rows.iter().map(|row| row.get::("id"))); + } + Ok(existing) + } + pub async fn count_in_workspace( &self, object_type: &str, diff --git a/crates/openshell-server/src/persistence/sqlite.rs b/crates/openshell-server/src/persistence/sqlite.rs index 8c434dcaa4..2cd27c702e 100644 --- a/crates/openshell-server/src/persistence/sqlite.rs +++ b/crates/openshell-server/src/persistence/sqlite.rs @@ -36,7 +36,10 @@ pub(super) fn embedded_migration_sql(version: i64) -> Option<&'static str> { } static IN_MEMORY_DB_SEQUENCE: AtomicU64 = AtomicU64::new(0); -use super::{DELETE_MANY_BATCH_SIZE, DRAFT_CHUNK_OBJECT_TYPE, POLICY_OBJECT_TYPE}; +use super::{ + DELETE_MANY_BATCH_SIZE, DRAFT_CHUNK_OBJECT_TYPE, EXISTING_IDS_BATCH_SIZE, POLICY_OBJECT_TYPE, +}; +use std::collections::HashSet; /// Pool ceiling for an on-disk database when the operator has not configured /// one. @@ -877,6 +880,32 @@ WHERE "object_type" = ?1 AND "id" = ?2 Ok(deleted) } + pub async fn existing_ids( + &self, + object_type: &str, + ids: &[String], + ) -> PersistenceResult> { + let mut existing = HashSet::with_capacity(ids.len()); + for ids in ids.chunks(EXISTING_IDS_BATCH_SIZE) { + let mut query = + QueryBuilder::::new("SELECT id FROM objects WHERE object_type = "); + query.push_bind(object_type).push(" AND id IN ("); + let mut separated = query.separated(", "); + for id in ids { + separated.push_bind(id); + } + separated.push_unseparated(")"); + + let rows = query + .build() + .fetch_all(&self.pool) + .await + .map_err(|e| map_db_error(&e))?; + existing.extend(rows.iter().map(|row| row.get::("id"))); + } + Ok(existing) + } + pub async fn count_in_workspace( &self, object_type: &str, diff --git a/crates/openshell-server/src/persistence/tests.rs b/crates/openshell-server/src/persistence/tests.rs index 40d2c315c3..b5a5c4944d 100644 --- a/crates/openshell-server/src/persistence/tests.rs +++ b/crates/openshell-server/src/persistence/tests.rs @@ -882,6 +882,40 @@ async fn delete_many_is_bounded_idempotent_and_type_scoped() { assert!(store.get("provider", "other-type").await.unwrap().is_some()); } +#[tokio::test] +async fn existing_ids_spans_batches_and_is_type_scoped() { + let store = test_store().await; + let mut ids: Vec = (0..=super::EXISTING_IDS_BATCH_SIZE) + .map(|idx| format!("missing-{idx}")) + .collect(); + for id in ["sandbox-first", "sandbox-last"] { + store + .put("sandbox", id, id, "default", b"payload", None) + .await + .unwrap(); + } + store + .put( + "provider", + "other-type", + "other-type", + "default", + b"payload", + None, + ) + .await + .unwrap(); + ids.insert(0, "sandbox-first".to_string()); + ids.extend(["other-type".to_string(), "sandbox-last".to_string()]); + + let existing = store.existing_ids("sandbox", &ids).await.unwrap(); + assert_eq!( + existing, + std::collections::HashSet::from(["sandbox-first".to_string(), "sandbox-last".to_string()]) + ); + assert!(store.existing_ids("sandbox", &[]).await.unwrap().is_empty()); +} + #[tokio::test] async fn file_backed_sqlite_bulk_delete_allows_concurrent_control_reads() { use std::sync::Arc; diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index cb3ecd40dc..db787f19a2 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -5,6 +5,7 @@ use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Duration, Instant}; use openshell_core::proto::{SandboxLogLine, SandboxStreamEvent}; use openshell_ocsf::OCSF_TARGET; @@ -58,6 +59,8 @@ struct PerSandbox { /// 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, + /// Last publish or subscribe, for the orphan sweeper's idle check. + last_active: Instant, } impl PerSandbox { @@ -66,16 +69,27 @@ impl PerSandbox { sender: None, tail: VecDeque::new(), last_trimmed_seq: 0, + last_active: Instant::now(), } } fn subscribe(&mut self) -> broadcast::Receiver> { + self.last_active = Instant::now(); self.sender .get_or_insert_with(|| broadcast::channel(BROADCAST_CAPACITY).0) .subscribe() } + /// No live receivers and no activity for at least `min_idle`. + fn is_idle(&self, min_idle: Duration) -> bool { + self.sender + .as_ref() + .is_none_or(|sender| sender.receiver_count() == 0) + && self.last_active.elapsed() >= min_idle + } + fn push(&mut self, cursored: CursoredEvent, tail_cap: usize) { + self.last_active = Instant::now(); let cursored = Arc::new(cursored); if let Some(sender) = &self.sender { let _ = sender.send(Arc::clone(&cursored)); @@ -368,6 +382,32 @@ impl TracingLogBus { SeqAllocator::is_closed_locked(&self.seq.lock(), sandbox_id) } + /// Sandboxes buffered on either bus that are idle on every bus holding + /// them: no live receivers and no publish or subscribe for `min_idle`. + pub(crate) fn idle_sandbox_ids(&self, min_idle: Duration) -> Vec { + let mut idle: HashMap = self + .inner + .lock() + .expect("tracing bus lock poisoned") + .per_id + .iter() + .map(|(id, per)| (id.clone(), per.is_idle(min_idle))) + .collect(); + for (id, per) in &self + .platform_event_bus + .inner + .lock() + .expect("platform event bus lock poisoned") + .per_id + { + let entry = idle.entry(id.clone()).or_insert(true); + *entry &= per.is_idle(min_idle); + } + idle.into_iter() + .filter_map(|(id, idle)| idle.then_some(id)) + .collect() + } + /// Remove all bus entries for the given sandbox id, including the platform /// event bus that shares this bus's cursor allocator. /// From 921bb7e080d46d52103a41353f687c36a307a415 Mon Sep 17 00:00:00 2001 From: divesh Date: Tue, 10 Nov 2026 18:28:28 -0800 Subject: [PATCH 2/2] fix(server): sweep watched buffers of deleted sandboxes (#5401) A logs-only WatchSandbox never learns of a deletion served by another replica, so skipping buffers with live receivers kept them forever and left the watch hanging. Check every quiet buffer; closing one ends its watch streams. Signed-off-by: divesh --- crates/openshell-server/src/bus_sweeper.rs | 30 ++++++---- crates/openshell-server/src/grpc/sandbox.rs | 64 +++++++++++++++++++++ crates/openshell-server/src/tracing_bus.rs | 10 ++-- 3 files changed, 86 insertions(+), 18 deletions(-) diff --git a/crates/openshell-server/src/bus_sweeper.rs b/crates/openshell-server/src/bus_sweeper.rs index 445d826408..89c4446e90 100644 --- a/crates/openshell-server/src/bus_sweeper.rs +++ b/crates/openshell-server/src/bus_sweeper.rs @@ -24,7 +24,8 @@ use crate::tracing_bus::TracingLogBus; pub const SWEEP_INTERVAL: Duration = Duration::from_secs(30); /// Buffers active more recently than this are not checked. A sandbox that is -/// still publishing or watched almost certainly still exists. +/// still publishing almost certainly still exists. Watched buffers are checked +/// too, and closing one ends its watch streams. pub const MIN_IDLE: Duration = Duration::from_mins(2); /// Spawn the sweeper. Only multi-replica deployments need it: a single replica @@ -59,7 +60,7 @@ pub fn spawn(store: Arc, bus: TracingLogBus, mut shutdown_rx: watch::Rece /// A sandbox is closed only when its record is missing on two sweeps in a /// row; `suspects` carries the ids missing on the previous one. A store error /// closes nothing and leaves `suspects` unchanged. -async fn sweep_once( +pub async fn sweep_once( store: &Store, bus: &TracingLogBus, min_idle: Duration, @@ -165,32 +166,37 @@ mod tests { } #[tokio::test] - async fn active_buffers_are_not_checked() { + async fn recently_active_buffers_are_not_checked() { let store = crate::persistence::test_store().await; let bus = TracingLogBus::new(); publish_log(&bus, "sb-gone"); let mut suspects = HashSet::new(); - // Recently active: skipped until it has been idle for `min_idle`. for _ in 0..2 { let swept = sweep_once(&store, &bus, Duration::from_hours(1), &mut suspects).await; assert_eq!(swept.unwrap(), 0); } - - // A live receiver on either bus keeps the sandbox out of the sweep. - let platform_rx = bus.platform_event_bus.subscribe("sb-gone"); - for _ in 0..2 { - let swept = sweep_once(&store, &bus, Duration::ZERO, &mut suspects).await; - assert_eq!(swept.unwrap(), 0); - } assert!(!bus.is_closed("sb-gone")); + } - drop(platform_rx); + #[tokio::test] + async fn watched_buffers_of_deleted_sandbox_are_freed() { + let store = crate::persistence::test_store().await; + let bus = TracingLogBus::new(); + publish_log(&bus, "sb-gone"); + let mut log_rx = bus.subscribe("sb-gone"); + + let mut suspects = HashSet::new(); for expected in [0, 1] { let swept = sweep_once(&store, &bus, Duration::ZERO, &mut suspects).await; assert_eq!(swept.unwrap(), expected); } + assert!(bus.is_closed("sb-gone")); + assert!(matches!( + log_rx.recv().await, + Err(tokio::sync::broadcast::error::RecvError::Closed) + )); } #[tokio::test] diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 448cbbbe09..7a0290dc5a 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -5373,6 +5373,70 @@ mod tests { ); } + #[tokio::test] + async fn logs_only_watch_ends_when_sweeper_frees_sandbox_deleted_elsewhere() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("swept", 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, + ..Default::default() + }), + ) + .await + .unwrap(); + let mut stream = response.into_inner(); + + // A delivered log line proves the watch has subscribed. + seed_log_lines(&state, &id, 1); + loop { + let event = stream.next().await.unwrap().unwrap(); + if !event.cursor.is_empty() { + break; + } + } + + // Another replica deletes the record; this one never runs delete cleanup. + state + .store + .delete(Sandbox::object_type(), &id) + .await + .unwrap(); + let mut suspects = HashSet::new(); + for expected in [0, 1] { + let closed = crate::bus_sweeper::sweep_once( + &state.store, + &state.tracing_log_bus, + std::time::Duration::ZERO, + &mut suspects, + ) + .await + .unwrap(); + assert_eq!(closed, expected); + } + + let ended = tokio::time::timeout(std::time::Duration::from_secs(5), async { + while let Some(item) = stream.next().await { + if item.is_err() { + break; + } + } + }) + .await; + assert!( + ended.is_ok(), + "logs-only watch should end once its buffers are freed" + ); + } + /// 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 { diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index db787f19a2..23ddfee5fe 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -80,12 +80,10 @@ impl PerSandbox { .subscribe() } - /// No live receivers and no activity for at least `min_idle`. + /// No publish or subscribe for at least `min_idle`. Live receivers don't + /// count: a logs-only watch never learns of a deletion served elsewhere. fn is_idle(&self, min_idle: Duration) -> bool { - self.sender - .as_ref() - .is_none_or(|sender| sender.receiver_count() == 0) - && self.last_active.elapsed() >= min_idle + self.last_active.elapsed() >= min_idle } fn push(&mut self, cursored: CursoredEvent, tail_cap: usize) { @@ -383,7 +381,7 @@ impl TracingLogBus { } /// Sandboxes buffered on either bus that are idle on every bus holding - /// them: no live receivers and no publish or subscribe for `min_idle`. + /// them: no publish or subscribe for `min_idle`. pub(crate) fn idle_sandbox_ids(&self, min_idle: Duration) -> Vec { let mut idle: HashMap = self .inner