fix(server): free deleted sandboxes' buffers on every replica

crates/openshell-server/src/bus_sweeper.rs

@@ -0,0 +1,217 @@
+// 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 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
+/// runs every delete cleanup itself.
+pub fn spawn(store: Arc<Store>, bus: TracingLogBus, mut shutdown_rx: watch::Receiver<bool>) {
+    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.
+pub async fn sweep_once(
+    store: &Store,
+    bus: &TracingLogBus,
+    min_idle: Duration,
+    suspects: &mut HashSet<String>,
+) -> PersistenceResult<usize> {
+    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 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();
+
+        for _ in 0..2 {
+            let swept = sweep_once(&store, &bus, Duration::from_hours(1), &mut suspects).await;
+            assert_eq!(swept.unwrap(), 0);
+        }
+        assert!(!bus.is_closed("sb-gone"));
+    }
+
+    #[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]
+    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"));
+    }
+}

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 {

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,15 +1283,20 @@ 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(),
             state.sandbox_watch_bus.clone(),
             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(),

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<T> = Result<T, PersistenceError>;
 /// 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<HashSet<String>> {
+        store_dispatch_traced!(self.existing_ids(object_type, ids))
+    }
+
     /// Count objects of a given type within a workspace.
     #[tracing::instrument(
         name = "store",