From bcd77b446980dc50fc47839c2128b404656f8ea2 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Tue, 13 Oct 2026 21:29:43 +0000 Subject: [PATCH] api, cluster-controller: sort the tests that use the Direct substrate as a fixture (tests only) Production and Kind run the Kubernetes substrate, but several api and cluster-controller tests build a Direct substrate because it needs no apiserver. Sort the tests that construct it by what they actually depend on: - `stale_heartbeat_direct_worker_is_not_empty_cluster` pins Direct's own inventory SQL (a registered worker with a stale heartbeat still counts as eligible), so it moves, byte for byte, from `cluster_admission::tests` into `substrate::direct::cluster_admission_tests`. The admission suite's `direct_substrate` and `enabled_config` become `pub(crate)` inside that `#[cfg(test)]` module. Its `admission` helper takes a bucket type private to `cluster_admission`, so the new module builds the same bucketless admission through `ClusterSandboxAdmission::with_config`. - Tests in which Direct only fills a constructor slot, or answers a question whose answer is the same on Kubernetes (zero or some eligible nodes, or `managed_by` bound into a query with no Kubernetes-specific term), get a short comment saying Direct is a placeholder. - Tests of shared logic whose outcome or path differs on Kubernetes are left unchanged, for a later change to run on a Kubernetes fixture. No production behavior, schema, wire format, metric or configuration changes. The only edits outside test files are a `#[cfg(test)]` module declaration and `pub(crate)` on an existing `#[cfg(test)]` module declaration. The test count is unchanged. --- crates/libs/api/src/cluster_admission/mod.rs | 2 +- .../libs/api/src/cluster_admission/tests.rs | 50 ++----------- .../lifecycle/characterization_tests.rs | 5 ++ .../controllers/sandbox/lifecycle/tests.rs | 4 + .../sandbox/queries/create_postgres_tests.rs | 1 + .../src/controllers/sandbox/spaced/tests.rs | 1 + .../sandbox/updates/postgres_tests.rs | 1 + .../api/src/controllers/substrate/direct.rs | 2 + .../direct/cluster_admission_tests.rs | 73 +++++++++++++++++++ .../controllers/substrate/kubernetes/tests.rs | 10 +++ .../api/src/controllers/template/tests.rs | 1 + .../src/public_sandbox_operations/tests.rs | 3 + .../src/public_operation_dispatcher/tests.rs | 3 + .../lifecycle_writer_tests.rs | 2 + .../controller/server/src/warm_pool/tests.rs | 5 ++ 15 files changed, 117 insertions(+), 46 deletions(-) create mode 100644 crates/libs/api/src/controllers/substrate/direct/cluster_admission_tests.rs diff --git a/crates/libs/api/src/cluster_admission/mod.rs b/crates/libs/api/src/cluster_admission/mod.rs index aa45b85f21..d5f91823ad 100644 --- a/crates/libs/api/src/cluster_admission/mod.rs +++ b/crates/libs/api/src/cluster_admission/mod.rs @@ -31,7 +31,7 @@ use crate::controllers::substrate::{ mod redis_bucket; #[cfg(test)] -mod tests; +pub(crate) mod tests; use redis_bucket::{ConsumeOutcome, RedisTokenBucket}; diff --git a/crates/libs/api/src/cluster_admission/tests.rs b/crates/libs/api/src/cluster_admission/tests.rs index 4f6fa5a076..d917fae584 100644 --- a/crates/libs/api/src/cluster_admission/tests.rs +++ b/crates/libs/api/src/cluster_admission/tests.rs @@ -18,7 +18,10 @@ use crate::controllers::substrate::{Substrate, direct::DirectSubstrate}; use pplx_space_test_env::TestResult; -fn direct_substrate(pool: &PgPool) -> Substrate { +// Direct is only a placeholder capacity source for most of this module's tests: +// they need zero or some eligible nodes. `capacity_probe_failure_fails_open` +// needs a count that fails, which only Direct's can. +pub(crate) fn direct_substrate(pool: &PgPool) -> Substrate { Substrate::Direct(Box::new(DirectSubstrate::new(pool.clone()))) } @@ -47,7 +50,7 @@ fn admission_with_flags( } } -fn enabled_config() -> ClusterSandboxAdmissionConfig { +pub(crate) fn enabled_config() -> ClusterSandboxAdmissionConfig { ClusterSandboxAdmissionConfig { enabled: true, refill_rps: 6.0, @@ -741,49 +744,6 @@ async fn fails_open_when_no_redis_configured() -> TestResult { Ok(()) } -#[tokio::test] -async fn stale_heartbeat_direct_worker_is_not_empty_cluster() -> TestResult { - let Some(db) = try_pg_test_db().await? else { - return Ok(()); - }; - // A registered worker whose status heartbeat is stale is still - // placement-eligible via direct placement's last-known fallback, so the - // empty-cluster guard must count it rather than reject as `empty_cluster`. - let node_id = NodeId::new(); - sqlx::query( - "INSERT INTO nodes (id, address, role, managed_by, state) - VALUES ($1, '127.0.0.1', 'worker', 'space', 'registered')", - ) - .bind(node_id) - .execute(&db.pool) - .await?; - sqlx::query( - "INSERT INTO node_direct (node_id, name, node_api_endpoint) - VALUES ($1, 'stale-worker', '127.0.0.1:9802')", - ) - .bind(node_id) - .execute(&db.pool) - .await?; - sqlx::query( - "INSERT INTO node_status_snapshots ( - node_id, observed_at, stale_after, reachable, role, - scheduling_status, services, updated_at - ) - VALUES ($1, NOW() - INTERVAL '1 hour', NOW() - INTERVAL '30 minutes', true, 'worker', 'ready', '[]'::jsonb, NOW() - INTERVAL '1 hour')", - ) - .bind(node_id) - .execute(&db.pool) - .await?; - - // No bucket → the only way this returns Ok is if the guard counted the - // stale worker (fail open); a false empty_cluster would return EmptyCluster. - let admission = admission(direct_substrate(&db.pool), enabled_config(), None); - admission - .admit_activation(ClusterActivationClass::Cold) - .await?; - Ok(()) -} - #[tokio::test] async fn exhausted_bucket_rate_limits_activation() -> TestResult { let Some(db) = try_pg_test_db().await? else { diff --git a/crates/libs/api/src/controllers/sandbox/lifecycle/characterization_tests.rs b/crates/libs/api/src/controllers/sandbox/lifecycle/characterization_tests.rs index ca8681a6e9..edd712a405 100644 --- a/crates/libs/api/src/controllers/sandbox/lifecycle/characterization_tests.rs +++ b/crates/libs/api/src/controllers/sandbox/lifecycle/characterization_tests.rs @@ -314,6 +314,7 @@ async fn wake_refuses_teardown_and_terminal_lifecycles_before_admission() -> Tes async fn non_lifecycle_actions_are_bad_requests_and_write_nothing() -> TestResult { let db = pg_test_db_with_max_connections(2).await?; let pool = &db.pool; + // Direct is only a placeholder here: this test never reaches the substrate. let controller = direct_controller(pool); let internal_labels = [ SandboxAction::Access, @@ -366,6 +367,7 @@ async fn non_lifecycle_actions_are_bad_requests_and_write_nothing() -> TestResul async fn drain_gated_actions_conflict_and_leave_the_row() -> TestResult { let db = pg_test_db_with_max_connections(3).await?; let pool = &db.pool; + // Direct is only a placeholder here: this test never reaches the substrate. let controller = direct_controller(pool); let node_id = NodeId::new(); sqlx::query(sql!( @@ -933,6 +935,8 @@ async fn stop_and_kill_on_a_draining_node_still_reach_the_worker() -> TestResult async fn untracked_start_and_restore_validate_then_charge_admission() -> TestResult { let db = pg_test_db_with_max_connections(3).await?; let pool = &db.pool; + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. let controller = controller_with_empty_cluster_admission(pool); for action in [SandboxAction::Start, SandboxAction::Restore] { @@ -1070,6 +1074,7 @@ async fn insert_observed_state( async fn observed_only_delete_without_an_observation_is_not_found() -> TestResult { let db = pg_test_db_with_max_connections(2).await?; let pool = &db.pool; + // Direct is only a placeholder here: this test never reaches the substrate. let controller = direct_controller(pool); let error = controller diff --git a/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs b/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs index cd7aab0487..514f536dcf 100644 --- a/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs +++ b/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs @@ -210,6 +210,8 @@ async fn rejected_activation_emits_no_orphaned_requested_event() -> TestResult { .await? .expect("sandbox row exists"); + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. let make_substrate = || { crate::controllers::substrate::Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), @@ -535,6 +537,8 @@ async fn invalid_activation_options_reject_before_admission_and_requested_event( // Admission that would reject (empty cluster) — the invalid request must // still fail with a 400 before this ever runs. + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. let make_substrate = || { crate::controllers::substrate::Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), diff --git a/crates/libs/api/src/controllers/sandbox/queries/create_postgres_tests.rs b/crates/libs/api/src/controllers/sandbox/queries/create_postgres_tests.rs index 6364aa25f9..5737c431c5 100644 --- a/crates/libs/api/src/controllers/sandbox/queries/create_postgres_tests.rs +++ b/crates/libs/api/src/controllers/sandbox/queries/create_postgres_tests.rs @@ -494,6 +494,7 @@ fn test_federated_owner(upstream_org_id: &str) -> TestResult { }) } +// Direct is only a placeholder: no test using this depends on substrate behavior. fn direct_sandbox_controller(pool: &PgPool) -> SandboxController { SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), diff --git a/crates/libs/api/src/controllers/sandbox/spaced/tests.rs b/crates/libs/api/src/controllers/sandbox/spaced/tests.rs index bc42f890cf..27b4ef17ee 100644 --- a/crates/libs/api/src/controllers/sandbox/spaced/tests.rs +++ b/crates/libs/api/src/controllers/sandbox/spaced/tests.rs @@ -240,6 +240,7 @@ async fn reconnect_node_api_is_blocked_by_access_drain_gate() -> TestResult { let db = pg_test_db().await?; let node_api = start_fake_reconnect_node_api().await?; let sandbox = insert_direct_sandbox_on_node(&db.pool, node_api.addr, Some("draining")).await?; + // Direct is only a placeholder here: this test never reaches the substrate. let substrate = Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), )); diff --git a/crates/libs/api/src/controllers/sandbox/updates/postgres_tests.rs b/crates/libs/api/src/controllers/sandbox/updates/postgres_tests.rs index 2122406e3e..8758e54e07 100644 --- a/crates/libs/api/src/controllers/sandbox/updates/postgres_tests.rs +++ b/crates/libs/api/src/controllers/sandbox/updates/postgres_tests.rs @@ -70,6 +70,7 @@ fn metadata_request(metadata: Option) -> UpdateSandboxRequest { } } +// Direct is only a placeholder: the tests using this never reach the substrate. fn direct_sandbox_controller(pool: &PgPool) -> SandboxController { SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), diff --git a/crates/libs/api/src/controllers/substrate/direct.rs b/crates/libs/api/src/controllers/substrate/direct.rs index 9d1c7af18b..450773645f 100644 --- a/crates/libs/api/src/controllers/substrate/direct.rs +++ b/crates/libs/api/src/controllers/substrate/direct.rs @@ -4217,5 +4217,7 @@ fn ensure_direct_create_supported(fork_disk_restore: bool) -> Result<(), Substra Ok(()) } +#[cfg(test)] +mod cluster_admission_tests; #[cfg(test)] mod tests; diff --git a/crates/libs/api/src/controllers/substrate/direct/cluster_admission_tests.rs b/crates/libs/api/src/controllers/substrate/direct/cluster_admission_tests.rs new file mode 100644 index 0000000000..5b473b5ca1 --- /dev/null +++ b/crates/libs/api/src/controllers/substrate/direct/cluster_admission_tests.rs @@ -0,0 +1,73 @@ +//! Cluster admission over Direct's own inventory count +//! (`DirectSubstrate::eligible_start_node_count`), which counts a registered +//! worker with a stale heartbeat as eligible, matching direct placement's +//! last-known-worker fallback. Kubernetes counts schedulable nodes from its +//! node cache instead. + +use pplx_space_common_core::types::NodeId; +use pplx_space_db_test_fixtures::try_pg_test_db; +use pplx_space_test_env::TestResult; + +use crate::cluster_admission::tests::{direct_substrate, enabled_config}; +use crate::cluster_admission::{ + ClusterActivationClass, ClusterSandboxAdmission, ClusterSandboxAdmissionConfig, +}; +use crate::controllers::substrate::Substrate; + +/// Stands in for `cluster_admission::tests::admission`, whose bucket parameter +/// is a type private to `cluster_admission`. The test passes no bucket, and +/// `with_config` builds none from a config without a Redis URL. +fn admission( + substrate: Substrate, + config: ClusterSandboxAdmissionConfig, + _bucket: Option, +) -> ClusterSandboxAdmission { + ClusterSandboxAdmission::with_config( + substrate, + config, + pplx_space_feature_flags::FeatureFlagClient::disabled(), + ) +} + +#[tokio::test] +async fn stale_heartbeat_direct_worker_is_not_empty_cluster() -> TestResult { + let Some(db) = try_pg_test_db().await? else { + return Ok(()); + }; + // A registered worker whose status heartbeat is stale is still + // placement-eligible via direct placement's last-known fallback, so the + // empty-cluster guard must count it rather than reject as `empty_cluster`. + let node_id = NodeId::new(); + sqlx::query( + "INSERT INTO nodes (id, address, role, managed_by, state) + VALUES ($1, '127.0.0.1', 'worker', 'space', 'registered')", + ) + .bind(node_id) + .execute(&db.pool) + .await?; + sqlx::query( + "INSERT INTO node_direct (node_id, name, node_api_endpoint) + VALUES ($1, 'stale-worker', '127.0.0.1:9802')", + ) + .bind(node_id) + .execute(&db.pool) + .await?; + sqlx::query( + "INSERT INTO node_status_snapshots ( + node_id, observed_at, stale_after, reachable, role, + scheduling_status, services, updated_at + ) + VALUES ($1, NOW() - INTERVAL '1 hour', NOW() - INTERVAL '30 minutes', true, 'worker', 'ready', '[]'::jsonb, NOW() - INTERVAL '1 hour')", + ) + .bind(node_id) + .execute(&db.pool) + .await?; + + // No bucket → the only way this returns Ok is if the guard counted the + // stale worker (fail open); a false empty_cluster would return EmptyCluster. + let admission = admission(direct_substrate(&db.pool), enabled_config(), None); + admission + .admit_activation(ClusterActivationClass::Cold) + .await?; + Ok(()) +} diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs b/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs index 7bac960336..76a8a3f635 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs @@ -5932,6 +5932,8 @@ mod kubernetes_tests { /// bucket needed. fn rejecting_admission(pool: &PgPool) -> crate::cluster_admission::ClusterSandboxAdmission { crate::cluster_admission::ClusterSandboxAdmission::with_config( + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. crate::controllers::substrate::Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), )), @@ -9403,6 +9405,8 @@ mod kubernetes_tests { ); let admission = std::sync::Arc::new( crate::cluster_admission::ClusterSandboxAdmission::with_config( + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool), )), @@ -9465,6 +9469,8 @@ mod kubernetes_tests { ); let admission = std::sync::Arc::new( crate::cluster_admission::ClusterSandboxAdmission::with_config( + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), )), @@ -9607,6 +9613,8 @@ mod kubernetes_tests { ))); let admission = std::sync::Arc::new( crate::cluster_admission::ClusterSandboxAdmission::with_config( + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), )), @@ -9668,6 +9676,8 @@ mod kubernetes_tests { mock_kubernetes_client(Arc::clone(&captured_pod_creates)), ))); let admission = crate::cluster_admission::ClusterSandboxAdmission::with_config( + // Direct is only a placeholder capacity source: it reports zero eligible nodes, + // as a synced, empty Kubernetes node cache would. Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), )), diff --git a/crates/libs/api/src/controllers/template/tests.rs b/crates/libs/api/src/controllers/template/tests.rs index 20a37f5248..1de9e9f19f 100644 --- a/crates/libs/api/src/controllers/template/tests.rs +++ b/crates/libs/api/src/controllers/template/tests.rs @@ -386,6 +386,7 @@ async fn template_controller_fixture() -> TestDbResult SandboxController { SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), diff --git a/crates/libs/cluster/controller/server/src/public_operation_dispatcher/tests.rs b/crates/libs/cluster/controller/server/src/public_operation_dispatcher/tests.rs index 82cf4c7e68..5ec054f1b0 100644 --- a/crates/libs/cluster/controller/server/src/public_operation_dispatcher/tests.rs +++ b/crates/libs/cluster/controller/server/src/public_operation_dispatcher/tests.rs @@ -687,6 +687,8 @@ async fn native_executor_requeues_create_when_cluster_admission_rejects() -> Tes attempts: 1, claim_generation: 1, }; + // Direct is only a placeholder: admission sees zero eligible nodes, as it would + // from a synced, empty Kubernetes node cache, and rejects before any create. let substrate = Substrate::Direct(Box::new( pplx_space_api::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), )); @@ -790,6 +792,7 @@ async fn native_executor_persists_exemptions_on_policy_update_and_template_reset .bind(org_id) .execute(&db.pool) .await?; + // Direct is only a placeholder here: this test never reaches the substrate. let substrate = Substrate::Direct(Box::new( pplx_space_api::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), )); diff --git a/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/lifecycle_writer_tests.rs b/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/lifecycle_writer_tests.rs index 68823c711a..9674aa16cc 100644 --- a/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/lifecycle_writer_tests.rs +++ b/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/lifecycle_writer_tests.rs @@ -46,6 +46,8 @@ fn age() -> PositiveDuration { PositiveDuration::new(Duration::from_secs(60)).expect("non-zero") } +// Direct is only a placeholder: it supplies `managed_by`, and only the converge +// CAS would decide differently for a Kubernetes label (see the module doc). fn direct_sink(pool: &PgPool) -> SubstrateSweepSink { SubstrateSweepSink::new( pool.clone(), diff --git a/crates/libs/cluster/controller/server/src/warm_pool/tests.rs b/crates/libs/cluster/controller/server/src/warm_pool/tests.rs index 6593b11d14..03c20ddf73 100644 --- a/crates/libs/cluster/controller/server/src/warm_pool/tests.rs +++ b/crates/libs/cluster/controller/server/src/warm_pool/tests.rs @@ -155,6 +155,7 @@ fn reconcile_stage_test_reconciler_inner() -> Reconciler { }, node_store, store: WarmPoolStore::new(postgres.clone()), + // Direct is only a placeholder: the tests using this never reach the substrate. substrate: Arc::new(Substrate::Direct(Box::new(DirectSubstrate::new(postgres)))), namespace: "space".to_string(), pod_config: pool_pod_config(), @@ -205,6 +206,7 @@ fn reconcile_test_reconciler( }, node_store, store, + // Direct is only a placeholder: the tests using this never reach the substrate. substrate: Arc::new(Substrate::Direct(Box::new(DirectSubstrate::new(postgres)))), namespace: "space".to_string(), pod_config: pool_pod_config(), @@ -879,6 +881,7 @@ async fn observe_warming_slots_retries_not_started_probe_and_promotes_slot() -> }, node_store, store: store.clone(), + // Direct is only a placeholder here: this test never reaches the substrate. substrate: Arc::new(Substrate::Direct(Box::new(DirectSubstrate::new( db.pool.clone(), )))), @@ -971,6 +974,7 @@ async fn observe_warming_slots_keeps_young_slot_missing_from_cache() -> TestResu }, node_store, store: store.clone(), + // Direct is only a placeholder here: this test never reaches the substrate. substrate: Arc::new(Substrate::Direct(Box::new(DirectSubstrate::new( db.pool.clone(), )))), @@ -1328,6 +1332,7 @@ async fn reconcile_once_starts_preparing_republished_exact_release() -> TestResu }, node_store, store: store.clone(), + // Direct is only a placeholder here: this test never reaches the substrate. substrate: Arc::new(Substrate::Direct(Box::new(DirectSubstrate::new( db.pool.clone(), )))),