[restore] retire an abandoned restore's execution in the revert that abandons it

architecture-baseline.toml

@@ -158,7 +158,7 @@
 "crates/libs/api/src/controllers/substrate/kubernetes/store/execution_rows.rs" = 1
 "crates/libs/api/src/controllers/substrate/kubernetes/store/pod_placement.rs" = 1
 "crates/libs/api/src/controllers/substrate/kubernetes/store/pod_tracking.rs" = 5
-"crates/libs/api/src/controllers/substrate/kubernetes/store/restore_lease.rs" = 5
+"crates/libs/api/src/controllers/substrate/kubernetes/store/restore_lease.rs" = 8
 "crates/libs/api/src/controllers/substrate/kubernetes/store/suspend.rs" = 8
 "crates/libs/api/src/controllers/substrate/mod.rs" = 1
 "crates/libs/cluster/controller/server/src/drained_node_terminator/store.rs" = 2
@@ -169,7 +169,7 @@
 "crates/libs/cluster/controller/server/src/started_sandbox_repair/store.rs" = 3
 "crates/libs/cluster/controller/server/src/suspended_execution_reaper/store.rs" = 1
 "crates/libs/cluster/controller/server/src/tombstone_retention.rs" = 1
-"crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/store.rs" = 10
+"crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/store.rs" = 12
 "crates/libs/internal/api/server/src/data_plane_sessions/service_route.rs" = 1
 "crates/libs/internal/api/server/src/published_state.rs" = 9
 "crates/libs/internal/api/server/src/sandbox_quarantine_store.rs" = 3

crates/libs/api/src/controllers/sandbox/ingress/tests/kubernetes.rs

@@ -123,3 +123,64 @@ async fn stored_ingress_access_resumes_paused_execution_before_reporting_ready()
     }
     Ok(())
 }
