diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 754605d174..243098eb96 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/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_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_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::("sb-skip-busy") + .await + .unwrap() + .unwrap(); + assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning)); + let expired = runtime + .store + .get_message::("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 = (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::(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_id) + .await .unwrap() .unwrap(); + assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning)); } #[tokio::test] diff --git a/crates/openshell-server/src/compute/mutation_guard.rs b/crates/openshell-server/src/compute/mutation_guard.rs index e8733015d6..0d137885a5 100644 --- a/crates/openshell-server/src/compute/mutation_guard.rs +++ b/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, diff --git a/crates/openshell-server/src/compute/provisioning_deadline.rs b/crates/openshell-server/src/compute/provisioning_deadline.rs index 7b7608d38d..cb351e64d0 100644 --- a/crates/openshell-server/src/compute/provisioning_deadline.rs +++ b/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::(&record.id) diff --git a/crates/openshell-server/src/gateway_metrics.rs b/crates/openshell-server/src/gateway_metrics.rs index 16734df8b8..4945818d52 100644 --- a/crates/openshell-server/src/gateway_metrics.rs +++ b/crates/openshell-server/src/gateway_metrics.rs @@ -35,6 +35,8 @@ pub const RELAY_EXPIRED_TOTAL: &str = "openshell_server_relay_expired_total"; pub const ROUTED_REQUEST_ATTEMPTS_TOTAL: &str = "openshell_server_routed_request_attempts_total"; pub const MUTATION_LOCK_TIMEOUTS_TOTAL: &str = "openshell_server_mutation_lock_timeouts_total"; pub const MUTATION_LOCK_ERRORS_TOTAL: &str = "openshell_server_mutation_lock_errors_total"; +pub const PROVISIONING_DEADLINE_SKIPS_TOTAL: &str = + "openshell_server_provisioning_deadline_skips_total"; // Histograms (explicit buckets, see BUCKETED_HISTOGRAMS) pub const RELAY_CLAIM_DURATION_SECONDS: &str = "openshell_server_relay_claim_duration_seconds"; pub const PEER_REQUEST_DURATION_SECONDS: &str = "openshell_server_peer_request_duration_seconds"; @@ -48,6 +50,7 @@ const LABEL_GRPC_CODE: &str = "grpc_code"; const LABEL_RELAY_KIND: &str = "relay_kind"; const LABEL_ROUTE: &str = "route"; const LABEL_SCOPE: &str = "scope"; +const LABEL_OVERDUE: &str = "overdue"; /// Buckets for the new latency histograms, 1 ms to 15 s. The top buckets cover the 10 s relay /// claim and lock timeouts and the 15 s routed-relay wait. @@ -288,6 +291,11 @@ pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: Unit::Count, "Mutation lock acquisitions that failed without timing out, such as a lock connection PostgreSQL did not open." ); + describe_counter!( + PROVISIONING_DEADLINE_SKIPS_TOTAL, + Unit::Count, + "Times the provisioning deadline scan skipped a sandbox because its local mutation locks were busy." + ); describe_histogram!( RELAY_CLAIM_DURATION_SECONDS, Unit::Seconds, @@ -334,6 +342,10 @@ pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: counter!(MUTATION_LOCK_TIMEOUTS_TOTAL, LABEL_SCOPE => scope.label()).increment(0); counter!(MUTATION_LOCK_ERRORS_TOTAL, LABEL_SCOPE => scope.label()).increment(0); } + for overdue in [false, true] { + counter!(PROVISIONING_DEADLINE_SKIPS_TOTAL, LABEL_OVERDUE => overdue_label(overdue)) + .increment(0); + } if let Some(capacity) = mutation_lock_connections { describe_gauge!( MUTATION_LOCK_CONNECTIONS_IN_USE, @@ -447,6 +459,17 @@ pub fn record_lock_error(scope: LockScope) { counter!(MUTATION_LOCK_ERRORS_TOTAL, LABEL_SCOPE => scope.label()).increment(1); } +/// The provisioning deadline scan skipped a sandbox whose local mutation locks stayed busy. An +/// `overdue` sandbox was already past its deadline, so the skip delays its expiry. +pub fn record_provisioning_deadline_skip(overdue: bool) { + counter!(PROVISIONING_DEADLINE_SKIPS_TOTAL, LABEL_OVERDUE => overdue_label(overdue)) + .increment(1); +} + +const fn overdue_label(overdue: bool) -> &'static str { + if overdue { "true" } else { "false" } +} + /// Counts one local relay setup or outbound peer attempt exactly once, and times peer requests. /// A relay succeeds when the supervisor claims it, on either route. Dropping an unfinished /// timer (the caller gave up) records `local_error` / `cancelled`. @@ -637,6 +660,14 @@ mod tests { "openshell_server_mutation_lock_errors_total{scope=\"sandbox\"}", 0, ), + ( + "openshell_server_provisioning_deadline_skips_total{overdue=\"false\"}", + 0, + ), + ( + "openshell_server_provisioning_deadline_skips_total{overdue=\"true\"}", + 0, + ), ] { assert_eq!(metrics.value(series), Some(expected), "{series}"); } diff --git a/crates/openshell-server/src/persistence/mutation_lock.rs b/crates/openshell-server/src/persistence/mutation_lock.rs index 835e0b1867..9a8e6c2201 100644 --- a/crates/openshell-server/src/persistence/mutation_lock.rs +++ b/crates/openshell-server/src/persistence/mutation_lock.rs @@ -13,7 +13,7 @@ //! | providers and workspace-scoped profiles | S(global) X(workspace) | //! | one sandbox, admin or supervisor | S(global) S(workspace) X(sandbox) | //! | lifecycle, driver watch, reconcile (process-local only) | S(global) X(sandbox) | -//! | provisioning-deadline reconcile (process-local only) | S(global) S(workspace) X(sandbox) | +//! | provisioning-deadline reconcile (process-local only, skips a busy sandbox) | S(global) S(workspace) X(sandbox) | //! //! Ordering rules, which make the scheme deadlock-free: //! diff --git a/docs/observability/gateway-metrics.mdx b/docs/observability/gateway-metrics.mdx index 12bf904db1..f04e272243 100644 --- a/docs/observability/gateway-metrics.mdx +++ b/docs/observability/gateway-metrics.mdx @@ -200,6 +200,7 @@ Mutation locks: | `openshell_server_mutation_lock_hold_seconds` | Histogram | `scope` | Time a mutation lock was held, recorded when it is released, including requests cancelled while they held it. A conflicting acquisition times out after 10 seconds, so the gateway also logs `mutation lock held longer than the lock wait timeout` with `scope` and `held_ms` when it releases a lock held longer than that. | | `openshell_server_mutation_lock_timeouts_total` | Counter | `scope` | Acquisitions that waited longer than 10 seconds, including background work such as supervisor disconnect handling and provider credential refresh. API requests fail with `UNAVAILABLE` and reason `MUTATION_LOCK_TIMEOUT`, and [API Errors](/sdk/api-errors) describes how to retry them. A lock connection that PostgreSQL does not open, although at least a second of the wait remained, counts in `openshell_server_mutation_lock_errors_total` instead. | | `openshell_server_mutation_lock_errors_total` | Counter | `scope` | Acquisitions that failed without timing out: PostgreSQL did not open a lock connection, or a lock statement failed. API requests fail with `INTERNAL`. | +| `openshell_server_provisioning_deadline_skips_total` | Counter | `overdue` | Times the provisioning deadline check skipped a sandbox because another operation on this replica held its locks. The check runs about once a second, waits about 500 milliseconds in total for busy locks, skips sandboxes whose locks are still busy, and retries them on the next check, so a sandbox counts once for each check it misses. In a multi-replica deployment, only the replica that runs background reconciliation records skips. A steady rate with `overdue="true"` means expiry is delayed by a global policy, settings, or platform profile change, a provider or profile change in the sandbox's workspace, or another operation on the sandbox. The gateway logs each skip at debug level with `sandbox_id`. | | `openshell_server_mutation_lock_connections_in_use` | Gauge | None | PostgreSQL lock-pool connections checked out on this replica: holding mutation locks, waiting for a PostgreSQL lock, or being returned to the pool, which can take up to 5 seconds. Exported only with PostgreSQL. | | `openshell_server_mutation_lock_connections_capacity` | Gauge | None | Lock-pool size per replica, `4` unless [`database_lock_max_connections`](/how-it-works/gateways/configuration#database-pool) sets it. Exported only with PostgreSQL. | @@ -239,6 +240,8 @@ The labels take these values: `openshell_server_grpc_requests_total` uses that name for the numeric status code. - `scope` is `global`, `workspace`, or `sandbox`. +- `overdue` is `true` when the skipped sandbox was already past its + provisioning deadline, otherwise `false`. Gauges and counters exist from startup. They start at `0`, except `openshell_server_relay_pending_capacity` and diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index cd1cbed4ef..ef8f96b1f1 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -537,7 +537,9 @@ workspace-profile, and other replicas' global changes delay them only while another operation on the same sandbox waits behind the change, holding that sandbox's key. Provisioning-deadline expiry also takes its workspace key locally, so provider and workspace-profile changes on the -same replica can delay it. Each +same replica can delay it; +`openshell_server_provisioning_deadline_skips_total{overdue="true"}` counts +the checks that skipped an overdue sandbox. Each replica takes its advisory locks on a dedicated lock pool of PostgreSQL connections, one per guard, 4 by default (gateway config `database_lock_max_connections`), so at most that many guarded