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, }