From bed155f762eec74bd1ca16480e06eecbdd971f2e Mon Sep 17 00:00:00 2001 From: Matt Jane Date: Mon, 26 Oct 2026 23:24:21 -0700 Subject: [PATCH 1/2] fix(compute): take over driver operations abandoned by a stopped gateway A claimed driver operation stays pending until the process that claimed it records the driver's response. When that gateway process stopped first, as in a node rotation that interrupted an automatic restart, nothing could ever settle it: the restart sweep, retries and deletion all refused the sandbox ("previous driver operation is still pending"), and the agent stayed down until the store rows were deleted by hand. Upstream tracks the same bug as NVIDIA/OpenShell#4409. An in-process driver's operation cannot outlive its process, apart from requests already sent, which the backend commits or rejects within its own request timeout. So: - Each claim records its owner, a per-process incarnation (SandboxProvisioning.driver_operation_owner), and every gateway process refreshes a presence row every 10 seconds. - A process that cannot refresh its presence for a minute fences itself and abandons the operations it owns, before dispatch if already fenced. - The reconciler takes over a pending operation once its owner's presence is 3 minutes old, or after watching it for 3 minutes when the owner left no presence (operations claimed before this change). Taking over settles it with an unknown outcome and rotates its ID, so a late response is ignored. - An interrupted automatic restart becomes due again, and the restart sweep starts a fresh runtime generation, stopping what the abandoned operation left behind first. Drivers running in their own process keep the original contract. Signed-off-by: Matt Jane Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_014dzrkBzoiGwdAtxRmRxtR1 --- crates/openshell-cli/src/run.rs | 6 + crates/openshell-server/src/compute/mod.rs | 209 +++++++++- .../src/compute/operation_owner.rs | 387 ++++++++++++++++++ .../src/compute/provisioning_operation.rs | 22 +- crates/openshell-server/src/storage_proto.rs | 7 +- docs/how-it-works/sandboxes/overview.mdx | 4 +- proto/openshell.proto | 9 +- sdk/go/proto/openshellv1/openshell.pb.go | 25 +- 8 files changed, 655 insertions(+), 14 deletions(-) create mode 100644 crates/openshell-server/src/compute/operation_owner.rs diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index c5314e7454..f235cb47e7 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -3082,6 +3082,7 @@ fn sandbox_to_json(sandbox: &Sandbox) -> serde_json::Value { "cleanup_retry_time": record.cleanup_retry_time.as_ref().map(ToString::to_string), "driver_operation_pending": record.driver_operation_pending, "driver_operation_id": record.driver_operation_id, + "driver_operation_owner": record.driver_operation_owner, })); serde_json::json!({ "id": sandbox.object_id(), @@ -8464,6 +8465,7 @@ mod tests { Some(openshell_core::proto::SandboxProvisioning { driver_operation_pending: pending, driver_operation_id: "operation-1".into(), + driver_operation_owner: "gateway-0/process".into(), ..Default::default() }); assert_eq!( @@ -8474,6 +8476,10 @@ mod tests { super::sandbox_to_json(&sandbox)["provisioning"]["driver_operation_id"], "operation-1" ); + assert_eq!( + super::sandbox_to_json(&sandbox)["provisioning"]["driver_operation_owner"], + "gateway-0/process" + ); } } diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 104ee001a5..d6287a8c0d 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -5,6 +5,7 @@ pub mod driver_config; pub mod lease; +mod operation_owner; pub mod provisioning_deadline; mod provisioning_operation; pub mod rootfs_tar; @@ -687,6 +688,8 @@ pub struct ComputeRuntime { Arc>>>, restart_notify: Arc, ssh_identities: Arc>, + /// This process as the owner of claimed driver operations. + operation_owner: Arc, } pub struct SandboxSyncGuard { @@ -781,6 +784,7 @@ impl ComputeRuntime { supervisor_sessions, sync_lock: Arc::new(Mutex::new(())), lifecycle_gates: Arc::new(LifecycleGateRegistry::default()), + operation_owner: Arc::new(operation_owner::OperationOwner::new(&lease::replica_id())), replica_id: lease::replica_id(), rootfs_tar_staging, restart_authority: Arc::new(OnceLock::new()), @@ -3137,6 +3141,11 @@ impl ComputeRuntime { ) { let _ = self.restart_authority.set(authority); let runtime = Arc::new(self.clone()); + tokio::spawn( + self.operation_owner + .clone() + .heartbeat_loop(self.store.clone(), shutdown_rx.clone()), + ); if self.store.is_single_replica() { let deadline_runtime = runtime.clone(); let deadline_shutdown = shutdown_rx.clone(); @@ -3160,6 +3169,13 @@ impl ComputeRuntime { } reconcile_runtime.reconcile_loop(reconcile_shutdown).await; }); + let takeover_runtime = runtime.clone(); + let takeover_shutdown = shutdown_rx.clone(); + tokio::spawn(async move { + takeover_runtime + .operation_takeover_loop(takeover_shutdown) + .await; + }); tokio::spawn(async move { runtime.restart_loop(shutdown_rx).await; }); @@ -3917,6 +3933,12 @@ impl ComputeRuntime { runtime.restart_loop(cancel_rx).await; }); + let runtime = self.clone(); + let cancel_rx = cancel_tx.subscribe(); + let takeover_handle = tokio::spawn(async move { + runtime.operation_takeover_loop(cancel_rx).await; + }); + loop { tokio::select! { () = tokio::time::sleep(LEASE_RENEWAL_INTERVAL) => { @@ -3945,6 +3967,7 @@ impl ComputeRuntime { let _ = reconcile_handle.await; let _ = deadline_handle.await; let _ = restart_handle.await; + let _ = takeover_handle.await; return; } } @@ -3956,6 +3979,7 @@ impl ComputeRuntime { let _ = reconcile_handle.await; let _ = deadline_handle.await; let _ = restart_handle.await; + let _ = takeover_handle.await; info!(replica = %lease.replica_id(), "reconciler lease lost — returning to standby"); } @@ -4145,7 +4169,7 @@ impl ComputeRuntime { { identity.write(&mut metadata.annotations); } - provisioning_operation::claim_record(sandbox); + provisioning_operation::claim_record(sandbox, self.operation_owner.id()); let status = sandbox.status.get_or_insert_with(Default::default); set_next_restart_at_ms(status, 0); upsert_ready_condition( @@ -7605,6 +7629,7 @@ pub fn new_test_runtime_with_driver( sync_lock: Arc::new(Mutex::new(())), lifecycle_gates: Arc::new(LifecycleGateRegistry::default()), replica_id: "test-replica".to_string(), + operation_owner: Arc::new(operation_owner::OperationOwner::new("test-replica")), rootfs_tar_staging: Arc::new(rootfs_tar::RootfsTarStagingRegistry::disabled()), restart_authority: Arc::new(OnceLock::new()), restart_notify: Arc::new(Notify::new()), @@ -8699,6 +8724,7 @@ mod tests { sync_lock: Arc::new(Mutex::new(())), lifecycle_gates: Arc::new(LifecycleGateRegistry::default()), replica_id: "test-replica".to_string(), + operation_owner: Arc::new(operation_owner::OperationOwner::new("test-replica")), rootfs_tar_staging: Arc::new(rootfs_tar::RootfsTarStagingRegistry::disabled()), restart_authority: Arc::new(OnceLock::new()), restart_notify: Arc::new(Notify::new()), @@ -18930,4 +18956,185 @@ mod tests { assert_eq!(starting.object_id(), "sb-ttl"); assert_eq!(starting.phase(), i32::from(SandboxPhase::Starting)); } + + /// A runtime-lost restart that is scheduled but not yet claimed. + fn lost_runtime_restart(id: &str, name: &str) -> Sandbox { + let mut sandbox = sandbox_record(id, name, SandboxPhase::Ready); + sandbox.status = Some(SandboxStatus { + phase: SandboxPhase::Ready as i32, + main_process_instance_id: "instance-1".into(), + provisioning: Some(provisioning_deadline::new_record( + openshell_core::time::now_ms(), + )), + ..Default::default() + }); + apply_runtime_lost_restart(&mut sandbox, openshell_core::time::now_ms()); + sandbox + } + + /// A restart claimed by `owner`, whose gateway process then stopped. + fn abandoned_restart(id: &str, name: &str, owner: &str) -> Sandbox { + let mut sandbox = lost_runtime_restart(id, name); + set_next_restart_at_ms(sandbox.status.as_mut().unwrap(), 0); + provisioning_operation::claim_record(&mut sandbox, owner); + sandbox + } + + async fn stored(runtime: &ComputeRuntime, id: &str) -> Sandbox { + runtime + .store + .get_message::(id) + .await + .unwrap() + .unwrap() + } + + fn operation_owner_of(sandbox: &Sandbox) -> &str { + &sandbox + .status + .as_ref() + .unwrap() + .provisioning + .as_ref() + .unwrap() + .driver_operation_owner + } + + #[tokio::test] + async fn restart_abandoned_by_a_stopped_gateway_is_taken_over_and_retried() { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver.clone()).await; + let sandbox = abandoned_restart("sb-orphan", "orphan", "gateway-0/stopped"); + runtime.store.put_message(&sandbox).await.unwrap(); + + // Without a takeover nothing ever retries the abandoned restart. + runtime.restart_due_sandboxes().await.unwrap(); + assert_eq!(driver.stop_calls(), 0); + + // The owner left no presence. It may not have heartbeated yet, so + // its operation is held for a full takeover delay first. + let mut unknown = operation_owner::UnknownOwners::default(); + let first_seen = Instant::now(); + let now_ms = openshell_core::time::now_ms(); + for elapsed in [ + Duration::ZERO, + operation_owner::ORPHAN_AFTER + .checked_sub(Duration::from_secs(1)) + .unwrap(), + ] { + runtime + .take_over_orphaned_operations(&mut unknown, first_seen + elapsed, now_ms) + .await + .unwrap(); + assert!(provisioning_deadline::driver_operation_pending( + &stored(&runtime, "sb-orphan").await + )); + } + runtime + .take_over_orphaned_operations( + &mut unknown, + first_seen + operation_owner::ORPHAN_AFTER, + now_ms, + ) + .await + .unwrap(); + let taken = stored(&runtime, "sb-orphan").await; + assert!(!provisioning_deadline::driver_operation_pending(&taken)); + assert!(!provisioning_operation::same_operation(&taken, &sandbox)); + assert_eq!(operation_owner_of(&taken), ""); + assert!(is_automatic_restart_transition(&taken)); + assert!(restart_is_due( + taken.status.as_ref().unwrap(), + openshell_core::time::now_ms() + )); + + runtime.restart_due_sandboxes().await.unwrap(); + assert_eq!(driver.stop_calls(), 1); + assert_eq!(driver.start_calls(), 1); + let restarted = stored(&runtime, "sb-orphan").await; + assert!(!provisioning_deadline::driver_operation_pending(&restarted)); + assert_eq!(restarted.phase(), SandboxPhase::Starting as i32); + } + + #[tokio::test] + async fn operation_is_taken_over_once_its_owner_stops_heartbeating() { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver.clone()).await; + let owner = operation_owner::OperationOwner::new("gateway-0"); + owner.heartbeat(&runtime.store).await.unwrap(); + let sandbox = abandoned_restart("sb-owned", "owned", owner.id()); + runtime.store.put_message(&sandbox).await.unwrap(); + assert_eq!( + runtime + .delete_sandbox("default", "owned") + .await + .unwrap_err() + .code(), + Code::FailedPrecondition + ); + + let mut unknown = operation_owner::UnknownOwners::default(); + let now = Instant::now(); + let heartbeat_ms = openshell_core::time::now_ms(); + let orphan_after_ms = i64::try_from(operation_owner::ORPHAN_AFTER.as_millis()).unwrap(); + runtime + .take_over_orphaned_operations( + &mut unknown, + now, + heartbeat_ms + orphan_after_ms - 1_000, + ) + .await + .unwrap(); + assert!(provisioning_deadline::driver_operation_pending( + &stored(&runtime, "sb-owned").await + )); + + runtime + .take_over_orphaned_operations(&mut unknown, now, heartbeat_ms + orphan_after_ms) + .await + .unwrap(); + assert!(!provisioning_deadline::driver_operation_pending( + &stored(&runtime, "sb-owned").await + )); + runtime.delete_sandbox("default", "owned").await.unwrap(); + } + + #[tokio::test] + async fn fenced_gateway_abandons_the_driver_operation_it_owns() { + let driver = ControlledDriver::new(); + driver.block_start(); + let runtime = test_runtime(driver.clone()).await; + runtime + .store + .put_message(&lost_runtime_restart("sb-fenced", "fenced")) + .await + .unwrap(); + let restarting = runtime.clone(); + let restart = + tokio::spawn(async move { restarting.restart_sandbox_runtime("sb-fenced").await }); + driver.start_started.notified().await; + let pending = wait_driver_pending(&runtime, "sb-fenced", true).await; + assert_eq!(operation_owner_of(&pending), runtime.operation_owner.id()); + + // A process never takes over its own operations, however old. + runtime + .take_over_orphaned_operations( + &mut operation_owner::UnknownOwners::default(), + Instant::now() + operation_owner::ORPHAN_AFTER * 2, + openshell_core::time::now_ms() + 3_600_000, + ) + .await + .unwrap(); + assert!(provisioning_deadline::driver_operation_pending( + &stored(&runtime, "sb-fenced").await + )); + + runtime.operation_owner.set_fenced(true); + let error = restart.await.unwrap().unwrap_err(); + assert!(error.contains("lost driver operation ownership"), "{error}"); + let settled = wait_driver_pending(&runtime, "sb-fenced", false).await; + assert!(provisioning_operation::same_operation(&settled, &pending)); + assert!(next_restart_at_ms(settled.status.as_ref().unwrap()) > 0); + assert_eq!(driver.start_calls(), 1); + } } diff --git a/crates/openshell-server/src/compute/operation_owner.rs b/crates/openshell-server/src/compute/operation_owner.rs new file mode 100644 index 0000000000..de0831cd39 --- /dev/null +++ b/crates/openshell-server/src/compute/operation_owner.rs @@ -0,0 +1,387 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Take over driver operations whose owning gateway process has stopped. +//! +//! A claimed driver operation stays pending until its owner records the +//! driver's response. If the owning process exits first, nothing else may +//! settle it, so the sandbox can never be retried, restarted or deleted. +//! +//! With an in-process driver the operation cannot outlive its process, apart +//! from API requests already sent, which the backend commits or rejects within +//! its own request timeout. Each process therefore heartbeats a presence row +//! keyed by a per-process incarnation, so a restarted container is a new owner +//! even though it keeps its replica ID. A process that cannot heartbeat for +//! [`FENCE_AFTER`] abandons its own driver operations. The reconciler takes +//! over a pending operation once its owner's presence is [`ORPHAN_AFTER`] old; +//! the gap between the two covers requests that were already in flight. +//! +//! Taking over settles the operation with an unknown outcome and rotates its +//! ID, as a driver error would. An automatic restart becomes due again, and +//! the restart sweep starts a fresh runtime generation, which stops whatever +//! the abandoned operation left behind first. +//! +//! Drivers running in a separate process can keep working after the gateway +//! stops, so their pending operations keep the original contract. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use openshell_core::proto::{Sandbox, SandboxCondition, SandboxPhase}; +use openshell_core::{ObjectId, ObjectName}; +use prost::Message; +use tokio::sync::watch; +use tracing::{info, warn}; + +use super::{ + ComputeRuntime, emit_reconcile_audit, is_automatic_restart_transition, next_restart_at_ms, + provisioning_deadline, provisioning_operation, sandbox_provisioning_attempt_id, + sandbox_resource_version, set_next_restart_at_ms, upsert_ready_condition, +}; +use crate::persistence::{ObjectType, PersistenceError, Store, WriteCondition}; + +const PRESENCE_OBJECT_TYPE: &str = "gateway_process_presence"; +const PAGE_SIZE: u32 = 1000; + +/// How often each gateway process refreshes its presence. +pub(super) const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10); +/// A process that has not refreshed its presence for this long abandons the +/// driver operations it owns. +pub(super) const FENCE_AFTER: Duration = Duration::from_mins(1); +/// A pending operation whose owner has not refreshed its presence for this +/// long is taken over. Exceeds [`FENCE_AFTER`] by more than the Kubernetes API +/// server's request timeout. +pub(super) const ORPHAN_AFTER: Duration = Duration::from_mins(3); +/// Presence rows of processes gone for this long are deleted. +const PRESENCE_RETENTION: Duration = Duration::from_hours(24); + +fn millis(duration: Duration) -> i64 { + i64::try_from(duration.as_millis()).unwrap_or(i64::MAX) +} + +/// This gateway process's identity as a driver operation owner. +#[derive(Debug)] +pub(super) struct OperationOwner { + id: String, + fenced: watch::Sender, +} + +impl OperationOwner { + pub(super) fn new(replica_id: &str) -> Self { + Self { + id: format!("{replica_id}/{}", uuid::Uuid::new_v4()), + fenced: watch::Sender::new(false), + } + } + + pub(super) fn id(&self) -> &str { + &self.id + } + + pub(super) fn is_fenced(&self) -> bool { + *self.fenced.borrow() + } + + pub(super) fn set_fenced(&self, fenced: bool) { + self.fenced.send_if_modified(|current| { + let changed = *current != fenced; + *current = fenced; + changed + }); + } + + /// Resolves once this process may no longer drive operations it owns. + pub(super) async fn wait_fenced(&self) { + let mut fenced = self.fenced.subscribe(); + // The sender lives as long as `self`, so the wait only ends on fencing. + let _ = fenced.wait_for(|fenced| *fenced).await; + } + + pub(super) async fn heartbeat(&self, store: &Store) -> Result<(), PersistenceError> { + store + .put_if( + PRESENCE_OBJECT_TYPE, + &self.id, + &self.id, + "", + b"{}", + None, + WriteCondition::Unconditional, + ) + .await + .map(|_| ()) + } + + /// Refresh this process's presence until shutdown, fencing its operations + /// while the store cannot be reached. The row is left behind at shutdown: + /// its age is what lets the reconciler take over sooner. + pub(super) async fn heartbeat_loop( + self: Arc, + store: Arc, + mut shutdown: watch::Receiver, + ) { + let mut last_heartbeat = Instant::now(); + loop { + match self.heartbeat(&store).await { + Ok(()) => { + last_heartbeat = Instant::now(); + if self.is_fenced() { + info!(owner = %self.id, "Driver operation ownership restored"); + } + self.set_fenced(false); + } + Err(error) => { + warn!(owner = %self.id, %error, "Failed to refresh gateway process presence"); + if last_heartbeat.elapsed() >= FENCE_AFTER && !self.is_fenced() { + warn!( + owner = %self.id, + "Abandoning owned driver operations; the reconciler will take them over" + ); + self.set_fenced(true); + } + } + } + tokio::select! { + () = tokio::time::sleep(HEARTBEAT_INTERVAL) => {} + _ = shutdown.changed() => return, + } + } + } +} + +/// Pending operations whose owner's presence could not be found, keyed by +/// sandbox and operation ID, with when this process first saw them. Covers +/// operations claimed before owners were recorded and owners that have not +/// written their first heartbeat yet. +#[derive(Debug, Default)] +pub(super) struct UnknownOwners(HashMap<(String, String), Instant>); + +impl ComputeRuntime { + pub(super) async fn operation_takeover_loop( + self: Arc, + mut cancel: watch::Receiver, + ) { + let mut unknown = UnknownOwners::default(); + loop { + if let Err(error) = self + .take_over_orphaned_operations( + &mut unknown, + Instant::now(), + openshell_core::time::now_ms(), + ) + .await + { + warn!(%error, "Driver operation takeover sweep failed"); + } + tokio::select! { + () = tokio::time::sleep(HEARTBEAT_INTERVAL) => {} + _ = cancel.changed() => return, + } + } + } + + /// `now` measures how long an owner has been unknown on this process's + /// clock; `now_ms` is compared with the store's heartbeat times. + pub(super) async fn take_over_orphaned_operations( + &self, + unknown: &mut UnknownOwners, + now: Instant, + now_ms: i64, + ) -> Result<(), String> { + if self.driver_process.is_some() { + return Ok(()); + } + let presence = self.read_presence(now_ms).await?; + let mut still_unknown = HashSet::new(); + let mut offset = 0; + loop { + let records = self + .store + .list_by_type(Sandbox::object_type(), PAGE_SIZE, offset) + .await + .map_err(|error| error.to_string())?; + let page_len = records.len(); + for record in records { + let Ok(sandbox) = Sandbox::decode(record.payload.as_slice()) else { + continue; + }; + let Some(provisioning) = sandbox + .status + .as_ref() + .and_then(|status| status.provisioning.as_ref()) + else { + continue; + }; + if !provisioning.driver_operation_pending + || sandbox_provisioning_attempt_id(&sandbox).is_none() + { + continue; + } + let owner = provisioning.driver_operation_owner.as_str(); + if owner == self.operation_owner.id() { + continue; + } + let orphaned = if let Some(&updated_at_ms) = presence.get(owner) { + now_ms.saturating_sub(updated_at_ms) >= millis(ORPHAN_AFTER) + } else { + let key = ( + sandbox.object_id().to_string(), + provisioning.driver_operation_id.clone(), + ); + let first_seen = *unknown.0.entry(key.clone()).or_insert(now); + still_unknown.insert(key); + now.saturating_duration_since(first_seen) >= ORPHAN_AFTER + }; + if orphaned && let Err(error) = self.take_over_operation(&sandbox).await { + warn!( + sandbox_id = %sandbox.object_id(), + sandbox_name = %sandbox.object_name(), + %error, + "Failed to take over abandoned driver operation" + ); + } + } + if page_len < PAGE_SIZE as usize { + break; + } + offset += u32::try_from(page_len).expect("takeover page length is capped"); + } + unknown.0.retain(|key, _| still_unknown.contains(key)); + Ok(()) + } + + /// Latest heartbeat per owner. Deletes presence rows past retention. + async fn read_presence(&self, now_ms: i64) -> Result, String> { + let mut presence = HashMap::new(); + let mut offset = 0; + loop { + let records = self + .store + .list_by_type(PRESENCE_OBJECT_TYPE, PAGE_SIZE, offset) + .await + .map_err(|error| error.to_string())?; + let page_len = records.len(); + for record in records { + if now_ms.saturating_sub(record.updated_at_ms) >= millis(PRESENCE_RETENTION) { + // Another sweep or a late heartbeat may win; either is fine. + let _ = self + .store + .delete_if(PRESENCE_OBJECT_TYPE, &record.id, record.resource_version) + .await; + } + presence.insert(record.id, record.updated_at_ms); + } + if page_len < PAGE_SIZE as usize { + break; + } + offset += u32::try_from(page_len).expect("presence page length is capped"); + } + Ok(presence) + } + + /// Settle an abandoned operation with an unknown outcome. Holds the + /// lifecycle gate so it cannot interleave with a local lifecycle change. + async fn take_over_operation(&self, observed: &Sandbox) -> Result<(), String> { + let sandbox_id = observed.object_id(); + let _lifecycle_guard = self.lifecycle_gates.lock_for(sandbox_id).await; + let Some(current) = self + .store + .get_message::(sandbox_id) + .await + .map_err(|error| error.to_string())? + else { + return Ok(()); + }; + if !provisioning_deadline::driver_operation_pending(¤t) + || !provisioning_operation::same_operation(¤t, observed) + { + return Ok(()); + } + let abandoned_by = current + .status + .as_ref() + .and_then(|status| status.provisioning.as_ref()) + .map(|record| record.driver_operation_owner.clone()) + .unwrap_or_default(); + let now_ms = openshell_core::time::now_ms(); + let restart = is_automatic_restart_transition(¤t); + let updated = match self + .store + .update_message_cas::( + sandbox_id, + sandbox_resource_version(¤t), + |sandbox| { + if let Some(record) = sandbox + .status + .as_mut() + .and_then(|status| status.provisioning.as_mut()) + { + record.driver_operation_pending = false; + // A late commit from the abandoned operation would + // carry its ID; a new one makes it stale. + record.driver_operation_id = uuid::Uuid::new_v4().to_string(); + record.driver_operation_owner.clear(); + // As in settlement: the abandoned operation may have + // created compute after any earlier cleanup ran. + record.cleanup_completed_time = None; + } + // A claimed restart (no restart time) is retried now with a + // fresh generation. An armed one is already due at its + // readiness deadline. + if restart + && sandbox + .status + .as_ref() + .is_some_and(|status| next_restart_at_ms(status) == 0) + { + let status = sandbox.status.get_or_insert_with(Default::default); + set_next_restart_at_ms(status, now_ms); + upsert_ready_condition( + &mut sandbox.status, + SandboxCondition { + r#type: "Ready".to_string(), + status: "False".to_string(), + reason: "SandboxRestartFailed".to_string(), + message: "Sandbox restart was interrupted when its gateway \ + process stopped; retrying with a new runtime" + .to_string(), + transition_time: None, + }, + ); + } + }, + ) + .await + { + Ok(updated) => updated, + // Someone else changed the sandbox; the next sweep re-evaluates it. + Err(PersistenceError::Conflict { .. }) => return Ok(()), + Err(error) => return Err(error.to_string()), + }; + warn!( + sandbox_id, + sandbox_name = %updated.object_name(), + abandoned_by = %abandoned_by, + phase = ?SandboxPhase::try_from(updated.phase()).unwrap_or(SandboxPhase::Unknown), + "Took over a driver operation abandoned by a stopped gateway process" + ); + emit_reconcile_audit( + openshell_ocsf::EntityActivityId::Update, + sandbox_id, + updated.object_name(), + format!( + "sandbox {} driver operation taken over from a stopped gateway process", + updated.object_name() + ), + vec![ + ("operation", serde_json::Value::from("take_over_operation")), + ("abandoned_by", serde_json::Value::from(abandoned_by)), + ], + ); + self.sandbox_index.update_from_sandbox(&updated); + self.sandbox_watch_bus.notify(sandbox_id); + self.restart_notify.notify_one(); + Ok(()) + } +} diff --git a/crates/openshell-server/src/compute/provisioning_operation.rs b/crates/openshell-server/src/compute/provisioning_operation.rs index 764c136cc4..30afe3d132 100644 --- a/crates/openshell-server/src/compute/provisioning_operation.rs +++ b/crates/openshell-server/src/compute/provisioning_operation.rs @@ -39,7 +39,8 @@ fn operation_id(sandbox: &Sandbox) -> Option<&str> { /// Called only inside a CAS that has checked the prior operation is settled. /// Legacy rows without an attempt retain their existing lifecycle contract. -pub(super) fn claim_record(sandbox: &mut Sandbox) { +/// `owner` is the claiming process; see [`super::operation_owner`]. +pub(super) fn claim_record(sandbox: &mut Sandbox, owner: &str) { if let Some(record) = sandbox .status .as_mut() @@ -47,6 +48,7 @@ pub(super) fn claim_record(sandbox: &mut Sandbox) { { record.driver_operation_pending = true; record.driver_operation_id = uuid::Uuid::new_v4().to_string(); + record.driver_operation_owner = owner.to_string(); } } @@ -129,7 +131,18 @@ impl ComputeRuntime { // Dropping a JoinHandle detaches its task. Do not abort it on a monitor // error or deadline: the remote side may still commit the request. let worker = async move { - let result = operation.await; + // A fenced process can no longer prove it owns the operation, and + // the reconciler may take it over. Stop driving it, before dispatch + // if already fenced. Requests already sent may still commit; the + // takeover delay covers them. + let fenced = runtime.operation_owner.clone(); + let result = tokio::select! { + biased; + () = fenced.wait_fenced() => Err(Status::unavailable( + "gateway process lost driver operation ownership; outcome remains unknown", + )), + result = operation => result, + }; let settled = if tracked { let settled = runtime .settle_provisioning_operation(&owned_attempt) @@ -233,7 +246,7 @@ impl ComputeRuntime { .update_message_cas::( starting.object_id(), sandbox_resource_version(¤t), - claim_record, + |sandbox| claim_record(sandbox, self.operation_owner.id()), ) .await { @@ -254,7 +267,8 @@ impl ComputeRuntime { async fn settle_provisioning_operation(&self, starting: &Sandbox) -> Result { // Keep the observed response in this task across transient persistence // failures. A process crash still leaves the durable claim intact; - // neither lease expiry nor backend absence can safely clear it. + // neither lease expiry nor backend absence can safely clear it. Only + // the owner's stopped heartbeat can (see `operation_owner`). loop { let result = self.try_settle_provisioning_operation(starting).await; match result { diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index dd89406cb7..d27cc90358 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -146,10 +146,13 @@ mod tests { // Legacy payloads decode empty owners; the gateway rebuilds their authority // from effective policy rather than trusting persisted owner stamps. // Operator credential export adds four public messages and no durable types. + // SandboxProvisioning.driver_operation_owner adds one string. Legacy + // payloads decode an empty owner, which the takeover sweep watches for the + // full takeover delay before treating it as gone. const PUBLIC_RPC_SCHEMA_SHA256: &str = - "5ae8334d9f6212a00c38ba965ff62e567d2846b15f4df8faf85954023da542e7"; + "644d9dd5df95f763784b21c1dd941e7f0f0e5e31cb342394f07b2c8a4c9a2fd3"; const DURABLE_SCHEMA_SHA256: &str = - "5281143e2825539af6148785c91f9df67f654eb74d215f57940af9314867a55f"; + "5e929e2f6e07dc976859e38cec5f6af3734fcdb741a7345a6a8988bc7f3135c6"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = "761dea31a521b0650840fe2a823ad6e36a265ed323ba4506889781d630df0ee3"; // Encoded with the schema that omitted stable_placeholder. Keep these diff --git a/docs/how-it-works/sandboxes/overview.mdx b/docs/how-it-works/sandboxes/overview.mdx index 940117980e..08b88b20dc 100644 --- a/docs/how-it-works/sandboxes/overview.mdx +++ b/docs/how-it-works/sandboxes/overview.mdx @@ -977,7 +977,9 @@ Preparation expiry sets the sandbox to `Error` with reason `ImagePreparationTime The gateway keeps each submitted create or start operation running after its caller disconnects, its deadline expires, or monitoring fails. This includes recovery and the stop/start sequence for automatic restart. While `driver_operation_pending` is true, explicit start, stop, and deletion are blocked; automatic restart keeps its schedule and waits. Timeout cleanup can try to stop partial compute, but it cannot declare completion until the original driver call returns and a subsequent stop succeeds. This prevents a stop that observed missing compute from allowing a retry just before the original create finishes. Uploaded rootfs archives remain available to the owned operation. -An actual driver response clears the pending flag, including ordinary rejection errors. The gateway retains `driver_operation_id` so a delayed response handler cannot change newer recovery work on the same provisioning attempt. Both fields appear in the `provisioning` JSON object. Backend behavior after a transport error still follows the driver's existing contract. If the owning gateway crashes before recording the response, the flag and any protected upload can remain indefinitely; restart and age alone cannot establish that the operation finished. There is currently no API to resolve that lost ownership, so automatic cleanup completion, retry, and deletion remain blocked for that record. +An actual driver response clears the pending flag, including ordinary rejection errors. The gateway retains `driver_operation_id` so a delayed response handler cannot change newer recovery work on the same provisioning attempt. Both fields appear in the `provisioning` JSON object, with `driver_operation_owner`, the gateway process that claimed the operation. Backend behavior after a transport error still follows the driver's existing contract. + +If the owning gateway process stops before recording the response, an in-process driver such as Kubernetes cannot still be working on the operation: only requests it had already sent can commit, within the backend's request timeout. Every gateway process refreshes a presence record every 10 seconds, and a process that cannot refresh it for 60 seconds abandons the operations it owns. The reconciler takes over a pending operation once its owner's presence is 3 minutes old, or after watching it for 3 minutes when the owner left no presence. Taking over settles the operation with an unknown outcome and gives it a new `driver_operation_id`, so retry and deletion work again and a late response is ignored. An interrupted automatic restart becomes due again and starts a fresh runtime generation, stopping whatever the abandoned operation left behind first. Restart and age alone still cannot establish that an operation of a driver running in its own process finished, so for those drivers the flag and any protected upload can remain indefinitely, and automatic cleanup completion, retry, and deletion remain blocked for that record. For `ImagePreparationTimedOut`, inspect image preparation and supervisor startup diagnostics and check whether the configured preparation budget is sufficient. For `ProvisioningTimedOut`, repair the rejected configuration. In either case, wait for cleanup to complete, then explicitly restart: diff --git a/proto/openshell.proto b/proto/openshell.proto index a1481289c5..8d58d628e2 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -3884,7 +3884,9 @@ message SandboxProvisioning { google.protobuf.Timestamp admission_start_time = 13; // An owned driver create/start future is still pending. Claimed before // dispatch and cleared on its actual success or error response. Monitor - // interruption, owner loss, and StopSandbox/NotFound do not clear it. + // interruption and StopSandbox/NotFound do not clear it. With an in-process + // driver, the reconciler clears it once the owning gateway process has + // stopped heartbeating (see driver_operation_owner). // While set, cleanup cannot complete and retry/deletion remain blocked. // This does not strengthen a driver's transport-error settlement contract. bool driver_operation_pending = 14; @@ -3892,6 +3894,11 @@ message SandboxProvisioning { // restart. Retained after settlement to reject delayed callbacks from an // earlier operation on the same attempt and runtime generation. string driver_operation_id = 15; + // Gateway process incarnation that claimed the operation. An in-process + // driver's operation cannot outlive that process, so a pending operation + // whose owner stopped heartbeating is taken over by the reconciler. Empty + // on operations claimed before ownership was recorded. + string driver_operation_owner = 16; } // Create-time request to expose one loopback HTTP service in a sandbox. diff --git a/sdk/go/proto/openshellv1/openshell.pb.go b/sdk/go/proto/openshellv1/openshell.pb.go index 246e692420..9c2d10900a 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -18018,7 +18018,9 @@ type SandboxProvisioning struct { AdmissionStartTime *timestamppb.Timestamp `protobuf:"bytes,13,opt,name=admission_start_time,json=admissionStartTime,proto3" json:"admission_start_time,omitempty"` // An owned driver create/start future is still pending. Claimed before // dispatch and cleared on its actual success or error response. Monitor - // interruption, owner loss, and StopSandbox/NotFound do not clear it. + // interruption and StopSandbox/NotFound do not clear it. With an in-process + // driver, the reconciler clears it once the owning gateway process has + // stopped heartbeating (see driver_operation_owner). // While set, cleanup cannot complete and retry/deletion remain blocked. // This does not strengthen a driver's transport-error settlement contract. DriverOperationPending bool `protobuf:"varint,14,opt,name=driver_operation_pending,json=driverOperationPending,proto3" json:"driver_operation_pending,omitempty"` @@ -18026,8 +18028,13 @@ type SandboxProvisioning struct { // restart. Retained after settlement to reject delayed callbacks from an // earlier operation on the same attempt and runtime generation. DriverOperationId string `protobuf:"bytes,15,opt,name=driver_operation_id,json=driverOperationId,proto3" json:"driver_operation_id,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Gateway process incarnation that claimed the operation. An in-process + // driver's operation cannot outlive that process, so a pending operation + // whose owner stopped heartbeating is taken over by the reconciler. Empty + // on operations claimed before ownership was recorded. + DriverOperationOwner string `protobuf:"bytes,16,opt,name=driver_operation_owner,json=driverOperationOwner,proto3" json:"driver_operation_owner,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *SandboxProvisioning) Reset() { @@ -18165,6 +18172,13 @@ func (x *SandboxProvisioning) GetDriverOperationId() string { return "" } +func (x *SandboxProvisioning) GetDriverOperationOwner() string { + if x != nil { + return x.DriverOperationOwner + } + return "" +} + // Create-time request to expose one loopback HTTP service in a sandbox. type SandboxServiceExposure struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -19574,7 +19588,7 @@ const file_openshell_proto_rawDesc = "" + "\x04path\x18\x04 \x01(\tR\x04path\x12=\n" + "\vlast_result\x18\x05 \x01(\x0e2\x1c.openshell.v1.EndpointResultR\n" + "lastResult\x12H\n" + - "\x12last_reported_time\x18j \x01(\v2\x1a.google.protobuf.TimestampR\x10lastReportedTimeJ\x04\b\x06\x10\aR\x10last_reported_at\"\xd5\a\n" + + "\x12last_reported_time\x18j \x01(\v2\x1a.google.protobuf.TimestampR\x10lastReportedTimeJ\x04\b\x06\x10\aR\x10last_reported_at\"\x8b\b\n" + "\x13SandboxProvisioning\x12\x1d\n" + "\n" + "attempt_id\x18\x01 \x01(\tR\tattemptId\x126\n" + @@ -19592,7 +19606,8 @@ const file_openshell_proto_rawDesc = "" + "\x14preparation_deadline\x18\f \x01(\v2\x1a.google.protobuf.TimestampR\x13preparationDeadline\x12L\n" + "\x14admission_start_time\x18\r \x01(\v2\x1a.google.protobuf.TimestampR\x12admissionStartTime\x128\n" + "\x18driver_operation_pending\x18\x0e \x01(\bR\x16driverOperationPending\x12.\n" + - "\x13driver_operation_id\x18\x0f \x01(\tR\x11driverOperationId\"\xaa\x01\n" + + "\x13driver_operation_id\x18\x0f \x01(\tR\x11driverOperationId\x124\n" + + "\x16driver_operation_owner\x18\x10 \x01(\tR\x14driverOperationOwner\"\xaa\x01\n" + "\x16SandboxServiceExposure\x12\x18\n" + "\aservice\x18\x01 \x01(\tR\aservice\x12\x1f\n" + "\vtarget_port\x18\x02 \x01(\rR\n" + From 2cf523be263ca2aac9d76f9db351b9474c187f9d Mon Sep 17 00:00:00 2001 From: Matt Jane Date: Mon, 26 Oct 2026 23:51:25 -0700 Subject: [PATCH 2/2] fix(compute): fence on elapsed time and judge owners on the reconciler's clock Review findings on the operation takeover: - A heartbeat that hung (a database host gone without a reset) never failed, so the owner never fenced while another replica could take it over. Each heartbeat now times out after 5s, and the fence is driven by time since the last successful heartbeat began. - Presence was judged by comparing the owner's write timestamp with the reconciler's wall clock, so skew could take over a live owner. The sweep now records when each owner's presence row last changed on its own clock. - An older gateway rewriting a sandbox during a rolling upgrade drops the unknown owner field, and the empty-owner delay was as short as any other. Operations with no recorded owner now wait 15 minutes. The takeover delay is 4 minutes, leaving margin over a 75s fence plus in-flight requests. - Takeover applied to unmanaged out-of-process drivers, because only managed ones set driver_process. The runtime now records an in-process driver explicitly. - Graceful shutdown now fences, so in-flight operations settle rather than wait for a takeover. New tests cover a hanging heartbeat, unfencing, shutdown, an owner kept alive by heartbeats, the unowned delay, and a late response after takeover. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_014dzrkBzoiGwdAtxRmRxtR1 Signed-off-by: Matt Jane --- crates/openshell-server/src/compute/mod.rs | 212 +++++++++++++++--- .../src/compute/operation_owner.rs | 182 ++++++++++----- crates/openshell-server/src/lib.rs | 1 + docs/how-it-works/sandboxes/overview.mdx | 2 +- proto/openshell.proto | 3 +- sdk/go/proto/openshellv1/openshell.pb.go | 3 +- 6 files changed, 305 insertions(+), 98 deletions(-) diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index d6287a8c0d..5f0f2f4eff 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -690,6 +690,9 @@ pub struct ComputeRuntime { ssh_identities: Arc>, /// This process as the owner of claimed driver operations. operation_owner: Arc, + /// The driver runs inside this process, so its operations cannot outlive + /// it. Only then are abandoned operations taken over. + driver_in_process: bool, } pub struct SandboxSyncGuard { @@ -785,6 +788,7 @@ impl ComputeRuntime { sync_lock: Arc::new(Mutex::new(())), lifecycle_gates: Arc::new(LifecycleGateRegistry::default()), operation_owner: Arc::new(operation_owner::OperationOwner::new(&lease::replica_id())), + driver_in_process: false, replica_id: lease::replica_id(), rootfs_tar_staging, restart_authority: Arc::new(OnceLock::new()), @@ -793,6 +797,12 @@ impl ComputeRuntime { }) } + /// Record that the driver runs inside this gateway process. + pub(crate) fn with_driver_in_process(mut self) -> Self { + self.driver_in_process = true; + self + } + pub(crate) fn configure_ssh_identities( &self, credentials: crate::credentials::CredentialRuntime, @@ -7630,6 +7640,7 @@ pub fn new_test_runtime_with_driver( lifecycle_gates: Arc::new(LifecycleGateRegistry::default()), replica_id: "test-replica".to_string(), operation_owner: Arc::new(operation_owner::OperationOwner::new("test-replica")), + driver_in_process: true, rootfs_tar_staging: Arc::new(rootfs_tar::RootfsTarStagingRegistry::disabled()), restart_authority: Arc::new(OnceLock::new()), restart_notify: Arc::new(Notify::new()), @@ -8725,6 +8736,7 @@ mod tests { lifecycle_gates: Arc::new(LifecycleGateRegistry::default()), replica_id: "test-replica".to_string(), operation_owner: Arc::new(operation_owner::OperationOwner::new("test-replica")), + driver_in_process: true, rootfs_tar_staging: Arc::new(rootfs_tar::RootfsTarStagingRegistry::disabled()), restart_authority: Arc::new(OnceLock::new()), restart_notify: Arc::new(Notify::new()), @@ -19000,6 +19012,13 @@ mod tests { .driver_operation_owner } + async fn assert_pending(runtime: &ComputeRuntime, id: &str, pending: bool) { + assert_eq!( + provisioning_deadline::driver_operation_pending(&stored(runtime, id).await), + pending + ); + } + #[tokio::test] async fn restart_abandoned_by_a_stopped_gateway_is_taken_over_and_retried() { let driver = ControlledDriver::new(); @@ -19011,11 +19030,10 @@ mod tests { runtime.restart_due_sandboxes().await.unwrap(); assert_eq!(driver.stop_calls(), 0); - // The owner left no presence. It may not have heartbeated yet, so - // its operation is held for a full takeover delay first. - let mut unknown = operation_owner::UnknownOwners::default(); + // The owner's presence never changes (here it has none), so its + // operation is taken over once that has lasted the takeover delay. + let mut observations = operation_owner::Observations::default(); let first_seen = Instant::now(); - let now_ms = openshell_core::time::now_ms(); for elapsed in [ Duration::ZERO, operation_owner::ORPHAN_AFTER @@ -19023,18 +19041,15 @@ mod tests { .unwrap(), ] { runtime - .take_over_orphaned_operations(&mut unknown, first_seen + elapsed, now_ms) + .take_over_orphaned_operations(&mut observations, first_seen + elapsed) .await .unwrap(); - assert!(provisioning_deadline::driver_operation_pending( - &stored(&runtime, "sb-orphan").await - )); + assert_pending(&runtime, "sb-orphan", true).await; } runtime .take_over_orphaned_operations( - &mut unknown, + &mut observations, first_seen + operation_owner::ORPHAN_AFTER, - now_ms, ) .await .unwrap(); @@ -19057,7 +19072,7 @@ mod tests { } #[tokio::test] - async fn operation_is_taken_over_once_its_owner_stops_heartbeating() { + async fn heartbeating_owner_keeps_its_operation_until_its_presence_stops_changing() { let driver = ControlledDriver::new(); let runtime = test_runtime(driver.clone()).await; let owner = operation_owner::OperationOwner::new("gateway-0"); @@ -19073,32 +19088,159 @@ mod tests { Code::FailedPrecondition ); - let mut unknown = operation_owner::UnknownOwners::default(); - let now = Instant::now(); - let heartbeat_ms = openshell_core::time::now_ms(); - let orphan_after_ms = i64::try_from(operation_owner::ORPHAN_AFTER.as_millis()).unwrap(); + let mut observations = operation_owner::Observations::default(); + let start = Instant::now(); + let orphan_after = operation_owner::ORPHAN_AFTER; + runtime + .take_over_orphaned_operations(&mut observations, start) + .await + .unwrap(); + // Heartbeats restart the clock however old the owner's own timestamps + // look, so skew between replicas cannot cause a takeover. + owner.heartbeat(&runtime.store).await.unwrap(); + runtime + .take_over_orphaned_operations(&mut observations, start + orphan_after) + .await + .unwrap(); + assert_pending(&runtime, "sb-owned", true).await; runtime .take_over_orphaned_operations( - &mut unknown, - now, - heartbeat_ms + orphan_after_ms - 1_000, + &mut observations, + start + + (orphan_after * 2) + .checked_sub(Duration::from_secs(1)) + .unwrap(), ) .await .unwrap(); - assert!(provisioning_deadline::driver_operation_pending( - &stored(&runtime, "sb-owned").await - )); + assert_pending(&runtime, "sb-owned", true).await; runtime - .take_over_orphaned_operations(&mut unknown, now, heartbeat_ms + orphan_after_ms) + .take_over_orphaned_operations(&mut observations, start + orphan_after * 2) .await .unwrap(); - assert!(!provisioning_deadline::driver_operation_pending( - &stored(&runtime, "sb-owned").await - )); + assert_pending(&runtime, "sb-owned", false).await; runtime.delete_sandbox("default", "owned").await.unwrap(); } + #[tokio::test] + async fn operation_without_a_recorded_owner_waits_the_longer_delay() { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver.clone()).await; + // Claimed before owners were recorded, or erased by an older gateway + // rewriting the record during a rolling upgrade. + let sandbox = abandoned_restart("sb-unowned", "unowned", ""); + runtime.store.put_message(&sandbox).await.unwrap(); + let mut observations = operation_owner::Observations::default(); + let start = Instant::now(); + for elapsed in [ + Duration::ZERO, + operation_owner::ORPHAN_AFTER, + operation_owner::UNOWNED_AFTER + .checked_sub(Duration::from_secs(1)) + .unwrap(), + ] { + runtime + .take_over_orphaned_operations(&mut observations, start + elapsed) + .await + .unwrap(); + assert_pending(&runtime, "sb-unowned", true).await; + } + runtime + .take_over_orphaned_operations( + &mut observations, + start + operation_owner::UNOWNED_AFTER, + ) + .await + .unwrap(); + assert_pending(&runtime, "sb-unowned", false).await; + } + + #[tokio::test] + async fn late_response_from_the_previous_owner_cannot_settle_a_taken_over_operation() { + let driver = ControlledDriver::new(); + driver.block_start(); + let runtime = test_runtime(driver.clone()).await; + let mut other = runtime.clone(); + other.sync_lock = Arc::new(Mutex::new(())); + other.lifecycle_gates = Arc::new(LifecycleGateRegistry::default()); + other.operation_owner = Arc::new(operation_owner::OperationOwner::new("gateway-1")); + runtime + .store + .put_message(&lost_runtime_restart("sb-late", "late")) + .await + .unwrap(); + let restarting = runtime.clone(); + let restart = + tokio::spawn(async move { restarting.restart_sandbox_runtime("sb-late").await }); + driver.start_started.notified().await; + wait_driver_pending(&runtime, "sb-late", true).await; + + // The owner wrote no presence here, so to the other replica it is gone. + let mut observations = operation_owner::Observations::default(); + let start = Instant::now(); + for elapsed in [Duration::ZERO, operation_owner::ORPHAN_AFTER] { + other + .take_over_orphaned_operations(&mut observations, start + elapsed) + .await + .unwrap(); + } + let taken = stored(&runtime, "sb-late").await; + assert!(!provisioning_deadline::driver_operation_pending(&taken)); + + driver.release_start(); + assert!(restart.await.unwrap().is_err()); + let after = stored(&runtime, "sb-late").await; + assert!(provisioning_operation::same_operation(&after, &taken)); + assert!(!provisioning_deadline::driver_operation_pending(&after)); + assert_eq!(driver.start_calls(), 1); + } + + #[tokio::test(start_paused = true)] + async fn heartbeat_that_hangs_fences_and_recovery_unfences() { + use std::sync::atomic::{AtomicBool, Ordering}; + let owner = Arc::new(operation_owner::OperationOwner::new("gateway-0")); + let hang = Arc::new(AtomicBool::new(true)); + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let started = tokio::time::Instant::now(); + let heartbeats = tokio::spawn({ + let owner = owner.clone(); + let hang = hang.clone(); + async move { + owner + .run_heartbeats( + || { + let hang = hang.load(Ordering::SeqCst); + async move { + if hang { + std::future::pending::<()>().await; + } + Ok(()) + } + }, + shutdown_rx, + ) + .await; + } + }); + + owner.wait_fenced().await; + let fenced_after = started.elapsed(); + assert!( + fenced_after >= operation_owner::FENCE_AFTER, + "{fenced_after:?}" + ); + assert!(fenced_after <= Duration::from_secs(75), "{fenced_after:?}"); + + hang.store(false, Ordering::SeqCst); + tokio::time::sleep(operation_owner::HEARTBEAT_INTERVAL * 2).await; + assert!(!owner.is_fenced()); + + shutdown_tx.send(true).unwrap(); + heartbeats.await.unwrap(); + assert!(owner.is_fenced(), "shutdown fences in-flight operations"); + } + #[tokio::test] async fn fenced_gateway_abandons_the_driver_operation_it_owns() { let driver = ControlledDriver::new(); @@ -19117,17 +19259,15 @@ mod tests { assert_eq!(operation_owner_of(&pending), runtime.operation_owner.id()); // A process never takes over its own operations, however old. - runtime - .take_over_orphaned_operations( - &mut operation_owner::UnknownOwners::default(), - Instant::now() + operation_owner::ORPHAN_AFTER * 2, - openshell_core::time::now_ms() + 3_600_000, - ) - .await - .unwrap(); - assert!(provisioning_deadline::driver_operation_pending( - &stored(&runtime, "sb-fenced").await - )); + let mut observations = operation_owner::Observations::default(); + let start = Instant::now(); + for elapsed in [Duration::ZERO, operation_owner::UNOWNED_AFTER * 2] { + runtime + .take_over_orphaned_operations(&mut observations, start + elapsed) + .await + .unwrap(); + } + assert_pending(&runtime, "sb-fenced", true).await; runtime.operation_owner.set_fenced(true); let error = restart.await.unwrap().unwrap_err(); diff --git a/crates/openshell-server/src/compute/operation_owner.rs b/crates/openshell-server/src/compute/operation_owner.rs index de0831cd39..5acf556ddb 100644 --- a/crates/openshell-server/src/compute/operation_owner.rs +++ b/crates/openshell-server/src/compute/operation_owner.rs @@ -11,20 +11,37 @@ //! from API requests already sent, which the backend commits or rejects within //! its own request timeout. Each process therefore heartbeats a presence row //! keyed by a per-process incarnation, so a restarted container is a new owner -//! even though it keeps its replica ID. A process that cannot heartbeat for -//! [`FENCE_AFTER`] abandons its own driver operations. The reconciler takes -//! over a pending operation once its owner's presence is [`ORPHAN_AFTER`] old; -//! the gap between the two covers requests that were already in flight. +//! even though it keeps its replica ID. +//! +//! - A process whose last successful heartbeat began [`FENCE_AFTER`] ago, +//! because writes fail or hang, abandons the operations it owns. So does a +//! process that is shutting down. +//! - The reconciler takes over a pending operation once its owner's presence +//! row has not changed for [`ORPHAN_AFTER`], measured on the reconciler's +//! own clock, so clock skew between replicas cannot cause a takeover. The +//! gap between the two delays covers requests that were already in flight. +//! - An operation with no recorded owner, claimed before owners were recorded +//! or with its owner erased by an older gateway rewriting the record during +//! a rolling upgrade, is taken over after [`UNOWNED_AFTER`], longer than any +//! driver operation runs. //! //! Taking over settles the operation with an unknown outcome and rotates its //! ID, as a driver error would. An automatic restart becomes due again, and //! the restart sweep starts a fresh runtime generation, which stops whatever -//! the abandoned operation left behind first. +//! the abandoned operation left behind first. Other operations are not +//! re-driven: the sandbox can be retried or deleted again, and its +//! provisioning deadline, if armed, still applies. +//! +//! Residual risk: a process frozen for longer than [`ORPHAN_AFTER`] (for +//! example a paused VM) cannot fence itself, and may resume one driver call it +//! had already started after the takeover. The restart that follows the +//! takeover stops stale generations first, which bounds the damage. //! -//! Drivers running in a separate process can keep working after the gateway -//! stops, so their pending operations keep the original contract. +//! Drivers that do not run inside the gateway process can keep working after +//! it stops, so their pending operations keep the original contract. use std::collections::{HashMap, HashSet}; +use std::future::Future; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -46,13 +63,19 @@ const PAGE_SIZE: u32 = 1000; /// How often each gateway process refreshes its presence. pub(super) const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10); -/// A process that has not refreshed its presence for this long abandons the -/// driver operations it owns. +/// A heartbeat write that takes longer than this counts as failed. +pub(super) const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(5); +/// A process whose last successful heartbeat began this long ago abandons the +/// driver operations it owns. With [`HEARTBEAT_INTERVAL`] and +/// [`HEARTBEAT_TIMEOUT`] it notices within 75 seconds. pub(super) const FENCE_AFTER: Duration = Duration::from_mins(1); -/// A pending operation whose owner has not refreshed its presence for this -/// long is taken over. Exceeds [`FENCE_AFTER`] by more than the Kubernetes API -/// server's request timeout. -pub(super) const ORPHAN_AFTER: Duration = Duration::from_mins(3); +/// A pending operation whose owner's presence has not changed for this long +/// is taken over. Exceeds the 75 seconds a process may take to fence itself +/// by more than a Kubernetes client timeout (30s) plus the API server's +/// request timeout (60s). +pub(super) const ORPHAN_AFTER: Duration = Duration::from_mins(4); +/// A pending operation with no recorded owner is taken over after this long. +pub(super) const UNOWNED_AFTER: Duration = Duration::from_mins(15); /// Presence rows of processes gone for this long are deleted. const PRESENCE_RETENTION: Duration = Duration::from_hours(24); @@ -98,7 +121,7 @@ impl OperationOwner { let _ = fenced.wait_for(|fenced| *fenced).await; } - pub(super) async fn heartbeat(&self, store: &Store) -> Result<(), PersistenceError> { + pub(super) async fn heartbeat(&self, store: &Store) -> Result<(), String> { store .put_if( PRESENCE_OBJECT_TYPE, @@ -111,65 +134,91 @@ impl OperationOwner { ) .await .map(|_| ()) + .map_err(|error| error.to_string()) } - /// Refresh this process's presence until shutdown, fencing its operations - /// while the store cannot be reached. The row is left behind at shutdown: - /// its age is what lets the reconciler take over sooner. + /// Refresh this process's presence until shutdown. The row is left behind + /// at shutdown; it simply stops changing. pub(super) async fn heartbeat_loop( self: Arc, store: Arc, - mut shutdown: watch::Receiver, + shutdown: watch::Receiver, ) { - let mut last_heartbeat = Instant::now(); + let owner = self.clone(); + self.run_heartbeats(|| owner.heartbeat(&store), shutdown) + .await; + } + + /// Fence on elapsed time since the last successful heartbeat began, so a + /// write that hangs fences as surely as one that fails. + pub(super) async fn run_heartbeats( + &self, + mut heartbeat: F, + mut shutdown: watch::Receiver, + ) where + F: FnMut() -> Fut, + Fut: Future>, + { + let mut last_success = tokio::time::Instant::now(); loop { - match self.heartbeat(&store).await { - Ok(()) => { - last_heartbeat = Instant::now(); + let started = tokio::time::Instant::now(); + match tokio::time::timeout(HEARTBEAT_TIMEOUT, heartbeat()).await { + Ok(Ok(())) => { + last_success = started; if self.is_fenced() { info!(owner = %self.id, "Driver operation ownership restored"); } self.set_fenced(false); } - Err(error) => { + Ok(Err(error)) => { warn!(owner = %self.id, %error, "Failed to refresh gateway process presence"); - if last_heartbeat.elapsed() >= FENCE_AFTER && !self.is_fenced() { - warn!( - owner = %self.id, - "Abandoning owned driver operations; the reconciler will take them over" - ); - self.set_fenced(true); - } } + Err(_) => { + warn!(owner = %self.id, "Timed out refreshing gateway process presence"); + } + } + if last_success.elapsed() >= FENCE_AFTER && !self.is_fenced() { + warn!( + owner = %self.id, + "Abandoning owned driver operations; the reconciler will take them over" + ); + self.set_fenced(true); } tokio::select! { () = tokio::time::sleep(HEARTBEAT_INTERVAL) => {} - _ = shutdown.changed() => return, + changed = shutdown.changed() => { + if changed.is_err() || *shutdown.borrow() { + // Settle in-flight operations now rather than leave + // them to a takeover once this process is gone. + self.set_fenced(true); + return; + } + } } } } } -/// Pending operations whose owner's presence could not be found, keyed by -/// sandbox and operation ID, with when this process first saw them. Covers -/// operations claimed before owners were recorded and owners that have not -/// written their first heartbeat yet. +/// What the takeover sweep has seen, on this process's clock. #[derive(Debug, Default)] -pub(super) struct UnknownOwners(HashMap<(String, String), Instant>); +pub(super) struct Observations { + /// Per owner: its presence row's version and when that last changed. + /// `None` while the owner has no row. + owners: HashMap, Instant)>, + /// Pending operations with no recorded owner, by sandbox and operation + /// ID, with when they were first seen. + unowned: HashMap<(String, String), Instant>, +} impl ComputeRuntime { pub(super) async fn operation_takeover_loop( self: Arc, mut cancel: watch::Receiver, ) { - let mut unknown = UnknownOwners::default(); + let mut observations = Observations::default(); loop { if let Err(error) = self - .take_over_orphaned_operations( - &mut unknown, - Instant::now(), - openshell_core::time::now_ms(), - ) + .take_over_orphaned_operations(&mut observations, Instant::now()) .await { warn!(%error, "Driver operation takeover sweep failed"); @@ -181,19 +230,17 @@ impl ComputeRuntime { } } - /// `now` measures how long an owner has been unknown on this process's - /// clock; `now_ms` is compared with the store's heartbeat times. pub(super) async fn take_over_orphaned_operations( &self, - unknown: &mut UnknownOwners, + observations: &mut Observations, now: Instant, - now_ms: i64, ) -> Result<(), String> { - if self.driver_process.is_some() { + if !self.driver_in_process { return Ok(()); } - let presence = self.read_presence(now_ms).await?; - let mut still_unknown = HashSet::new(); + let presence = self.read_presence().await?; + let mut seen_owners = HashSet::new(); + let mut seen_unowned = HashSet::new(); let mut offset = 0; loop { let records = self @@ -222,16 +269,25 @@ impl ComputeRuntime { if owner == self.operation_owner.id() { continue; } - let orphaned = if let Some(&updated_at_ms) = presence.get(owner) { - now_ms.saturating_sub(updated_at_ms) >= millis(ORPHAN_AFTER) - } else { + let orphaned = if owner.is_empty() { let key = ( sandbox.object_id().to_string(), provisioning.driver_operation_id.clone(), ); - let first_seen = *unknown.0.entry(key.clone()).or_insert(now); - still_unknown.insert(key); - now.saturating_duration_since(first_seen) >= ORPHAN_AFTER + let first_seen = *observations.unowned.entry(key.clone()).or_insert(now); + seen_unowned.insert(key); + now.saturating_duration_since(first_seen) >= UNOWNED_AFTER + } else { + let version = presence.get(owner).copied(); + let observed = observations + .owners + .entry(owner.to_string()) + .or_insert((version, now)); + if observed.0 != version { + *observed = (version, now); + } + seen_owners.insert(owner.to_string()); + now.saturating_duration_since(observed.1) >= ORPHAN_AFTER }; if orphaned && let Err(error) = self.take_over_operation(&sandbox).await { warn!( @@ -247,12 +303,20 @@ impl ComputeRuntime { } offset += u32::try_from(page_len).expect("takeover page length is capped"); } - unknown.0.retain(|key, _| still_unknown.contains(key)); + observations + .owners + .retain(|owner, _| seen_owners.contains(owner)); + observations + .unowned + .retain(|key, _| seen_unowned.contains(key)); Ok(()) } - /// Latest heartbeat per owner. Deletes presence rows past retention. - async fn read_presence(&self, now_ms: i64) -> Result, String> { + /// Each owner's presence version (resource version, update time). Deletes + /// rows untouched for [`PRESENCE_RETENTION`]; that comparison is the only + /// one made against another process's clock, and a day dwarfs any skew. + async fn read_presence(&self) -> Result, String> { + let now_ms = openshell_core::time::now_ms(); let mut presence = HashMap::new(); let mut offset = 0; loop { @@ -270,7 +334,7 @@ impl ComputeRuntime { .delete_if(PRESENCE_OBJECT_TYPE, &record.id, record.resource_version) .await; } - presence.insert(record.id, record.updated_at_ms); + presence.insert(record.id, (record.resource_version, record.updated_at_ms)); } if page_len < PAGE_SIZE as usize { break; diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 28d4a28589..6677eb9d48 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -1683,6 +1683,7 @@ async fn build_compute_runtime( supervisor_sessions, ) .await + .map(ComputeRuntime::with_driver_in_process) .map_err(|error| { Error::execution(format!("failed to create compute runtime: {error}")) })?, diff --git a/docs/how-it-works/sandboxes/overview.mdx b/docs/how-it-works/sandboxes/overview.mdx index 08b88b20dc..e5dac3cf00 100644 --- a/docs/how-it-works/sandboxes/overview.mdx +++ b/docs/how-it-works/sandboxes/overview.mdx @@ -979,7 +979,7 @@ The gateway keeps each submitted create or start operation running after its cal An actual driver response clears the pending flag, including ordinary rejection errors. The gateway retains `driver_operation_id` so a delayed response handler cannot change newer recovery work on the same provisioning attempt. Both fields appear in the `provisioning` JSON object, with `driver_operation_owner`, the gateway process that claimed the operation. Backend behavior after a transport error still follows the driver's existing contract. -If the owning gateway process stops before recording the response, an in-process driver such as Kubernetes cannot still be working on the operation: only requests it had already sent can commit, within the backend's request timeout. Every gateway process refreshes a presence record every 10 seconds, and a process that cannot refresh it for 60 seconds abandons the operations it owns. The reconciler takes over a pending operation once its owner's presence is 3 minutes old, or after watching it for 3 minutes when the owner left no presence. Taking over settles the operation with an unknown outcome and gives it a new `driver_operation_id`, so retry and deletion work again and a late response is ignored. An interrupted automatic restart becomes due again and starts a fresh runtime generation, stopping whatever the abandoned operation left behind first. Restart and age alone still cannot establish that an operation of a driver running in its own process finished, so for those drivers the flag and any protected upload can remain indefinitely, and automatic cleanup completion, retry, and deletion remain blocked for that record. +If the owning gateway process stops before recording the response, an in-process driver such as Kubernetes cannot still be working on the operation: only requests it had already sent can commit, within the backend's request timeout. Every gateway process refreshes a presence record every 10 seconds. A process abandons the operations it owns when it shuts down, or when its last successful refresh began a minute ago, whether refreshes fail or hang. The reconciler takes over a pending operation once its owner's presence record has not changed for 4 minutes, measured on the reconciler's own clock, so clock skew between replicas cannot trigger it. An operation with no recorded owner, from an older gateway, is taken over after 15 minutes. Taking over settles the operation with an unknown outcome and gives it a new `driver_operation_id`, so retry and deletion work again and a late response is ignored. An interrupted automatic restart becomes due again and starts a fresh runtime generation, stopping whatever the abandoned operation left behind first; other operations are not re-driven, but their provisioning deadline still applies. A gateway process frozen for longer than the takeover delay cannot abandon its work, so it may complete one driver call it had already started. For drivers that do not run inside the gateway process, restart and age alone still cannot establish that an operation finished, so the flag and any protected upload can remain indefinitely, and automatic cleanup completion, retry, and deletion remain blocked for that record. For `ImagePreparationTimedOut`, inspect image preparation and supervisor startup diagnostics and check whether the configured preparation budget is sufficient. For `ProvisioningTimedOut`, repair the rejected configuration. In either case, wait for cleanup to complete, then explicitly restart: diff --git a/proto/openshell.proto b/proto/openshell.proto index 8d58d628e2..433d49eecc 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -3897,7 +3897,8 @@ message SandboxProvisioning { // Gateway process incarnation that claimed the operation. An in-process // driver's operation cannot outlive that process, so a pending operation // whose owner stopped heartbeating is taken over by the reconciler. Empty - // on operations claimed before ownership was recorded. + // on operations claimed before ownership was recorded, or rewritten by a + // gateway that predates this field; those wait a longer takeover delay. string driver_operation_owner = 16; } diff --git a/sdk/go/proto/openshellv1/openshell.pb.go b/sdk/go/proto/openshellv1/openshell.pb.go index 9c2d10900a..270e674e78 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -18031,7 +18031,8 @@ type SandboxProvisioning struct { // Gateway process incarnation that claimed the operation. An in-process // driver's operation cannot outlive that process, so a pending operation // whose owner stopped heartbeating is taken over by the reconciler. Empty - // on operations claimed before ownership was recorded. + // on operations claimed before ownership was recorded, or rewritten by a + // gateway that predates this field; those wait a longer takeover delay. DriverOperationOwner string `protobuf:"bytes,16,opt,name=driver_operation_owner,json=driverOperationOwner,proto3" json:"driver_operation_owner,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache