From 1a95969f39987efd4c6d381359bff80a89471113 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Mon, 12 Oct 2026 19:47:30 +0000 Subject: [PATCH] api: let tests inject the Kubernetes readiness and suspend budgets KubernetesSubstrate now carries a SubstrateBudgets value (new kubernetes/budgets.rs) for six wall-clock budgets that were compile-time constants at their use sites: - create_ready (CREATE_READY_BUDGET): create_execution's outer readiness budget, its timeout message, and the phase deadline of a create that joins an in-flight activation; - create_ready_phase_margin (READY_WAIT_PHASE_MARGIN), as reserved inside create_ready; - preserve_decision (PRESERVE_DECISION_BUDGET); - suspend_durability (TEARDOWN_SUSPEND_DURABILITY_BUDGET): suspend_execution's deadline and the budget named in its timeout error; - pooled_split_receipt_wait (WARM_POOL_SETUP_RPC_TIMEOUT), as the started warm-pool adoption's split-receipt deadline; - teardown_worker_connect (POD_WORKER_CONNECT_TIMEOUT), as teardown's connect bound for the pod's worker. Production constructs SubstrateBudgets::default(), which is exactly those constants, so deadlines, error messages and logs are unchanged. The create and suspend timeout messages print their budget with Duration's Debug, which renders the 135 s and 300 s defaults as before ("135s", "300s") and a test's sub-second budget as, say, "600ms" rather than "0s"; unit tests pin both renderings. The free ready_wait_bounds used by wake, start, restore and access delegates to the same arithmetic with the default values, and every other use of the shared constants still reads them directly. Tests through the real entry points, with Postgres, the scripted apiserver and the scripted worker real and the budgets shortened: - create_execution readiness failures: when the outer budget cuts off a hung pod read, a scheduled, reachable pod is kept as started on its node, a terminating pod is deleted, a preserve decision that outlives its budget ends in a delete, and a vanished pod is left with its claim to reconciliation; a vanished pod is likewise left after node-mgr probes answer Unavailable; lock contention after the pod exists stays retryable only when the cleanup delete was accepted and is demoted to Failed when it was rejected; - a create that joins an in-flight activation returns the owner's pod without a lease of its own, or times out within the create budget without touching the owner's execution; - suspend_execution keeps polling transient failures and Pending until durable within its budget; past the budget, while Pending or before a node-api pod resolves, it times out and keeps its claim. The split-receipt refusal test drops from about 5 s to 0.3 s. The unreachable-worker teardown case sets a 300 ms connect budget and asserts the retire returns within POD_WORKER_CONNECT_TIMEOUT. On hosts that drop its TEST-NET SYN that proves the substrate's budget bounds the dial, and the case drops from about 3.2 s to 0.4 s; where the host answers ENETUNREACH the dial fails at once and the bound holds trivially. New and converted node-api tests use owned token files rather than the host's token paths. --- .../substrate/kubernetes/adapter.rs | 4 +- .../kubernetes/adapter/substrate_adapter.rs | 37 +- .../substrate/kubernetes/adapter/suspend.rs | 7 +- .../substrate/kubernetes/adapter/teardown.rs | 2 +- .../substrate/kubernetes/adapter/tests.rs | 71 ++ .../adapter_characterization_tests.rs | 1 + .../create_and_teardown.rs | 19 +- .../create_readiness.rs | 660 ++++++++++++++++++ .../hydration_and_suspend.rs | 139 ++++ .../pooled_adoption.rs | 12 +- .../substrate/kubernetes/budgets.rs | 76 ++ .../controllers/substrate/kubernetes/mod.rs | 6 + .../kubernetes/scripted_test_doubles.rs | 3 + .../substrate/kubernetes/support.rs | 8 +- .../controllers/substrate/kubernetes/tests.rs | 3 + 15 files changed, 1020 insertions(+), 28 deletions(-) create mode 100644 crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_readiness.rs create mode 100644 crates/libs/api/src/controllers/substrate/kubernetes/budgets.rs diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter.rs index 77a5fc7081..23d382f51d 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter.rs @@ -49,6 +49,8 @@ pub(super) use node_eligibility::{ use pooled::pooled_metadata_update_error_is_reclaimable; pub(super) use pooled::pooled_route_identity_adoption_request; use pooled::{pooled_wake_request, wait_for_pooled_split_receipt_capability}; +#[cfg(test)] +use substrate_adapter::create_ready_timeout_error; use suspend::await_generation_suspend; #[cfg(test)] use suspend::{await_suspend_continuation, poll_generation_suspend}; @@ -66,7 +68,7 @@ pub(super) const WARM_POOL_SETUP_RPC_TIMEOUT: Duration = Duration::from_secs(5); // Planned teardown waits for the exact snapshot upload to become remotely // durable before retiring the pod. Full-codec artifacts can outlive the // generic node-RPC budget, so this path needs its own outer deadline. -const TEARDOWN_SUSPEND_DURABILITY_BUDGET: Duration = Duration::from_secs(300); +pub(super) const TEARDOWN_SUSPEND_DURABILITY_BUDGET: Duration = Duration::from_secs(300); const MAX_LOGGED_SUSPEND_PENDING_REASON_BYTES: usize = 512; const MAX_SUSPEND_PENDING_REASON_LOGS: u8 = 3; const SUSPEND_PENDING_REASON_LOG_INTERVAL: Duration = Duration::from_secs(30); diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/substrate_adapter.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/substrate_adapter.rs index d801fedd34..7575b3a00c 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/substrate_adapter.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/substrate_adapter.rs @@ -135,7 +135,7 @@ impl SubstrateAdapter for KubernetesSubstrate { ) .await?; let execution = fence.execution; - let (_, phase_deadline) = ready_wait_bounds(super::super::CREATE_READY_BUDGET); + let (_, phase_deadline) = self.budgets.ready_wait_bounds(self.budgets.create_ready); let (pod_ip, binding) = self .wait_for_pod_proxy_reachable( request.metric_identity, @@ -444,7 +444,8 @@ impl SubstrateAdapter for KubernetesSubstrate { // The inner waits clamp to `phase_deadline` so their phase-tagged // timeouts fire first; the outer timeout is the backstop for the // stretches between them. - let (create_budget, phase_deadline) = ready_wait_bounds(super::super::CREATE_READY_BUDGET); + let (create_budget, phase_deadline) = + self.budgets.ready_wait_bounds(self.budgets.create_ready); let select_outcome = match tokio::time::timeout(create_budget, async { let (pod_ip, binding) = self .wait_for_pod_proxy_reachable( @@ -496,16 +497,7 @@ impl SubstrateAdapter for KubernetesSubstrate { .await { Ok(outcome) => outcome, - // Neutral message: whether the work is preserved (pod left running, - // retry the same sandbox) or torn down (pod never scheduled) is - // decided by the preserve check below, so don't promise either here. - Err(_elapsed) => Err(SubstrateError::Timeout { - message: format!( - "sandbox create did not reach readiness within {}s", - super::super::CREATE_READY_BUDGET.as_secs() - ), - phase: None, - }), + Err(_elapsed) => Err(create_ready_timeout_error(self.budgets.create_ready)), }; let binding = match select_outcome { Ok(binding) => binding, @@ -528,7 +520,7 @@ impl SubstrateAdapter for KubernetesSubstrate { // the latency cliff. if create_wait_should_preserve_node_work(&err) && let Ok(Some(node_id)) = tokio::time::timeout( - super::super::PRESERVE_DECISION_BUDGET, + self.budgets.preserve_decision, self.preservable_create_node(sandbox_uuid, &pod_name), ) .await @@ -1236,7 +1228,7 @@ impl SubstrateAdapter for KubernetesSubstrate { &self, sandbox_id: SandboxId, ) -> Result, SubstrateError> { - let suspend_deadline = Instant::now() + TEARDOWN_SUSPEND_DURABILITY_BUDGET; + let suspend_deadline = Instant::now() + self.budgets.suspend_durability; SandboxEvent::new( sandbox_id, SandboxEventType::SuspendRequested, @@ -1384,6 +1376,7 @@ impl SubstrateAdapter for KubernetesSubstrate { expected_generation, &endpoint_addr, suspend_deadline, + self.budgets.suspend_durability, ) .await { @@ -2621,7 +2614,7 @@ impl SubstrateAdapter for KubernetesSubstrate { let require_pooled_adoption_node_id = if wait_until_ready { wait_for_pooled_split_receipt_capability( &mut node_api, - Instant::now() + WARM_POOL_SETUP_RPC_TIMEOUT, + Instant::now() + self.budgets.pooled_split_receipt_wait, ) .await?; true @@ -3093,3 +3086,17 @@ impl SubstrateAdapter for KubernetesSubstrate { Ok(Some(dedupe_inventory_entries(candidates))) } } + +/// `create_execution`'s error when its outer readiness budget expires. +/// +/// Neutral message: whether the work is preserved (pod left running, retry the +/// same sandbox) or torn down (pod never scheduled) is decided by the preserve +/// check after it, so don't promise either here. `Duration`'s `Debug` renders +/// a whole-second budget such as the production default as `135s`, and keeps +/// a test's sub-second budget legible. +pub(super) fn create_ready_timeout_error(create_ready: Duration) -> SubstrateError { + SubstrateError::Timeout { + message: format!("sandbox create did not reach readiness within {create_ready:?}"), + phase: None, + } +} diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/suspend.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/suspend.rs index d486d81137..4de514544b 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/suspend.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/suspend.rs @@ -43,6 +43,7 @@ pub(super) async fn await_generation_suspend( execution_generation: ExecutionGeneration, endpoint_addr: &str, deadline: Instant, + durability_budget: Duration, ) -> Result { poll_generation_suspend( |request| { @@ -53,6 +54,7 @@ pub(super) async fn await_generation_suspend( execution_generation, endpoint_addr, deadline, + durability_budget, ) .await } @@ -63,6 +65,8 @@ pub(super) async fn poll_generation_suspend( execution_generation: ExecutionGeneration, endpoint_addr: &str, deadline: Instant, + // The budget `deadline` was derived from, named in the timeout error. + durability_budget: Duration, ) -> Result where F: FnMut( @@ -106,8 +110,7 @@ where .await .map_err(|_| SubstrateError::Timeout { message: format!( - "generation-bound suspend for sandbox {sandbox_id} generation {execution_generation} did not become durable within {}s", - TEARDOWN_SUSPEND_DURABILITY_BUDGET.as_secs(), + "generation-bound suspend for sandbox {sandbox_id} generation {execution_generation} did not become durable within {durability_budget:?}", ), phase: None, })? { diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/teardown.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/teardown.rs index c1a8570357..259baf9753 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/teardown.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/teardown.rs @@ -358,7 +358,7 @@ impl KubernetesSubstrate { let client = Self::pod_worker_client( pod_ip, self.k8s.sandbox_proxy_port, - POD_WORKER_CONNECT_TIMEOUT, + self.budgets.teardown_worker_connect, ) .await; if let Some(mut client) = client { diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/tests.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/tests.rs index fcafd1c1c3..09743f42fb 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter/tests.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter/tests.rs @@ -66,6 +66,7 @@ async fn poll_suspend_responses( generation, "http://node-api", Instant::now() + budget, + budget, ) .await; let summaries: Vec<_> = capture @@ -181,6 +182,75 @@ async fn generation_suspend_bounds_retained_reason_on_a_utf8_boundary() { assert!(!result.unwrap_err().to_string().contains('界')); } +// The budget-derived timeout messages print the budget with `Duration`'s +// `Debug`. For each production default that is byte-identical to the earlier +// `{}s` of `as_secs()`; a test's sub-second budget no longer reads `0s`. +#[test] +fn create_ready_timeout_message_names_the_budget() { + let default = SubstrateBudgets::DEFAULT.create_ready; + for (budget, expected) in [ + ( + default, + format!( + "sandbox create did not reach readiness within {}s", + default.as_secs() + ), + ), + ( + Duration::from_millis(500), + "sandbox create did not reach readiness within 500ms".to_string(), + ), + ] { + match create_ready_timeout_error(budget) { + SubstrateError::Timeout { + message, + phase: None, + } => assert_eq!(message, expected), + other => panic!("unexpected create timeout error {other:?}"), + } + } + assert_eq!( + format!("{default:?}"), + "135s", + "the production create budget's rendering" + ); +} + +#[tokio::test(start_paused = true)] +async fn generation_suspend_timeout_message_names_the_budget() { + let default = SubstrateBudgets::DEFAULT.suspend_durability; + let sandbox_id = SandboxId::new(); + let prefix = format!( + "generation-bound suspend for sandbox {sandbox_id} generation 1 did not become durable within" + ); + for (budget, expected) in [ + (default, format!("{prefix} {}s", default.as_secs())), + (Duration::from_millis(600), format!("{prefix} 600ms")), + ] { + let result = poll_generation_suspend( + |_request| std::future::pending(), + sandbox_id, + ExecutionGeneration::new(1).unwrap(), + "http://node-api", + Instant::now() + budget, + budget, + ) + .await; + match result { + Err(SubstrateError::Timeout { + message, + phase: None, + }) => assert_eq!(message, expected), + other => panic!("unexpected suspend timeout result {other:?}"), + } + } + assert_eq!( + format!("{default:?}"), + "300s", + "the production suspend budget's rendering" + ); +} + #[tokio::test(start_paused = true)] async fn generation_suspend_polling_preserves_success_and_terminal_failure() { let artifact_id = LocalArtifactId::new(); @@ -280,6 +350,7 @@ async fn generation_suspend_reason_change_logs_are_rate_limited_and_capped() { ExecutionGeneration::new(1).unwrap(), "http://node-api", Instant::now() + Duration::from_secs(120), + Duration::from_secs(120), ) .await; assert!(matches!(result, Err(SubstrateError::Timeout { .. }))); diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests.rs index 5c1ff03749..e9e04dea2c 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests.rs @@ -10,6 +10,7 @@ //! carries a TEST-NET address that nothing answers on. mod create_and_teardown; +mod create_readiness; mod hydration_and_suspend; mod pooled_adoption; diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_and_teardown.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_and_teardown.rs index c1b43aab94..438619e6f9 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_and_teardown.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_and_teardown.rs @@ -176,7 +176,7 @@ enum TeardownCase { async fn teardown_keeps_execution_tracked_when_the_node_cannot_finish( #[case] case: TeardownCase, ) -> TestResult { - if !crate::test_support::enter_sim_worker_process() { + if !crate::test_support::enter_sim_worker_process_with_owned_token_files() { return Ok(()); } let db = pplx_space_db_test_fixtures::pg_test_db().await?; @@ -231,7 +231,7 @@ async fn teardown_keeps_execution_tracked_when_the_node_cannot_finish( .ensure_default_running(&sandbox_id.to_string()) .await?; let (client, log) = teardown_kube_client(); - let substrate = substrate_with_store_client_proxy_port_and_pod( + let mut substrate = substrate_with_store_client_proxy_port_and_pod( db.pool.clone(), client, worker.port, @@ -244,7 +244,16 @@ async fn teardown_keeps_execution_tracked_when_the_node_cannot_finish( Some(pod_ip), )), ); + if let TeardownCase::RetireWorkerUnreachable = case { + // The TEST-NET dial never connects. Where the host silently drops the + // SYN, this budget ends it, and the bound below proves the substrate's + // budget rather than the production constant is the one teardown + // uses. Where the host answers `ENETUNREACH`, the dial fails at once + // and the bound holds trivially. + substrate.budgets.teardown_worker_connect = Duration::from_millis(300); + } + let started = Instant::now(); let outcome = match case { TeardownCase::RetireWorkerUnreachable | TeardownCase::RetireStopFails @@ -272,6 +281,12 @@ async fn teardown_keeps_execution_tracked_when_the_node_cannot_finish( ) => {} (case, outcome) => panic!("{case:?}: unexpected teardown outcome {outcome:?}"), } + if let TeardownCase::RetireWorkerUnreachable = case { + assert!( + started.elapsed() < POD_WORKER_CONNECT_TIMEOUT, + "the substrate's teardown connect budget, not the production one, must end the dial" + ); + } assert!( requests_matching(&log, &Method::DELETE).is_empty(), "{case:?}: the pod must not be deleted" diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_readiness.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_readiness.rs new file mode 100644 index 0000000000..b7464e3510 --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/create_readiness.rs @@ -0,0 +1,660 @@ +//! `create_execution` after its pod exists: the arms that run when the +//! readiness wait fails, and a create that joins an activation already in +//! flight. +//! +//! The substrate's [`SubstrateBudgets`] are shortened so the outer readiness +//! budget and the preserve decision expire after a second, while Postgres, +//! the scripted apiserver and the scripted worker stay real. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use axum::http::{Method, StatusCode}; +use futures_util::FutureExt as _; +use k8s_openapi::api::core::v1::{Node, NodeAddress, NodeStatus}; +use pplx_space_common_core::types::{ + NodeId, SandboxActivationAttemptId, SandboxTransitionTargetState, +}; +use sqlx::{Postgres, Transaction}; + +use super::super::scripted_test_doubles::{ + CallObserver, KubeReply, KubeRequest, ScriptedReply, ScriptedWorker, ScriptedWorkerConfig, + kube_status, pod_path, pod_reply, requests_matching, scripted_kube_client_async, + start_scripted_worker, +}; +use super::super::tests::{insert_test_sandbox_row, substrate_with_inventory_caches_and_pool}; +use super::*; + +const SPACED_WAIT_READY: &str = "WorkerApi/SpacedWaitReady"; +const PODS_PATH: &str = "/api/v1/namespaces/space/pods"; +/// A node the apiserver names but the substrate's node cache does not hold. +const UNCACHED_NODE: &str = "node-uncached"; + +/// Shortened create budgets: the inner readiness phases get 500 ms and the +/// outer budget fires 500 ms later. A readiness wait parked on a pod read +/// that never answers reaches that read after the 250 ms reflector wait, well +/// inside the outer budget even on a loaded runner. +fn test_create_budgets() -> SubstrateBudgets { + SubstrateBudgets { + create_ready: Duration::from_secs(1), + create_ready_phase_margin: Duration::from_millis(500), + preserve_decision: Duration::from_secs(1), + ..SubstrateBudgets::DEFAULT + } +} + +/// The production create budgets with a shortened preserve decision, for +/// cases that end on their own phase well inside the create budget. +fn phase_ending_budgets() -> SubstrateBudgets { + SubstrateBudgets { + preserve_decision: Duration::from_secs(1), + ..SubstrateBudgets::DEFAULT + } +} + +/// The worker node, cached with the UID and address placement records. +fn worker_node() -> Node { + Node { + metadata: kube::api::ObjectMeta { + name: Some(WORKER_NODE.to_string()), + uid: Some(format!("{WORKER_NODE}-uid")), + ..Default::default() + }, + status: Some(NodeStatus { + addresses: Some(vec![NodeAddress { + type_: "InternalIP".to_string(), + address: "127.0.0.1".to_string(), + }]), + ..Default::default() + }), + ..Default::default() + } +} + +fn worker_node_id() -> NodeId { + space_k8s_node_id(WORKER_NODE, SpaceNodeRole::Worker.as_str()) +} + +/// A substrate whose created pods are reachable on `worker`, with only +/// [`WORKER_NODE`] in its node cache. +fn substrate_on(pool: PgPool, client: Client, worker: &ScriptedWorker) -> KubernetesSubstrate { + let mut substrate = + substrate_with_inventory_caches_and_pool(Vec::new(), vec![worker_node()], pool); + substrate.k8s.client = client; + substrate.k8s.sandbox_proxy_port = worker.port; + substrate.budgets = test_create_budgets(); + substrate +} + +/// A running pod for `sandbox_id`'s first generation on `node`, at the +/// loopback address its worker listens on. +fn running_pod(pod_name: &str, sandbox_id: SandboxId, node: Option<&str>) -> Pod { + let mut pod = sandbox_pod( + pod_name, + &format!("{pod_name}-uid"), + sandbox_id, + 1, + "Running", + Some("127.0.0.1"), + ); + pod.spec.as_mut().expect("sandbox pod has a spec").node_name = node.map(str::to_string); + pod +} + +fn requested_pod_name(request: &KubeRequest) -> String { + request + .body + .as_ref() + .and_then(|pod| pod.pointer("/metadata/name")) + .and_then(serde_json::Value::as_str) + .expect("a pod create names its pod") + .to_string() +} + +async fn active_attempt(pool: &PgPool, sandbox_id: SandboxId) -> Result, BoxError> { + Ok(sqlx::query_scalar(sql!( + "SELECT activation_attempt_id FROM sandbox_k8s_executions + WHERE sandbox_id = $1 AND active" + )) + .bind(sandbox_id) + .fetch_one(pool) + .await?) +} + +// ── readiness failure ─────────────────────────────────────────────────────── + +/// How the readiness wait of a create whose pod was accepted fails. +#[derive(Debug, Clone, Copy)] +enum ReadinessFailure { + /// Readiness's first pod read never answers. No readiness phase bounds + /// that read, so only the outer create budget ends the wait, with a + /// timeout that names no phase. + OuterBudget, + /// Every guest probe answers gRPC `Unavailable` until the command-ready + /// phase times out. + NodeMgrUnavailable, + /// The first execution read loses a lock race: a table lock is held from + /// the pod create until the cleanup delete arrives. + LockContention, +} + +/// What the apiserver reports for the pod once readiness has failed. +#[derive(Debug, Clone, Copy)] +enum PodAfterFailure { + /// Scheduled on the cached worker node, its proxy still reachable. + ScheduledReachable, + /// Scheduled on a node missing from the cache whose apiserver read never + /// answers, so the preserve decision cannot finish. + ScheduledOnUncachedNode, + /// Marked for deletion. + Terminating, + /// Gone. + Absent, +} + +#[derive(Debug, Clone, Copy)] +enum CleanupDelete { + Accepted, + Rejected, +} + +#[derive(Debug, Clone, Copy)] +enum ExpectedEnd { + /// Returned as started on the worker node, without readiness; the pod + /// and its execution are kept. + PreservedOnNode, + /// The pod is deleted, its execution retired, and the error returned. + TornDown(ExpectedError), + /// Nothing is deleted or retired: cleanup is left to reconciliation, and + /// the error returned. + Deferred(ExpectedError), +} + +#[derive(Debug, Clone, Copy)] +enum ExpectedError { + /// The outer budget's own timeout, which carries no phase. + TimeoutWithoutPhase, + /// The command-ready phase's timeout. + GuestNotReadyTimeout, + /// Still retryable: the cleanup delete was accepted. + LockContention, + /// Demoted from contention: nothing is removing the pod, so a retry would + /// orphan it. + Failed, +} + +impl ExpectedError { + fn matches(self, error: &SubstrateError) -> bool { + match self { + Self::TimeoutWithoutPhase => { + matches!(error, SubstrateError::Timeout { phase: None, .. }) + } + Self::GuestNotReadyTimeout => matches!( + error, + SubstrateError::Timeout { + phase: Some(WaitReadyTimeoutPhase::GuestNotReady), + .. + } + ), + Self::LockContention => matches!(error, SubstrateError::LockContention { .. }), + Self::Failed => matches!(error, SubstrateError::Failed { .. }), + } + } +} + +/// Apiserver state the scripted replies read and the test's observers change. +#[derive(Default)] +struct ApiserverState { + pod_name: Mutex>, + readiness_failed: AtomicBool, + /// Held from the pod create to the cleanup delete in the contention case. + table_lock: tokio::sync::Mutex>>, +} + +impl ApiserverState { + async fn release_table_lock(&self) { + if let Some(lock) = self.table_lock.lock().await.take() { + lock.rollback() + .await + .expect("release the execution table lock"); + } + } +} + +/// A create whose readiness wait fails gives back exactly what the failure +/// arm owns: a pod that is scheduled and reachable when the outer budget +/// expires is kept and returned as started; one that is terminating, or whose +/// preserve decision outlives its budget, is deleted with its execution +/// retired; one that vanished after the wait timed out, on the outer budget or +/// after node-mgr probes answered `Unavailable`, is left, with its claim, to +/// reconciliation. A contended read after the pod exists stays retryable only +/// when the cleanup delete was accepted. +#[rstest::rstest] +#[case::outer_budget_preserves_a_reachable_pod( + ReadinessFailure::OuterBudget, + PodAfterFailure::ScheduledReachable, + CleanupDelete::Accepted, + ExpectedEnd::PreservedOnNode +)] +#[case::outer_budget_tears_down_a_terminating_pod( + ReadinessFailure::OuterBudget, + PodAfterFailure::Terminating, + CleanupDelete::Accepted, + ExpectedEnd::TornDown(ExpectedError::TimeoutWithoutPhase) +)] +#[case::slow_preserve_decision_tears_down( + ReadinessFailure::OuterBudget, + PodAfterFailure::ScheduledOnUncachedNode, + CleanupDelete::Accepted, + ExpectedEnd::TornDown(ExpectedError::TimeoutWithoutPhase) +)] +#[case::outer_budget_defers_cleanup_of_a_vanished_pod( + ReadinessFailure::OuterBudget, + PodAfterFailure::Absent, + CleanupDelete::Accepted, + ExpectedEnd::Deferred(ExpectedError::TimeoutWithoutPhase) +)] +#[case::unreachable_node_mgr_defers_cleanup_of_a_vanished_pod( + ReadinessFailure::NodeMgrUnavailable, + PodAfterFailure::Absent, + CleanupDelete::Accepted, + ExpectedEnd::Deferred(ExpectedError::GuestNotReadyTimeout) +)] +#[case::contention_with_accepted_cleanup_stays_retryable( + ReadinessFailure::LockContention, + PodAfterFailure::ScheduledReachable, + CleanupDelete::Accepted, + ExpectedEnd::TornDown(ExpectedError::LockContention) +)] +#[case::contention_with_rejected_cleanup_is_demoted( + ReadinessFailure::LockContention, + PodAfterFailure::ScheduledReachable, + CleanupDelete::Rejected, + ExpectedEnd::TornDown(ExpectedError::Failed) +)] +#[tokio::test(flavor = "current_thread")] +async fn create_readiness_failure_keeps_or_releases_only_what_its_arm_owns( + #[case] failure: ReadinessFailure, + #[case] pod_after_failure: PodAfterFailure, + #[case] cleanup_delete: CleanupDelete, + #[case] expected: ExpectedEnd, +) -> TestResult { + if !crate::test_support::enter_sim_worker_process_with_owned_token_files() { + return Ok(()); + } + let db = pplx_space_db_test_fixtures::pg_test_db().await?; + let sandbox_id = SandboxId::new(); + insert_test_sandbox_row(&db.pool, sandbox_id, "creating").await?; + + let state = Arc::new(ApiserverState::default()); + let probe_reply = match failure { + ReadinessFailure::NodeMgrUnavailable => { + ScriptedReply::Status(tonic::Status::unavailable("node-mgr unreachable")) + } + // Readiness fails before any probe. + ReadinessFailure::OuterBudget | ReadinessFailure::LockContention => ScriptedReply::Hang, + }; + let observed = Arc::clone(&state); + let readiness_failed: CallObserver = Arc::new(move || { + observed.readiness_failed.store(true, Ordering::SeqCst); + async {}.boxed() + }); + let worker = start_scripted_worker(ScriptedWorkerConfig { + observers: [(SPACED_WAIT_READY, readiness_failed)].into(), + ..ScriptedWorkerConfig::new([(SPACED_WAIT_READY, probe_reply)]) + }) + .await?; + + let reply_state = Arc::clone(&state); + let lock_pool = db.pool.clone(); + let (client, log) = scripted_kube_client_async(move |request| { + let state = Arc::clone(&reply_state); + let lock_pool = lock_pool.clone(); + async move { + let pod_name = state.pod_name.lock().unwrap().clone(); + let at_pod = pod_name.as_deref().map(pod_path); + match request.method { + Method::POST if request.path == PODS_PATH => { + let pod_name = requested_pod_name(&request); + *state.pod_name.lock().unwrap() = Some(pod_name.clone()); + if let ReadinessFailure::LockContention = failure { + let mut lock = lock_pool.begin().await.expect("begin the table lock"); + sqlx::query(sql!( + "LOCK TABLE sandbox_k8s_executions IN ACCESS EXCLUSIVE MODE" + )) + .execute(&mut *lock) + .await + .expect("lock the execution table"); + *state.table_lock.lock().await = Some(lock); + } + // Recording the pod's UID and IP moves the execution out + // of `creating`, so readiness reads the pod rather than + // watching it. No node yet: placement is recorded only + // after readiness. + pod_reply(&running_pod(&pod_name, sandbox_id, None)) + } + Method::GET if Some(&request.path) == at_pod.as_ref() => { + let pod_name = pod_name.expect("the pod was created"); + // Readiness makes the first read of the pod; the reads + // after it see the pod as it is once readiness failed. + let first_read = !state.readiness_failed.load(Ordering::SeqCst); + match failure { + ReadinessFailure::OuterBudget if first_read => { + state.readiness_failed.store(true, Ordering::SeqCst); + return KubeReply::Hang; + } + _ => {} + } + let after_failure = state.readiness_failed.load(Ordering::SeqCst); + match (after_failure, pod_after_failure) { + (true, PodAfterFailure::Absent) => { + kube_status(StatusCode::NOT_FOUND, "NotFound") + } + (true, PodAfterFailure::Terminating) => { + let mut pod = running_pod(&pod_name, sandbox_id, Some(WORKER_NODE)); + pod.metadata.deletion_timestamp = + Some(k8s_openapi::apimachinery::pkg::apis::meta::v1::Time( + jiff::Timestamp::now(), + )); + pod_reply(&pod) + } + (true, PodAfterFailure::ScheduledOnUncachedNode) => { + pod_reply(&running_pod(&pod_name, sandbox_id, Some(UNCACHED_NODE))) + } + _ => pod_reply(&running_pod(&pod_name, sandbox_id, Some(WORKER_NODE))), + } + } + Method::GET if request.path == format!("/api/v1/nodes/{UNCACHED_NODE}") => { + KubeReply::Hang + } + Method::DELETE if Some(&request.path) == at_pod.as_ref() => { + state.release_table_lock().await; + match cleanup_delete { + CleanupDelete::Accepted => kube_status(StatusCode::OK, "Success"), + CleanupDelete::Rejected => { + kube_status(StatusCode::INTERNAL_SERVER_ERROR, "InternalError") + } + } + } + _ => kube_status(StatusCode::BAD_REQUEST, "BadRequest"), + } + } + }); + let pool = match failure { + // Bound lock waits, as an application login's role-level + // `lock_timeout` does, so the held table lock aborts the read. + ReadinessFailure::LockContention => { + sqlx::postgres::PgPoolOptions::new() + .max_connections(4) + .after_connect(|conn, _| { + Box::pin(async move { + sqlx::query("SET lock_timeout = '100ms'") + .execute(conn) + .await?; + Ok(()) + }) + }) + .connect(&db.database_url) + .await? + } + ReadinessFailure::OuterBudget | ReadinessFailure::NodeMgrUnavailable => db.pool.clone(), + }; + let mut substrate = substrate_on(pool, client, &worker); + match failure { + // A data-plane timeout far past the budget, so only the outer budget + // can end the wait. + ReadinessFailure::OuterBudget => { + substrate.k8s.create_data_plane_ready_timeout = Duration::from_secs(60); + } + // These end on their own phase; the create budget stays out of reach. + ReadinessFailure::NodeMgrUnavailable => { + substrate.budgets = phase_ending_budgets(); + substrate.k8s.create_command_ready_timeout = Duration::from_millis(500); + } + ReadinessFailure::LockContention => { + substrate.budgets = phase_ending_budgets(); + } + } + let inputs = CreateInputs::new(); + let template_ref = inputs.template_ref(); + let mut request = inputs.create_request(sandbox_id, &template_ref); + request.wait_until_ready = true; + + let started = Instant::now(); + let outcome = + tokio::time::timeout(Duration::from_secs(10), substrate.create_execution(request)) + .await + .map_err(|_| format!("{failure:?}/{pod_after_failure:?}: the create must end"))?; + let elapsed = started.elapsed(); + state.release_table_lock().await; + + assert!( + elapsed < PRESERVE_DECISION_BUDGET, + "{failure:?}/{pod_after_failure:?}: the shortened budgets must end the create, took \ + {elapsed:?}" + ); + assert!( + substrate.task_tracker.is_empty(), + "no cleanup is left running in the background" + ); + let pod_name = state + .pod_name + .lock() + .unwrap() + .clone() + .expect("the pod was created"); + let deletes = requests_matching(&log, &Method::DELETE); + let active = active_execution_count(&db.pool, sandbox_id).await?; + match expected { + ExpectedEnd::PreservedOnNode => { + let outcome = + outcome.map_err(|error| format!("expected a preserved create: {error:?}"))?; + assert_eq!(outcome.initial_lifecycle, "started"); + assert_eq!(outcome.initial_placement, "on_node"); + assert_eq!(outcome.active_node_id, Some(worker_node_id())); + assert!(!outcome.command_ready, "readiness was never observed"); + assert!(outcome.binding.is_none()); + assert!( + deletes.is_empty(), + "the preserved pod must not be deleted: {deletes:?}" + ); + assert_eq!(active, 1, "the preserved pod's execution stays tracked"); + let (state, _, pod_uid, pod_ip) = execution_row(&db.pool, sandbox_id, 1).await?; + assert_eq!( + (state.as_str(), pod_uid, pod_ip.as_deref()), + ( + "running", + Some(format!("{pod_name}-uid")), + Some("127.0.0.1") + ) + ); + assert_eq!(substrate.cancel_tracker.current_count(), 0); + } + ExpectedEnd::TornDown(error) => { + let outcome = outcome.err().ok_or("expected the create to fail")?; + assert!( + error.matches(&outcome), + "expected {error:?}, got {outcome:?}" + ); + let [delete] = deletes.as_slice() else { + panic!("expected one pod delete, got {deletes:?}"); + }; + assert_eq!(delete.path, pod_path(&pod_name)); + assert_eq!(active, 0, "the execution must be retired"); + assert_eq!( + substrate.cancel_tracker.current_count(), + 0, + "a synchronous cleanup is not a cancellation" + ); + } + ExpectedEnd::Deferred(error) => { + let outcome = outcome.err().ok_or("expected the create to fail")?; + assert!( + error.matches(&outcome), + "expected {error:?}, got {outcome:?}" + ); + assert!(deletes.is_empty(), "nothing may be deleted: {deletes:?}"); + assert_eq!( + active, 1, + "the claim is left for the orphaned-sandbox reconciler" + ); + assert_eq!( + substrate.cancel_tracker.current_count(), + 1, + "the deferred cleanup still counts toward cancel-storm backpressure" + ); + } + } + Ok(()) +} + +// ── joined create ─────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, Copy)] +enum OwnerPod { + /// The owner's pod is running and its proxy reachable. + Reachable, + /// The owner's pod never gets an IP. + NeverAssigned, +} + +/// A create that finds another create's activation in flight joins it rather +/// than creating a pod: it returns the owner's pod once that pod is +/// reachable, without a lease of its own, and when the owner's pod never +/// becomes reachable it times out within the create budget, leaving the +/// owner's execution and activation untouched. +#[rstest::rstest] +#[case::returns_the_owners_reachable_pod(OwnerPod::Reachable)] +#[case::times_out_without_disturbing_the_owner(OwnerPod::NeverAssigned)] +#[tokio::test(flavor = "current_thread")] +async fn joined_create_waits_on_the_owners_pod(#[case] owner_pod: OwnerPod) -> TestResult { + if !crate::test_support::enter_sim_worker_process_with_owned_token_files() { + return Ok(()); + } + let db = pplx_space_db_test_fixtures::pg_test_db().await?; + let sandbox_id = SandboxId::new(); + insert_test_sandbox_row(&db.pool, sandbox_id, "creating").await?; + let worker = start_scripted_worker(ScriptedWorkerConfig::new([])).await?; + let pod_name: Arc>> = Arc::default(); + let reply_pod_name = Arc::clone(&pod_name); + let (client, log) = scripted_kube_client_async(move |request| { + let pod_name = reply_pod_name.lock().unwrap().clone(); + async move { + match (request.method, pod_name) { + (Method::GET, Some(pod_name)) if request.path == pod_path(&pod_name) => { + match owner_pod { + OwnerPod::Reachable => { + pod_reply(&running_pod(&pod_name, sandbox_id, Some(WORKER_NODE))) + } + OwnerPod::NeverAssigned => pod_reply(&sandbox_pod( + &pod_name, + &format!("{pod_name}-uid"), + sandbox_id, + 1, + "Pending", + None, + )), + } + } + // The per-pod watch never reports an IP. + (Method::GET, _) if request.path == PODS_PATH => KubeReply::Hang, + _ => kube_status(StatusCode::BAD_REQUEST, "BadRequest"), + } + } + }); + let mut substrate = substrate_on(db.pool.clone(), client, &worker); + // A phase timeout far past the budget, so only the create budget can end + // the joined wait. + substrate.k8s.create_data_plane_ready_timeout = Duration::from_secs(60); + + // Another create owns the activation. + let owner_attempt = SandboxActivationAttemptId::new(); + let owner = substrate + .store + .begin_activation_execution_row( + sandbox_id, + owner_attempt, + SandboxTransitionTargetState::Started, + ) + .await?; + *pod_name.lock().unwrap() = Some(owner.pod_name.clone()); + if let OwnerPod::Reachable = owner_pod { + substrate + .store + .persist_pod_metadata( + sandbox_id, + owner.generation, + Some(&format!("{}-uid", owner.pod_name)), + Some("127.0.0.1"), + ) + .await?; + } + let inputs = CreateInputs::new(); + let template_ref = inputs.template_ref(); + + let started = Instant::now(); + let outcome = tokio::time::timeout( + Duration::from_secs(10), + substrate.create_execution(inputs.create_request(sandbox_id, &template_ref)), + ) + .await + .map_err(|_| format!("{owner_pod:?}: the joined create must end"))?; + + assert!( + requests_matching(&log, &Method::POST).is_empty() + && requests_matching(&log, &Method::DELETE).is_empty(), + "{owner_pod:?}: a joined create neither creates nor deletes a pod" + ); + assert_eq!( + active_attempt(&db.pool, sandbox_id).await?, + Some(owner_attempt.as_uuid()), + "{owner_pod:?}: the owner keeps its activation" + ); + assert!( + substrate.task_tracker.is_empty(), + "{owner_pod:?}: a joined create leaves nothing running in the background" + ); + match owner_pod { + OwnerPod::Reachable => { + let outcome = + outcome.map_err(|error| format!("expected a joined create: {error:?}"))?; + assert!( + outcome.lease_completion.is_none(), + "the lease is the owner's" + ); + assert_eq!(outcome.initial_lifecycle, "started"); + assert_eq!(outcome.active_node_id, Some(worker_node_id())); + assert!(outcome.command_ready); + let binding = outcome + .binding + .ok_or("a joined create returns the owner's binding")?; + assert_eq!( + binding.execution_object_id, + format!("{}-uid", owner.pod_name) + ); + assert_eq!(binding.execution_generation, 1); + } + OwnerPod::NeverAssigned => { + let outcome = outcome + .err() + .ok_or("expected the joined create to time out")?; + assert!( + matches!( + outcome, + SubstrateError::Timeout { + phase: Some(WaitReadyTimeoutPhase::PodNeverReady), + .. + } + ), + "{outcome:?}" + ); + let elapsed = started.elapsed(); + assert!( + elapsed < Duration::from_secs(5), + "the create budget, not the data-plane timeout, must end the wait: {elapsed:?}" + ); + assert_eq!(active_execution_count(&db.pool, sandbox_id).await?, 1); + } + } + Ok(()) +} diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/hydration_and_suspend.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/hydration_and_suspend.rs index 0b9e820d7a..495e719637 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/hydration_and_suspend.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/hydration_and_suspend.rs @@ -596,3 +596,142 @@ async fn suspend_execution_keeps_the_pod_when_the_node_refuses() -> TestResult { assert_eq!(active_execution_count(&db.pool, sandbox_id).await?, 1); Ok(()) } + +#[derive(Debug, Clone, Copy)] +enum SuspendDurability { + /// Transient node-api failures and Pending observations end in a durable + /// suspend. + PendingThenDurable, + /// The node reports Pending until the durability budget expires. + PendingPastTheBudget, + /// No ready node-api pod serves the sandbox's node before the budget + /// expires. + NodeApiNeverResolves, +} + +/// `suspend_execution` polls a Pending generation suspend through transient +/// node-api failures until it is durable, within the substrate's durability +/// budget. When the budget expires first, while the node is still Pending or +/// before a node-api pod is found, the suspend times out and keeps its claim: +/// the sandbox stays `suspending`, the execution tracked and the pod in place, +/// for a retry to join. +#[rstest::rstest] +#[case::pending_then_durable(SuspendDurability::PendingThenDurable)] +#[case::pending_past_the_budget(SuspendDurability::PendingPastTheBudget)] +#[case::node_api_never_resolves(SuspendDurability::NodeApiNeverResolves)] +#[tokio::test(flavor = "current_thread")] +async fn suspend_execution_polls_the_node_within_its_durability_budget( + #[case] durability: SuspendDurability, +) -> TestResult { + if !crate::test_support::enter_sim_worker_process_with_owned_token_files() { + return Ok(()); + } + let db = pplx_space_db_test_fixtures::pg_test_db().await?; + let sandbox_id = SandboxId::new(); + let pod_name = "space-sbx-suspend-budget-g1"; + let node_artifact = LocalArtifactId::new(); + insert_test_sandbox_row(&db.pool, sandbox_id, "started").await?; + insert_execution( + &db.pool, + sandbox_id, + ExecutionSeed::running(pod_name, "suspend-budget-uid"), + ) + .await?; + let pending = ScriptedReply::message(&sbx::PrepareTeardownSuspendResponse { + outcome: sbx::TeardownSuspendOutcome::Pending as i32, + reason: None, + published_artifact: None, + }); + let durable = ScriptedReply::message(&sbx::PrepareTeardownSuspendResponse { + outcome: sbx::TeardownSuspendOutcome::Durable as i32, + reason: None, + published_artifact: Some(sbx::PublishedArtifact { + local_artifact_id: node_artifact.to_string(), + artifact_kind: sbx::ArtifactKind::Full as i32, + origin: sbx::ArtifactOrigin::Suspend as i32, + snapshot_name: None, + }), + }); + let suspend_reply = match durability { + SuspendDurability::PendingThenDurable => ScriptedReply::Sequence(Arc::from([ + ScriptedReply::Status(tonic::Status::unavailable("node-api restarting")), + pending.clone(), + durable, + ])), + SuspendDurability::PendingPastTheBudget | SuspendDurability::NodeApiNeverResolves => { + pending + } + }; + let worker = start_scripted_worker(ScriptedWorkerConfig::new([( + "WorkerApi/PrepareTeardownSuspend", + suspend_reply, + )])) + .await?; + let (client, log) = scripted_kube_client(move |request| { + if request.path != pod_path(pod_name) { + return kube_status(StatusCode::BAD_REQUEST, "BadRequest"); + } + match request.method { + Method::DELETE => kube_status(StatusCode::OK, "Success"), + _ => kube_status(StatusCode::NOT_FOUND, "NotFound"), + } + }); + let mut substrate = match durability { + SuspendDurability::NodeApiNeverResolves => { + substrate_with_store_and_client(db.pool.clone(), client) + } + SuspendDurability::PendingThenDurable | SuspendDurability::PendingPastTheBudget => { + substrate_with_node_api_on(&db.pool, client, worker.port) + } + }; + substrate.k8s.k8s_blocking_delete_timeout = BLOCKING_DELETE_TIMEOUT; + substrate.budgets.suspend_durability = match durability { + // Room for three observations paced 250 ms apart. + SuspendDurability::PendingThenDurable => Duration::from_secs(2), + SuspendDurability::PendingPastTheBudget | SuspendDurability::NodeApiNeverResolves => { + Duration::from_millis(600) + } + }; + + let started = Instant::now(); + let outcome = tokio::time::timeout( + Duration::from_secs(10), + substrate.suspend_execution_tracked(sandbox_id), + ) + .await + .map_err(|_| format!("{durability:?}: the suspend must end within its budget"))?; + let elapsed = started.elapsed(); + + match durability { + // The commit and the fenced pod retirement after a durable reply are + // pinned by `suspend_execution_commits_the_nodes_artifact_before_retiring_the_pod`, + // and the poll's per-reply handling by `adapter::tests`; this case + // adds that the entry point keeps polling within its budget. + SuspendDurability::PendingThenDurable => { + let returned = outcome?.expect("Kubernetes suspend always returns its artifact"); + assert_eq!(returned.id, node_artifact); + } + SuspendDurability::PendingPastTheBudget | SuspendDurability::NodeApiNeverResolves => { + assert!( + matches!(outcome, Err(SubstrateError::Timeout { .. })), + "{durability:?}: expected a timeout, got {outcome:?}" + ); + assert!( + elapsed < Duration::from_secs(5), + "{durability:?}: the substrate's durability budget, not the production one, \ + must end the suspend: {elapsed:?}" + ); + assert!( + log.lock().unwrap().is_empty(), + "{durability:?}: the pod must not be retired" + ); + assert_eq!( + sandbox_lifecycle(&db.pool, sandbox_id).await?, + "suspending", + "{durability:?}: the suspend keeps its claim for a retry" + ); + assert_eq!(active_execution_count(&db.pool, sandbox_id).await?, 1); + } + } + Ok(()) +} diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs index 36f472d0b0..a389bda6bd 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs @@ -527,7 +527,7 @@ fn substrate_with_claimed_pod_store(pool: &PgPool, worker: &ScriptedWorker) -> K /// not advertise split operation receipts before applying any setup. #[tokio::test(flavor = "current_thread")] async fn started_pooled_adoption_refuses_a_worker_without_split_receipts() -> TestResult { - if !crate::test_support::enter_sim_worker_process() { + if !crate::test_support::enter_sim_worker_process_with_owned_token_files() { return Ok(()); } let inputs = PooledInputs::new(); @@ -537,9 +537,17 @@ async fn started_pooled_adoption_refuses_a_worker_without_split_receipts() -> Te }) .await?; - let outcome = prepare(&substrate_without_store(&worker), &inputs, true).await; + let mut substrate = substrate_without_store(&worker); + substrate.budgets.pooled_split_receipt_wait = Duration::from_millis(300); + + let started = Instant::now(); + let outcome = prepare(&substrate, &inputs, true).await; assert!(outcome.is_err(), "an incapable worker must be refused"); + assert!( + started.elapsed() < adapter::WARM_POOL_SETUP_RPC_TIMEOUT, + "the substrate's split-receipt budget, not the production one, must end the wait" + ); let receipt_checks = worker .received("Health/Check") .into_iter() diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/budgets.rs b/crates/libs/api/src/controllers/substrate/kubernetes/budgets.rs new file mode 100644 index 0000000000..6acc1b6a7d --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/kubernetes/budgets.rs @@ -0,0 +1,76 @@ +//! Wall-clock budgets a [`KubernetesSubstrate`](super::KubernetesSubstrate) +//! carries as values rather than reading from its constants, so a test can +//! shorten them while Postgres, the scripted apiserver and the scripted worker +//! stay real. + +use std::time::Duration; + +use tokio::time::Instant; + +/// The budgets of `create_execution`'s readiness wait and preserve decision, +/// `suspend_execution`'s durability wait, started warm-pool adoption's +/// split-receipt wait and teardown's worker connect. +/// +/// Production uses [`Default`], which is exactly the constants each field +/// names, so production deadlines, error messages and logs are unchanged. +/// Other paths that share a constant (wake, start and access readiness, and +/// the other warm-pool setup RPCs) still read it directly. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct SubstrateBudgets { + /// [`super::CREATE_READY_BUDGET`]: the outer bound on a create's + /// readiness wait. Its phase deadline also bounds the readiness phases of + /// a create that joins an in-flight activation; that joined wait has no + /// outer bound. + pub(super) create_ready: Duration, + /// [`super::READY_WAIT_PHASE_MARGIN`], as reserved inside + /// [`Self::create_ready`]. + pub(super) create_ready_phase_margin: Duration, + /// [`super::PRESERVE_DECISION_BUDGET`]. + pub(super) preserve_decision: Duration, + /// [`super::adapter::TEARDOWN_SUSPEND_DURABILITY_BUDGET`]. + pub(super) suspend_durability: Duration, + /// [`super::adapter::WARM_POOL_SETUP_RPC_TIMEOUT`], as the deadline for a + /// started adoption's worker to advertise split operation receipts. + pub(super) pooled_split_receipt_wait: Duration, + /// [`super::POD_WORKER_CONNECT_TIMEOUT`], as teardown's connect bound for + /// the pod's worker. + pub(super) teardown_worker_connect: Duration, +} + +impl SubstrateBudgets { + pub(super) const DEFAULT: Self = Self { + create_ready: super::CREATE_READY_BUDGET, + create_ready_phase_margin: super::READY_WAIT_PHASE_MARGIN, + preserve_decision: super::PRESERVE_DECISION_BUDGET, + suspend_durability: super::adapter::TEARDOWN_SUSPEND_DURABILITY_BUDGET, + pooled_split_receipt_wait: super::adapter::WARM_POOL_SETUP_RPC_TIMEOUT, + teardown_worker_connect: super::POD_WORKER_CONNECT_TIMEOUT, + }; + + /// [`super::support::ready_wait_bounds`] within these budgets: the outer + /// backstop and the phase-aware inner deadline for a readiness wait of + /// `requested`. The inner wait is clamped to + /// `create_ready − create_ready_phase_margin`, and the outer backstop is + /// one margin later, so it never exceeds `create_ready` while the margin + /// fits inside it. + pub(super) fn ready_wait_bounds(&self, requested: Duration) -> (Duration, Instant) { + // The defaults are statically asserted to leave room. A test-shortened + // pair must too, or the backstop would outlast `create_ready`. + debug_assert!( + self.create_ready_phase_margin <= self.create_ready, + "the phase margin must fit inside the create budget" + ); + let inner = requested.min( + self.create_ready + .saturating_sub(self.create_ready_phase_margin), + ); + let outer = inner + self.create_ready_phase_margin; + (outer, Instant::now() + inner) + } +} + +impl Default for SubstrateBudgets { + fn default() -> Self { + Self::DEFAULT + } +} diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/mod.rs b/crates/libs/api/src/controllers/substrate/kubernetes/mod.rs index 4ec02c44df..1a6d84fe7c 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/mod.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/mod.rs @@ -44,6 +44,7 @@ use tonic::{Code, transport::Channel}; use tracing::Instrument; use uuid::Uuid; +mod budgets; mod caches; mod cancel_rate_tracker; mod pod_reflector; @@ -51,6 +52,7 @@ mod restore_waiters; mod sandbox_pod_view; mod store; +use budgets::SubstrateBudgets; pub use pod_reflector::ReflectorReady; use sandbox_pod_view::NodeNetworkSlots; pub use sandbox_pod_view::{ @@ -822,6 +824,9 @@ pub struct KubernetesSubstrate { /// directly so a join's re-check interval is settable in tests; /// production takes the default, which is those constants. recovery_join_backoff: execution::RecoveryJoinBackoffPolicy, + /// Readiness, preserve-decision, suspend and connect budgets. Production + /// takes the default, which is the constants each one names. + budgets: SubstrateBudgets, shutdown_token: CancellationToken, task_tracker: TaskTracker, } @@ -871,6 +876,7 @@ impl KubernetesSubstrate { node_group_placement, cancel_tracker: CancelRateTracker::with_default_budget(), recovery_join_backoff: execution::RecoveryJoinBackoffPolicy::default(), + budgets: SubstrateBudgets::default(), shutdown_token, task_tracker, })) diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs b/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs index f3c49180c8..16a39c6bc7 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs @@ -145,6 +145,8 @@ type CallLog = Arc>>; pub(super) enum ScriptedReply { Status(tonic::Status), Message(Bytes), + /// Never answer, for callers that bound the call with their own budget. + Hang, /// Answers the n-th call to its method with the n-th reply and every /// later call with the last one. An empty sequence scripts nothing. Sequence(Arc<[ScriptedReply]>), @@ -235,6 +237,7 @@ where match reply.as_ref().map(|reply| reply.for_attempt(attempt)) { Some(ScriptedReply::Status(status)) => Ok(status.clone().into_http()), Some(ScriptedReply::Message(message)) => Ok(grpc_unary_ok(message.clone())), + Some(ScriptedReply::Hang) => std::future::pending().await, Some(ScriptedReply::Sequence(_)) | None => { let request = http::Request::from_parts( parts, diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/support.rs b/crates/libs/api/src/controllers/substrate/kubernetes/support.rs index 8ef0760539..01690064b4 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/support.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/support.rs @@ -1345,11 +1345,9 @@ async fn release_claimed_execution( /// before the node-RPC channel timeout severs the connection. pub(super) fn ready_wait_bounds(requested: Duration) -> (Duration, Instant) { // `READY_WAIT_PHASE_MARGIN < CREATE_READY_BUDGET` is statically - // asserted next to the constants, so neither expression can underflow - // or exceed the channel-safe budget. - let inner = requested.min(CREATE_READY_BUDGET - READY_WAIT_PHASE_MARGIN); - let outer = inner + READY_WAIT_PHASE_MARGIN; - (outer, Instant::now() + inner) + // asserted next to the constants, so neither bound can underflow or + // exceed the channel-safe budget. + SubstrateBudgets::DEFAULT.ready_wait_bounds(requested) } /// Time remaining until `phase_deadline`, zero once it has passed. Used to diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs b/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs index 60554f426f..7bac960336 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs @@ -246,6 +246,7 @@ mod kubernetes_tests { // snapshots; resolution by `node_group_placement`'s own tests. node_group_placement: NodeGroupPlacementResolver::new(pool, false), recovery_join_backoff: RecoveryJoinBackoffPolicy::default(), + budgets: SubstrateBudgets::default(), shutdown_token, task_tracker, } @@ -1941,6 +1942,7 @@ mod kubernetes_tests { // snapshots; resolution by `node_group_placement`'s own tests. node_group_placement: NodeGroupPlacementResolver::new(pool, false), recovery_join_backoff: RecoveryJoinBackoffPolicy::default(), + budgets: SubstrateBudgets::default(), shutdown_token, task_tracker, } @@ -8924,6 +8926,7 @@ mod kubernetes_tests { // snapshots; resolution by `node_group_placement`'s own tests. node_group_placement: NodeGroupPlacementResolver::new(pool, false), recovery_join_backoff: RecoveryJoinBackoffPolicy::default(), + budgets: SubstrateBudgets::default(), shutdown_token, task_tracker, }