+
+/// Ingress to a `restoring` sandbox whose restore lease has lapsed while its
+/// execution still owns the restore (so `observe` reports no owner) reaches
+/// `access`, which must not dial the restoring pod: it joins the restore and
+/// reports ready only once the sandbox is `started`.
+#[tokio::test(flavor = "current_thread")]
+async fn stored_ingress_access_joins_a_restore_whose_lease_lapsed()
+-> pplx_space_test_env::TestResult {
+    use pplx_space_common_core::types::AutoLifecycleWakeSource;
+    use pplx_space_db_test_fixtures::try_pg_test_db;
+    use pplx_space_node_api_sim::SimWorkerHarness;
+    if !crate::test_support::enter_sim_worker_process() {
+        return Ok(());
+    }
+    let Some(db) = try_pg_test_db().await? else {
+        return Ok(());
+    };
+    let worker = SimWorkerHarness::builder(crate::test_support::kubernetes::WORKER_NODE)
+        .with_internal_api()
+        .start()
+        .await?;
+    let node = insert_kubernetes_worker_node(&db.pool).await?;
+    let id = SandboxId::new();
+    let mut policy = AutoLifecyclePolicy::disabled();
+    policy.wake.on.push(AutoLifecycleWakeSource::Access);
+    sqlx::query(sql!("INSERT INTO sandboxes (id, user_id, org_id, managed_by, lifecycle, placement, active_node_id, template_id, content_hash, auto_lifecycle_json, network_policy_generation, restore_lease_expires_at, restore_lease_token)
+        VALUES ($1, '00000000-0000-0000-0000-000000000010', '00000000-0000-0000-0000-000000000001', 'k8s', 'restoring', 'on_node', $2, 'seed', 'hash-seed', $3, 1, NOW() - INTERVAL '1 second', $4)"))
+        .bind(id.as_uuid()).bind(node).bind(sqlx::types::Json(&policy)).bind(uuid::Uuid::now_v7()).execute(&db.pool).await?;
+    let LiveExecution { pod, .. } =
+        insert_live_execution(&db.pool, id, RouteIdentity::Published).await?;
+    let (substrate, _) = substrate_over_pods(&db.pool, worker.port(), vec![pod])?;
+    let backend = StoredAccess {
+        pool: db.pool.clone(),
+        controller: SandboxController::new(kubernetes(substrate)),
+    };
+    assert!(
+        !backend.observe(id).await?.restore_owned,
+        "the lapsed lease leaves ingress with no owner to observe"
+    );
+    let owner_pool = db.pool.clone();
+    let owner = tokio::spawn(async move {
+        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+        sqlx::query(sql!(
+            "UPDATE sandboxes SET lifecycle = 'started' WHERE id = $1"
+        ))
+        .bind(id.as_uuid())
+        .execute(&owner_pool)
+        .await
+    });
+
+    let outcome = backend.access(id).await;
+    let lifecycle_at_return = fetch_sandbox(&db.pool, id).await?.unwrap().lifecycle;
+    owner.await??;
+
+    assert_eq!(outcome?, AccessOutcome::Ready);
+    assert_eq!(
+        lifecycle_at_return, "started",
+        "ingress must not reach a restoring pod"
+    );
+    Ok(())
+}

crates/libs/api/src/controllers/sandbox/lifecycle.rs

@@ -1,4 +1,4 @@
 use pplx_space_common_core::types::SandboxId;
 
 use super::*;
 use crate::controllers::db_contention_error;
@@ -379,7 +379,10 @@
 /// may continue, but this stale completion must not replace `suspending` with
 /// `started`. A substrate whose restore-shaped wakes claim `restoring` before
 /// the node call and leave `restoring -> started` to this CAS (Kubernetes)
-/// also fences a finalized `suspended` state. Direct does not: its
+/// also fences a finalized `suspended` state. On Kubernetes the restore's
+/// owner usually committed `started` already (see
+/// `SubstrateCapabilities::caller_completes_claimed_restore`); this CAS then
+/// rewrites `started` and completes. Direct does not: its
 /// worker-local wake leaves its pre-wake lifecycle (`suspended`, or
 /// `restoring` kept by an earlier lease restore) for this CAS to advance, and
 /// its lease restore writes `started` before returning.
@@ -2168,7 +2171,11 @@
             }
         };
         if let Some(status) = status {
-            maybe_update_sandbox_lifecycle(pool, id, status.as_str()).await;
+            if !(status == SandboxLifecycleStatus::Suspended
+                && self.substrate.capabilities().suspend_commits_lifecycle)
+            {
+                maybe_update_sandbox_lifecycle(pool, id, status.as_str()).await;
+            }
             if status == SandboxLifecycleStatus::Started {
                 record_dormant_activation_if_needed(sandbox, metric_identity, action);
                 self.reconcile_open_sandbox_port_exposures_best_effort(pool, id)
@@ -2408,6 +2415,9 @@
 #[cfg(test)]
 mod characterization_tests;
 
+#[cfg(all(test, feature = "kubernetes"))]
+mod kubernetes_suspend_tests;
+
 // Coverage-gap acknowledgment: `SandboxController::control_runtime_client`'s
 // ordering invariant (drain check must run before any substrate call that can
 // implicit-recover a pod) is not unit-testable here because the `Substrate`

crates/libs/api/src/controllers/sandbox/lifecycle/kubernetes_suspend_tests.rs

@@ -0,0 +1,108 @@
+//! The suspend action's lifecycle write on the Kubernetes substrate, whose
+//! `suspend_execution` commits `suspended` itself.
+
+use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+
+use pplx_space_common_core::types::{LocalArtifactId, SandboxId};
+use pplx_space_db_test_fixtures::pg_test_db;
+use pplx_space_metrics::sandbox::api::SandboxApiMetricIdentity;
+use pplx_space_test_env::TestResult;
+
+use super::*;
+use crate::controllers::substrate::kubernetes::tests::{
+    deleted_pod_kubernetes_client, insert_k8s_execution, insert_test_sandbox_row,
+    substrate_with_store_and_client,
+};
+
+const POD_NAME: &str = "space-sbx-suspend-race-g1";
+const POD_UID: &str = "suspend-race-uid";
+
+/// A wake that claims the sandbox right after the substrate commits its
+/// suspend keeps the row. The suspend action writes no lifecycle of its own
+/// afterwards: a trailing `suspended` would put the woken, routed sandbox
+/// back to `suspended` while its new execution stays active, the shape whose
+/// pod holds the only copy of the sandbox's state.
+///
+/// A deferred trigger stands in for the wake: it moves the row to
+/// `restoring` when the transaction that committed `suspended` commits,
+/// which is after the substrate's write and before anything the controller
+/// does next.
+#[tokio::test(flavor = "current_thread")]
+async fn suspend_keeps_a_wake_that_claimed_after_the_suspend_committed() -> TestResult {
+    let db = pg_test_db().await?;
+    let pool = &db.pool;
+    let sandbox_id = SandboxId::new();
+    insert_test_sandbox_row(pool, sandbox_id, "suspending").await?;
+    // The node already committed the suspend, so the substrate only retires
+    // the pod and finalizes the row; it never reaches the node.
+    sqlx::query(sql!(
+        "UPDATE sandboxes
+            SET current_local_artifact_id = $2,
+                current_local_artifact_kind = 'full',
+                current_local_artifact_origin = 'suspend'
+          WHERE id = $1"
+    ))
+    .bind(sandbox_id)
+    .bind(LocalArtifactId::new())
+    .execute(pool)
+    .await?;
+    insert_k8s_execution(pool, sandbox_id, 1, POD_NAME, POD_UID, None, true).await?;
+    sqlx::query(sql!(
+        "UPDATE sandbox_k8s_executions SET state = 'suspended' WHERE sandbox_id = $1"
+    ))
+    .bind(sandbox_id)
+    .execute(pool)
+    .await?;
+    sqlx::query(sql!(
+        "CREATE FUNCTION wake_claims_after_suspend() RETURNS trigger AS $$
+         BEGIN
+             UPDATE sandboxes SET lifecycle = 'restoring' WHERE id = NEW.id;
+             RETURN NULL;
+         END;
+         $$ LANGUAGE plpgsql"
+    ))
+    .execute(pool)
+    .await?;
+    sqlx::query(sql!(
+        "CREATE CONSTRAINT TRIGGER wake_claims_after_suspend
+             AFTER UPDATE OF lifecycle ON sandboxes
+             DEFERRABLE INITIALLY DEFERRED
+             FOR EACH ROW
+             WHEN (OLD.lifecycle = 'suspending' AND NEW.lifecycle = 'suspended')
+             EXECUTE FUNCTION wake_claims_after_suspend()"
+    ))
+    .execute(pool)
+    .await?;
+
+    let mut substrate = substrate_with_store_and_client(
+        pool.clone(),
+        deleted_pod_kubernetes_client(POD_NAME, POD_UID, Arc::new(AtomicUsize::new(0))),
+    );
+    substrate.k8s.k8s_blocking_delete_timeout = std::time::Duration::from_millis(500);
+    let controller = SandboxController::new(
+        crate::controllers::substrate::Substrate::from_kubernetes(substrate),
+    );
+    let sandbox = crate::controllers::sandbox::fetch_sandbox(pool, sandbox_id)
+        .await?
+        .expect("sandbox row exists");
+
+    controller
+        .action(
+            pool,
+            sandbox_id,
+            Some(&sandbox),
+            SandboxApiMetricIdentity::native(SandboxAction::Suspend.as_str()),
+            SandboxAction::Suspend,
+            None,
+        )
+        .await?;
+
+    let lifecycle: String =
+        sqlx::query_scalar(sql!("SELECT lifecycle FROM sandboxes WHERE id = $1"))
+            .bind(sandbox_id)
+            .fetch_one(pool)
+            .await?;
+    assert_eq!(lifecycle, "restoring");
+    Ok(())
+}

crates/libs/api/src/controllers/sandbox/wake_completion.rs

@@ -25,6 +25,13 @@
 //! Process death during the redrive still loses the write; the sweeper
 //! remains the backstop, exactly as `CancelGuard` defers to the reconciler
 //! when it cannot clean up itself.
+//!
+//! On Kubernetes the restore's owner commits `started` in the substrate under
+//! its own lease token before the wake returns
+//! (`K8sStore::complete_owned_user_restore`, this module's CAS with the
+//! lease-expiry guard replaced by that token). The shield covers the wakes
+//! whose owner commit didn't land, and a redrive that finds the row already
+//! `started` has nothing left to do.
 
 use pplx_space_common_core::sql;
 use std::sync::{Arc, LazyLock};

crates/libs/api/src/controllers/substrate/kubernetes/adapter/substrate_adapter.rs

@@ -1,4 +1,4 @@
 //! `SubstrateAdapter` implementation for `KubernetesSubstrate`.
 
 use super::*;
 use pplx_space_common_core::types::SubstrateKind;
@@ -1497,31 +1497,54 @@
             .resolve_pod_instance(sandbox_id)
             .await
             .map_err(ControllerError::from)?
-            && let Ok(client) = self.access_kept_worker_client(sandbox_id, &pod).await
         {
-            return Ok(client);
+            if pod.sandbox_restoring {
+                return self
+                    .join_restore_before_dial(
+                        sandbox_id,
+                        pplx_space_metrics::sandbox::api::SandboxApiMetricIdentity::native(
+                            "runtime_client",
+                        ),
+                        Duration::from_secs(120),
+                        charge,
+                    )
+                    .await;
+            }
+            if let Ok(client) = self.access_kept_worker_client(sandbox_id, &pod).await {
+                return Ok(client);
+            }
         }
 
         let sandbox = self
             .store
             .require_sandbox(sandbox_id)
             .await
             .map_err(ControllerError::from)?;
+        if is_restoring_lifecycle(&sandbox.lifecycle) {
+            return self
+                .join_restore_before_dial(
+                    sandbox_id,
+                    pplx_space_metrics::sandbox::api::SandboxApiMetricIdentity::native(
+                        "runtime_client",
+                    ),
+                    Duration::from_secs(120),
+                    charge,
+                )
+                .await;
+        }
         if normalize_lifecycle(&sandbox_before_resolution.lifecycle) == "suspending"
             && is_suspended_lifecycle(&sandbox.lifecycle)
         {
-            let binding = self
-                .wake_execution_with_metric_identity(
+            return self
+                .wake_and_dial_once_started(
                     &sandbox,
                     Duration::from_secs(120),
                     pplx_space_metrics::sandbox::api::SandboxApiMetricIdentity::native(
                         "runtime_client",
                     ),
                     charge,
                 )
-                .await
-                .map_err(ControllerError::from)?;
-            return connect_worker(binding.binding.runtime_endpoint).await;
+                .await;
         }
         if is_wake_recovery_lifecycle(&sandbox.lifecycle)
             && !is_dormant_lifecycle(&sandbox.lifecycle)
@@ -1556,6 +1579,12 @@
         else {
             return Ok(None);
         };
+        if pod.sandbox_restoring {
+            // Never dial a restoring pod (see `join_restore_before_dial`).
+            // This path neither recovers nor waits: its callers are
+            // best-effort or turn "unreachable" into a retryable 503.
+            return Ok(None);
+        }
 
         match self.access_kept_worker_client(sandbox_id, &pod).await {
             Ok(client) => Ok(Some(client)),
@@ -1591,23 +1620,26 @@
     ) -> Result<RuntimeClient, ControllerError> {
         if is_suspended_lifecycle(&sandbox.lifecycle) {
             Self::should_allow_access_recovery(sandbox, activity)?;
-            let binding = self
-                .wake_execution_with_metric_identity(
+            return self
+                .wake_and_dial_once_started(
                     sandbox,
                     timeout,
                     metric_identity,
                     crate::controllers::substrate::RecoveryCharge::Exempt,
                 )
-                .await
-                .map_err(ControllerError::from)?;
-            return connect_worker(binding.binding.runtime_endpoint).await;
+                .await;
         }
 
         if let Some(pod) = self
             .resolve_pod_instance(sandbox.id)
             .await
             .map_err(ControllerError::from)?
         {
+            if pod.sandbox_restoring {
+                return self
+                    .join_restore_before_dial(sandbox.id, metric_identity, timeout, charge)
+                    .await;
+            }
             let pod_ip = pod.ip;
             // A preserved slow create was reported `started` before its guest
             // command API was confirmed ready. Force a (bounded) command-ready
@@ -1668,6 +1700,11 @@
             .await
             .map_err(ControllerError::from)?;
         Self::should_allow_access_recovery(&sandbox, activity)?;
+        if is_restoring_lifecycle(&sandbox.lifecycle) {
+            return self
+                .join_restore_before_dial(sandbox.id, metric_identity, timeout, charge)
+                .await;
+        }
         let binding = self
             .ensure_execution_ready_for_access(&sandbox, metric_identity, timeout, charge, None)
             .await?;

crates/libs/api/src/controllers/substrate/kubernetes/adapter/teardown.rs

@@ -597,7 +597,10 @@ impl KubernetesSubstrate {
 
     /// Idempotent best-effort delete of each tracked pod name. An already-gone
     /// pod is the expected case, not a failure.
-    async fn delete_tracked_pods(&self, pod_names: Vec<String>) -> Result<(), SubstrateError> {
+    pub(in crate::controllers::substrate::kubernetes) async fn delete_tracked_pods(
+        &self,
+        pod_names: Vec<String>,
+    ) -> Result<(), SubstrateError> {
         for pod_name in pod_names {
             if let Err(err) = delete_sandbox_pod(&self.k8s, &pod_name).await
                 && !is_not_found_error(&err)

crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests.rs

@@ -14,6 +14,7 @@ mod create_readiness;
 mod durable_snapshot;
 mod hydration_and_suspend;
 mod pooled_adoption;
+mod restoring_access;
 
 use std::collections::BTreeMap;