api: let tests inject the Kubernetes readiness and suspend budgets

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);

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

@@ -1,4 +1,4 @@
 //! `SubstrateAdapter` implementation for `KubernetesSubstrate`.
 
 use super::*;
 
@@ -135,7 +135,7 @@
                     )
                     .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 @@
         // 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 @@
         .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 @@
                 // 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 @@
         &self,
         sandbox_id: SandboxId,
     ) -> Result<Option<PublishedLocalArtifact>, 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 @@
                 expected_generation,
                 &endpoint_addr,
                 suspend_deadline,
+                self.budgets.suspend_durability,
             )
             .await
             {
@@ -2621,7 +2614,7 @@
         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 @@
         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,
+    }
+}

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<PublishedLocalArtifact, SubstrateError> {
     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<F, Fut>(
     execution_generation: ExecutionGeneration,
     endpoint_addr: &str,
     deadline: Instant,
+    // The budget `deadline` was derived from, named in the timeout error.
+    durability_budget: Duration,
 ) -> Result<PublishedLocalArtifact, SubstrateError>
 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,
             })? {

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 {

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 { .. })));

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;
 

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"