api, cluster-controller: sort the tests that use the Direct substrate as a fixture (tests only)

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

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 {

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

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()),

crates/libs/api/src/controllers/sandbox/queries/create_postgres_tests.rs

@@ -494,6 +494,7 @@ fn test_federated_owner(upstream_org_id: &str) -> TestResult<FederatedOwner> {
     })
 }
 
+// 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()),

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

crates/libs/api/src/controllers/sandbox/updates/postgres_tests.rs

@@ -70,6 +70,7 @@ fn metadata_request(metadata: Option<SandboxMetadata>) -> 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()),

crates/libs/api/src/controllers/substrate/direct.rs

@@ -1,4 +1,4 @@
 use pplx_space_common_core::sql;
 use pplx_space_common_svc::sql::contention;
 use std::fmt;
 use std::time::Duration;
@@ -4217,5 +4217,7 @@
     Ok(())
 }
 
+#[cfg(test)]
+mod cluster_admission_tests;
 #[cfg(test)]
 mod tests;

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<std::convert::Infallible>,
+) -> 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(())
+}

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()),
             )),

crates/libs/api/src/controllers/template/tests.rs

@@ -386,6 +386,7 @@ async fn template_controller_fixture() -> TestDbResult<TemplateControllerFixture
     .execute(&db.pool)
     .await?;
     let store = TemplateBuildJobsStore::new(db.pool.clone());
+    // Direct is only a placeholder: the tests using this fixture never reach the substrate.
     let substrate = Substrate::Direct(Box::new(
         crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()),
     ));

crates/libs/api/src/public_sandbox_operations/tests.rs

@@ -57,6 +57,9 @@ fn claimed_create_operation_for(
     }
 }
 
+// Direct is only a placeholder for the existing-row tests, which never reach the
+// substrate. The admitted re-execution test's `sandbox_k8s_executions` check can
+// only fail on Kubernetes.
 fn direct_sandbox_controller(pool: &PgPool) -> SandboxController {
     SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new(
         crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()),

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

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(),

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(),
         )))),