fix(server): bound the provisioning-deadline scan's lock wait

crates/openshell-server/src/compute/mod.rs

@@ -18876,40 +18876,283 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn provisioning_reconcile_waits_for_a_local_workspace_writer() {
+    async fn provisioning_reconcile_skips_a_sandbox_behind_a_workspace_writer() {
+        use crate::gateway_metrics::MetricsCapture;
+        use crate::persistence::MutationLockKey;
+        let metrics = MetricsCapture::install();
         let runtime = test_runtime(ControlledDriver::new()).await;
+        // Keys are taken in ascending order. With the sandbox key first, the
+        // scan holds X(sandbox) while it waits for S(workspace).
+        let workspace_key = MutationLockKey::Workspace("default").advisory_key();
+        let sandbox_id = (0..1000)
+            .map(|i| format!("sb-ws-writer-{i}"))
+            .find(|id| MutationLockKey::Sandbox(id).advisory_key() < workspace_key)
+            .expect("a sandbox key that sorts before the workspace key");
+        let sandbox_key = MutationLockKey::Sandbox(&sandbox_id).advisory_key();
         runtime
             .store
             .put_message(&sandbox_record(
-                "sb-ws-writer",
+                &sandbox_id,
                 "ws-writer",
                 SandboxPhase::Provisioning,
             ))
             .await
             .unwrap();
+        let now = openshell_core::time::now_ms();
+        runtime.reconcile_provisioning_deadlines(now).await.unwrap();
+
         // Reconcile re-derives configuration from provider and profile
-        // records, so it waits for their writers, which hold X(workspace).
+        // records, so it must not run under their writers, which hold
+        // X(workspace).
         let writer = runtime
             .mutation_guard(MutationScope::Workspace("default"))
             .await
             .unwrap();
-        let mut reconcile = {
+        let reconcile = {
             let runtime = runtime.clone();
-            let now = openshell_core::time::now_ms();
-            tokio::spawn(async move { runtime.reconcile_provisioning_deadlines(now).await })
+            tokio::spawn(async move {
+                runtime
+                    .reconcile_provisioning_deadlines(now + 300_000)
+                    .await
+            })
         };
-        assert!(
-            tokio::time::timeout(Duration::from_millis(100), &mut reconcile)
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while runtime.mutation_locks.key_references(sandbox_key) == 0
+                || runtime.mutation_locks.key_references(workspace_key) < 2
+            {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the scan holds the sandbox key and waits for the workspace key");
+
+        // Start, stop, and delete take the sandbox key; they must not wait
+        // for the writer through the scan.
+        let lifecycle = {
+            let runtime = runtime.clone();
+            let sandbox_id = sandbox_id.clone();
+            tokio::spawn(async move { drop(runtime.lock_sandbox_local(&sandbox_id).await) })
+        };
+        tokio::time::timeout(Duration::from_secs(5), lifecycle)
+            .await
+            .expect("a lifecycle lock proceeds while the writer holds the workspace")
+            .unwrap();
+        tokio::time::timeout(Duration::from_secs(5), reconcile)
+            .await
+            .expect("the scan skips the busy sandbox instead of waiting for the writer")
+            .unwrap()
+            .unwrap();
+        assert_eq!(runtime.mutation_locks.key_references(sandbox_key), 0);
+        assert_eq!(
+            metrics.value("openshell_server_provisioning_deadline_skips_total{overdue=\"true\"}"),
+            Some(1),
+            "the sandbox was past its deadline when the scan skipped it"
+        );
+        let skipped = runtime
+            .store
+            .get_message::<Sandbox>(&sandbox_id)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
+
+        drop(writer);
+        runtime
+            .reconcile_provisioning_deadlines(now + 300_000)
+            .await
+            .unwrap();
+        let expired = runtime
+            .store
+            .get_message::<Sandbox>(&sandbox_id)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(expired.phase(), i32::from(SandboxPhase::Error));
+    }
+
+    #[tokio::test]
+    async fn provisioning_reconcile_continues_past_a_skipped_sandbox() {
+        let runtime = test_runtime(ControlledDriver::new()).await;
+        // The scan orders by insert time, then name: the busy sandbox is stored
+        // first and named first, so a skip that ended the pass would leave the
+        // other sandbox alone.
+        let busy = sandbox_record("sb-skip-busy", "a-busy", SandboxPhase::Provisioning);
+        let mut other = sandbox_record("sb-skip-other", "b-other", SandboxPhase::Provisioning);
+        other.metadata.as_mut().unwrap().workspace = "alpha".to_string();
+        runtime.store.put_message(&busy).await.unwrap();
+        runtime.store.put_message(&other).await.unwrap();
+        let now = openshell_core::time::now_ms();
+        runtime.reconcile_provisioning_deadlines(now).await.unwrap();
+
+        let writer = runtime
+            .mutation_guard(MutationScope::Workspace("default"))
+            .await
+            .unwrap();
+        // Spawned: the lock-order check forbids nesting under the writer in this task.
+        let reconcile = {
+            let runtime = runtime.clone();
+            tokio::spawn(async move {
+                runtime
+                    .reconcile_provisioning_deadlines(now + 300_000)
+                    .await
+            })
+        };
+        tokio::time::timeout(Duration::from_secs(5), reconcile)
+            .await
+            .expect("the scan skips the busy sandbox instead of waiting for the writer")
+            .unwrap()
+            .unwrap();
+        let skipped = runtime
+            .store
+            .get_message::<Sandbox>("sb-skip-busy")
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
+        let expired = runtime
+            .store
+            .get_message::<Sandbox>("sb-skip-other")
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            expired.phase(),
+            i32::from(SandboxPhase::Error),
+            "a skipped sandbox must not end the pass"
+        );
+        drop(writer);
+    }
+
+    #[tokio::test]
+    async fn provisioning_reconcile_waits_once_per_pass_behind_a_busy_workspace() {
+        use super::provisioning_deadline::SCAN_LOCK_WAIT_BUDGET;
+        use crate::gateway_metrics::MetricsCapture;
+        let metrics = MetricsCapture::install();
+        let runtime = test_runtime(ControlledDriver::new()).await;
+        let ids: Vec<String> = (0..6).map(|i| format!("sb-budget-{i}")).collect();
+        for (i, id) in ids.iter().enumerate() {
+            runtime
+                .store
+                .put_message(&sandbox_record(
+                    id,
+                    &format!("budget-{i}"),
+                    SandboxPhase::Provisioning,
+                ))
                 .await
-                .is_err(),
-            "provisioning reconcile must wait for the workspace writer"
+                .unwrap();
+        }
+        let now = openshell_core::time::now_ms();
+        runtime.reconcile_provisioning_deadlines(now).await.unwrap();
+
+        // Every sandbox needs S(workspace), which the writer holds.
+        let writer = runtime
+            .mutation_guard(MutationScope::Workspace("default"))
+            .await
+            .unwrap();
+        let started = tokio::time::Instant::now();
+        // Spawned: the lock-order check forbids nesting under the writer in this task.
+        let reconcile = {
+            let runtime = runtime.clone();
+            tokio::spawn(async move { runtime.reconcile_provisioning_deadlines(now + 1_000).await })
+        };
+        tokio::time::timeout(Duration::from_secs(10), reconcile)
+            .await
+            .expect("the scan finishes while the writer holds the workspace")
+            .unwrap()
+            .unwrap();
+        let waited = started.elapsed();
+        assert!(
+            waited < SCAN_LOCK_WAIT_BUDGET * 4,
+            "one wait per sandbox would take {:?}; the pass took {waited:?}",
+            SCAN_LOCK_WAIT_BUDGET * 6
+        );
+        assert_eq!(
+            metrics.value("openshell_server_provisioning_deadline_skips_total{overdue=\"false\"}"),
+            Some(6)
         );
+        for id in &ids {
+            let skipped = runtime
+                .store
+                .get_message::<Sandbox>(id)
+                .await
+                .unwrap()
+                .unwrap();
+            assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
+        }
         drop(writer);
+    }
+
+    #[tokio::test]
+    async fn provisioning_reconcile_releases_the_workspace_behind_a_busy_sandbox() {
+        use crate::persistence::MutationLockKey;
+        let runtime = test_runtime(ControlledDriver::new()).await;
+        // With the workspace key first, the scan holds S(workspace) while it
+        // waits for X(sandbox), and a provider writer queues behind the scan.
+        let workspace_key = MutationLockKey::Workspace("default").advisory_key();
+        let sandbox_id = (0..1000)
+            .map(|i| format!("sb-busy-{i}"))
+            .find(|id| MutationLockKey::Sandbox(id).advisory_key() > workspace_key)
+            .expect("a sandbox key that sorts after the workspace key");
+        let sandbox_key = MutationLockKey::Sandbox(&sandbox_id).advisory_key();
+        runtime
+            .store
+            .put_message(&sandbox_record(
+                &sandbox_id,
+                "busy",
+                SandboxPhase::Provisioning,
+            ))
+            .await
+            .unwrap();
+        let now = openshell_core::time::now_ms();
+        runtime.reconcile_provisioning_deadlines(now).await.unwrap();
+
+        let lifecycle = runtime.lock_sandbox_local(&sandbox_id).await;
+        let reconcile = {
+            let runtime = runtime.clone();
+            tokio::spawn(async move {
+                runtime
+                    .reconcile_provisioning_deadlines(now + 300_000)
+                    .await
+            })
+        };
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while runtime.mutation_locks.key_references(workspace_key) == 0
+                || runtime.mutation_locks.key_references(sandbox_key) < 2
+            {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the scan holds the workspace key and waits for the sandbox key");
+
+        let writer = {
+            let runtime = runtime.clone();
+            tokio::spawn(async move {
+                drop(
+                    runtime
+                        .mutation_guard(MutationScope::Workspace("default"))
+                        .await
+                        .unwrap(),
+                );
+            })
+        };
+        tokio::time::timeout(Duration::from_secs(5), writer)
+            .await
+            .expect("a provider writer proceeds while the sandbox is busy")
+            .unwrap();
         tokio::time::timeout(Duration::from_secs(5), reconcile)
             .await
-            .expect("provisioning reconcile proceeds once the writer releases")
+            .expect("the scan skips the busy sandbox instead of waiting for it")
+            .unwrap()
+            .unwrap();
+        drop(lifecycle);
+        let skipped = runtime
+            .store
+            .get_message::<Sandbox>(&sandbox_id)
+            .await
             .unwrap()
             .unwrap();
+        assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
     }
 
     #[tokio::test]

crates/openshell-server/src/compute/mutation_guard.rs

@@ -452,6 +452,11 @@ impl ComputeRuntime {
     /// Local S(global) S(workspace) X(sandbox), for provisioning-deadline
     /// reconciliation, which re-derives configuration from provider and
     /// profile records and must not interleave with their local writers.
+    ///
+    /// The wait is unbounded. When the sandbox key sorts first, the caller
+    /// holds X(sandbox) while it waits behind a workspace writer, so a
+    /// background caller should bound the wait with a timeout; dropping the
+    /// future releases the keys already taken.
     pub(super) async fn lock_sandbox_local_in_workspace(
         &self,
         workspace: &str,

crates/openshell-server/src/compute/provisioning_deadline.rs

@@ -14,6 +14,16 @@ use serde::{Deserialize, Serialize};
 
 const REPAIR_WINDOW_MS: i64 = 300_000;
 
+/// How long one scan pass waits, in total, for sandboxes' local keys. A sandbox
+/// whose keys are still busy is skipped until the next pass. Keys are taken in
+/// ascending order, so the scan can hold X(sandbox) while it waits for
+/// S(workspace) behind a provider or profile writer, and the sandbox's
+/// lifecycle operations wait behind the scan. Once the budget is spent, the
+/// rest of the pass waits about a timer tick for each sandbox's keys, so a
+/// busy workspace or global key costs one wait per pass, not one per sandbox.
+/// Well under the mutation lock timeout.
+pub(super) const SCAN_LOCK_WAIT_BUDGET: std::time::Duration = std::time::Duration::from_millis(500);
+
 /// Opaque identity of a committed effective configuration change. A -> B -> A
 /// must use three different IDs even if the first and last content hashes match.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -456,6 +466,7 @@ impl super::ComputeRuntime {
             .collect_records(Sandbox::object_type(), ObjectListQuery::AllWorkspaces)
             .await
             .map_err(|error| error.to_string())?;
+        let mut lock_wait_left = SCAN_LOCK_WAIT_BUDGET;
         for record in records {
             let candidate =
                 Sandbox::decode(record.payload.as_slice()).map_err(|error| error.to_string())?;
@@ -469,9 +480,35 @@ impl super::ComputeRuntime {
             // Expiration can fence Starting while its driver RPC owns the
             // lifecycle gate. Cleanup waits for that gate; Error never waits
             // for compute I/O, matching the existing driver-observation fence.
-            let local_guard = self
-                .lock_sandbox_local_in_workspace(candidate.object_workspace(), &record.id)
-                .await;
+            // A timed-out acquisition releases the keys it already took. With
+            // no budget left, the timeout still polls the keys before its timer
+            // fires, and unlike `now_or_never` it yields instead of skipping
+            // free keys when the task's coop budget is spent.
+            let wait_started = tokio::time::Instant::now();
+            let acquired = tokio::time::timeout(
+                lock_wait_left,
+                self.lock_sandbox_local_in_workspace(candidate.object_workspace(), &record.id),
+            )
+            .await;
+            lock_wait_left = lock_wait_left.saturating_sub(wait_started.elapsed());
+            let Ok(local_guard) = acquired else {
+                // Brief contention is routine; a skip past the deadline delays
+                // expiry. The counter shows both without a warning per pass.
+                let overdue = candidate
+                    .status
+                    .as_ref()
+                    .and_then(|status| status.provisioning.as_ref())
+                    .and_then(|record| record.deadline.as_ref())
+                    .and_then(|deadline| timestamp_to_millis(deadline).ok())
+                    .is_some_and(|deadline| deadline <= now_ms);
+                crate::gateway_metrics::record_provisioning_deadline_skip(overdue);
+                tracing::debug!(
+                    sandbox_id = %record.id,
+                    overdue,
+                    "Provisioning deadline scan skipped a sandbox whose locks are busy"
+                );
+                continue;
+            };
             let Some(mut current) = self
                 .store
                 .get_message::<Sandbox>(&record.id)