diff --git a/crates/libs/api/src/controllers/substrate/execution_fence.rs b/crates/libs/api/src/controllers/substrate/execution_fence.rs new file mode 100644 index 000000000..ed3dff15d --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/execution_fence.rs @@ -0,0 +1,548 @@ +//! The identity fence that decides whether a recorded execution is the one a retire or teardown may +//! act on. + +use super::*; + +/// Pod identity a caller observed and wants a retire pinned to. +/// +/// `retire_execution` addresses a sandbox by id, resolves whatever pod is +/// tracked at call time, and tears it down. Callers reach that call after a +/// probe and a CAS, and the pod can change in between — recover, be replaced +/// by a same-name successor, or be superseded by a newer-generation execution. +/// A fence carries the identity the caller actually saw so the substrate can +/// refuse instead of acting on stale evidence. +/// +/// The name is required; the other two are what the caller happened to know, +/// and each one it has strengthens the fence: +/// +/// * **Name alone** may act only on an *absent* pod. It pins the teardown to +/// one execution — the control plane can answer "is this still the active +/// execution?", which catches a successor — but it can prove nothing about +/// a live object at that name, so a live pod refuses. This is the shape the +/// sweeper falls back to for an execution whose pod UID has not been +/// persisted yet, which is a real state: the UID is written after the pod is +/// created. It is what lets the sweep still collect rows whose pod is gone, +/// without ever deleting a pod whose identity nobody observed. The state is +/// self-clearing: once pod metadata persists, the next candidate select +/// carries a UID and the full fence applies. +/// * **`pod_uid`** adds the apiserver-enforced precondition, catching a +/// same-name pod that was replaced. +/// * **`resource_version`** catches the same pod having *moved* — bound, +/// status changed, recovered. Only a caller that read the live pod object +/// has one; `sandbox_k8s_executions` has no column for it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PodIdentityFence { + pub pod_name: PodName, + pub pod_uid: Option, + /// What the caller's execution row said about itself when it was read. + /// Distinct from [`Self::pod_uid`], which is what the caller saw of + /// the live object — a row can record a UID whose pod is long gone, and a + /// row re-claimed for a replacement records none while its predecessor's + /// pod is still there. + pub recorded: RecordedExecution, + pub resource_version: Option, + /// The activation attempt the caller selected. + /// + /// Neither the generation nor the pod name identifies an *attempt*: an + /// activation retry or a cross-node redrive reuses both, and the pod name + /// is derived from the generation. So a replacement attempt can look + /// identical to the one the caller observed — same generation, same name, + /// and before it materializes, the same absent UID. This is the only field + /// that tells them apart, which matters most where the rest of the fence + /// is weakest. + pub activation_attempt_id: Option, + /// The execution generation the caller selected, when it recorded one. + pub generation: Option, +} + +/// What the execution row records about itself. +/// +/// Grouped by *who observed it* rather than by field. Everything inside +/// [`Self::Read`] came from one read of one row, so it is all known or all +/// unknown together, and "the caller never looked" is one state rather than a +/// `None` on each field. That distinction is the whole point: for a row, each +/// of these absences is a *value* — a re-claimed row records no pod, a +/// completed activation records no attempt — so an `Option` per field would +/// let "I did not look" authorize exactly what "the row records nothing" +/// should be caught by. +/// +/// A new fact the row keeps about itself goes inside `Read`, where the +/// exhaustive destructure in [`ExecutionIdentity::reconcile`] forces it to be +/// given a rule. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecordedExecution { + /// The caller did not read the row, so it constrains nothing. + Unread, + /// What the row said when the caller read it. + Read { + /// Which pod the row claims as its own. Cleared by a replacement + /// claim, and left behind when a pod is deleted, so it is neither the + /// live pod's UID nor derivable from it. + pod_uid: Option, + /// The durable identity of the activation that owns this execution. + /// Outlives the ephemeral attempt id, which is cleared on completion — + /// so two activations that have both completed are otherwise + /// indistinguishable, and this is what tells them apart. + route_activation_id: Option, + }, +} + +impl RecordedExecution { + /// The row's pod claim. `Unread` is not a row state; only row-derived + /// values are ever bound into SQL, and those are never `Unread`. + #[must_use] + pub fn pod_uid(&self) -> Option<&PodUid> { + match self { + Self::Read { pod_uid, .. } => pod_uid.as_ref(), + Self::Unread => None, + } + } + + /// The row's durable activation identity, with the same caveat. + #[must_use] + pub fn route_activation_id(&self) -> Option { + match self { + Self::Read { + route_activation_id, + .. + } => *route_activation_id, + Self::Unread => None, + } + } +} + +/// What a caller saw of the pod object itself. +/// +/// A sum type rather than two more `Option`s because "I could not see the pod" +/// and "I saw that there is no pod" authorize opposite things, and an `Option` +/// spells them the same way. Every bug this fence has had came from a site +/// reading an absent field as "no expectation here" — so absence is spelled +/// out as a variant that has to be matched. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ObservedPod { + /// The caller had no view of the pod object. Nothing about a live pod can + /// be proven from such an observation, so it may act only on a pod that is + /// *still* absent. + Unobserved, + /// The caller looked and there was no pod at that name. + Absent, + /// The caller saw this exact object. + Seen { + pod_uid: PodUid, + /// Only a caller that read the live object has one; the execution + /// table has no column for it. + resource_version: Option, + /// Whether the pod has reached a phase it cannot leave. + /// + /// A pod that is merely running can still *change* — Pending going + /// Running keeps its UID, which is why a caller with no + /// resourceVersion may not act on one. A terminal pod cannot, so its + /// UID is a complete identification. That distinction is what keeps + /// terminal pods collectable at all: the filtered reflector drops + /// them, so no caller can ever observe one, and without this every + /// terminal pod would defer forever behind a name-only fence. + terminal: bool, + }, +} + +/// One execution's identity, as observed at a single instant. +/// +/// This type exists because "is this still the execution I observed?" was +/// being answered independently at six places — the pod-name check, the +/// attempt check, the live-pod check, the apiserver delete precondition, the +/// tracking-release CAS, and the UID backfill — and those six disagreed about +/// which fields matter *and* about what an absent field means. Every review +/// finding against this fence was one of those six picking a wrong subset or +/// reading absence as a wildcard. One value, compared in one place, is what +/// makes that class of bug unrepresentable rather than repeatedly patched. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ExecutionIdentity { + pub pod_name: PodName, + pub generation: Option, + pub activation_attempt_id: Option, + /// What the execution row said about itself when the caller read it. + /// + /// Distinct from [`Self::pod`], which is the live object, and not + /// derivable from it. This is the only part of the identity that can catch + /// a row changing underneath a caller whose pod is absent — the case where + /// there is no live object left to compare at all. It is also what the + /// database guards bind, so the values they enforce and the values + /// compared here cannot drift apart. + pub recorded: RecordedExecution, + pub pod: ObservedPod, +} + +impl ExecutionIdentity { + /// The single answer to "does `current` describe the execution I observed?" + /// + /// `self` is the observation; `current` is the world now. Every field is + /// destructured rather than accessed, so adding one to + /// [`ExecutionIdentity`] fails to compile here until it is given a rule — + /// the structural reason a future field cannot be silently left out of the + /// comparison the way `generation` originally was. + /// + /// Absence is a value everywhere except where it is *unobservable*: a + /// caller that never saw the pod cannot claim anything about a live one, + /// and a resourceVersion is only available to a caller that read the + /// object, so its absence constrains nothing. + pub fn reconcile(&self, current: &Self) -> Result<(), PodFenceMismatch> { + let Self { + pod_name, + generation, + activation_attempt_id, + recorded, + pod, + } = self; + + if pod_name != ¤t.pod_name { + return Err(PodFenceMismatch::ExecutionSuperseded); + } + if generation.is_some() && generation != ¤t.generation { + return Err(PodFenceMismatch::ExecutionSuperseded); + } + if activation_attempt_id != ¤t.activation_attempt_id { + return Err(PodFenceMismatch::ActivationAttemptChanged); + } + // A row that claimed no pod and now claims one has been taken by a + // create since the caller looked; one that claimed a pod and now + // claims none has been re-claimed for a replacement. Both are the row + // becoming someone else's, and when the pod is absent this is the only + // field left that can say so. + if let RecordedExecution::Read { + pod_uid, + route_activation_id, + } = recorded + { + if pod_uid.as_ref() != current.recorded.pod_uid() { + return Err(PodFenceMismatch::PodUidChanged); + } + if *route_activation_id != current.recorded.route_activation_id() { + return Err(PodFenceMismatch::ActivationAttemptChanged); + } + } + match (pod, ¤t.pod) { + // Nothing left at that name to protect: the delete would 404 and + // the rest of the retire is still wanted. Refusing here would + // strand every row whose pod is genuinely gone. Safe to permit + // only because the recorded-UID check above already rejected a row + // that was re-claimed while its replacement's pod had yet to + // appear — without that, an absent pod would make every remaining + // field match a row in the middle of becoming someone else's. + (_, ObservedPod::Absent) => Ok(()), + // A caller that never saw the pod is normally refused against a + // live one: a Pending pod going Running keeps its UID, so a UID + // alone cannot tell "the object I meant" from "the recovery I must + // not destroy". A *terminal* pod cannot recover, so its UID is a + // complete identification — and it has to be, because the filtered + // reflector drops terminal pods and no caller can ever observe + // one. Without this arm every terminal pod defers forever. + ( + ObservedPod::Unobserved, + ObservedPod::Seen { + pod_uid: current_uid, + terminal: true, + .. + }, + ) if recorded.pod_uid() == Some(current_uid) => Ok(()), + // No view of the pod, but there is one now. + (ObservedPod::Unobserved | ObservedPod::Absent, _) => { + Err(PodFenceMismatch::PodIdentityUnverified) + } + ( + ObservedPod::Seen { + pod_uid, + resource_version, + terminal: _, + }, + ObservedPod::Seen { + pod_uid: current_uid, + resource_version: current_version, + terminal: _, + }, + ) => { + if pod_uid != current_uid { + return Err(PodFenceMismatch::PodUidChanged); + } + if resource_version.is_some() && resource_version != current_version { + return Err(PodFenceMismatch::PodResourceVersionChanged); + } + Ok(()) + } + (ObservedPod::Seen { .. }, ObservedPod::Unobserved) => { + Err(PodFenceMismatch::PodIdentityUnverified) + } + } + } +} + +impl PodIdentityFence { + /// A fence on the pod name alone — supersession only. Prefer + /// [`Self::with_pod_uid`] whenever a UID is recorded. + pub fn new(pod_name: PodName) -> Self { + Self { + pod_name, + pod_uid: None, + resource_version: None, + activation_attempt_id: None, + generation: None, + recorded: RecordedExecution::Unread, + } + } + + /// Record what the caller's execution row said about itself, as read at + /// select time. + #[must_use] + pub fn with_recorded(mut self, recorded: RecordedExecution) -> Self { + self.recorded = recorded; + self + } + + #[must_use] + pub fn with_generation(mut self, generation: i64) -> Self { + self.generation = Some(generation); + self + } + + /// The fence as an [`ExecutionIdentity`] — the one shape every guard on + /// this path compares and binds from. + pub fn observed(&self) -> ExecutionIdentity { + ExecutionIdentity { + pod_name: self.pod_name.clone(), + generation: self.generation, + activation_attempt_id: self.activation_attempt_id, + recorded: self.recorded.clone(), + pod: match self.pod_uid.clone() { + Some(pod_uid) => ObservedPod::Seen { + pod_uid, + resource_version: self.resource_version.clone(), + // A caller's observation comes from the filtered reflector, + // which drops terminal pods, so it is never one. + terminal: false, + }, + None => ObservedPod::Unobserved, + }, + } + } + + #[must_use] + pub fn with_activation_attempt( + mut self, + activation_attempt_id: SandboxActivationAttemptId, + ) -> Self { + self.activation_attempt_id = Some(activation_attempt_id); + self + } + + #[must_use] + pub fn with_pod_uid(mut self, pod_uid: PodUid) -> Self { + self.pod_uid = Some(pod_uid); + self + } + + #[must_use] + pub fn with_resource_version(mut self, resource_version: PodResourceVersion) -> Self { + self.resource_version = Some(resource_version); + self + } +} + +/// What the caller observed about the sandbox's execution, and therefore what +/// the substrate has to re-prove before tearing anything down. +/// +/// "I saw no execution" is a real observation and gets its own variant rather +/// than collapsing into "no fence". Collapsing them is a fail-open: a row that +/// had no execution when it was selected is exactly the row a concurrent +/// create or wake is most likely to give one, and the unfenced teardown is the +/// most destructive path there is — it deletes every tracked pod name and +/// releases tracking unconditionally. Having seen nothing must not authorize +/// that. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RetireFence { + /// Nothing observed; retire whatever is tracked. Correct only for a caller + /// that has already made the pod provably irrelevant — it deleted the pod + /// itself under its own precondition first — not for one that simply has + /// no identity to hand. + Unobserved, + /// The caller observed the sandbox with *no* active execution. The retire + /// must refuse if one exists by the time it runs. + NoExecution, + /// The caller observed this pod. + Pod(PodIdentityFence), +} + +/// A pod's identity as the substrate currently sees it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObservedPodIdentity { + pub pod_uid: PodUid, + pub resource_version: Option, +} + +/// Arguments for [`SubstrateAdapter::retire_execution`]. Grouped rather than +/// positional for the same reason [`CreateExecutionRequest`] is. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RetireExecutionRequest { + pub sandbox_id: SandboxId, + pub fence: RetireFence, +} + +impl RetireExecutionRequest { + pub fn fenced(sandbox_id: SandboxId, pod_fence: PodIdentityFence) -> Self { + Self { + sandbox_id, + fence: RetireFence::Pod(pod_fence), + } + } +} + +/// Which half of a [`PodIdentityFence`] the substrate found violated. Carried +/// so callers can log and meter *why* they deferred without re-deriving it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PodFenceMismatch { + /// The fenced pod's execution row is no longer the sandbox's active + /// execution: a newer generation took over. Retiring now would delete the + /// successor's pod, which the caller never observed. + ExecutionSuperseded, + /// No execution row records the fenced pod name at all — or, for a caller + /// that observed no execution, one exists now. + ExecutionMissing, + /// A live pod exists at the fenced name with a different UID: the pod was + /// replaced. + PodUidChanged, + /// A different activation attempt owns the execution now. Same generation, + /// same pod name — a retry or a redrive replaced the attempt the caller + /// selected. + ActivationAttemptChanged, + /// The live pod is the same object but has moved on (bound, status + /// changed) since the caller observed it. + PodResourceVersionChanged, + /// The apiserver rejected the preconditioned delete (409). The pod changed + /// between the check above and the delete itself. + DeletePreconditionRejected, + /// The fence carries no pod UID and a live pod exists at that name, so + /// there is nothing to prove the object is the one the caller observed. + /// A name-only fence may only act on an absent pod. + PodIdentityUnverified, +} + +impl PodFenceMismatch { + /// Whether something the in-VM stop did *not* reach may now own the + /// sandbox. + /// + /// This decides whether [`TeardownProgress::RuntimeStopped`] lets a caller + /// leave its own state transition standing. The stop is addressed to the + /// execution the fence observed, so it proves the sandbox is down only + /// where the mismatch describes *that same object* having moved. Anything + /// else — a new execution, a new attempt, a different pod at the same + /// name, or an identity that could not be proven at all — may be running + /// untouched, and leaving a dormant or terminal lifecycle over it is the + /// stranded state nothing revisits. + /// + /// Exactly one variant is safe to answer `false` to, so the default is + /// `true` and each exception has to earn it. Answering `true` when the + /// sandbox happens to be down costs one re-evaluation on the next tick; + /// answering `false` when it is up strands a live sandbox. + #[must_use] + pub fn successor_took_over(self) -> bool { + match self { + // The same pod object, at a later resourceVersion: a Pending pod + // coming up is the canonical case. The stop reached it. + Self::PodResourceVersionChanged => false, + // A different pod at that name — the stop reached the old one and + // this one is untouched. + Self::PodUidChanged + // A different execution or attempt owns the sandbox now. + | Self::ExecutionSuperseded + | Self::ActivationAttemptChanged + | Self::ExecutionMissing + // The apiserver rejected the delete's precondition, which it + // returns as one 409 whether the pod moved or was replaced. Not + // worth another read to disambiguate: the conservative answer only + // costs a re-evaluation. + | Self::DeletePreconditionRejected + // Nothing could be proven about what is there. + | Self::PodIdentityUnverified => true, + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::ExecutionSuperseded => "execution_superseded", + Self::ExecutionMissing => "execution_missing", + Self::PodUidChanged => "pod_uid_changed", + Self::ActivationAttemptChanged => "activation_attempt_changed", + Self::PodResourceVersionChanged => "pod_resource_version_changed", + Self::DeletePreconditionRejected => "delete_precondition_rejected", + Self::PodIdentityUnverified => "pod_identity_unverified", + } + } +} + +impl std::fmt::Display for PodFenceMismatch { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Result of [`SubstrateAdapter::retire_execution`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RetireExecutionOutcome { + /// The execution was retired: node-local state preserved, pod(s) deleted + /// or already absent, pod tracking released. + Retired, + /// A supplied fence no longer matched. This is *defer*, not failure: the + /// row changed under the caller, and whatever changed it owns the sandbox + /// now. `progress` says whether anything had already been done by the time + /// that surfaced, which is what a caller needs to know before undoing a + /// transition it made to enable this call. + PodIdentityChanged { + mismatch: PodFenceMismatch, + progress: TeardownProgress, + }, +} + +impl RetireExecutionOutcome { + /// Whether a caller may undo a state transition it made to enable this + /// retire. + /// + /// One place, because three sweeper passes ask it and a subset that drifts + /// apart is how this fence's bugs have always looked. Two facts decide it, + /// and both have to hold: nothing destructive happened yet, *or* whatever + /// owns the sandbox now is something this teardown never reached. + /// + /// A retire that succeeded is not an undo case at all — the caller's + /// transition is exactly what it wanted. + #[must_use] + pub fn permits_undo(&self) -> bool { + match self { + Self::Retired => false, + Self::PodIdentityChanged { mismatch, progress } => { + !progress.mutated() || mismatch.successor_took_over() + } + } + } +} + +/// How far a refused teardown got before the mismatch surfaced. +/// +/// Not every refusal is free. The fence is re-proven at several points and +/// only the earliest run before anything is mutated — the pod delete's +/// precondition, in particular, is evaluated by the apiserver a round trip +/// after the in-VM stop. A caller that compensates its own state transition on +/// a refusal has to distinguish the two, or it will restore a lifecycle +/// claiming a running VM for a sandbox that is already stopped. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TeardownProgress { + /// No in-VM stop, no pod delete, no tracking release. + NothingMutated, + /// The observed pod was deleted under its precondition. + PodDeleted, + /// The in-VM stop already ran and its disk backup was enqueued. + RuntimeStopped, +} + +impl TeardownProgress { + #[must_use] + pub fn mutated(self) -> bool { + !matches!(self, Self::NothingMutated) + } +} diff --git a/crates/libs/api/src/controllers/substrate/local_artifact.rs b/crates/libs/api/src/controllers/substrate/local_artifact.rs new file mode 100644 index 000000000..6d0ac8243 --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/local_artifact.rs @@ -0,0 +1,101 @@ +//! The local artifact a sandbox publishes when it suspends, and the guard that refuses to delete +//! the current one. + +use super::*; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PublishedLocalArtifact { + pub id: LocalArtifactId, + pub kind: ArtifactKind, +} + +impl PublishedLocalArtifact { + pub(crate) fn from_published( + published: Option, + operation: &str, + expected_kind: ArtifactKind, + ) -> Result { + let published = published.ok_or_else(|| SubstrateError::Failed { + message: format!("blocking {operation} returned no local_artifact_id"), + })?; + let id = published + .local_artifact_id + .parse() + .map_err(|err| SubstrateError::Failed { + message: format!("blocking {operation} returned invalid local_artifact_id: {err}"), + })?; + let kind = + match pplx_space_sbx_mgr_client::proto::ArtifactKind::try_from(published.artifact_kind) + { + Ok(pplx_space_sbx_mgr_client::proto::ArtifactKind::Full) => ArtifactKind::Full, + Ok(pplx_space_sbx_mgr_client::proto::ArtifactKind::Disk) => ArtifactKind::Disk, + _ => { + return Err(SubstrateError::Failed { + message: format!("blocking {operation} returned unsupported artifact_kind"), + }); + } + }; + if kind != expected_kind { + return Err(SubstrateError::Failed { + message: format!( + "blocking {operation} returned {kind} artifact, expected {expected_kind}" + ), + }); + } + Ok(Self { id, kind }) + } + + pub(crate) fn from_suspend_response( + response: pplx_space_sbx_mgr_client::proto::SuspendResponse, + ) -> Result { + Self::from_published(response.published_artifact, "suspend", ArtifactKind::Full) + } + + pub(crate) fn from_committed_suspend_response( + response: pplx_space_sbx_mgr_client::proto::SuspendResponse, + sandbox_id: SandboxId, + substrate: &'static str, + ) -> Option { + match Self::from_suspend_response(response) { + Ok(published_artifact) => Some(published_artifact), + Err(err) => { + tracing::warn!( + sandbox_id = %sandbox_id, + substrate, + error = %err, + "sandbox suspend completed but returned unusable artifact metadata" + ); + None + } + } + } +} + +/// Call only after locking the sandbox row in the same transaction. Retention +/// takes that lock before changing a snapshot from `ready` to `deleting`, so a +/// later statement observes the committed winner and cannot pin a claimed +/// artifact for restore. +pub(crate) async fn reject_deleting_current_local_artifact( + tx: &mut Transaction<'_, Postgres>, + sandbox_id: SandboxId, + artifact: PublishedLocalArtifact, +) -> Result<(), SubstrateError> { + let deleting_snapshot = deleting_snapshot_matches_artifact("$1", "$2", "$3"); + let query = format!("SELECT {deleting_snapshot}"); + let deleting = sqlx::query_scalar::<_, bool>(sqlx::AssertSqlSafe(sql!(query.as_str()))) + .bind(sandbox_id.as_uuid()) + .bind(artifact.id.as_uuid()) + .bind(artifact.kind.as_str()) + .fetch_one(&mut **tx) + .await + .db_context("failed to check current artifact retention claim")?; + if deleting { + return Err(SubstrateError::Unavailable { + message: format!( + "local artifact {} is already claimed for snapshot deletion", + artifact.id + ), + }); + } + Ok(()) +} diff --git a/crates/libs/api/src/controllers/substrate/mod.rs b/crates/libs/api/src/controllers/substrate/mod.rs index 237fa6cac..c17221713 100644 --- a/crates/libs/api/src/controllers/substrate/mod.rs +++ b/crates/libs/api/src/controllers/substrate/mod.rs @@ -79,82 +79,19 @@ use crate::models::{ClusterPoolConfigRequest, ClusterPoolConfigResponse, Sandbox use sqlx::{PgPool, Postgres, Transaction}; -type RuntimeTransport = GrpcClientTrace>; - -/// Runtime RPCs and capability checks share the resolved worker connection. -#[derive(Clone, Debug)] -pub struct RuntimeClient { - worker: WorkerApiClient, - channel: Channel, - /// Set when `channel` is a cached channel that other callers share. - kept: Option, -} - -impl RuntimeClient { - fn new(channel: Channel) -> Self { - let transport = node_api_channel(channel.clone(), SpaceBearerToken::NodeApi); - Self { - worker: WorkerApiClient::new(transport), - channel, - kept: None, - } - } - - /// A client on a channel a cache keeps while `watch` is open. - #[cfg(feature = "kubernetes")] - fn kept(channel: Channel, watch: pplx_space_common_grpc::ConnectionWatch) -> Self { - Self { - kept: Some(watch), - ..Self::new(channel) - } - } - - /// Stop handing this client's cached channel to later callers, so the - /// next one dials afresh. Call it when the peer showed that new calls - /// should reach a new connection, such as a draining node-api; this - /// client and its clones keep working. A no-op for an uncached channel. - pub(crate) fn retire_kept_connection(&self) { - if let Some(watch) = &self.kept { - watch.close(); - } - } - - pub(super) async fn require_activation_policy_support( - &mut self, - timeout: Duration, - ) -> Result<(), ControllerError> { - require_activation_policy_support( - node_api_channel(self.channel.clone(), SpaceBearerToken::NodeApi), - timeout, - ) - .await - } - - fn internal_client(&self) -> RuntimeInternalClient { - WorkerApiInternalClient::new(node_api_channel( - self.channel.clone(), - SpaceBearerToken::NodeApiInternal, - )) - } -} - -impl std::ops::Deref for RuntimeClient { - type Target = WorkerApiClient; - - fn deref(&self) -> &Self::Target { - &self.worker - } -} - -impl std::ops::DerefMut for RuntimeClient { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.worker - } -} -pub type RuntimeInternalClient = - WorkerApiInternalClient>>; -pub type BuilderClient = - BuilderApiClient>>; +mod execution_fence; +mod local_artifact; +mod node_api_client; +mod pool_claims; +mod recovery_charge; +mod sandbox_state_probe; + +pub use execution_fence::*; +pub use local_artifact::*; +pub use node_api_client::*; +pub use pool_claims::*; +pub use recovery_charge::*; +pub use sandbox_state_probe::*; /// Outcome of a non-recovering guest command-readiness probe. /// @@ -205,55 +142,6 @@ fn probe_status_is_transient(code: tonic::Code) -> bool { ) } -/// Per-request node-RPC channel timeout (the tonic `Endpoint::timeout` on -/// worker/guest channels). This — NOT the ALB — is the binding internal cut -/// behind the create/restore latency cliff: a slow-but-not-stuck node RPC that -/// rides up to it is severed by tonic as `Code::Cancelled`, which surfaces as -/// an opaque HTTP 408 with no sandbox id and no retry semantics. Every API-side -/// blocking wait must finish, *and* a structured retryable response must flush, -/// below it. (The ECW ALB idle timeout is far higher — see -/// [`ALB_IDLE_TIMEOUT_SECS`] — so the ALB is not the binding constraint.) -/// -/// The 150s timeout leaves a 135s blocking budget after reserving 15s for -/// completing the response. -/// -/// The value lives in `common-core`'s `restore_ownership` module because the -/// placement crate's restore lease TTL must equal it; deriving both from one -/// constant is what keeps a future timeout bump from silently splitting the -/// lease-outlives-RPC invariant across crates. -pub(crate) const NODE_RPC_TIMEOUT: Duration = - pplx_space_common_core::restore_ownership::NODE_RPC_TIMEOUT; -pub(crate) const NODE_RPC_TIMEOUT_SECS: u64 = NODE_RPC_TIMEOUT.as_secs(); -const NODE_API_HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(10); -const NODE_API_HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(5); -/// Headroom below [`NODE_RPC_TIMEOUT`] for the API poll loop / wait wrapper to -/// detect exhaustion and return a structured, retryable response (carrying the -/// sandbox id + `Retry-After`) before tonic severs the channel with a bare -/// `Cancelled` 408. -const BLOCKING_WAIT_MARGIN_SECS: u64 = 15; -/// API-side blocking-wait budget for create/restore/wake/access, held below -/// [`NODE_RPC_TIMEOUT`] so a slow wait surfaces as a structured retryable -/// `SubstrateError::Timeout` before the per-request channel timeout fires. The -/// single budget every blocking lifecycle wait is sized against. -/// -/// `pub` (not `pub(crate)`) so out-of-crate callers that pass a -/// `ready_timeout` — e.g. the cluster-controller's node-drain restore — -/// can report the effective (clamped) per-attempt wait instead of -/// hand-copying this value into comments and logs. -pub const NODE_RPC_BLOCKING_BUDGET: Duration = - Duration::from_secs(NODE_RPC_TIMEOUT_SECS - BLOCKING_WAIT_MARGIN_SECS); -/// The ECW ALB idle timeout (`idle_timeout.timeout_seconds`), set via the AWS -/// LB controller annotation on the deployment's Ingress. It is the OUTER -/// ceiling, far above the node-RPC -/// timeout — documented here so the hierarchy is explicit and version-checked; -/// keep it in sync with the Helm values. The static assert pins -/// `NODE_RPC_TIMEOUT < ALB idle`, so the binding cut can never silently exceed -/// the gateway. -pub(crate) const ALB_IDLE_TIMEOUT_SECS: u64 = 1800; -const _: () = assert!( - NODE_RPC_TIMEOUT_SECS < ALB_IDLE_TIMEOUT_SECS, - "node-RPC channel timeout must stay below the ALB idle timeout", -); /// `Retry-After` hint attached to the structured response returned when a wait /// exhausts the budget. The node keeps converging in the background, so a /// short delay is enough for a follow-up to reach a ready sandbox. @@ -282,36 +170,6 @@ pub(crate) const LOCK_CONTENTION_RETRY_AFTER: Duration = Duration::from_secs(1); /// claim CAS. pub(crate) const RESTORE_RETRY_AFTER_FLOOR: Duration = Duration::from_secs(1); -pub(crate) fn node_api_endpoint( - addr: &str, - connect_timeout: Duration, - request_timeout: Duration, -) -> Result { - let uri = pplx_space_common_grpc::internal_transport::qualify_endpoint(addr); - Endpoint::from_shared(uri) - .map(|endpoint| configure_node_api_endpoint(endpoint, connect_timeout, request_timeout)) - .map_err(|err| SubstrateError::Failed { - message: format!("invalid node-api endpoint '{addr}': {err}"), - }) -} - -fn configure_node_api_endpoint( - endpoint: Endpoint, - connect_timeout: Duration, - request_timeout: Duration, -) -> Endpoint { - pplx_space_node_api_client::configure_node_api_endpoint( - endpoint, - NodeApiEndpointConfig::default() - .with_connect_timeout(connect_timeout) - .with_request_timeout(request_timeout) - .with_http2_keep_alive( - NODE_API_HTTP2_KEEP_ALIVE_INTERVAL, - NODE_API_HTTP2_KEEP_ALIVE_TIMEOUT, - ), - ) -} - /// Whether a `create_execution` wait error means "the pod is still coming up, /// leave it running so the node converges" (preserve node-side work) rather /// than "creation genuinely failed, tear it down". Only a [`SubstrateError:: @@ -453,37 +311,6 @@ impl CreateBootMode { } } -pub(crate) fn default_pool_eligible_for_request( - boot_mode: CreateBootMode, - backend: &str, - wait_until_ready: bool, - disk_mib: Option, - file_changes_enabled: bool, - implicit_wake_enabled: bool, -) -> bool { - backend == "fc" - && matches!( - boot_mode, - CreateBootMode::Fallthrough | CreateBootMode::Pool - ) - && disk_mib.is_none() - && !file_changes_enabled - && (wait_until_ready || implicit_wake_enabled) -} - -/// Combine a substrate's generic warm-pool eligibility with env handling. A cold -/// create bakes `req.env` into the sandbox at boot; if the substrate can't -/// reproduce that on a pooled sandbox at adoption (`substrate_supports_pool_env` -/// is false), an env-bearing create must take the cold path rather than silently -/// drop the requested variables. -pub(crate) fn pool_eligible_with_env( - substrate_eligible: bool, - substrate_supports_pool_env: bool, - has_request_env: bool, -) -> bool { - substrate_eligible && (substrate_supports_pool_env || !has_request_env) -} - #[cfg(test)] mod pool_eligibility_tests; @@ -493,80 +320,12 @@ pub struct ActivationOptions { pub ready_timeout: Duration, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct PublishedLocalArtifact { - pub id: LocalArtifactId, - pub kind: ArtifactKind, -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct EnqueuedArtifactBackup { pub artifact: PublishedLocalArtifact, pub attempt_id: BackupAttemptId, } -impl PublishedLocalArtifact { - pub(crate) fn from_published( - published: Option, - operation: &str, - expected_kind: ArtifactKind, - ) -> Result { - let published = published.ok_or_else(|| SubstrateError::Failed { - message: format!("blocking {operation} returned no local_artifact_id"), - })?; - let id = published - .local_artifact_id - .parse() - .map_err(|err| SubstrateError::Failed { - message: format!("blocking {operation} returned invalid local_artifact_id: {err}"), - })?; - let kind = - match pplx_space_sbx_mgr_client::proto::ArtifactKind::try_from(published.artifact_kind) - { - Ok(pplx_space_sbx_mgr_client::proto::ArtifactKind::Full) => ArtifactKind::Full, - Ok(pplx_space_sbx_mgr_client::proto::ArtifactKind::Disk) => ArtifactKind::Disk, - _ => { - return Err(SubstrateError::Failed { - message: format!("blocking {operation} returned unsupported artifact_kind"), - }); - } - }; - if kind != expected_kind { - return Err(SubstrateError::Failed { - message: format!( - "blocking {operation} returned {kind} artifact, expected {expected_kind}" - ), - }); - } - Ok(Self { id, kind }) - } - - pub(crate) fn from_suspend_response( - response: pplx_space_sbx_mgr_client::proto::SuspendResponse, - ) -> Result { - Self::from_published(response.published_artifact, "suspend", ArtifactKind::Full) - } - - pub(crate) fn from_committed_suspend_response( - response: pplx_space_sbx_mgr_client::proto::SuspendResponse, - sandbox_id: SandboxId, - substrate: &'static str, - ) -> Option { - match Self::from_suspend_response(response) { - Ok(published_artifact) => Some(published_artifact), - Err(err) => { - tracing::warn!( - sandbox_id = %sandbox_id, - substrate, - error = %err, - "sandbox suspend completed but returned unusable artifact metadata" - ); - None - } - } - } -} - pub(crate) async fn persist_optional_current_local_artifact( pool: &PgPool, sandbox_id: SandboxId, @@ -613,35 +372,6 @@ pub(crate) async fn persist_optional_current_local_artifact( Ok(()) } -/// Call only after locking the sandbox row in the same transaction. Retention -/// takes that lock before changing a snapshot from `ready` to `deleting`, so a -/// later statement observes the committed winner and cannot pin a claimed -/// artifact for restore. -pub(crate) async fn reject_deleting_current_local_artifact( - tx: &mut Transaction<'_, Postgres>, - sandbox_id: SandboxId, - artifact: PublishedLocalArtifact, -) -> Result<(), SubstrateError> { - let deleting_snapshot = deleting_snapshot_matches_artifact("$1", "$2", "$3"); - let query = format!("SELECT {deleting_snapshot}"); - let deleting = sqlx::query_scalar::<_, bool>(sqlx::AssertSqlSafe(sql!(query.as_str()))) - .bind(sandbox_id.as_uuid()) - .bind(artifact.id.as_uuid()) - .bind(artifact.kind.as_str()) - .fetch_one(&mut **tx) - .await - .db_context("failed to check current artifact retention claim")?; - if deleting { - return Err(SubstrateError::Unavailable { - message: format!( - "local artifact {} is already claimed for snapshot deletion", - artifact.id - ), - }); - } - Ok(()) -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ExecutionActivityKind { Access, @@ -719,550 +449,6 @@ pub struct CreateExecutionRequest<'a> { pub node_pool: Option<&'a NodePoolName>, } -/// Pod identity a caller observed and wants a retire pinned to. -/// -/// `retire_execution` addresses a sandbox by id, resolves whatever pod is -/// tracked at call time, and tears it down. Callers reach that call after a -/// probe and a CAS, and the pod can change in between — recover, be replaced -/// by a same-name successor, or be superseded by a newer-generation execution. -/// A fence carries the identity the caller actually saw so the substrate can -/// refuse instead of acting on stale evidence. -/// -/// The name is required; the other two are what the caller happened to know, -/// and each one it has strengthens the fence: -/// -/// * **Name alone** may act only on an *absent* pod. It pins the teardown to -/// one execution — the control plane can answer "is this still the active -/// execution?", which catches a successor — but it can prove nothing about -/// a live object at that name, so a live pod refuses. This is the shape the -/// sweeper falls back to for an execution whose pod UID has not been -/// persisted yet, which is a real state: the UID is written after the pod is -/// created. It is what lets the sweep still collect rows whose pod is gone, -/// without ever deleting a pod whose identity nobody observed. The state is -/// self-clearing: once pod metadata persists, the next candidate select -/// carries a UID and the full fence applies. -/// * **`pod_uid`** adds the apiserver-enforced precondition, catching a -/// same-name pod that was replaced. -/// * **`resource_version`** catches the same pod having *moved* — bound, -/// status changed, recovered. Only a caller that read the live pod object -/// has one; `sandbox_k8s_executions` has no column for it. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct PodIdentityFence { - pub pod_name: PodName, - pub pod_uid: Option, - /// What the caller's execution row said about itself when it was read. - /// Distinct from [`Self::pod_uid`], which is what the caller saw of - /// the live object — a row can record a UID whose pod is long gone, and a - /// row re-claimed for a replacement records none while its predecessor's - /// pod is still there. - pub recorded: RecordedExecution, - pub resource_version: Option, - /// The activation attempt the caller selected. - /// - /// Neither the generation nor the pod name identifies an *attempt*: an - /// activation retry or a cross-node redrive reuses both, and the pod name - /// is derived from the generation. So a replacement attempt can look - /// identical to the one the caller observed — same generation, same name, - /// and before it materializes, the same absent UID. This is the only field - /// that tells them apart, which matters most where the rest of the fence - /// is weakest. - pub activation_attempt_id: Option, - /// The execution generation the caller selected, when it recorded one. - pub generation: Option, -} - -/// What the execution row records about itself. -/// -/// Grouped by *who observed it* rather than by field. Everything inside -/// [`Self::Read`] came from one read of one row, so it is all known or all -/// unknown together, and "the caller never looked" is one state rather than a -/// `None` on each field. That distinction is the whole point: for a row, each -/// of these absences is a *value* — a re-claimed row records no pod, a -/// completed activation records no attempt — so an `Option` per field would -/// let "I did not look" authorize exactly what "the row records nothing" -/// should be caught by. -/// -/// A new fact the row keeps about itself goes inside `Read`, where the -/// exhaustive destructure in [`ExecutionIdentity::reconcile`] forces it to be -/// given a rule. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum RecordedExecution { - /// The caller did not read the row, so it constrains nothing. - Unread, - /// What the row said when the caller read it. - Read { - /// Which pod the row claims as its own. Cleared by a replacement - /// claim, and left behind when a pod is deleted, so it is neither the - /// live pod's UID nor derivable from it. - pod_uid: Option, - /// The durable identity of the activation that owns this execution. - /// Outlives the ephemeral attempt id, which is cleared on completion — - /// so two activations that have both completed are otherwise - /// indistinguishable, and this is what tells them apart. - route_activation_id: Option, - }, -} - -impl RecordedExecution { - /// The row's pod claim. `Unread` is not a row state; only row-derived - /// values are ever bound into SQL, and those are never `Unread`. - #[must_use] - pub fn pod_uid(&self) -> Option<&PodUid> { - match self { - Self::Read { pod_uid, .. } => pod_uid.as_ref(), - Self::Unread => None, - } - } - - /// The row's durable activation identity, with the same caveat. - #[must_use] - pub fn route_activation_id(&self) -> Option { - match self { - Self::Read { - route_activation_id, - .. - } => *route_activation_id, - Self::Unread => None, - } - } -} - -/// What a caller saw of the pod object itself. -/// -/// A sum type rather than two more `Option`s because "I could not see the pod" -/// and "I saw that there is no pod" authorize opposite things, and an `Option` -/// spells them the same way. Every bug this fence has had came from a site -/// reading an absent field as "no expectation here" — so absence is spelled -/// out as a variant that has to be matched. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum ObservedPod { - /// The caller had no view of the pod object. Nothing about a live pod can - /// be proven from such an observation, so it may act only on a pod that is - /// *still* absent. - Unobserved, - /// The caller looked and there was no pod at that name. - Absent, - /// The caller saw this exact object. - Seen { - pod_uid: PodUid, - /// Only a caller that read the live object has one; the execution - /// table has no column for it. - resource_version: Option, - /// Whether the pod has reached a phase it cannot leave. - /// - /// A pod that is merely running can still *change* — Pending going - /// Running keeps its UID, which is why a caller with no - /// resourceVersion may not act on one. A terminal pod cannot, so its - /// UID is a complete identification. That distinction is what keeps - /// terminal pods collectable at all: the filtered reflector drops - /// them, so no caller can ever observe one, and without this every - /// terminal pod would defer forever behind a name-only fence. - terminal: bool, - }, -} - -/// One execution's identity, as observed at a single instant. -/// -/// This type exists because "is this still the execution I observed?" was -/// being answered independently at six places — the pod-name check, the -/// attempt check, the live-pod check, the apiserver delete precondition, the -/// tracking-release CAS, and the UID backfill — and those six disagreed about -/// which fields matter *and* about what an absent field means. Every review -/// finding against this fence was one of those six picking a wrong subset or -/// reading absence as a wildcard. One value, compared in one place, is what -/// makes that class of bug unrepresentable rather than repeatedly patched. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ExecutionIdentity { - pub pod_name: PodName, - pub generation: Option, - pub activation_attempt_id: Option, - /// What the execution row said about itself when the caller read it. - /// - /// Distinct from [`Self::pod`], which is the live object, and not - /// derivable from it. This is the only part of the identity that can catch - /// a row changing underneath a caller whose pod is absent — the case where - /// there is no live object left to compare at all. It is also what the - /// database guards bind, so the values they enforce and the values - /// compared here cannot drift apart. - pub recorded: RecordedExecution, - pub pod: ObservedPod, -} - -impl ExecutionIdentity { - /// The single answer to "does `current` describe the execution I observed?" - /// - /// `self` is the observation; `current` is the world now. Every field is - /// destructured rather than accessed, so adding one to - /// [`ExecutionIdentity`] fails to compile here until it is given a rule — - /// the structural reason a future field cannot be silently left out of the - /// comparison the way `generation` originally was. - /// - /// Absence is a value everywhere except where it is *unobservable*: a - /// caller that never saw the pod cannot claim anything about a live one, - /// and a resourceVersion is only available to a caller that read the - /// object, so its absence constrains nothing. - pub fn reconcile(&self, current: &Self) -> Result<(), PodFenceMismatch> { - let Self { - pod_name, - generation, - activation_attempt_id, - recorded, - pod, - } = self; - - if pod_name != ¤t.pod_name { - return Err(PodFenceMismatch::ExecutionSuperseded); - } - if generation.is_some() && generation != ¤t.generation { - return Err(PodFenceMismatch::ExecutionSuperseded); - } - if activation_attempt_id != ¤t.activation_attempt_id { - return Err(PodFenceMismatch::ActivationAttemptChanged); - } - // A row that claimed no pod and now claims one has been taken by a - // create since the caller looked; one that claimed a pod and now - // claims none has been re-claimed for a replacement. Both are the row - // becoming someone else's, and when the pod is absent this is the only - // field left that can say so. - if let RecordedExecution::Read { - pod_uid, - route_activation_id, - } = recorded - { - if pod_uid.as_ref() != current.recorded.pod_uid() { - return Err(PodFenceMismatch::PodUidChanged); - } - if *route_activation_id != current.recorded.route_activation_id() { - return Err(PodFenceMismatch::ActivationAttemptChanged); - } - } - match (pod, ¤t.pod) { - // Nothing left at that name to protect: the delete would 404 and - // the rest of the retire is still wanted. Refusing here would - // strand every row whose pod is genuinely gone. Safe to permit - // only because the recorded-UID check above already rejected a row - // that was re-claimed while its replacement's pod had yet to - // appear — without that, an absent pod would make every remaining - // field match a row in the middle of becoming someone else's. - (_, ObservedPod::Absent) => Ok(()), - // A caller that never saw the pod is normally refused against a - // live one: a Pending pod going Running keeps its UID, so a UID - // alone cannot tell "the object I meant" from "the recovery I must - // not destroy". A *terminal* pod cannot recover, so its UID is a - // complete identification — and it has to be, because the filtered - // reflector drops terminal pods and no caller can ever observe - // one. Without this arm every terminal pod defers forever. - ( - ObservedPod::Unobserved, - ObservedPod::Seen { - pod_uid: current_uid, - terminal: true, - .. - }, - ) if recorded.pod_uid() == Some(current_uid) => Ok(()), - // No view of the pod, but there is one now. - (ObservedPod::Unobserved | ObservedPod::Absent, _) => { - Err(PodFenceMismatch::PodIdentityUnverified) - } - ( - ObservedPod::Seen { - pod_uid, - resource_version, - terminal: _, - }, - ObservedPod::Seen { - pod_uid: current_uid, - resource_version: current_version, - terminal: _, - }, - ) => { - if pod_uid != current_uid { - return Err(PodFenceMismatch::PodUidChanged); - } - if resource_version.is_some() && resource_version != current_version { - return Err(PodFenceMismatch::PodResourceVersionChanged); - } - Ok(()) - } - (ObservedPod::Seen { .. }, ObservedPod::Unobserved) => { - Err(PodFenceMismatch::PodIdentityUnverified) - } - } - } -} - -impl PodIdentityFence { - /// A fence on the pod name alone — supersession only. Prefer - /// [`Self::with_pod_uid`] whenever a UID is recorded. - pub fn new(pod_name: PodName) -> Self { - Self { - pod_name, - pod_uid: None, - resource_version: None, - activation_attempt_id: None, - generation: None, - recorded: RecordedExecution::Unread, - } - } - - /// Record what the caller's execution row said about itself, as read at - /// select time. - #[must_use] - pub fn with_recorded(mut self, recorded: RecordedExecution) -> Self { - self.recorded = recorded; - self - } - - #[must_use] - pub fn with_generation(mut self, generation: i64) -> Self { - self.generation = Some(generation); - self - } - - /// The fence as an [`ExecutionIdentity`] — the one shape every guard on - /// this path compares and binds from. - pub fn observed(&self) -> ExecutionIdentity { - ExecutionIdentity { - pod_name: self.pod_name.clone(), - generation: self.generation, - activation_attempt_id: self.activation_attempt_id, - recorded: self.recorded.clone(), - pod: match self.pod_uid.clone() { - Some(pod_uid) => ObservedPod::Seen { - pod_uid, - resource_version: self.resource_version.clone(), - // A caller's observation comes from the filtered reflector, - // which drops terminal pods, so it is never one. - terminal: false, - }, - None => ObservedPod::Unobserved, - }, - } - } - - #[must_use] - pub fn with_activation_attempt( - mut self, - activation_attempt_id: SandboxActivationAttemptId, - ) -> Self { - self.activation_attempt_id = Some(activation_attempt_id); - self - } - - #[must_use] - pub fn with_pod_uid(mut self, pod_uid: PodUid) -> Self { - self.pod_uid = Some(pod_uid); - self - } - - #[must_use] - pub fn with_resource_version(mut self, resource_version: PodResourceVersion) -> Self { - self.resource_version = Some(resource_version); - self - } -} - -/// What the caller observed about the sandbox's execution, and therefore what -/// the substrate has to re-prove before tearing anything down. -/// -/// "I saw no execution" is a real observation and gets its own variant rather -/// than collapsing into "no fence". Collapsing them is a fail-open: a row that -/// had no execution when it was selected is exactly the row a concurrent -/// create or wake is most likely to give one, and the unfenced teardown is the -/// most destructive path there is — it deletes every tracked pod name and -/// releases tracking unconditionally. Having seen nothing must not authorize -/// that. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum RetireFence { - /// Nothing observed; retire whatever is tracked. Correct only for a caller - /// that has already made the pod provably irrelevant — it deleted the pod - /// itself under its own precondition first — not for one that simply has - /// no identity to hand. - Unobserved, - /// The caller observed the sandbox with *no* active execution. The retire - /// must refuse if one exists by the time it runs. - NoExecution, - /// The caller observed this pod. - Pod(PodIdentityFence), -} - -/// A pod's identity as the substrate currently sees it. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ObservedPodIdentity { - pub pod_uid: PodUid, - pub resource_version: Option, -} - -/// Arguments for [`SubstrateAdapter::retire_execution`]. Grouped rather than -/// positional for the same reason [`CreateExecutionRequest`] is. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct RetireExecutionRequest { - pub sandbox_id: SandboxId, - pub fence: RetireFence, -} - -impl RetireExecutionRequest { - pub fn fenced(sandbox_id: SandboxId, pod_fence: PodIdentityFence) -> Self { - Self { - sandbox_id, - fence: RetireFence::Pod(pod_fence), - } - } -} - -/// Which half of a [`PodIdentityFence`] the substrate found violated. Carried -/// so callers can log and meter *why* they deferred without re-deriving it. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum PodFenceMismatch { - /// The fenced pod's execution row is no longer the sandbox's active - /// execution: a newer generation took over. Retiring now would delete the - /// successor's pod, which the caller never observed. - ExecutionSuperseded, - /// No execution row records the fenced pod name at all — or, for a caller - /// that observed no execution, one exists now. - ExecutionMissing, - /// A live pod exists at the fenced name with a different UID: the pod was - /// replaced. - PodUidChanged, - /// A different activation attempt owns the execution now. Same generation, - /// same pod name — a retry or a redrive replaced the attempt the caller - /// selected. - ActivationAttemptChanged, - /// The live pod is the same object but has moved on (bound, status - /// changed) since the caller observed it. - PodResourceVersionChanged, - /// The apiserver rejected the preconditioned delete (409). The pod changed - /// between the check above and the delete itself. - DeletePreconditionRejected, - /// The fence carries no pod UID and a live pod exists at that name, so - /// there is nothing to prove the object is the one the caller observed. - /// A name-only fence may only act on an absent pod. - PodIdentityUnverified, -} - -impl PodFenceMismatch { - /// Whether something the in-VM stop did *not* reach may now own the - /// sandbox. - /// - /// This decides whether [`TeardownProgress::RuntimeStopped`] lets a caller - /// leave its own state transition standing. The stop is addressed to the - /// execution the fence observed, so it proves the sandbox is down only - /// where the mismatch describes *that same object* having moved. Anything - /// else — a new execution, a new attempt, a different pod at the same - /// name, or an identity that could not be proven at all — may be running - /// untouched, and leaving a dormant or terminal lifecycle over it is the - /// stranded state nothing revisits. - /// - /// Exactly one variant is safe to answer `false` to, so the default is - /// `true` and each exception has to earn it. Answering `true` when the - /// sandbox happens to be down costs one re-evaluation on the next tick; - /// answering `false` when it is up strands a live sandbox. - #[must_use] - pub fn successor_took_over(self) -> bool { - match self { - // The same pod object, at a later resourceVersion: a Pending pod - // coming up is the canonical case. The stop reached it. - Self::PodResourceVersionChanged => false, - // A different pod at that name — the stop reached the old one and - // this one is untouched. - Self::PodUidChanged - // A different execution or attempt owns the sandbox now. - | Self::ExecutionSuperseded - | Self::ActivationAttemptChanged - | Self::ExecutionMissing - // The apiserver rejected the delete's precondition, which it - // returns as one 409 whether the pod moved or was replaced. Not - // worth another read to disambiguate: the conservative answer only - // costs a re-evaluation. - | Self::DeletePreconditionRejected - // Nothing could be proven about what is there. - | Self::PodIdentityUnverified => true, - } - } - - pub fn as_str(self) -> &'static str { - match self { - Self::ExecutionSuperseded => "execution_superseded", - Self::ExecutionMissing => "execution_missing", - Self::PodUidChanged => "pod_uid_changed", - Self::ActivationAttemptChanged => "activation_attempt_changed", - Self::PodResourceVersionChanged => "pod_resource_version_changed", - Self::DeletePreconditionRejected => "delete_precondition_rejected", - Self::PodIdentityUnverified => "pod_identity_unverified", - } - } -} - -impl std::fmt::Display for PodFenceMismatch { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(self.as_str()) - } -} - -/// Result of [`SubstrateAdapter::retire_execution`]. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum RetireExecutionOutcome { - /// The execution was retired: node-local state preserved, pod(s) deleted - /// or already absent, pod tracking released. - Retired, - /// A supplied fence no longer matched. This is *defer*, not failure: the - /// row changed under the caller, and whatever changed it owns the sandbox - /// now. `progress` says whether anything had already been done by the time - /// that surfaced, which is what a caller needs to know before undoing a - /// transition it made to enable this call. - PodIdentityChanged { - mismatch: PodFenceMismatch, - progress: TeardownProgress, - }, -} - -impl RetireExecutionOutcome { - /// Whether a caller may undo a state transition it made to enable this - /// retire. - /// - /// One place, because three sweeper passes ask it and a subset that drifts - /// apart is how this fence's bugs have always looked. Two facts decide it, - /// and both have to hold: nothing destructive happened yet, *or* whatever - /// owns the sandbox now is something this teardown never reached. - /// - /// A retire that succeeded is not an undo case at all — the caller's - /// transition is exactly what it wanted. - #[must_use] - pub fn permits_undo(&self) -> bool { - match self { - Self::Retired => false, - Self::PodIdentityChanged { mismatch, progress } => { - !progress.mutated() || mismatch.successor_took_over() - } - } - } -} - -/// How far a refused teardown got before the mismatch surfaced. -/// -/// Not every refusal is free. The fence is re-proven at several points and -/// only the earliest run before anything is mutated — the pod delete's -/// precondition, in particular, is evaluated by the apiserver a round trip -/// after the in-VM stop. A caller that compensates its own state transition on -/// a refusal has to distinguish the two, or it will restore a lifecycle -/// claiming a running VM for a sandbox that is already stopped. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum TeardownProgress { - /// No in-VM stop, no pod delete, no tracking release. - NothingMutated, - /// The observed pod was deleted under its precondition. - PodDeleted, - /// The in-VM stop already ran and its disk backup was enqueued. - RuntimeStopped, -} - -impl TeardownProgress { - #[must_use] - pub fn mutated(self) -> bool { - !matches!(self, Self::NothingMutated) - } -} - pub struct CreateExecutionOutcome { pub lease_completion: Option, pub sandbox_id: SandboxId, @@ -1279,230 +465,6 @@ pub struct CreateExecutionOutcome { pub command_ready: bool, } -pub struct PooledExecutionClaimRequest<'a> { - pub metric_identity: SandboxApiMetricIdentity, - pub actor: &'a ActorRef, - pub template: &'a str, - pub content_hash: Option<&'a str>, - /// Pre-resolved template release id. When supplied, `claim_pooled_execution` - /// skips the `template_releases` lookup the caller already performed. - pub template_release_id: Option, - pub runtime: &'a str, - pub vcpu: Option, - pub memory_mib: Option, - pub env: &'a BTreeMap, - pub auto_lifecycle: &'a AutoLifecyclePolicy, - pub network_policy_override: Option<&'a NetworkPolicyV1>, - pub network_policy_generation: u64, - pub template_network_policy: &'a NetworkPolicyV1, - pub template_network_policy_generation: u64, - /// Routes the claim to the matching encryptable pool partition. Encrypted - /// creates bypass the pool until claim-time re-homing can stamp and apply - /// the flag at adoption. - pub volume_encryption: bool, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct PooledExecutionPoolShape { - pub vcpu: u32, - pub memory_mib: u32, - pub disk_mib: Option, - pub disk_io_limits: pplx_space_common_core::types::DiskIoLimits, -} - -impl PooledExecutionPoolShape { - pub const fn new( - vcpu: u32, - memory_mib: u32, - disk_io_limits: pplx_space_common_core::types::DiskIoLimits, - ) -> Self { - Self { - vcpu, - memory_mib, - disk_mib: None, - disk_io_limits, - } - } - - pub const fn with_disk_mib( - vcpu: u32, - memory_mib: u32, - disk_mib: Option, - disk_io_limits: pplx_space_common_core::types::DiskIoLimits, - ) -> Self { - Self { - vcpu, - memory_mib, - disk_mib, - disk_io_limits, - } - } -} - -pub struct PooledExecutionClaim { - pub claim_id: Uuid, - pub pool_slot_id: Option, - pub sandbox_id: SandboxId, - pub content_hash: String, - pub pod_name: String, - pub pod_uid: Option, - pub execution_generation: u64, - pub pod_ip: String, - pub vcpu: Option, - pub memory_mib: Option, - pub disk_mib: Option, - pub disk_io_limits: pplx_space_common_core::types::DiskIoLimits, - pub active_node_id: Option, - /// Authoritative route identity returned while claiming the execution. - /// Substrates without route identity leave it unset. - pub route_identity: Option, -} - -/// A warm-pool slot claimed for a create when the create was admitted, before -/// it was queued. The create's public sandbox ID is that slot's reserved ID, -/// so the executor must adopt exactly this slot (or build cold under the same -/// ID) and never claim a different one. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct AdmittedPoolClaim { - pub claim_id: Uuid, - pub pool_slot_id: Uuid, -} - -/// A ready warm-pool slot claimed while a create is being admitted, in the -/// transaction that records the create. `sandbox_id` is the slot's reserved -/// ID, which becomes the create's public ID. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct AdmissionPoolClaim { - pub claim: AdmittedPoolClaim, - pub sandbox_id: SandboxId, -} - -/// What the substrate found for an [`AdmittedPoolClaim`]. -pub enum AdmittedPoolClaimLookup { - /// The admitted slot still carries this claim and passed the same pod - /// and placement checks an ordinary claim makes. The caller must persist - /// it with `persist_pooled_execution_claim` or compensate it with - /// `rollback_pooled_execution_claim`. - Adoptable { - claim: Box, - pool_shape: PooledExecutionPoolShape, - }, - /// No unbound slot owns a pod under this sandbox ID any more (the slot - /// was collected, or was bound and retired). A cold create under the same - /// ID is safe once it skips the generation the slot's pod used. - Released, - /// A slot still owns a pod under this sandbox ID that this create will - /// not adopt. It has been taken out of the claimable inventory, and the - /// pool's own cleanup collects it; until then a cold create under this ID - /// would leave that pod with nothing to collect it, so the caller waits. - ReleasePending, -} - -pub struct PooledExecutionClaimAttempt { - pub claim: Option, - /// Exact release-and-shape pool selected by this attempt. A miss can have - /// no shape when no pool exists for the resolved release; callers must not - /// manufacture one from optional create-request fields in that case. - pub pool_shape: Option, - /// Why this attempt missed, when the reason is one a retry cannot change. - /// - /// `None` for an ordinary miss — no slot, a stale slot, a pod that could - /// not be read. Those are all "try again later", which is what the pool's - /// own miss handling already assumes. - /// - /// The substrate does not decide what a refusal means for the caller, - /// because that depends on the boot mode it does not see: a fall-through - /// create builds cold and resolves the same boundary again, while a strict - /// `boot_mode=pool` create has nowhere to fall through to and would - /// otherwise report a permanent refusal as "no slot available yet". - pub refusal: Option, -} - -/// A pooled-claim miss that no retry can turn into a hit. -/// -/// The variant carries the *class* of the answer, so the controller renders it -/// rather than deciding it — only this module knows which condition occurred. -/// The rule is the status the condition warrants, which for every condition the -/// cold path *also* fails on means matching what the cold path answers. One -/// producer has no cold-path counterpart at all; see below. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum PooledClaimRefusal { - /// The sandbox's node-group boundary admits no pooled pod. A conflict: the - /// request is incompatible with this sandbox's placement. - /// - /// Two producers, and they reach the same status differently: - /// - /// * An **empty** candidate set (`NoCompatibleNodeGroup` — an org-specific - /// template whose org has no compatible dedicated group). The cold path - /// maps this to `Conflict`, so the two agree and the caller sees the same - /// 409 either way. - /// * A **dedicated-only** set. Here the cold path does not fail at all: it - /// renders the dedicated affinity and the create succeeds. There is no - /// cold-path status to match, and 409 is the answer specific to a strict - /// `boot_mode=pool` request — a pooled pod is built for no org and lands - /// on shared capacity, so no slot can ever satisfy this sandbox and no - /// retry changes that. - PlacementIncompatible { message: String }, - /// The boundary could not be computed at all, for a reason an operator has - /// to repair — a reaped `templates` row, or a predicate disagreement. The - /// cold path maps both to `Failed`, so this renders as a 500: it is a - /// control-plane fault, not something the caller expressed, and reporting - /// it as a conflict would take the only signal of a missing row out of 5xx - /// alerting. Terminal, so *not* a 503 either — a strict-pool client must - /// not be told to retry what only an operator can fix. - PlacementUnresolvable { message: String }, -} - -impl std::fmt::Debug for PooledExecutionClaimAttempt { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("PooledExecutionClaimAttempt") - .field("claimed", &self.claim.is_some()) - .field("pool_shape", &self.pool_shape) - .field("refusal", &self.refusal) - .finish() - } -} - -impl PooledExecutionClaimAttempt { - pub const fn claimed( - claim: PooledExecutionClaim, - pool_shape: PooledExecutionPoolShape, - ) -> Self { - Self { - claim: Some(claim), - pool_shape: Some(pool_shape), - refusal: None, - } - } - - pub const fn missed(pool_shape: Option) -> Self { - Self { - claim: None, - pool_shape, - refusal: None, - } - } - - /// A miss the caller may need to surface differently. See - /// [`PooledClaimRefusal`]. - pub const fn refused( - pool_shape: Option, - refusal: PooledClaimRefusal, - ) -> Self { - Self { - claim: None, - pool_shape, - refusal: Some(refusal), - } - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum PooledPrepareFailureCleanup { - FinalizeSandboxDbDelete, - DeferredToReconciler, -} - /// Inputs to resolve a node-api gRPC endpoint regardless of substrate. /// /// Substrates that store an endpoint directly populate `direct_endpoint`; @@ -2050,111 +1012,6 @@ pub(crate) fn suspend_rpc_error(err: tonic::Status) -> SubstrateError { } } -/// Capacity classes understood by the substrate admission boundary. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ActivationAdmissionClass { - /// A fresh execution without snapshot restoration. - Cold, - /// An execution restored from a durable sandbox snapshot. - Restore, -} - -/// Adapter-neutral rejection returned by activation admission. -#[derive(Debug, thiserror::Error)] -#[error("{message}")] -pub struct ActivationAdmissionRejection { - message: String, - retry_after: Option, - rate_limited: bool, -} - -impl ActivationAdmissionRejection { - pub(crate) fn new(message: String, retry_after: Option, rate_limited: bool) -> Self { - Self { - message, - retry_after, - rate_limited, - } - } -} - -/// Boxed admission decision future used by [`ActivationAdmission`]. -pub type ActivationAdmissionFuture<'a> = - Pin> + Send + 'a>>; - -/// Capacity admission capability supplied to substrate adapters. -pub trait ActivationAdmission: Send + Sync { - /// Admit or reject one capacity-spending activation. - fn admit(&self, class: ActivationAdmissionClass) -> ActivationAdmissionFuture<'_>; -} - -/// Cluster-activation charge policy for a substrate data-plane call that may -/// recreate a missing pod. Explicit at the substrate boundary so a -/// recovery-capable path can't silently bypass the limiter by omitting a handle, -/// and so "limiter unconfigured" is distinguished from "intentionally exempt". -/// -/// Recovery-capable paths (data-plane access, control-wake, rollback, resume) -/// pass [`RecoveryCharge::Enforce`]. Teardown and already-paid paths pass -/// [`RecoveryCharge::Exempt`]. Read paths that must -/// never recreate a pod should call `reachable_runtime_client` instead of -/// passing a charge policy at all. -// Owned (`Arc`) rather than borrowed: the wake path hands the charge to a -// detached coalesced run that outlives the borrowing request, so the policy -// must not carry a caller lifetime. -#[derive(Clone)] -pub enum RecoveryCharge { - /// Charge activation admission when this path wins ownership of capacity- - /// spending work. Carries `None` only when the process has no limiter - /// configured (then it's a no-op). - Enforce(Option>), - /// Never charge here: a teardown/cleanup path, or one whose activation was - /// already charged upstream. - Exempt, -} - -impl RecoveryCharge { - pub(crate) fn enforce(admission: Option<&Arc>) -> Self - where - T: ActivationAdmission + 'static, - { - Self::Enforce( - admission.map(|admission| Arc::clone(admission) as Arc), - ) - } - - async fn charge(self, class: ActivationAdmissionClass) -> Result<(), SubstrateError> { - let RecoveryCharge::Enforce(Some(admission)) = self else { - return Ok(()); - }; - admission - .admit(class) - .await - .map_err(|error| SubstrateError::AdmissionRejected { - retry_after: error.retry_after, - rate_limited: error.rate_limited, - message: error.to_string(), - }) - } - - /// Charge a Restore activation for a freshly-won restore claim. No-op - /// for [`RecoveryCharge::Exempt`] or an unconfigured limiter. Exactly - /// the claim-winner pays: joining followers never call this, and a - /// lease takeover pays even when its request entered while another - /// restore was in flight. - pub(crate) async fn charge_restore_claim(self) -> Result<(), SubstrateError> { - self.charge(ActivationAdmissionClass::Restore).await - } - - /// Charge a Cold activation if this policy enforces one. No-op for - /// [`RecoveryCharge::Exempt`] or an unconfigured limiter. Only the - /// Kubernetes substrate spends deferred cold admission, so this is unused - /// without that feature. - #[cfg(feature = "kubernetes")] - pub(crate) async fn charge_cold_activation(self) -> Result<(), SubstrateError> { - self.charge(ActivationAdmissionClass::Cold).await - } -} - /// How the substrate addresses a node for node-api calls: which field of a /// [`NodeApiTarget`] must be present. Must match the field the substrate's /// [`SubstrateAdapter::resolve_node_api_endpoint`] requires. @@ -2824,17 +1681,6 @@ where } } -fn retrying_worker_node_api( - endpoint_addr: &str, - request_timeout: Duration, -) -> Result { - let endpoint = node_api_endpoint(endpoint_addr, Duration::from_secs(3), request_timeout)?; - Ok(NodeApiClient::with_service_name( - "worker-node-api", - endpoint, - )) -} - #[derive(Clone)] pub enum Substrate { Direct(Box), @@ -3082,159 +1928,6 @@ pub struct NodeSchedulabilityRequest<'a> { pub substrate_node_id: &'a str, } -pub async fn connect_node_api_client_for_target( - substrate: &A, - target: &NodeApiTarget<'_>, - request_timeout: Duration, -) -> Result<(String, RuntimeClient), SubstrateError> -where - A: SubstrateAdapter + ?Sized, -{ - let endpoint_addr = substrate.resolve_node_api_endpoint(target).await?; - let endpoint = node_api_endpoint(&endpoint_addr, Duration::from_secs(3), request_timeout)?; - let client = connect_worker(endpoint) - .await - .map_err(|err| SubstrateError::Unavailable { - message: format!("node-api not reachable at {endpoint_addr}: {err}"), - })?; - Ok((endpoint_addr, client)) -} - -/// Per-sandbox liveness/salvage summary observed by one node-api -/// `WorkerApi/SandboxState` probe (which proxies the on-node sbx-mgr `State` -/// RPC). Verify-before-destroy reconcilers use this to decide whether a stale -/// control-plane row still has a live VM or salvageable node-local state -/// behind it before taking any destructive action. Live observations carry -/// the node-local lifecycle status when it parses, so repair loops can tell a -/// stale control-plane `failed` row from a node-local `failed` row. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ObservedSandboxState { - /// The Firecracker process is alive on the node — the VM is running. - VmAlive { - node_status: Option, - }, - /// Known to the on-node sbx-mgr, VM not running, and a current-instance - /// local artifact pointer is recorded: the node holds salvageable state - /// that may not be backed up to durable storage yet. - DormantWithArtifact { - node_status: Option, - }, - /// Known to the on-node sbx-mgr, VM not running, and no current-instance - /// local artifact pointer. - DormantNoArtifact { - node_status: Option, - }, - /// The on-node sbx-mgr has no record of the sandbox (`missing_ids`). - Missing, -} - -impl ObservedSandboxState { - pub const fn as_str(self) -> &'static str { - match self { - Self::VmAlive { .. } => "vm_alive", - Self::DormantWithArtifact { .. } => "dormant_with_artifact", - Self::DormantNoArtifact { .. } => "dormant_no_artifact", - Self::Missing => "missing", - } - } - - pub const fn node_status(self) -> Option { - match self { - Self::VmAlive { node_status } - | Self::DormantWithArtifact { node_status } - | Self::DormantNoArtifact { node_status } => node_status, - Self::Missing => None, - } - } -} - -impl std::fmt::Display for ObservedSandboxState { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(self.as_str()) - } -} - -/// Batch-probe the liveness of `sandbox_ids` on one node via the node-api -/// `WorkerApi/SandboxState` RPC. Read-only: nothing on the node changes. -/// -/// Ids the node answered for (present or explicitly missing) appear in the -/// returned map; an id absent from the map means the node's answer did not -/// cover it (a malformed id echo, or a skew where the node returned neither -/// presence nor absence) and callers must treat it as inconclusive rather -/// than missing. The latency contract is the channel timeout the caller -/// configured on `client` — this function adds no waiting of its own. -pub async fn sandbox_states_with_client( - client: &mut RuntimeClient, - endpoint_addr: &str, - sandbox_ids: &[SandboxId], -) -> Result, SubstrateError> { - let response = client - .sandbox_state(pplx_space_sbx_mgr_client::proto::StateRequest { - sandbox_ids: sandbox_ids.iter().map(SandboxId::to_string).collect(), - }) - .await - .map_err(|err| SubstrateError::Unavailable { - message: format!("node-api sandbox-state probe at {endpoint_addr} failed: {err}"), - })? - .into_inner(); - - Ok(observed_sandbox_states(response)) -} - -fn observed_sandbox_states( - response: pplx_space_sbx_mgr_client::proto::StateResponse, -) -> BTreeMap { - let mut states = BTreeMap::new(); - for info in response.sandboxes { - let Ok(sandbox_id) = info.sandbox_id.parse::() else { - // Not one of ours (sbx-mgr ids are UUIDs); leave it out of the - // map so the caller classifies the requested id as inconclusive. - continue; - }; - let node_status = match info.observed_lifecycle() { - Ok(status) => status.label().parse::().ok(), - Err(error) => { - tracing::warn!( - sandbox_id = %sandbox_id, - lifecycle = ?info.lifecycle, - error = %error, - "node-api sandbox-state probe returned unparseable sandbox status" - ); - // Unknown vocabulary is not evidence of a dead sandbox. Omit - // it so the requested-id probe remains inconclusive. - continue; - } - }; - let state = if info.firecracker_alive { - ObservedSandboxState::VmAlive { node_status } - } else if info.current_local_artifact_id.is_some() { - ObservedSandboxState::DormantWithArtifact { node_status } - } else { - ObservedSandboxState::DormantNoArtifact { node_status } - }; - states.insert(sandbox_id, state); - } - for missing in response.missing_ids { - if let Ok(sandbox_id) = missing.parse::() { - states.insert(sandbox_id, ObservedSandboxState::Missing); - } - } - states -} - -pub async fn connect_retrying_node_api_client_for_target( - substrate: &A, - target: &NodeApiTarget<'_>, - request_timeout: Duration, -) -> Result<(String, NodeApiClient), SubstrateError> -where - A: SubstrateAdapter + ?Sized, -{ - let endpoint_addr = substrate.resolve_node_api_endpoint(target).await?; - let client = retrying_worker_node_api(&endpoint_addr, request_timeout)?; - Ok((endpoint_addr, client)) -} - pub async fn enqueue_artifact_backup_with_client( client: &mut RuntimeClient, endpoint_addr: &str, @@ -4248,154 +2941,5 @@ impl SubstrateAdapter for Substrate { } } -/// Wrap a connected node-api `Channel` with the bearer-token interceptor (so a -/// local token read/mount failure fails the call with `Status::unavailable`, not a -/// header-less request the node rejects as a 401) and the client trace layer. -/// Every node-api client is built through this so auth is attached in one place. -pub fn node_api_channel( - channel: Channel, - kind: SpaceBearerToken, -) -> GrpcClientTrace> { - wrap_grpc_client_channel(InterceptedService::new( - channel, - SpaceBearerInterceptor::new(kind), - )) -} - -/// Apply the node-api internal transport to `endpoint` for the TCP connect -/// helpers, carrying the TLS handshake bound the transport applies. -async fn node_api_connect_endpoint( - endpoint: Endpoint, -) -> Result { - pplx_space_common_grpc::internal_transport::client_connect_endpoint( - pplx_space_common_grpc::internal_transport::Service::NodeApi, - endpoint, - ) - .await - .map_err(ControllerError::node_api_transport) -} - -pub(super) async fn require_activation_policy_support( - transport: RuntimeTransport, - timeout: Duration, -) -> Result<(), ControllerError> { - let mut health = tonic_health::pb::health_client::HealthClient::new(transport); - let mut request = tonic::Request::new(tonic_health::pb::HealthCheckRequest { - service: pplx_space_sbx_mgr_client::ACTIVATION_POLICY_HEALTH_SERVICE.to_owned(), - }); - request.set_timeout(timeout); - match tokio::time::timeout(timeout, health.check(request)).await { - Ok(Ok(response)) - if response.get_ref().status - == tonic_health::pb::health_check_response::ServingStatus::Serving as i32 => - { - Ok(()) - } - Ok(Ok(_)) => Err(ControllerError::service_unavailable( - "worker does not support activation policy", - )), - Ok(Err(error)) => Err(ControllerError::service_unavailable(format!( - "worker activation policy capability check failed: {error}" - ))), - Err(_) => Err(ControllerError::service_unavailable( - "worker activation policy capability check timed out", - )), - } -} - -pub async fn connect_worker(endpoint: Endpoint) -> Result { - let channel = node_api_connect_endpoint(endpoint) - .await? - .connect_with_retry( - "worker-node-api", - "worker-node-api", - pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, - ) - .await - .map_err(|err| { - ControllerError::service_unavailable(format!("node-api not reachable: {err}")) - })?; - Ok(RuntimeClient::new(channel)) -} - -/// HTTP/2 connection window for a kept worker connection: 16 default-sized -/// (2 MiB) stream windows. -#[cfg(feature = "kubernetes")] -const KEPT_WORKER_CONNECTION_WINDOW_BYTES: u32 = 32 * 1024 * 1024; - -/// [`connect_worker`] for a channel a cache keeps: the same dial, plus the -/// watch that closes once the connection it was dialed with ends. -#[cfg(feature = "kubernetes")] -async fn connect_worker_watched( - endpoint: Endpoint, -) -> Result<(Channel, pplx_space_common_grpc::ConnectionWatch), ControllerError> { - // A kept connection carries every call to its pod, so it needs its own - // liveness check: ping it while idle too, so a peer lost without a FIN is - // torn down before the next caller reuses it, rather than only once a - // call is in flight. - // - // It also carries concurrent streams that used to get a connection each. - // A reader that stops polling holds up to one stream window of the - // connection window; a window several streams wide keeps a few stalled - // readers from blocking the pod's other calls. Buffering stays bounded by - // the stream windows, as it was across separate connections. - let endpoint = endpoint - .keep_alive_while_idle(true) - .initial_connection_window_size(KEPT_WORKER_CONNECTION_WINDOW_BYTES); - node_api_connect_endpoint(endpoint) - .await? - .connect_with_retry_watched( - "worker-node-api", - "worker-node-api", - pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, - ) - .await - .map_err(|err| { - ControllerError::service_unavailable(format!("node-api not reachable: {err}")) - }) -} - -pub async fn connect_worker_internal( - endpoint: Endpoint, -) -> Result { - let channel = node_api_connect_endpoint(endpoint) - .await? - .connect_with_retry( - "worker-node-api-internal", - "worker-node-api-internal", - pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, - ) - .await - .map_err(|err| { - ControllerError::service_unavailable(format!("node-api not reachable: {err}")) - })?; - Ok(WorkerApiInternalClient::new(wrap_grpc_client_channel( - InterceptedService::new( - channel, - SpaceBearerInterceptor::new(SpaceBearerToken::NodeApiInternal), - ), - ))) -} - -pub async fn connect_builder(endpoint: Endpoint) -> Result { - let channel = node_api_connect_endpoint(endpoint) - .await? - .connect_with_retry( - "builder-node-api", - "builder-node-api", - pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, - ) - .await - .map_err(|err| { - ControllerError::service_unavailable(format!("node-api not reachable: {err}")) - })?; - Ok(BuilderApiClient::new(wrap_grpc_client_channel( - InterceptedService::new( - channel, - SpaceBearerInterceptor::new(SpaceBearerToken::NodeApi), - ), - ))) -} - #[cfg(test)] mod tests; diff --git a/crates/libs/api/src/controllers/substrate/node_api_client.rs b/crates/libs/api/src/controllers/substrate/node_api_client.rs new file mode 100644 index 000000000..d30ad2f25 --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/node_api_client.rs @@ -0,0 +1,361 @@ +//! Clients for a node's node-api (worker, runtime, builder) and the RPC budgets they share. + +use super::*; + +pub(super) type RuntimeTransport = + GrpcClientTrace>; + +/// Runtime RPCs and capability checks share the resolved worker connection. +#[derive(Clone, Debug)] +pub struct RuntimeClient { + pub(super) worker: WorkerApiClient, + pub(super) channel: Channel, + /// Set when `channel` is a cached channel that other callers share. + pub(super) kept: Option, +} + +impl RuntimeClient { + pub(super) fn new(channel: Channel) -> Self { + let transport = node_api_channel(channel.clone(), SpaceBearerToken::NodeApi); + Self { + worker: WorkerApiClient::new(transport), + channel, + kept: None, + } + } + + /// A client on a channel a cache keeps while `watch` is open. + #[cfg(feature = "kubernetes")] + pub(super) fn kept(channel: Channel, watch: pplx_space_common_grpc::ConnectionWatch) -> Self { + Self { + kept: Some(watch), + ..Self::new(channel) + } + } + + /// Stop handing this client's cached channel to later callers, so the + /// next one dials afresh. Call it when the peer showed that new calls + /// should reach a new connection, such as a draining node-api; this + /// client and its clones keep working. A no-op for an uncached channel. + pub(crate) fn retire_kept_connection(&self) { + if let Some(watch) = &self.kept { + watch.close(); + } + } + + pub(in crate::controllers) async fn require_activation_policy_support( + &mut self, + timeout: Duration, + ) -> Result<(), ControllerError> { + require_activation_policy_support( + node_api_channel(self.channel.clone(), SpaceBearerToken::NodeApi), + timeout, + ) + .await + } + + pub(super) fn internal_client(&self) -> RuntimeInternalClient { + WorkerApiInternalClient::new(node_api_channel( + self.channel.clone(), + SpaceBearerToken::NodeApiInternal, + )) + } +} + +impl std::ops::Deref for RuntimeClient { + type Target = WorkerApiClient; + + fn deref(&self) -> &Self::Target { + &self.worker + } +} + +impl std::ops::DerefMut for RuntimeClient { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.worker + } +} + +pub type RuntimeInternalClient = + WorkerApiInternalClient>>; + +pub type BuilderClient = + BuilderApiClient>>; + +/// Per-request node-RPC channel timeout (the tonic `Endpoint::timeout` on +/// worker/guest channels). This — NOT the ALB — is the binding internal cut +/// behind the create/restore latency cliff: a slow-but-not-stuck node RPC that +/// rides up to it is severed by tonic as `Code::Cancelled`, which surfaces as +/// an opaque HTTP 408 with no sandbox id and no retry semantics. Every API-side +/// blocking wait must finish, *and* a structured retryable response must flush, +/// below it. (The ECW ALB idle timeout is far higher — see +/// [`ALB_IDLE_TIMEOUT_SECS`] — so the ALB is not the binding constraint.) +/// +/// The 150s timeout leaves a 135s blocking budget after reserving 15s for +/// completing the response. +/// +/// The value lives in `common-core`'s `restore_ownership` module because the +/// placement crate's restore lease TTL must equal it; deriving both from one +/// constant is what keeps a future timeout bump from silently splitting the +/// lease-outlives-RPC invariant across crates. +pub(crate) const NODE_RPC_TIMEOUT: Duration = + pplx_space_common_core::restore_ownership::NODE_RPC_TIMEOUT; + +pub(crate) const NODE_RPC_TIMEOUT_SECS: u64 = NODE_RPC_TIMEOUT.as_secs(); + +pub(super) const NODE_API_HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(10); + +pub(super) const NODE_API_HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(5); + +/// Headroom below [`NODE_RPC_TIMEOUT`] for the API poll loop / wait wrapper to +/// detect exhaustion and return a structured, retryable response (carrying the +/// sandbox id + `Retry-After`) before tonic severs the channel with a bare +/// `Cancelled` 408. +pub(super) const BLOCKING_WAIT_MARGIN_SECS: u64 = 15; + +/// API-side blocking-wait budget for create/restore/wake/access, held below +/// [`NODE_RPC_TIMEOUT`] so a slow wait surfaces as a structured retryable +/// `SubstrateError::Timeout` before the per-request channel timeout fires. The +/// single budget every blocking lifecycle wait is sized against. +/// +/// `pub` (not `pub(crate)`) so out-of-crate callers that pass a +/// `ready_timeout` — e.g. the cluster-controller's node-drain restore — +/// can report the effective (clamped) per-attempt wait instead of +/// hand-copying this value into comments and logs. +pub const NODE_RPC_BLOCKING_BUDGET: Duration = + Duration::from_secs(NODE_RPC_TIMEOUT_SECS - BLOCKING_WAIT_MARGIN_SECS); + +/// The ECW ALB idle timeout (`idle_timeout.timeout_seconds`), set via the AWS +/// LB controller annotation on the deployment's Ingress. It is the OUTER +/// ceiling, far above the node-RPC +/// timeout — documented here so the hierarchy is explicit and version-checked; +/// keep it in sync with the Helm values. The static assert pins +/// `NODE_RPC_TIMEOUT < ALB idle`, so the binding cut can never silently exceed +/// the gateway. +pub(crate) const ALB_IDLE_TIMEOUT_SECS: u64 = 1800; + +const _: () = assert!( + NODE_RPC_TIMEOUT_SECS < ALB_IDLE_TIMEOUT_SECS, + "node-RPC channel timeout must stay below the ALB idle timeout", +); + +pub(crate) fn node_api_endpoint( + addr: &str, + connect_timeout: Duration, + request_timeout: Duration, +) -> Result { + let uri = pplx_space_common_grpc::internal_transport::qualify_endpoint(addr); + Endpoint::from_shared(uri) + .map(|endpoint| configure_node_api_endpoint(endpoint, connect_timeout, request_timeout)) + .map_err(|err| SubstrateError::Failed { + message: format!("invalid node-api endpoint '{addr}': {err}"), + }) +} + +pub(super) fn configure_node_api_endpoint( + endpoint: Endpoint, + connect_timeout: Duration, + request_timeout: Duration, +) -> Endpoint { + pplx_space_node_api_client::configure_node_api_endpoint( + endpoint, + NodeApiEndpointConfig::default() + .with_connect_timeout(connect_timeout) + .with_request_timeout(request_timeout) + .with_http2_keep_alive( + NODE_API_HTTP2_KEEP_ALIVE_INTERVAL, + NODE_API_HTTP2_KEEP_ALIVE_TIMEOUT, + ), + ) +} + +pub(super) fn retrying_worker_node_api( + endpoint_addr: &str, + request_timeout: Duration, +) -> Result { + let endpoint = node_api_endpoint(endpoint_addr, Duration::from_secs(3), request_timeout)?; + Ok(NodeApiClient::with_service_name( + "worker-node-api", + endpoint, + )) +} + +pub async fn connect_node_api_client_for_target( + substrate: &A, + target: &NodeApiTarget<'_>, + request_timeout: Duration, +) -> Result<(String, RuntimeClient), SubstrateError> +where + A: SubstrateAdapter + ?Sized, +{ + let endpoint_addr = substrate.resolve_node_api_endpoint(target).await?; + let endpoint = node_api_endpoint(&endpoint_addr, Duration::from_secs(3), request_timeout)?; + let client = connect_worker(endpoint) + .await + .map_err(|err| SubstrateError::Unavailable { + message: format!("node-api not reachable at {endpoint_addr}: {err}"), + })?; + Ok((endpoint_addr, client)) +} + +pub async fn connect_retrying_node_api_client_for_target( + substrate: &A, + target: &NodeApiTarget<'_>, + request_timeout: Duration, +) -> Result<(String, NodeApiClient), SubstrateError> +where + A: SubstrateAdapter + ?Sized, +{ + let endpoint_addr = substrate.resolve_node_api_endpoint(target).await?; + let client = retrying_worker_node_api(&endpoint_addr, request_timeout)?; + Ok((endpoint_addr, client)) +} + +/// Wrap a connected node-api `Channel` with the bearer-token interceptor (so a +/// local token read/mount failure fails the call with `Status::unavailable`, not a +/// header-less request the node rejects as a 401) and the client trace layer. +/// Every node-api client is built through this so auth is attached in one place. +pub fn node_api_channel( + channel: Channel, + kind: SpaceBearerToken, +) -> GrpcClientTrace> { + wrap_grpc_client_channel(InterceptedService::new( + channel, + SpaceBearerInterceptor::new(kind), + )) +} + +/// Apply the node-api internal transport to `endpoint` for the TCP connect +/// helpers, carrying the TLS handshake bound the transport applies. +pub(super) async fn node_api_connect_endpoint( + endpoint: Endpoint, +) -> Result { + pplx_space_common_grpc::internal_transport::client_connect_endpoint( + pplx_space_common_grpc::internal_transport::Service::NodeApi, + endpoint, + ) + .await + .map_err(ControllerError::node_api_transport) +} + +pub(in crate::controllers) async fn require_activation_policy_support( + transport: RuntimeTransport, + timeout: Duration, +) -> Result<(), ControllerError> { + let mut health = tonic_health::pb::health_client::HealthClient::new(transport); + let mut request = tonic::Request::new(tonic_health::pb::HealthCheckRequest { + service: pplx_space_sbx_mgr_client::ACTIVATION_POLICY_HEALTH_SERVICE.to_owned(), + }); + request.set_timeout(timeout); + match tokio::time::timeout(timeout, health.check(request)).await { + Ok(Ok(response)) + if response.get_ref().status + == tonic_health::pb::health_check_response::ServingStatus::Serving as i32 => + { + Ok(()) + } + Ok(Ok(_)) => Err(ControllerError::service_unavailable( + "worker does not support activation policy", + )), + Ok(Err(error)) => Err(ControllerError::service_unavailable(format!( + "worker activation policy capability check failed: {error}" + ))), + Err(_) => Err(ControllerError::service_unavailable( + "worker activation policy capability check timed out", + )), + } +} + +pub async fn connect_worker(endpoint: Endpoint) -> Result { + let channel = node_api_connect_endpoint(endpoint) + .await? + .connect_with_retry( + "worker-node-api", + "worker-node-api", + pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, + ) + .await + .map_err(|err| { + ControllerError::service_unavailable(format!("node-api not reachable: {err}")) + })?; + Ok(RuntimeClient::new(channel)) +} + +/// HTTP/2 connection window for a kept worker connection: 16 default-sized +/// (2 MiB) stream windows. +#[cfg(feature = "kubernetes")] +pub(super) const KEPT_WORKER_CONNECTION_WINDOW_BYTES: u32 = 32 * 1024 * 1024; + +/// [`connect_worker`] for a channel a cache keeps: the same dial, plus the +/// watch that closes once the connection it was dialed with ends. +#[cfg(feature = "kubernetes")] +pub(super) async fn connect_worker_watched( + endpoint: Endpoint, +) -> Result<(Channel, pplx_space_common_grpc::ConnectionWatch), ControllerError> { + // A kept connection carries every call to its pod, so it needs its own + // liveness check: ping it while idle too, so a peer lost without a FIN is + // torn down before the next caller reuses it, rather than only once a + // call is in flight. + // + // It also carries concurrent streams that used to get a connection each. + // A reader that stops polling holds up to one stream window of the + // connection window; a window several streams wide keeps a few stalled + // readers from blocking the pod's other calls. Buffering stays bounded by + // the stream windows, as it was across separate connections. + let endpoint = endpoint + .keep_alive_while_idle(true) + .initial_connection_window_size(KEPT_WORKER_CONNECTION_WINDOW_BYTES); + node_api_connect_endpoint(endpoint) + .await? + .connect_with_retry_watched( + "worker-node-api", + "worker-node-api", + pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, + ) + .await + .map_err(|err| { + ControllerError::service_unavailable(format!("node-api not reachable: {err}")) + }) +} + +pub async fn connect_worker_internal( + endpoint: Endpoint, +) -> Result { + let channel = node_api_connect_endpoint(endpoint) + .await? + .connect_with_retry( + "worker-node-api-internal", + "worker-node-api-internal", + pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, + ) + .await + .map_err(|err| { + ControllerError::service_unavailable(format!("node-api not reachable: {err}")) + })?; + Ok(WorkerApiInternalClient::new(wrap_grpc_client_channel( + InterceptedService::new( + channel, + SpaceBearerInterceptor::new(SpaceBearerToken::NodeApiInternal), + ), + ))) +} + +pub async fn connect_builder(endpoint: Endpoint) -> Result { + let channel = node_api_connect_endpoint(endpoint) + .await? + .connect_with_retry( + "builder-node-api", + "builder-node-api", + pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY, + ) + .await + .map_err(|err| { + ControllerError::service_unavailable(format!("node-api not reachable: {err}")) + })?; + Ok(BuilderApiClient::new(wrap_grpc_client_channel( + InterceptedService::new( + channel, + SpaceBearerInterceptor::new(SpaceBearerToken::NodeApi), + ), + ))) +} diff --git a/crates/libs/api/src/controllers/substrate/pool_claims.rs b/crates/libs/api/src/controllers/substrate/pool_claims.rs new file mode 100644 index 000000000..fd2c45745 --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/pool_claims.rs @@ -0,0 +1,258 @@ +//! Warm-pool eligibility and the request, claim and refusal types of a pooled execution claim. + +use super::*; + +pub(crate) fn default_pool_eligible_for_request( + boot_mode: CreateBootMode, + backend: &str, + wait_until_ready: bool, + disk_mib: Option, + file_changes_enabled: bool, + implicit_wake_enabled: bool, +) -> bool { + backend == "fc" + && matches!( + boot_mode, + CreateBootMode::Fallthrough | CreateBootMode::Pool + ) + && disk_mib.is_none() + && !file_changes_enabled + && (wait_until_ready || implicit_wake_enabled) +} + +/// Combine a substrate's generic warm-pool eligibility with env handling. A cold +/// create bakes `req.env` into the sandbox at boot; if the substrate can't +/// reproduce that on a pooled sandbox at adoption (`substrate_supports_pool_env` +/// is false), an env-bearing create must take the cold path rather than silently +/// drop the requested variables. +pub(crate) fn pool_eligible_with_env( + substrate_eligible: bool, + substrate_supports_pool_env: bool, + has_request_env: bool, +) -> bool { + substrate_eligible && (substrate_supports_pool_env || !has_request_env) +} + +pub struct PooledExecutionClaimRequest<'a> { + pub metric_identity: SandboxApiMetricIdentity, + pub actor: &'a ActorRef, + pub template: &'a str, + pub content_hash: Option<&'a str>, + /// Pre-resolved template release id. When supplied, `claim_pooled_execution` + /// skips the `template_releases` lookup the caller already performed. + pub template_release_id: Option, + pub runtime: &'a str, + pub vcpu: Option, + pub memory_mib: Option, + pub env: &'a BTreeMap, + pub auto_lifecycle: &'a AutoLifecyclePolicy, + pub network_policy_override: Option<&'a NetworkPolicyV1>, + pub network_policy_generation: u64, + pub template_network_policy: &'a NetworkPolicyV1, + pub template_network_policy_generation: u64, + /// Routes the claim to the matching encryptable pool partition. Encrypted + /// creates bypass the pool until claim-time re-homing can stamp and apply + /// the flag at adoption. + pub volume_encryption: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PooledExecutionPoolShape { + pub vcpu: u32, + pub memory_mib: u32, + pub disk_mib: Option, + pub disk_io_limits: pplx_space_common_core::types::DiskIoLimits, +} + +impl PooledExecutionPoolShape { + pub const fn new( + vcpu: u32, + memory_mib: u32, + disk_io_limits: pplx_space_common_core::types::DiskIoLimits, + ) -> Self { + Self { + vcpu, + memory_mib, + disk_mib: None, + disk_io_limits, + } + } + + pub const fn with_disk_mib( + vcpu: u32, + memory_mib: u32, + disk_mib: Option, + disk_io_limits: pplx_space_common_core::types::DiskIoLimits, + ) -> Self { + Self { + vcpu, + memory_mib, + disk_mib, + disk_io_limits, + } + } +} + +pub struct PooledExecutionClaim { + pub claim_id: Uuid, + pub pool_slot_id: Option, + pub sandbox_id: SandboxId, + pub content_hash: String, + pub pod_name: String, + pub pod_uid: Option, + pub execution_generation: u64, + pub pod_ip: String, + pub vcpu: Option, + pub memory_mib: Option, + pub disk_mib: Option, + pub disk_io_limits: pplx_space_common_core::types::DiskIoLimits, + pub active_node_id: Option, + /// Authoritative route identity returned while claiming the execution. + /// Substrates without route identity leave it unset. + pub route_identity: Option, +} + +/// A warm-pool slot claimed for a create when the create was admitted, before +/// it was queued. The create's public sandbox ID is that slot's reserved ID, +/// so the executor must adopt exactly this slot (or build cold under the same +/// ID) and never claim a different one. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AdmittedPoolClaim { + pub claim_id: Uuid, + pub pool_slot_id: Uuid, +} + +/// A ready warm-pool slot claimed while a create is being admitted, in the +/// transaction that records the create. `sandbox_id` is the slot's reserved +/// ID, which becomes the create's public ID. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AdmissionPoolClaim { + pub claim: AdmittedPoolClaim, + pub sandbox_id: SandboxId, +} + +/// What the substrate found for an [`AdmittedPoolClaim`]. +pub enum AdmittedPoolClaimLookup { + /// The admitted slot still carries this claim and passed the same pod + /// and placement checks an ordinary claim makes. The caller must persist + /// it with `persist_pooled_execution_claim` or compensate it with + /// `rollback_pooled_execution_claim`. + Adoptable { + claim: Box, + pool_shape: PooledExecutionPoolShape, + }, + /// No unbound slot owns a pod under this sandbox ID any more (the slot + /// was collected, or was bound and retired). A cold create under the same + /// ID is safe once it skips the generation the slot's pod used. + Released, + /// A slot still owns a pod under this sandbox ID that this create will + /// not adopt. It has been taken out of the claimable inventory, and the + /// pool's own cleanup collects it; until then a cold create under this ID + /// would leave that pod with nothing to collect it, so the caller waits. + ReleasePending, +} + +pub struct PooledExecutionClaimAttempt { + pub claim: Option, + /// Exact release-and-shape pool selected by this attempt. A miss can have + /// no shape when no pool exists for the resolved release; callers must not + /// manufacture one from optional create-request fields in that case. + pub pool_shape: Option, + /// Why this attempt missed, when the reason is one a retry cannot change. + /// + /// `None` for an ordinary miss — no slot, a stale slot, a pod that could + /// not be read. Those are all "try again later", which is what the pool's + /// own miss handling already assumes. + /// + /// The substrate does not decide what a refusal means for the caller, + /// because that depends on the boot mode it does not see: a fall-through + /// create builds cold and resolves the same boundary again, while a strict + /// `boot_mode=pool` create has nowhere to fall through to and would + /// otherwise report a permanent refusal as "no slot available yet". + pub refusal: Option, +} + +/// A pooled-claim miss that no retry can turn into a hit. +/// +/// The variant carries the *class* of the answer, so the controller renders it +/// rather than deciding it — only this module knows which condition occurred. +/// The rule is the status the condition warrants, which for every condition the +/// cold path *also* fails on means matching what the cold path answers. One +/// producer has no cold-path counterpart at all; see below. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PooledClaimRefusal { + /// The sandbox's node-group boundary admits no pooled pod. A conflict: the + /// request is incompatible with this sandbox's placement. + /// + /// Two producers, and they reach the same status differently: + /// + /// * An **empty** candidate set (`NoCompatibleNodeGroup` — an org-specific + /// template whose org has no compatible dedicated group). The cold path + /// maps this to `Conflict`, so the two agree and the caller sees the same + /// 409 either way. + /// * A **dedicated-only** set. Here the cold path does not fail at all: it + /// renders the dedicated affinity and the create succeeds. There is no + /// cold-path status to match, and 409 is the answer specific to a strict + /// `boot_mode=pool` request — a pooled pod is built for no org and lands + /// on shared capacity, so no slot can ever satisfy this sandbox and no + /// retry changes that. + PlacementIncompatible { message: String }, + /// The boundary could not be computed at all, for a reason an operator has + /// to repair — a reaped `templates` row, or a predicate disagreement. The + /// cold path maps both to `Failed`, so this renders as a 500: it is a + /// control-plane fault, not something the caller expressed, and reporting + /// it as a conflict would take the only signal of a missing row out of 5xx + /// alerting. Terminal, so *not* a 503 either — a strict-pool client must + /// not be told to retry what only an operator can fix. + PlacementUnresolvable { message: String }, +} + +impl std::fmt::Debug for PooledExecutionClaimAttempt { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PooledExecutionClaimAttempt") + .field("claimed", &self.claim.is_some()) + .field("pool_shape", &self.pool_shape) + .field("refusal", &self.refusal) + .finish() + } +} + +impl PooledExecutionClaimAttempt { + pub const fn claimed( + claim: PooledExecutionClaim, + pool_shape: PooledExecutionPoolShape, + ) -> Self { + Self { + claim: Some(claim), + pool_shape: Some(pool_shape), + refusal: None, + } + } + + pub const fn missed(pool_shape: Option) -> Self { + Self { + claim: None, + pool_shape, + refusal: None, + } + } + + /// A miss the caller may need to surface differently. See + /// [`PooledClaimRefusal`]. + pub const fn refused( + pool_shape: Option, + refusal: PooledClaimRefusal, + ) -> Self { + Self { + claim: None, + pool_shape, + refusal: Some(refusal), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PooledPrepareFailureCleanup { + FinalizeSandboxDbDelete, + DeferredToReconciler, +} diff --git a/crates/libs/api/src/controllers/substrate/recovery_charge.rs b/crates/libs/api/src/controllers/substrate/recovery_charge.rs new file mode 100644 index 000000000..20cf0a3d8 --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/recovery_charge.rs @@ -0,0 +1,111 @@ +//! Activation admission and the recovery charge a cold or restored activation pays. + +use super::*; + +/// Capacity classes understood by the substrate admission boundary. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ActivationAdmissionClass { + /// A fresh execution without snapshot restoration. + Cold, + /// An execution restored from a durable sandbox snapshot. + Restore, +} + +/// Adapter-neutral rejection returned by activation admission. +#[derive(Debug, thiserror::Error)] +#[error("{message}")] +pub struct ActivationAdmissionRejection { + pub(super) message: String, + pub(super) retry_after: Option, + pub(super) rate_limited: bool, +} + +impl ActivationAdmissionRejection { + pub(crate) fn new(message: String, retry_after: Option, rate_limited: bool) -> Self { + Self { + message, + retry_after, + rate_limited, + } + } +} + +/// Boxed admission decision future used by [`ActivationAdmission`]. +pub type ActivationAdmissionFuture<'a> = + Pin> + Send + 'a>>; + +/// Capacity admission capability supplied to substrate adapters. +pub trait ActivationAdmission: Send + Sync { + /// Admit or reject one capacity-spending activation. + fn admit(&self, class: ActivationAdmissionClass) -> ActivationAdmissionFuture<'_>; +} + +/// Cluster-activation charge policy for a substrate data-plane call that may +/// recreate a missing pod. Explicit at the substrate boundary so a +/// recovery-capable path can't silently bypass the limiter by omitting a handle, +/// and so "limiter unconfigured" is distinguished from "intentionally exempt". +/// +/// Recovery-capable paths (data-plane access, control-wake, rollback, resume) +/// pass [`RecoveryCharge::Enforce`]. Teardown and already-paid paths pass +/// [`RecoveryCharge::Exempt`]. Read paths that must +/// never recreate a pod should call `reachable_runtime_client` instead of +/// passing a charge policy at all. +// Owned (`Arc`) rather than borrowed: the wake path hands the charge to a +// detached coalesced run that outlives the borrowing request, so the policy +// must not carry a caller lifetime. +#[derive(Clone)] +pub enum RecoveryCharge { + /// Charge activation admission when this path wins ownership of capacity- + /// spending work. Carries `None` only when the process has no limiter + /// configured (then it's a no-op). + Enforce(Option>), + /// Never charge here: a teardown/cleanup path, or one whose activation was + /// already charged upstream. + Exempt, +} + +impl RecoveryCharge { + pub(crate) fn enforce(admission: Option<&Arc>) -> Self + where + T: ActivationAdmission + 'static, + { + Self::Enforce( + admission.map(|admission| Arc::clone(admission) as Arc), + ) + } + + pub(super) async fn charge( + self, + class: ActivationAdmissionClass, + ) -> Result<(), SubstrateError> { + let RecoveryCharge::Enforce(Some(admission)) = self else { + return Ok(()); + }; + admission + .admit(class) + .await + .map_err(|error| SubstrateError::AdmissionRejected { + retry_after: error.retry_after, + rate_limited: error.rate_limited, + message: error.to_string(), + }) + } + + /// Charge a Restore activation for a freshly-won restore claim. No-op + /// for [`RecoveryCharge::Exempt`] or an unconfigured limiter. Exactly + /// the claim-winner pays: joining followers never call this, and a + /// lease takeover pays even when its request entered while another + /// restore was in flight. + pub(crate) async fn charge_restore_claim(self) -> Result<(), SubstrateError> { + self.charge(ActivationAdmissionClass::Restore).await + } + + /// Charge a Cold activation if this policy enforces one. No-op for + /// [`RecoveryCharge::Exempt`] or an unconfigured limiter. Only the + /// Kubernetes substrate spends deferred cold admission, so this is unused + /// without that feature. + #[cfg(feature = "kubernetes")] + pub(crate) async fn charge_cold_activation(self) -> Result<(), SubstrateError> { + self.charge(ActivationAdmissionClass::Cold).await + } +} diff --git a/crates/libs/api/src/controllers/substrate/sandbox_state_probe.rs b/crates/libs/api/src/controllers/substrate/sandbox_state_probe.rs new file mode 100644 index 000000000..56404610a --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/sandbox_state_probe.rs @@ -0,0 +1,125 @@ +//! Reads of the sandbox states a node reports, used to verify before destroying. + +use super::*; + +/// Per-sandbox liveness/salvage summary observed by one node-api +/// `WorkerApi/SandboxState` probe (which proxies the on-node sbx-mgr `State` +/// RPC). Verify-before-destroy reconcilers use this to decide whether a stale +/// control-plane row still has a live VM or salvageable node-local state +/// behind it before taking any destructive action. Live observations carry +/// the node-local lifecycle status when it parses, so repair loops can tell a +/// stale control-plane `failed` row from a node-local `failed` row. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ObservedSandboxState { + /// The Firecracker process is alive on the node — the VM is running. + VmAlive { + node_status: Option, + }, + /// Known to the on-node sbx-mgr, VM not running, and a current-instance + /// local artifact pointer is recorded: the node holds salvageable state + /// that may not be backed up to durable storage yet. + DormantWithArtifact { + node_status: Option, + }, + /// Known to the on-node sbx-mgr, VM not running, and no current-instance + /// local artifact pointer. + DormantNoArtifact { + node_status: Option, + }, + /// The on-node sbx-mgr has no record of the sandbox (`missing_ids`). + Missing, +} + +impl ObservedSandboxState { + pub const fn as_str(self) -> &'static str { + match self { + Self::VmAlive { .. } => "vm_alive", + Self::DormantWithArtifact { .. } => "dormant_with_artifact", + Self::DormantNoArtifact { .. } => "dormant_no_artifact", + Self::Missing => "missing", + } + } + + pub const fn node_status(self) -> Option { + match self { + Self::VmAlive { node_status } + | Self::DormantWithArtifact { node_status } + | Self::DormantNoArtifact { node_status } => node_status, + Self::Missing => None, + } + } +} + +impl std::fmt::Display for ObservedSandboxState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Batch-probe the liveness of `sandbox_ids` on one node via the node-api +/// `WorkerApi/SandboxState` RPC. Read-only: nothing on the node changes. +/// +/// Ids the node answered for (present or explicitly missing) appear in the +/// returned map; an id absent from the map means the node's answer did not +/// cover it (a malformed id echo, or a skew where the node returned neither +/// presence nor absence) and callers must treat it as inconclusive rather +/// than missing. The latency contract is the channel timeout the caller +/// configured on `client` — this function adds no waiting of its own. +pub async fn sandbox_states_with_client( + client: &mut RuntimeClient, + endpoint_addr: &str, + sandbox_ids: &[SandboxId], +) -> Result, SubstrateError> { + let response = client + .sandbox_state(pplx_space_sbx_mgr_client::proto::StateRequest { + sandbox_ids: sandbox_ids.iter().map(SandboxId::to_string).collect(), + }) + .await + .map_err(|err| SubstrateError::Unavailable { + message: format!("node-api sandbox-state probe at {endpoint_addr} failed: {err}"), + })? + .into_inner(); + + Ok(observed_sandbox_states(response)) +} + +pub(super) fn observed_sandbox_states( + response: pplx_space_sbx_mgr_client::proto::StateResponse, +) -> BTreeMap { + let mut states = BTreeMap::new(); + for info in response.sandboxes { + let Ok(sandbox_id) = info.sandbox_id.parse::() else { + // Not one of ours (sbx-mgr ids are UUIDs); leave it out of the + // map so the caller classifies the requested id as inconclusive. + continue; + }; + let node_status = match info.observed_lifecycle() { + Ok(status) => status.label().parse::().ok(), + Err(error) => { + tracing::warn!( + sandbox_id = %sandbox_id, + lifecycle = ?info.lifecycle, + error = %error, + "node-api sandbox-state probe returned unparseable sandbox status" + ); + // Unknown vocabulary is not evidence of a dead sandbox. Omit + // it so the requested-id probe remains inconclusive. + continue; + } + }; + let state = if info.firecracker_alive { + ObservedSandboxState::VmAlive { node_status } + } else if info.current_local_artifact_id.is_some() { + ObservedSandboxState::DormantWithArtifact { node_status } + } else { + ObservedSandboxState::DormantNoArtifact { node_status } + }; + states.insert(sandbox_id, state); + } + for missing in response.missing_ids { + if let Ok(sandbox_id) = missing.parse::() { + states.insert(sandbox_id, ObservedSandboxState::Missing); + } + } + states +} diff --git a/infra/space/kind/charts/space-node-api-local/values.yaml b/infra/space/kind/charts/space-node-api-local/values.yaml index 594af3b67..0e9ab01a0 100644 --- a/infra/space/kind/charts/space-node-api-local/values.yaml +++ b/infra/space/kind/charts/space-node-api-local/values.yaml @@ -44,7 +44,7 @@ auth: - space-builder-node-api # WorkerApiInternal (audience space-node-api-internal): only the api's # substrate dials it, from the api and the cluster-controller - # (internal_client, substrate/mod.rs; + # (internal_client, substrate/node_api_client.rs; # substrate/kubernetes/pod_ops/worker_clients.rs). SPACE_NODE_API_INTERNAL_AUTH_ALLOWED_SERVICE_ACCOUNTS: - space-api