From 3112b4712a262490884e4cad22a53f1901f0f55d Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Fri, 16 Oct 2026 21:46:59 +0000 Subject: [PATCH] api, cluster-controller: make the Substrate variants private to the substrate module `Substrate` becomes a struct over a private `Inner` enum, so only `controllers::substrate` and its submodules can construct or match the Direct and Kubernetes variants. A new `match substrate { Substrate::Direct(..) => .. }` in shared code, tests included, now fails to compile (E0599); the inner enum and its field are private (E0603, E0616). Shared code asks `capabilities()` for static facts, calls `SubstrateAdapter` or inherent methods for behavior, and uses `as_kubernetes()`, binding its payload, for Kubernetes-only state. `SubstrateAdapter::managed_by()` still names the running substrate for SQL row scope and wire labels. Earlier phase 3 changes had already converted every production variant match outside the module, so the compiler found only construction sites. They now use `Substrate::from_direct` and `Substrate::from_kubernetes`: `from_env_with_lifecycle` and the test fixtures in the api crate and the cluster controller. The forwarding impls inside the module match `&self.inner`. The cluster controller's leader-election choice was the remaining identity check: it matched `as_kubernetes()` as `Some(_)` and built its lease client from the environment. It now reads the new `SubstrateCapabilities::contends_for_leader_lease` (Kubernetes true, Direct false) through `LeaderElectionConfig::contends`, which also applies the kill-switch. The decision is unchanged, and a table test covers both substrates with election on and off. Two archcheck `pattern_ratchet` instances fence identity checks. `kubernetes_access` counts every `as_kubernetes(` call outside the substrate module against a baseline (today one, in `reconcilers.rs`, which binds the payload), so a new caller in any form needs a reviewed baseline entry. `capability_identity` rejects naming `SubstrateCapabilities::DIRECT` or `::KUBERNETES` in production code outside the two substrates' `capabilities()`, since comparing `capabilities()` with a const is another identity check. No production behavior, wire format, schema, metric or configuration changes. The substrate `AGENTS.md` now says the variants are private, how shared code builds a `Substrate`, how to add a capability, that `as_kubernetes()` must bind and use its payload, what the two ratchets enforce, and that `managed_by()` is never parsed to decide behavior: the transient-lifecycle sweeper's `failed_active_execution_scope` is the only parse, and new fixed-scope SQL returns early by reading a capability or reusing that helper. The api crate's `AGENTS.md` points at `capabilities()` and `as_kubernetes()` instead of `Substrate::Direct`. --- ARCHITECTURE.toml | 51 +++ architecture-baseline.toml | 4 + crates/libs/api/AGENTS.md | 6 +- .../src/controllers/sandbox/ingress/tests.rs | 4 +- .../controllers/sandbox/lifecycle/tests.rs | 27 +- .../wake_completion_postgres_tests.rs | 12 +- .../sandbox/queries/create_postgres_tests.rs | 4 +- .../src/controllers/sandbox/spaced/tests.rs | 4 +- .../sandbox/updates/postgres_tests.rs | 4 +- .../sandbox/updates/transport_tests.rs | 4 +- .../api/src/controllers/substrate/AGENTS.md | 57 ++- .../src/controllers/substrate/direct/tests.rs | 4 +- .../hydration_and_suspend.rs | 4 +- .../controllers/substrate/kubernetes/tests.rs | 68 ++-- .../kubernetes/tests/substrate_scope.rs | 8 +- .../libs/api/src/controllers/substrate/mod.rs | 346 ++++++++++-------- .../substrate/runtime_rpc/tests.rs | 2 +- .../api/src/controllers/substrate/tests.rs | 12 +- .../api/src/controllers/template/tests.rs | 12 +- .../src/public_sandbox_operations/tests.rs | 15 +- .../routes/admin/node_maintenance/tests.rs | 4 +- .../routes/admin/nodes/drain_handler_tests.rs | 4 +- .../libs/api/src/routes/admin/nodes/tests.rs | 4 +- .../api/src/routes/admin/sandboxes/tests.rs | 4 +- crates/libs/api/src/routes/e2b/tests.rs | 6 +- .../sandboxes/admission_test_fixture.rs | 2 +- crates/libs/api/src/test_support.rs | 2 +- .../server/src/leader_election/mod.rs | 9 + .../server/src/leader_election/tests/mod.rs | 33 +- .../libs/cluster/controller/server/src/lib.rs | 126 +++---- .../src/public_operation_dispatcher/tests.rs | 8 +- .../controller/server/src/reconcilers.rs | 2 +- .../lifecycle_writer_tests.rs | 4 +- .../src/transient_lifecycle_sweeper/store.rs | 2 +- .../controller/server/src/warm_pool/tests.rs | 16 +- 35 files changed, 510 insertions(+), 364 deletions(-) diff --git a/ARCHITECTURE.toml b/ARCHITECTURE.toml index 701fbec26..6fb023c69 100644 --- a/ARCHITECTURE.toml +++ b/ARCHITECTURE.toml @@ -47,6 +47,57 @@ mode = "advisory" [rules.pattern_ratchet.direct_tables.stale_entries] # entries with fewer matches, none, or gone mode = "advisory" +# Shared code reaches Kubernetes-only state through `Substrate::as_kubernetes()`, +# binding and using the payload. A presence test on it, in any form, only asks +# which substrate is running, which belongs in a `SubstrateCapabilities` field +# (crates/libs/api/src/controllers/substrate/AGENTS.md). A line matcher cannot +# tell the two apart, so every call outside the substrate module counts, and a +# new caller needs a reviewed baseline entry. +[rules.pattern_ratchet.kubernetes_access] +patterns = ['\bas_kubernetes\s*\('] +allow = ["crates/libs/api/src/controllers/substrate/**"] + +[rules.pattern_ratchet.kubernetes_access.no_crossing] +mode = "required" + +[rules.pattern_ratchet.kubernetes_access.no_new_entries] +mode = "advisory" + +[rules.pattern_ratchet.kubernetes_access.no_raised_caps] +mode = "advisory" + +[rules.pattern_ratchet.kubernetes_access.listed_no_growth] +mode = "required" + +[rules.pattern_ratchet.kubernetes_access.stale_entries] +mode = "required" + +# `SubstrateCapabilities` derives `PartialEq` and its per-substrate consts are +# public, so comparing `capabilities()` with one of them is another way to ask +# which substrate is running. Only the two substrates' `capabilities()` name +# them in production code; tests are outside the inventory. +[rules.pattern_ratchet.capability_identity] +patterns = ['\bSubstrateCapabilities\s*::\s*(?:DIRECT|KUBERNETES)\b'] +allow = [ + "crates/libs/api/src/controllers/substrate/direct.rs", + "crates/libs/api/src/controllers/substrate/kubernetes/adapter/substrate_adapter.rs", +] + +[rules.pattern_ratchet.capability_identity.no_crossing] +mode = "required" + +[rules.pattern_ratchet.capability_identity.no_new_entries] +mode = "advisory" + +[rules.pattern_ratchet.capability_identity.no_raised_caps] +mode = "advisory" + +[rules.pattern_ratchet.capability_identity.listed_no_growth] +mode = "required" + +[rules.pattern_ratchet.capability_identity.stale_entries] +mode = "required" + # Rootfs copies must preserve NoCoW and sparse extents through common-io's # reflink helper. A line matcher cannot infer a path variable's contents, so # conservatively fence all new plain filesystem copies and cp launches. diff --git a/architecture-baseline.toml b/architecture-baseline.toml index b8f20e98d..36848d33a 100644 --- a/architecture-baseline.toml +++ b/architecture-baseline.toml @@ -127,6 +127,10 @@ "crates/libs/internal/api/server/src/lib.rs" = 1 "crates/libs/internal/api/server/src/published_state.rs" = 4 "crates/libs/sandbox/store/src/lib.rs" = 2 + +[pattern_ratchet.kubernetes_access] +# Spawns the Kubernetes-only reconcilers from the bound `KubernetesSubstrate`. +"crates/libs/cluster/controller/server/src/reconcilers.rs" = 1 [pattern_ratchet.rootfs_copy] # Local nspawn and privileged builder copies are deferred; builders stay CoW. "crates/apps/local/node-mgr/svc/src/snapshot.rs" = 2 diff --git a/crates/libs/api/AGENTS.md b/crates/libs/api/AGENTS.md index fa4599fed..52ecf2905 100644 --- a/crates/libs/api/AGENTS.md +++ b/crates/libs/api/AGENTS.md @@ -1,8 +1,8 @@ # `pplx-space-api` -Before changing the substrate, or shared code that branches on the substrate -(`Substrate::Direct`, `managed_by()`) or names a Direct-only table, read -`src/controllers/substrate/AGENTS.md`. +Before changing the substrate, or shared code that depends on the substrate +(`capabilities()`, `as_kubernetes()`, `managed_by()`) or names a Direct-only +table, read `src/controllers/substrate/AGENTS.md`. ## Changes diff --git a/crates/libs/api/src/controllers/sandbox/ingress/tests.rs b/crates/libs/api/src/controllers/sandbox/ingress/tests.rs index 00a8a6e04..2959ec533 100644 --- a/crates/libs/api/src/controllers/sandbox/ingress/tests.rs +++ b/crates/libs/api/src/controllers/sandbox/ingress/tests.rs @@ -285,9 +285,9 @@ async fn stored_ingress_access_resumes_paused_execution_before_reporting_ready() .await?; let backend = StoredAccess { pool: db.pool.clone(), - controller: SandboxController::new(Substrate::Direct(Box::new( + controller: SandboxController::new(Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - ))), + )), }; for (wake_enabled, pause_after_observation, worker_allows_access) in [ (true, false, true), diff --git a/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs b/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs index 514f536dc..b103bbecd 100644 --- a/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs +++ b/crates/libs/api/src/controllers/sandbox/lifecycle/tests.rs @@ -213,9 +213,9 @@ async fn rejected_activation_emits_no_orphaned_requested_event() -> TestResult { // 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::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )) + ) }; let admission = crate::cluster_admission::ClusterSandboxAdmission::with_config( make_substrate(), @@ -287,9 +287,9 @@ async fn direct_start_restore_spends_deferred_admission_at_entry() -> TestResult .await? .expect("sandbox row exists"); let make_substrate = || { - crate::controllers::substrate::Substrate::Direct(Box::new( + crate::controllers::substrate::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )) + ) }; let admission = crate::cluster_admission::ClusterSandboxAdmission::with_config( make_substrate(), @@ -540,9 +540,9 @@ async fn invalid_activation_options_reject_before_admission_and_requested_event( // 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::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )) + ) }; let admission = crate::cluster_admission::ClusterSandboxAdmission::with_config( make_substrate(), @@ -748,9 +748,9 @@ async fn exhausting_the_finalize_retry_budget_stays_retryable() -> TestResult { .await? .expect("sandbox row exists"); let make_substrate = || { - crate::controllers::substrate::Substrate::Direct(Box::new( + crate::controllers::substrate::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )) + ) }; let admission = crate::cluster_admission::ClusterSandboxAdmission::with_config( make_substrate(), @@ -838,10 +838,9 @@ async fn a_contended_delete_precheck_stays_retryable() -> TestResult { .execute(&mut *blocker) .await?; - let controller = - SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( - crate::controllers::substrate::direct::DirectSubstrate::new(bounded.clone()), - ))); + let controller = SandboxController::new(crate::controllers::substrate::Substrate::from_direct( + crate::controllers::substrate::direct::DirectSubstrate::new(bounded.clone()), + )); let error = controller .delete_control_plane(&bounded, sandbox_id, Some(&sandbox)) .await @@ -921,9 +920,9 @@ async fn resume_uses_resolved_runtime_without_async_observation() -> TestResult record.execution_generation = generation; worker.state().insert_new(record).await?; let controller = - SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( + SandboxController::new(crate::controllers::substrate::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - ))); + )); let mut client = controller .substrate .runtime_client(id, crate::controllers::substrate::RecoveryCharge::Exempt) diff --git a/crates/libs/api/src/controllers/sandbox/lifecycle/wake_completion_postgres_tests.rs b/crates/libs/api/src/controllers/sandbox/lifecycle/wake_completion_postgres_tests.rs index 50801052a..98e726c4c 100644 --- a/crates/libs/api/src/controllers/sandbox/lifecycle/wake_completion_postgres_tests.rs +++ b/crates/libs/api/src/controllers/sandbox/lifecycle/wake_completion_postgres_tests.rs @@ -60,14 +60,14 @@ async fn wake_completion_cas_outcome_for_every_prior_lifecycle() -> TestResult { // `wake_completion_outcome_maps_to_the_wake_response`; the Kubernetes // side, like the Restore arm, is observable only in the race. assert_eq!( - Substrate::Direct(Box::new(DirectSubstrate::new(pool.clone()))).capabilities(), + Substrate::from_direct(DirectSubstrate::new(pool.clone())).capabilities(), SubstrateCapabilities::DIRECT ); #[cfg(feature = "kubernetes")] assert_eq!( - Substrate::Kubernetes(Box::new( + Substrate::from_kubernetes( crate::controllers::substrate::kubernetes::tests::substrate_with_store(pool.clone()), - )) + ) .capabilities(), SubstrateCapabilities::KUBERNETES ); @@ -208,10 +208,8 @@ async fn wake_completion_redrive_on_kubernetes_hydrates_before_completing() -> T kube_status(StatusCode::NOT_FOUND, "NotFound") } }); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( - pool.clone(), - client, - ))); + let substrate = + Substrate::from_kubernetes(substrate_with_store_and_client(pool.clone(), client)); // A zero budget makes exactly one attempt: read, routability check, // fenced CAS. 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 41200a7d5..8218e5326 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 @@ -496,9 +496,9 @@ 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( + SandboxController::new(crate::controllers::substrate::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), - ))) + )) } async fn install_failing_federated_grant_triggers(pool: &PgPool) -> TestResult { diff --git a/crates/libs/api/src/controllers/sandbox/spaced/tests.rs b/crates/libs/api/src/controllers/sandbox/spaced/tests.rs index a0a7cd503..6b5a2d664 100644 --- a/crates/libs/api/src/controllers/sandbox/spaced/tests.rs +++ b/crates/libs/api/src/controllers/sandbox/spaced/tests.rs @@ -241,9 +241,9 @@ async fn reconnect_node_api_is_blocked_by_access_drain_gate() -> TestResult { 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( + let substrate = Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )); + ); let node_api_addr = node_api.addr.to_string(); let endpoint = crate::controllers::substrate::node_api_endpoint( &node_api_addr, 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 8758e54e0..9f359c735 100644 --- a/crates/libs/api/src/controllers/sandbox/updates/postgres_tests.rs +++ b/crates/libs/api/src/controllers/sandbox/updates/postgres_tests.rs @@ -72,9 +72,9 @@ 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( + SandboxController::new(crate::controllers::substrate::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), - ))) + )) } /// The whole path: request in, row written, and the value visible through diff --git a/crates/libs/api/src/controllers/sandbox/updates/transport_tests.rs b/crates/libs/api/src/controllers/sandbox/updates/transport_tests.rs index 67a8fe6cd..37a6cb7e2 100644 --- a/crates/libs/api/src/controllers/sandbox/updates/transport_tests.rs +++ b/crates/libs/api/src/controllers/sandbox/updates/transport_tests.rs @@ -259,9 +259,9 @@ impl Fixture { .execute(&db.pool) .await?; let controller = - SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( + SandboxController::new(crate::controllers::substrate::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - ))); + )); Ok(Self { db, controller, diff --git a/crates/libs/api/src/controllers/substrate/AGENTS.md b/crates/libs/api/src/controllers/substrate/AGENTS.md index dc40f450a..790e04ee5 100644 --- a/crates/libs/api/src/controllers/substrate/AGENTS.md +++ b/crates/libs/api/src/controllers/substrate/AGENTS.md @@ -57,30 +57,52 @@ For new work: arm give both substrates the default and skip the Kubernetes override. Only a hook shared code calls on every path may default to "nothing to do", as `retire_fence_mismatch` does, and it still needs its arm. -- Where shared code needs a static fact about the substrate, add a - `SubstrateCapabilities` field, set it in both `SubstrateCapabilities::DIRECT` - and `::KUBERNETES`, and read it through `capabilities()`; where the - substrate must act, add a required method (above). Don't match - `Substrate::Direct` or branch behavior on `managed_by()`. Matching - `Substrate::Kubernetes(k)` to reach Kubernetes-only state is fine. +- `Substrate`'s variants are private to this module (`Inner` in + `mod.rs`), so shared code cannot match them; it builds one with + `Substrate::from_direct` or `Substrate::from_kubernetes`. Where shared code + needs a static fact about the substrate, add a `SubstrateCapabilities` + field with a doc comment stating what it means, set it in both + `SubstrateCapabilities::DIRECT` and `::KUBERNETES`, and read it through + `capabilities()`; where the substrate must act, add a required method + (above). Don't widen the variants' visibility. +- Code that needs Kubernetes-only state (its client, caches, or a reconciler + typed on `KubernetesSubstrate`) uses `as_kubernetes()` and binds and uses + the payload. A presence test on it (`Some(_)`, `if let Some(_)`, + `.is_some()`, `.is_none()` and the like) only asks which substrate is + running: add a capability instead, as the cluster controller's leader + election reads `contends_for_leader_lease`. Comparing `capabilities()` with + `SubstrateCapabilities::DIRECT` or `::KUBERNETES` asks the same and is + equally off limits. archcheck enforces two line counts over production code + (tests are outside them). `pattern_ratchet.kubernetes_access` counts every + `as_kubernetes(` call outside this module, in any form, against its + `architecture-baseline.toml` entry, so a new caller needs a reviewed + baseline entry. `pattern_ratchet.capability_identity` rejects any mention + of the two consts outside `direct.rs` and + `kubernetes/adapter/substrate_adapter.rs`. +- `managed_by()` names the running substrate for SQL row scope and wire + labels. Never parse it to decide behavior. The only parse is + `failed_active_execution_scope` in `transient_lifecycle_sweeper/store.rs`, + which turns the running label into the fixed SQL scope of the sweeper's + failed-active-execution cleanup; don't add another. - Scoping a query to the rows a substrate owns is required, not debt. In new SQL, bind the substrate value instead of a `'k8s'`/`'space'` literal: `managed_by = $n` with `.bind(substrate.managed_by())` for the running substrate's rows, as the transient-lifecycle sweeper does, or a fixed `SubstrateKind::Kubernetes`/`::Space` where code targets one substrate's rows. When a fixed-scope statement may run under another substrate, return - early in Rust instead of also ANDing the running substrate's value, as the - sweeper's failed-active-execution cleanup does. Existing literals, such as - the shared `routable_execution_or_non_k8s` fragment, are debt. Never choose - a runtime endpoint from a row's label. + early in Rust instead of also ANDing the running substrate's value: read a + capability, or reuse `failed_active_execution_scope` where its scope fits, + as the sweeper's failed-active-execution cleanup does. Existing literals, + such as the shared `routable_execution_or_non_k8s` fragment, are debt. + Never choose a runtime endpoint from a row's label. - `node_direct`, `sandbox_direct_route_identities` and `sandbox_direct_route_provenance` belong to Direct. Outside tests, write them only from `direct.rs` and the Direct node registry; read them through an existing query that already joins them rather than adding a join. archcheck's `pattern_ratchet.direct_tables` rule enforces this (advisory today). -- Other existing substrate matches, Direct-table joins in shared code and - `Unsupported` trait defaults (`list_pool_configs`, `upsert_pool_config`, +- Existing Direct-table joins in shared code and `Unsupported` trait + defaults (`list_pool_configs`, `upsert_pool_config`, `disable_pool_config`) are debt, not precedent. ## Tests @@ -93,11 +115,12 @@ restore, and pooled manifests in one place. Keep narrow tests only for behavior that is hard to express clearly in the broad YAML contract test. -Use `Substrate::Direct` only to test Direct behavior (tests under `direct/`) or -behavior that genuinely doesn't depend on the substrate. Test shared lifecycle -behavior Kubernetes deployments depend on against a `Substrate::Kubernetes` -built with the `kubernetes::tests` helpers (`substrate_with_store` and its -variants), as `routes/public/sandboxes/admission_test_fixture.rs` does. +Use `Substrate::from_direct` only to test Direct behavior (tests under +`direct/`) or behavior that genuinely doesn't depend on the substrate. Test +shared lifecycle behavior Kubernetes deployments depend on against a +`Substrate::from_kubernetes` built with the `kubernetes::tests` helpers +(`substrate_with_store` and its variants), as +`routes/public/sandboxes/admission_test_fixture.rs` does. Adapter tests inside `kubernetes/` use `scripted_test_doubles.rs`, as `adapter_characterization_tests/` does. Both compile only with `--features kubernetes`: CI's `--workspace --all-features` run includes them, diff --git a/crates/libs/api/src/controllers/substrate/direct/tests.rs b/crates/libs/api/src/controllers/substrate/direct/tests.rs index 2747e3758..73282785d 100644 --- a/crates/libs/api/src/controllers/substrate/direct/tests.rs +++ b/crates/libs/api/src/controllers/substrate/direct/tests.rs @@ -2751,7 +2751,7 @@ async fn direct_create_fixture() -> TestResult> { .fetch_one(&postgres.pool) .await?; let controller = crate::controllers::sandbox::SandboxController::new( - super::super::Substrate::Direct(Box::new(DirectSubstrate::new(postgres.pool.clone()))), + super::super::Substrate::from_direct(DirectSubstrate::new(postgres.pool.clone())), ); Ok(Some(DirectCreateFixture { postgres, @@ -2953,7 +2953,7 @@ async fn live_cpu_update_preserves_allocation_across_cold_restart() -> TestResul .execute(&fixture.pool) .await?; let controller = crate::controllers::sandbox::SandboxController::new( - super::super::Substrate::Direct(Box::new(DirectSubstrate::new(fixture.pool.clone()))), + super::super::Substrate::from_direct(DirectSubstrate::new(fixture.pool.clone())), ); for quota in [2, 8] { let sandbox = crate::controllers::sandbox::fetch_sandbox(&fixture.pool, sandbox_id) 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 95da125e5..5b664be03 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 @@ -345,9 +345,9 @@ async fn routability_check_answers_with_the_hydration_verdict( HydrationPod::Forbidden => kube_status(StatusCode::FORBIDDEN, "Forbidden"), HydrationPod::PriorGeneration => unreachable!("no case scripts a predecessor's pod"), }); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_stale_cached_pod( + let substrate = Substrate::from_kubernetes(substrate_with_stale_cached_pod( &db.pool, client, sandbox_id, - ))); + )); let check = substrate .ensure_routable_execution(sandbox_id, Duration::from_secs(5)) diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs b/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs index 5370d69e4..2a59a0dd3 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/tests.rs @@ -3044,7 +3044,7 @@ mod kubernetes_tests { #[tokio::test(flavor = "current_thread")] async fn substrate_enum_forwards_observed_delete_to_kubernetes_override() { - let substrate = Substrate::Kubernetes(Box::new(substrate_with_unreachable_store())); + let substrate = Substrate::from_kubernetes(substrate_with_unreachable_store()); let target = NodeApiTarget { managed_by: SpaceNodeManagedBy::Kubernetes, role: SpaceNodeRole::Worker, @@ -3090,13 +3090,13 @@ mod kubernetes_tests { let sim_worker = start_sim_worker().await?; let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); - let controller = SandboxController::new(Substrate::Kubernetes(Box::new( + let controller = SandboxController::new(Substrate::from_kubernetes( substrate_with_store_client_and_proxy_port( pool.clone(), running_pod_kubernetes_client(Arc::clone(&captured_pod_creates), "127.0.0.1"), sim_worker.port(), ), - ))) + )) .with_route_ed25519_authority(Some(test_route_ed25519_authority())); let sandbox = fetch_sandbox(&pool, sandbox_id) .await? @@ -5607,7 +5607,7 @@ mod kubernetes_tests { ); publish_completed_restore(&pool, sandbox_id).await?; let (substrate, request_count) = substrate_with_failed_pod_creates(pool.clone()); - let controller = SandboxController::new(Substrate::Kubernetes(Box::new(substrate))); + let controller = SandboxController::new(Substrate::from_kubernetes(substrate)); // Hold the row lock so the joined follower pauses after observing the // owner's binding but before its controller completion write lands. @@ -5817,7 +5817,7 @@ mod kubernetes_tests { let release_request = Arc::new(Semaphore::new(0)); let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); let sim_worker = start_sim_worker().await?; - let controller = SandboxController::new(Substrate::Kubernetes(Box::new( + let controller = SandboxController::new(Substrate::from_kubernetes( substrate_with_store_client_and_proxy_port( pool.clone(), blocking_running_pod_kubernetes_client( @@ -5828,7 +5828,7 @@ mod kubernetes_tests { ), sim_worker.port(), ), - ))); + )); // Act: once Restore has claimed ownership and reached Pod creation, // queue Delete behind the materialization fence, then let it retire @@ -5938,9 +5938,9 @@ mod kubernetes_tests { 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::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), - )), + ), crate::cluster_admission::ClusterSandboxAdmissionConfig { enabled: true, refill_rps: 1.0, @@ -5967,10 +5967,10 @@ mod kubernetes_tests { .execute(&pool) .await?; let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), mock_kubernetes_client(Arc::clone(&captured_pod_creates)), - ))); + )); let controller = SandboxController::with_cluster_admission(substrate, rejecting_admission(&pool)); @@ -6010,10 +6010,10 @@ mod kubernetes_tests { .await? .expect("inserted sandbox must be visible"); let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), mock_kubernetes_client(Arc::clone(&captured_pod_creates)), - ))); + )); let controller = SandboxController::with_cluster_admission(substrate, rejecting_admission(&pool)); @@ -6067,14 +6067,14 @@ mod kubernetes_tests { let request_count = Arc::new(AtomicUsize::new(0)); let request_started = Arc::new(Semaphore::new(0)); let release_request = Arc::new(Semaphore::new(0)); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), blocking_failed_pod_create_kubernetes_client( Arc::clone(&request_count), Arc::clone(&request_started), Arc::clone(&release_request), ), - ))); + )); let controller = SandboxController::with_cluster_admission(substrate, admission); // Act. @@ -6132,14 +6132,14 @@ mod kubernetes_tests { let request_count = Arc::new(AtomicUsize::new(0)); let request_started = Arc::new(Semaphore::new(0)); let release_request = Arc::new(Semaphore::new(0)); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), blocking_failed_pod_create_kubernetes_client( Arc::clone(&request_count), Arc::clone(&request_started), Arc::clone(&release_request), ), - ))); + )); let controller = SandboxController::with_cluster_admission(substrate, rejecting_admission(&pool)); @@ -6196,14 +6196,14 @@ mod kubernetes_tests { let request_count = Arc::new(AtomicUsize::new(0)); let request_started = Arc::new(Semaphore::new(0)); let release_request = Arc::new(Semaphore::new(0)); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), blocking_failed_pod_create_kubernetes_client( Arc::clone(&request_count), Arc::clone(&request_started), Arc::clone(&release_request), ), - ))); + )); let controller = SandboxController::with_cluster_admission(substrate, rejecting_admission(&pool)); @@ -6770,13 +6770,13 @@ mod kubernetes_tests { let sim_worker = start_sim_worker().await?; let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); - let controller = SandboxController::new(Substrate::Kubernetes(Box::new( + let controller = SandboxController::new(Substrate::from_kubernetes( substrate_with_store_client_and_proxy_port( pool.clone(), running_pod_kubernetes_client(Arc::clone(&captured_pod_creates), "127.0.0.1"), sim_worker.port(), ), - ))); + )); let sandbox = fetch_sandbox(&pool, sandbox_id) .await? .expect("inserted sandbox must be visible"); @@ -9412,9 +9412,9 @@ mod kubernetes_tests { 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( + Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool), - )), + ), crate::cluster_admission::ClusterSandboxAdmissionConfig { enabled: true, refill_rps: 6.0, @@ -9476,9 +9476,9 @@ mod kubernetes_tests { 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( + Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), - )), + ), crate::cluster_admission::ClusterSandboxAdmissionConfig { enabled: true, refill_rps: 6.0, @@ -9554,9 +9554,9 @@ mod kubernetes_tests { // Direct is only a placeholder capacity source: its empty nodes inventory reports // zero eligible nodes, as a synced, empty Kubernetes node cache would, so admission // rejects with EmptyCluster (no Redis needed). - Substrate::Direct(Box::new( + Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), - )), + ), crate::cluster_admission::ClusterSandboxAdmissionConfig { enabled: true, refill_rps: 6.0, @@ -9613,17 +9613,17 @@ mod kubernetes_tests { insert_test_sandbox_row(&pool, sandbox_id, "started").await?; let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), mock_kubernetes_client(Arc::clone(&captured_pod_creates)), - ))); + )); 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( + Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), - )), + ), crate::cluster_admission::ClusterSandboxAdmissionConfig { enabled: true, refill_rps: 6.0, @@ -9677,15 +9677,15 @@ mod kubernetes_tests { insert_test_sandbox_row(&pool, sandbox_id, "started").await?; let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), 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()), + Substrate::from_direct(crate::controllers::substrate::direct::DirectSubstrate::new( + pool.clone(), )), crate::cluster_admission::ClusterSandboxAdmissionConfig { enabled: true, diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/tests/substrate_scope.rs b/crates/libs/api/src/controllers/substrate/kubernetes/tests/substrate_scope.rs index e2af95edb..459ecb97d 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/tests/substrate_scope.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/tests/substrate_scope.rs @@ -34,15 +34,15 @@ async fn control_runtime_client_wakes_only_kubernetes_rows_after_an_unreachable_ .expect("sandbox row"); let captured_pod_creates = Arc::new(Mutex::new(Vec::new())); - let substrate = Substrate::Kubernetes(Box::new(substrate_with_store_and_client( + let substrate = Substrate::from_kubernetes(substrate_with_store_and_client( pool.clone(), 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()), + Substrate::from_direct(crate::controllers::substrate::direct::DirectSubstrate::new( + pool.clone(), )), crate::cluster_admission::ClusterSandboxAdmissionConfig { enabled: true, diff --git a/crates/libs/api/src/controllers/substrate/mod.rs b/crates/libs/api/src/controllers/substrate/mod.rs index c17221713..f8b672445 100644 --- a/crates/libs/api/src/controllers/substrate/mod.rs +++ b/crates/libs/api/src/controllers/substrate/mod.rs @@ -1065,6 +1065,13 @@ pub struct SubstrateCapabilities { /// and force-upload actions refuse before reading the node, except /// `:undrain`, which only clears a `drain_phase` an earlier drain left. pub supports_node_drain: bool, + /// The cluster controller may run as several replicas, so each one + /// contends for a leader `Lease` on the apiserver and only the holder runs + /// the reconcilers (unless the leader-election kill-switch is off). When + /// false, the controller is a single local process and runs the + /// reconcilers unconditionally. The lease client comes from the process + /// environment, not from the substrate. + pub contends_for_leader_lease: bool, } impl SubstrateCapabilities { @@ -1075,6 +1082,7 @@ impl SubstrateCapabilities { resolves_cold_create_shape: false, supports_node_pool_placement: false, supports_node_drain: false, + contends_for_leader_lease: false, }; pub const KUBERNETES: Self = Self { @@ -1084,6 +1092,7 @@ impl SubstrateCapabilities { resolves_cold_create_shape: true, supports_node_pool_placement: true, supports_node_drain: true, + contends_for_leader_lease: true, }; } @@ -1681,31 +1690,58 @@ where } } +/// The running substrate. Shared code asks it what it supports +/// ([`SubstrateAdapter::capabilities`]) and asks it to act (the +/// [`SubstrateAdapter`] and inherent methods) rather than branching on which +/// substrate it is; [`SubstrateAdapter::managed_by`] still names it, for SQL +/// row scope and wire labels. The variants live in the private `Inner`, so +/// only this module can match them; code outside it builds a `Substrate` with +/// `from_direct` or `from_kubernetes`. #[derive(Clone)] -pub enum Substrate { +pub struct Substrate { + inner: Inner, +} + +#[derive(Clone)] +enum Inner { Direct(Box), #[cfg(feature = "kubernetes")] Kubernetes(Box), } impl Substrate { + /// Run on the Direct substrate. + pub fn from_direct(substrate: direct::DirectSubstrate) -> Self { + Self { + inner: Inner::Direct(Box::new(substrate)), + } + } + + /// Run on the Kubernetes substrate. + #[cfg(feature = "kubernetes")] + pub fn from_kubernetes(substrate: kubernetes::KubernetesSubstrate) -> Self { + Self { + inner: Inner::Kubernetes(Box::new(substrate)), + } + } + /// The Kubernetes substrate, for code that needs Kubernetes-only state /// (its client, caches, or reconcilers typed on it). Behavior shared code /// asks a substrate to perform belongs on [`SubstrateAdapter`], and static /// facts on [`SubstrateCapabilities`]. #[cfg(feature = "kubernetes")] pub fn as_kubernetes(&self) -> Option<&kubernetes::KubernetesSubstrate> { - match self { - Self::Direct(_) => None, - Self::Kubernetes(substrate) => Some(substrate), + match &self.inner { + Inner::Direct(_) => None, + Inner::Kubernetes(substrate) => Some(substrate), } } pub fn live_node_names(&self) -> Option> { - match self { - Self::Direct(_) => None, + match &self.inner { + Inner::Direct(_) => None, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.live_node_names(), + Inner::Kubernetes(substrate) => substrate.live_node_names(), } } @@ -1713,10 +1749,10 @@ impl Substrate { &self, target: &NodeApiTarget<'_>, ) -> Result { - match self { - Self::Direct(substrate) => substrate.resolve_node_api_endpoint(target).await, + match &self.inner { + Inner::Direct(substrate) => substrate.resolve_node_api_endpoint(target).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.try_resolve_node_api_endpoint(target), + Inner::Kubernetes(substrate) => substrate.try_resolve_node_api_endpoint(target), } } @@ -1734,7 +1770,7 @@ impl Substrate { match kubernetes::KubernetesSubstrate::try_from_env(pool, shutdown_token, task_tracker) .await { - Ok(Some(substrate)) => return Ok(Self::Kubernetes(Box::new(substrate))), + Ok(Some(substrate)) => return Ok(Self::from_kubernetes(substrate)), Ok(None) => {} Err(err) => return Err(err), } @@ -1742,9 +1778,9 @@ impl Substrate { #[cfg(not(feature = "kubernetes"))] drop((shutdown_token, task_tracker)); - Ok(Self::Direct(Box::new(direct::DirectSubstrate::new( + Ok(Self::from_direct(direct::DirectSubstrate::new( pool.clone(), - )))) + ))) } pub(crate) async fn reserve_rollback_route_identity( @@ -1752,13 +1788,13 @@ impl Substrate { sandbox_id: SandboxId, ) -> Result, SubstrateError> { - match self { - Self::Direct(substrate) => substrate + match &self.inner { + Inner::Direct(substrate) => substrate .reserve_rollback_route_identity(sandbox_id) .await .map(Some), #[cfg(feature = "kubernetes")] - Self::Kubernetes(_) => Ok(None), + Inner::Kubernetes(_) => Ok(None), } } @@ -1767,14 +1803,14 @@ impl Substrate { sandbox_id: SandboxId, route_identity: Option, ) -> Result<(), SubstrateError> { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .persist_rollback_route_identity(sandbox_id, route_identity) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(_) => Ok(()), + Inner::Kubernetes(_) => Ok(()), } } @@ -1850,10 +1886,10 @@ impl Substrate { /// traffic before its cache is warm. The direct substrate has no such /// cache and is always ready. pub fn readiness(&self) -> ReflectorReady { - match self { - Self::Direct(_) => ReflectorReady::always_ready(), + match &self.inner { + Inner::Direct(_) => ReflectorReady::always_ready(), #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.readiness(), + Inner::Kubernetes(substrate) => substrate.readiness(), } } @@ -1872,10 +1908,10 @@ impl Substrate { /// the caller can fail open rather than treating an unknown as an empty /// cluster. `Some(0)` is a confident "no start-capable nodes". pub async fn eligible_start_node_count(&self) -> Result, SubstrateError> { - match self { - Self::Direct(substrate) => Ok(Some(substrate.eligible_start_node_count().await?)), + match &self.inner { + Inner::Direct(substrate) => Ok(Some(substrate.eligible_start_node_count().await?)), #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => Ok(substrate.eligible_start_node_count()), + Inner::Kubernetes(substrate) => Ok(substrate.eligible_start_node_count()), } } @@ -1897,13 +1933,13 @@ impl Substrate { _request: NodeSchedulabilityRequest<'_>, _desired: NodeSchedulability, ) -> Result<(), SubstrateError> { - match self { - Self::Direct(_) => Err(SubstrateError::Unsupported { + match &self.inner { + Inner::Direct(_) => Err(SubstrateError::Unsupported { message: "the active substrate does not support setting node schedulability" .to_string(), }), #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate.set_node_schedulability(_request, _desired).await } } @@ -2289,18 +2325,18 @@ fn backup_source_to_proto( impl SubstrateAdapter for Substrate { fn managed_by(&self) -> &'static str { - match self { - Self::Direct(substrate) => substrate.managed_by(), + match &self.inner { + Inner::Direct(substrate) => substrate.managed_by(), #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.managed_by(), + Inner::Kubernetes(substrate) => substrate.managed_by(), } } fn capabilities(&self) -> SubstrateCapabilities { - match self { - Self::Direct(substrate) => substrate.capabilities(), + match &self.inner { + Inner::Direct(substrate) => substrate.capabilities(), #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.capabilities(), + Inner::Kubernetes(substrate) => substrate.capabilities(), } } @@ -2308,10 +2344,10 @@ impl SubstrateAdapter for Substrate { &self, request: CreateExecutionRequest<'_>, ) -> Result { - match self { - Self::Direct(substrate) => substrate.create_execution(request).await, + match &self.inner { + Inner::Direct(substrate) => substrate.create_execution(request).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.create_execution(request).await, + Inner::Kubernetes(substrate) => substrate.create_execution(request).await, } } @@ -2320,14 +2356,14 @@ impl SubstrateAdapter for Substrate { sandbox_id: SandboxId, completion: Option, ) { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .complete_activation_lease(sandbox_id, completion) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .complete_activation_lease(sandbox_id, completion) .await @@ -2336,10 +2372,10 @@ impl SubstrateAdapter for Substrate { } async fn delete_execution(&self, sandbox_id: SandboxId) -> Result<(), SubstrateError> { - match self { - Self::Direct(substrate) => substrate.delete_execution(sandbox_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.delete_execution(sandbox_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.delete_execution(sandbox_id).await, + Inner::Kubernetes(substrate) => substrate.delete_execution(sandbox_id).await, } } @@ -2347,10 +2383,10 @@ impl SubstrateAdapter for Substrate { &self, request: RetireExecutionRequest, ) -> Result { - match self { - Self::Direct(substrate) => substrate.retire_execution(request).await, + match &self.inner { + Inner::Direct(substrate) => substrate.retire_execution(request).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.retire_execution(request).await, + Inner::Kubernetes(substrate) => substrate.retire_execution(request).await, } } @@ -2359,10 +2395,12 @@ impl SubstrateAdapter for Substrate { sandbox_id: SandboxId, fence: &RetireFence, ) -> Result, SubstrateError> { - match self { - Self::Direct(substrate) => substrate.retire_fence_mismatch(sandbox_id, fence).await, + match &self.inner { + Inner::Direct(substrate) => substrate.retire_fence_mismatch(sandbox_id, fence).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.retire_fence_mismatch(sandbox_id, fence).await, + Inner::Kubernetes(substrate) => { + substrate.retire_fence_mismatch(sandbox_id, fence).await + } } } @@ -2370,10 +2408,10 @@ impl SubstrateAdapter for Substrate { &self, pod_names: &[PodName], ) -> HashMap { - match self { - Self::Direct(substrate) => substrate.observe_pod_identities(pod_names).await, + match &self.inner { + Inner::Direct(substrate) => substrate.observe_pod_identities(pod_names).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.observe_pod_identities(pod_names).await, + Inner::Kubernetes(substrate) => substrate.observe_pod_identities(pod_names).await, } } @@ -2382,14 +2420,14 @@ impl SubstrateAdapter for Substrate { sandbox_id: SandboxId, target: &NodeApiTarget<'_>, ) -> Result<(), SubstrateError> { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .delete_observed_execution(sandbox_id, target) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .delete_observed_execution(sandbox_id, target) .await @@ -2404,14 +2442,14 @@ impl SubstrateAdapter for Substrate { options: ActivationOptions, charge: RecoveryCharge, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .start_execution(metric_identity, sandbox_id, options, charge) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .start_execution(metric_identity, sandbox_id, options, charge) .await @@ -2426,14 +2464,14 @@ impl SubstrateAdapter for Substrate { options: ActivationOptions, charge: RecoveryCharge, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .restore_execution(metric_identity, sandbox_id, options, charge) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .restore_execution(metric_identity, sandbox_id, options, charge) .await @@ -2446,14 +2484,14 @@ impl SubstrateAdapter for Substrate { sandbox_id: SandboxId, budget: Duration, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .ensure_routable_execution(sandbox_id, budget) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .ensure_routable_execution(sandbox_id, budget) .await @@ -2465,10 +2503,10 @@ impl SubstrateAdapter for Substrate { &self, sandbox_id: SandboxId, ) -> Result, SubstrateError> { - match self { - Self::Direct(substrate) => substrate.suspend_execution(sandbox_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.suspend_execution(sandbox_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.suspend_execution_tracked(sandbox_id).await, + Inner::Kubernetes(substrate) => substrate.suspend_execution_tracked(sandbox_id).await, } } @@ -2479,14 +2517,14 @@ impl SubstrateAdapter for Substrate { source: BackupSource, target: &NodeApiTarget<'_>, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .enqueue_artifact_backup(sandbox_id, expected_kind, source, target) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .enqueue_artifact_backup(sandbox_id, expected_kind, source, target) .await @@ -2503,8 +2541,8 @@ impl SubstrateAdapter for Substrate { source_storage_uri: String, destination_storage_uri: String, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { SubstrateAdapter::create_snapshot_from_durable_artifact( substrate.as_ref(), target, @@ -2517,7 +2555,7 @@ impl SubstrateAdapter for Substrate { .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { SubstrateAdapter::create_snapshot_from_durable_artifact( substrate.as_ref(), target, @@ -2540,14 +2578,14 @@ impl SubstrateAdapter for Substrate { wake_activity: Option, charge: RecoveryCharge, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .wake_execution(metric_identity, sandbox, timeout, wake_activity, charge) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .wake_execution(metric_identity, sandbox, timeout, wake_activity, charge) .await @@ -2560,10 +2598,10 @@ impl SubstrateAdapter for Substrate { sandbox_id: SandboxId, charge: RecoveryCharge, ) -> Result { - match self { - Self::Direct(substrate) => substrate.runtime_client(sandbox_id, charge).await, + match &self.inner { + Inner::Direct(substrate) => substrate.runtime_client(sandbox_id, charge).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.runtime_client(sandbox_id, charge).await, + Inner::Kubernetes(substrate) => substrate.runtime_client(sandbox_id, charge).await, } } @@ -2571,10 +2609,10 @@ impl SubstrateAdapter for Substrate { &self, sandbox_id: SandboxId, ) -> Result, ControllerError> { - match self { - Self::Direct(substrate) => substrate.reachable_runtime_client(sandbox_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.reachable_runtime_client(sandbox_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.reachable_runtime_client(sandbox_id).await, + Inner::Kubernetes(substrate) => substrate.reachable_runtime_client(sandbox_id).await, } } @@ -2582,10 +2620,10 @@ impl SubstrateAdapter for Substrate { &self, sandbox_id: SandboxId, ) -> Result, ControllerError> { - match self { - Self::Direct(substrate) => substrate.probe_runtime_client(sandbox_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.probe_runtime_client(sandbox_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.probe_runtime_client(sandbox_id).await, + Inner::Kubernetes(substrate) => substrate.probe_runtime_client(sandbox_id).await, } } @@ -2597,14 +2635,14 @@ impl SubstrateAdapter for Substrate { timeout: Duration, charge: RecoveryCharge, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .access_runtime_client(metric_identity, sandbox, activity, timeout, charge) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .access_runtime_client(metric_identity, sandbox, activity, timeout, charge) .await @@ -2613,10 +2651,10 @@ impl SubstrateAdapter for Substrate { } async fn sandbox_ip(&self, sandbox_id: SandboxId) -> Result, ControllerError> { - match self { - Self::Direct(substrate) => substrate.sandbox_ip(sandbox_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.sandbox_ip(sandbox_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.sandbox_ip(sandbox_id).await, + Inner::Kubernetes(substrate) => substrate.sandbox_ip(sandbox_id).await, } } @@ -2627,14 +2665,14 @@ impl SubstrateAdapter for Substrate { timeout: Duration, charge: RecoveryCharge, ) -> Result, ControllerError> { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .access_sandbox_ip(metric_identity, sandbox, timeout, charge) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .access_sandbox_ip(metric_identity, sandbox, timeout, charge) .await @@ -2643,10 +2681,10 @@ impl SubstrateAdapter for Substrate { } async fn builder_client(&self) -> Result { - match self { - Self::Direct(substrate) => substrate.builder_client().await, + match &self.inner { + Inner::Direct(substrate) => substrate.builder_client().await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.builder_client().await, + Inner::Kubernetes(substrate) => substrate.builder_client().await, } } @@ -2654,10 +2692,10 @@ impl SubstrateAdapter for Substrate { &self, node_id: NodeId, ) -> Result { - match self { - Self::Direct(substrate) => substrate.builder_client_for_node(node_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.builder_client_for_node(node_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.builder_client_for_node(node_id).await, + Inner::Kubernetes(substrate) => substrate.builder_client_for_node(node_id).await, } } @@ -2670,8 +2708,8 @@ impl SubstrateAdapter for Substrate { file_changes_enabled: bool, implicit_wake_enabled: bool, ) -> bool { - match self { - Self::Direct(substrate) => substrate.pool_eligible_for_request( + match &self.inner { + Inner::Direct(substrate) => substrate.pool_eligible_for_request( boot_mode, backend, wait_until_ready, @@ -2680,7 +2718,7 @@ impl SubstrateAdapter for Substrate { implicit_wake_enabled, ), #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.pool_eligible_for_request( + Inner::Kubernetes(substrate) => substrate.pool_eligible_for_request( boot_mode, backend, wait_until_ready, @@ -2692,10 +2730,10 @@ impl SubstrateAdapter for Substrate { } fn pool_supports_request_env(&self) -> bool { - match self { - Self::Direct(substrate) => substrate.pool_supports_request_env(), + match &self.inner { + Inner::Direct(substrate) => substrate.pool_supports_request_env(), #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.pool_supports_request_env(), + Inner::Kubernetes(substrate) => substrate.pool_supports_request_env(), } } @@ -2703,10 +2741,10 @@ impl SubstrateAdapter for Substrate { &self, request: &PooledExecutionClaimRequest<'_>, ) -> Result { - match self { - Self::Direct(substrate) => substrate.claim_pooled_execution(request).await, + match &self.inner { + Inner::Direct(substrate) => substrate.claim_pooled_execution(request).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.claim_pooled_execution(request).await, + Inner::Kubernetes(substrate) => substrate.claim_pooled_execution(request).await, } } @@ -2716,14 +2754,14 @@ impl SubstrateAdapter for Substrate { admitted: AdmittedPoolClaim, sandbox_id: SandboxId, ) -> Result { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .lookup_admitted_pooled_execution(request, admitted, sandbox_id) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .lookup_admitted_pooled_execution(request, admitted, sandbox_id) .await @@ -2735,10 +2773,10 @@ impl SubstrateAdapter for Substrate { &self, sandbox_id: SandboxId, ) -> Result<(), ControllerError> { - match self { - Self::Direct(substrate) => substrate.skip_admitted_pool_generation(sandbox_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.skip_admitted_pool_generation(sandbox_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate.skip_admitted_pool_generation(sandbox_id).await } } @@ -2748,10 +2786,10 @@ impl SubstrateAdapter for Substrate { &self, sandbox_id: SandboxId, ) -> Result { - match self { - Self::Direct(substrate) => substrate.admitted_pool_cleanup_pending(sandbox_id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.admitted_pool_cleanup_pending(sandbox_id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate.admitted_pool_cleanup_pending(sandbox_id).await } } @@ -2762,14 +2800,14 @@ impl SubstrateAdapter for Substrate { tx: &mut Transaction<'_, Postgres>, template_release_id: Uuid, ) -> Result, ControllerError> { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .claim_admission_pool_slot(tx, template_release_id) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .claim_admission_pool_slot(tx, template_release_id) .await @@ -2783,14 +2821,14 @@ impl SubstrateAdapter for Substrate { request: &PooledExecutionClaimRequest<'_>, claim: &PooledExecutionClaim, ) -> Result<(), ControllerError> { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .persist_pooled_execution_claim(tx, request, claim) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .persist_pooled_execution_claim(tx, request, claim) .await @@ -2802,10 +2840,10 @@ impl SubstrateAdapter for Substrate { &self, claim: &PooledExecutionClaim, ) -> Result<(), ControllerError> { - match self { - Self::Direct(substrate) => substrate.rollback_pooled_execution_claim(claim).await, + match &self.inner { + Inner::Direct(substrate) => substrate.rollback_pooled_execution_claim(claim).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.rollback_pooled_execution_claim(claim).await, + Inner::Kubernetes(substrate) => substrate.rollback_pooled_execution_claim(claim).await, } } @@ -2813,10 +2851,10 @@ impl SubstrateAdapter for Substrate { &self, claim: &PooledExecutionClaim, ) -> Result { - match self { - Self::Direct(substrate) => substrate.cleanup_failed_pooled_prepare(claim).await, + match &self.inner { + Inner::Direct(substrate) => substrate.cleanup_failed_pooled_prepare(claim).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.cleanup_failed_pooled_prepare(claim).await, + Inner::Kubernetes(substrate) => substrate.cleanup_failed_pooled_prepare(claim).await, } } @@ -2825,14 +2863,14 @@ impl SubstrateAdapter for Substrate { claim: &PooledExecutionClaim, policy: &AutoLifecyclePolicy, ) -> Result<(), ControllerError> { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .apply_pooled_execution_auto_lifecycle(claim, policy) .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .apply_pooled_execution_auto_lifecycle(claim, policy) .await @@ -2849,8 +2887,8 @@ impl SubstrateAdapter for Substrate { wait_until_ready: bool, timeout: Duration, ) -> Result<(), ControllerError> { - match self { - Self::Direct(substrate) => { + match &self.inner { + Inner::Direct(substrate) => { substrate .prepare_pooled_execution( metric_identity, @@ -2863,7 +2901,7 @@ impl SubstrateAdapter for Substrate { .await } #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => { + Inner::Kubernetes(substrate) => { substrate .prepare_pooled_execution( metric_identity, @@ -2879,10 +2917,10 @@ impl SubstrateAdapter for Substrate { } async fn list_pool_configs(&self) -> Result, ControllerError> { - match self { - Self::Direct(substrate) => substrate.list_pool_configs().await, + match &self.inner { + Inner::Direct(substrate) => substrate.list_pool_configs().await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.list_pool_configs().await, + Inner::Kubernetes(substrate) => substrate.list_pool_configs().await, } } @@ -2890,10 +2928,10 @@ impl SubstrateAdapter for Substrate { &self, request: ClusterPoolConfigRequest, ) -> Result { - match self { - Self::Direct(substrate) => substrate.upsert_pool_config(request).await, + match &self.inner { + Inner::Direct(substrate) => substrate.upsert_pool_config(request).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.upsert_pool_config(request).await, + Inner::Kubernetes(substrate) => substrate.upsert_pool_config(request).await, } } @@ -2901,10 +2939,10 @@ impl SubstrateAdapter for Substrate { &self, id: Uuid, ) -> Result { - match self { - Self::Direct(substrate) => substrate.disable_pool_config(id).await, + match &self.inner { + Inner::Direct(substrate) => substrate.disable_pool_config(id).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.disable_pool_config(id).await, + Inner::Kubernetes(substrate) => substrate.disable_pool_config(id).await, } } @@ -2912,20 +2950,20 @@ impl SubstrateAdapter for Substrate { &self, target: &NodeApiTarget<'_>, ) -> Result { - match self { - Self::Direct(substrate) => substrate.resolve_node_api_endpoint(target).await, + match &self.inner { + Inner::Direct(substrate) => substrate.resolve_node_api_endpoint(target).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.resolve_node_api_endpoint(target).await, + Inner::Kubernetes(substrate) => substrate.resolve_node_api_endpoint(target).await, } } async fn select_ready_worker_node_api_target( &self, ) -> Result { - match self { - Self::Direct(substrate) => substrate.select_ready_worker_node_api_target().await, + match &self.inner { + Inner::Direct(substrate) => substrate.select_ready_worker_node_api_target().await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.select_ready_worker_node_api_target().await, + Inner::Kubernetes(substrate) => substrate.select_ready_worker_node_api_target().await, } } @@ -2933,10 +2971,10 @@ impl SubstrateAdapter for Substrate { &self, role_filter: Option<&str>, ) -> Result>, SubstrateError> { - match self { - Self::Direct(substrate) => substrate.list_inventory_nodes(role_filter).await, + match &self.inner { + Inner::Direct(substrate) => substrate.list_inventory_nodes(role_filter).await, #[cfg(feature = "kubernetes")] - Self::Kubernetes(substrate) => substrate.list_inventory_nodes(role_filter).await, + Inner::Kubernetes(substrate) => substrate.list_inventory_nodes(role_filter).await, } } } diff --git a/crates/libs/api/src/controllers/substrate/runtime_rpc/tests.rs b/crates/libs/api/src/controllers/substrate/runtime_rpc/tests.rs index 8c81f281e..63c9f8c03 100644 --- a/crates/libs/api/src/controllers/substrate/runtime_rpc/tests.rs +++ b/crates/libs/api/src/controllers/substrate/runtime_rpc/tests.rs @@ -11,7 +11,7 @@ async fn fixture() -> (RuntimeClient, Substrate) { pool.close().await; ( RuntimeClient::new(channel), - Substrate::Direct(Box::new(DirectSubstrate::new(pool))), + Substrate::from_direct(DirectSubstrate::new(pool)), ) } diff --git a/crates/libs/api/src/controllers/substrate/tests.rs b/crates/libs/api/src/controllers/substrate/tests.rs index 56ec64b79..4e39c7dff 100644 --- a/crates/libs/api/src/controllers/substrate/tests.rs +++ b/crates/libs/api/src/controllers/substrate/tests.rs @@ -668,13 +668,13 @@ async fn node_drain_capability_matches_set_node_schedulability() { .connect_lazy("postgres://postgres@127.0.0.1:1/postgres") .expect("lazy pool should parse"); #[cfg_attr(not(feature = "kubernetes"), allow(unused_mut))] - let mut substrates = vec![Substrate::Direct(Box::new(direct::DirectSubstrate::new( + let mut substrates = vec![Substrate::from_direct(direct::DirectSubstrate::new( pool.clone(), - )))]; + ))]; #[cfg(feature = "kubernetes")] - substrates.push(Substrate::Kubernetes(Box::new( + substrates.push(Substrate::from_kubernetes( kubernetes::tests::substrate_with_store(pool), - ))); + )); for substrate in substrates { let result = substrate @@ -711,7 +711,7 @@ async fn node_addressing_names_the_field_endpoint_resolution_reads() { let k8s_node_name = "node-a"; #[cfg_attr(not(feature = "kubernetes"), allow(unused_mut))] let mut cases = vec![( - Substrate::Direct(Box::new(direct::DirectSubstrate::new(pool.clone()))), + Substrate::from_direct(direct::DirectSubstrate::new(pool.clone())), NodeManagedBy::Space, stored_endpoint.to_string(), )]; @@ -731,7 +731,7 @@ async fn node_addressing_names_the_field_endpoint_resolution_reads() { ); let pod_endpoint = format!("192.0.2.20:{}", substrate.k8s.node_api_port); cases.push(( - Substrate::Kubernetes(Box::new(substrate)), + Substrate::from_kubernetes(substrate), NodeManagedBy::Kubernetes, pod_endpoint, )); diff --git a/crates/libs/api/src/controllers/template/tests.rs b/crates/libs/api/src/controllers/template/tests.rs index bc64954aa..7dc4992b9 100644 --- a/crates/libs/api/src/controllers/template/tests.rs +++ b/crates/libs/api/src/controllers/template/tests.rs @@ -188,9 +188,9 @@ async fn build_submission_grants_the_base_before_dispatch( #[future] base_release_fixture: TestDbResult, ) -> TestDbResult<()> { let BaseReleaseFixture { db, owner, .. } = base_release_fixture.await?; - let substrate = Substrate::Direct(Box::new( + let substrate = Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )); + ); let controller = TemplateController::new(substrate, db.pool.clone()); let submission = base_submission("private-base:stable".into(), OrgId::from_uuid(owner)); @@ -396,9 +396,9 @@ async fn template_controller_fixture() -> TestDbResult TestDbResult<()> { let db = pg_test_db().await?; - let substrate = Substrate::Direct(Box::new( + let substrate = Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )); + ); let controller = TemplateController::new(substrate, db.pool.clone()); let insert_stale_grant = || { sqlx::query(sql!( diff --git a/crates/libs/api/src/public_sandbox_operations/tests.rs b/crates/libs/api/src/public_sandbox_operations/tests.rs index b57cb8183..55a9b1286 100644 --- a/crates/libs/api/src/public_sandbox_operations/tests.rs +++ b/crates/libs/api/src/public_sandbox_operations/tests.rs @@ -66,9 +66,9 @@ fn claimed_create_operation_for( // trait default `false` while Kubernetes reads the reserved slot, so a slot whose // cleanup is pending yields workflow progress instead of `Succeeded`. fn direct_sandbox_controller(pool: &PgPool) -> SandboxController { - SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( + SandboxController::new(crate::controllers::substrate::Substrate::from_direct( crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()), - ))) + )) } #[cfg(feature = "kubernetes")] @@ -151,14 +151,14 @@ async fn native_create_executor_preserves_operator_policy_across_replacements() (true, true, false), // Legacy federated payload: no saved selection. (true, true, true), ] { - let substrate = crate::controllers::substrate::Substrate::Kubernetes(Box::new( + let substrate = crate::controllers::substrate::Substrate::from_kubernetes( substrate_with_store_client_proxy_port_and_pod( db.pool.clone(), running_pod_kubernetes_client(Arc::clone(&captured_pods), "127.0.0.1"), pod_listener.local_addr()?.port(), None, ), - )); + ); let controller = SandboxController::new(substrate) .with_egress_profiles(profiles.clone()) .with_tls_interception_exemptions(configured.then(|| exemptions.clone())); @@ -761,10 +761,9 @@ async fn create_executor_retries_unpinned_transient_substrate_failure() { claim_generation: 1, traceparent: None, }; - let controller = - SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new( - crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - ))); + let controller = SandboxController::new(crate::controllers::substrate::Substrate::from_direct( + crate::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), + )); let executor = CreateSandboxOperationExecutor::new(db.pool.clone(), controller); let result = executor.execute(&operation).await; diff --git a/crates/libs/api/src/routes/admin/node_maintenance/tests.rs b/crates/libs/api/src/routes/admin/node_maintenance/tests.rs index 3b6a3ee61..fdf2e4510 100644 --- a/crates/libs/api/src/routes/admin/node_maintenance/tests.rs +++ b/crates/libs/api/src/routes/admin/node_maintenance/tests.rs @@ -165,11 +165,11 @@ async fn node_registry_delete_follows_the_running_substrate() -> TestResult { { let mut kubernetes = AppState::from_pool(db.pool.clone()).await?; kubernetes.sandbox_controller = crate::controllers::sandbox::SandboxController::new( - crate::controllers::substrate::Substrate::Kubernetes(Box::new( + crate::controllers::substrate::Substrate::from_kubernetes( crate::controllers::substrate::kubernetes::tests::substrate_with_store( db.pool.clone(), ), - )), + ), ); cases.push(( kubernetes, diff --git a/crates/libs/api/src/routes/admin/nodes/drain_handler_tests.rs b/crates/libs/api/src/routes/admin/nodes/drain_handler_tests.rs index 0611d8b14..c652289b8 100644 --- a/crates/libs/api/src/routes/admin/nodes/drain_handler_tests.rs +++ b/crates/libs/api/src/routes/admin/nodes/drain_handler_tests.rs @@ -67,9 +67,9 @@ async fn kubernetes_state(pool: &PgPool) -> TestResult { ) .build(); let mut state = AppState::from_pool(pool.clone()).await?; - state.sandbox_controller = SandboxController::new(Substrate::Kubernetes(Box::new( + state.sandbox_controller = SandboxController::new(Substrate::from_kubernetes( substrate_with_store_and_client(pool.clone(), client), - ))); + )); Ok(state) } diff --git a/crates/libs/api/src/routes/admin/nodes/tests.rs b/crates/libs/api/src/routes/admin/nodes/tests.rs index 5913f3571..e9f03d88d 100644 --- a/crates/libs/api/src/routes/admin/nodes/tests.rs +++ b/crates/libs/api/src/routes/admin/nodes/tests.rs @@ -2076,12 +2076,12 @@ async fn kubernetes_node_drain_and_cordon_converge_the_node() -> TestResult { .build(); let mut state = AppState::from_pool(db.pool.clone()).await?; state.sandbox_controller = crate::controllers::sandbox::SandboxController::new( - crate::controllers::substrate::Substrate::Kubernetes(Box::new( + crate::controllers::substrate::Substrate::from_kubernetes( crate::controllers::substrate::kubernetes::tests::substrate_with_store_and_client( db.pool.clone(), client, ), - )), + ), ); let actor = inventory_test_actor(crate::routes::authz::cluster_managing_org_id(&db.pool).await?); diff --git a/crates/libs/api/src/routes/admin/sandboxes/tests.rs b/crates/libs/api/src/routes/admin/sandboxes/tests.rs index 01a154611..798aa4135 100644 --- a/crates/libs/api/src/routes/admin/sandboxes/tests.rs +++ b/crates/libs/api/src/routes/admin/sandboxes/tests.rs @@ -575,11 +575,11 @@ async fn admin_pinned_create_follows_the_running_substrate() -> TestResult { { let mut kubernetes = AppState::from_pool(db.pool.clone()).await?; kubernetes.sandbox_controller = crate::controllers::sandbox::SandboxController::new( - crate::controllers::substrate::Substrate::Kubernetes(Box::new( + crate::controllers::substrate::Substrate::from_kubernetes( crate::controllers::substrate::kubernetes::tests::substrate_with_store( db.pool.clone(), ), - )), + ), ); cases.push((kubernetes, StatusCode::ACCEPTED)); } diff --git a/crates/libs/api/src/routes/e2b/tests.rs b/crates/libs/api/src/routes/e2b/tests.rs index e11822e1b..af1db2c40 100644 --- a/crates/libs/api/src/routes/e2b/tests.rs +++ b/crates/libs/api/src/routes/e2b/tests.rs @@ -341,7 +341,7 @@ async fn build_kubernetes_app_state_fixture( worker.port(), initial_pod, ); - let substrate = crate::controllers::substrate::Substrate::Kubernetes(Box::new(kubernetes)); + let substrate = crate::controllers::substrate::Substrate::from_kubernetes(kubernetes); let mut state = AppState::from_pool(pool.clone()).await?; configure_test_ecw_credential_keyring(&mut state); state.sandbox_controller = @@ -385,7 +385,7 @@ async fn missing_recovery_app_state_fixture() -> TestResult ); let mut state = fixture.state.clone(); state.sandbox_controller = crate::controllers::sandbox::SandboxController::new( - crate::controllers::substrate::Substrate::Kubernetes(Box::new(kubernetes)), + crate::controllers::substrate::Substrate::from_kubernetes(kubernetes), ); let org = fetch_local_org(&fixture.pool).await?; let app = e2b_api_routes() diff --git a/crates/libs/api/src/routes/public/sandboxes/admission_test_fixture.rs b/crates/libs/api/src/routes/public/sandboxes/admission_test_fixture.rs index 1995bea90..7d032c007 100644 --- a/crates/libs/api/src/routes/public/sandboxes/admission_test_fixture.rs +++ b/crates/libs/api/src/routes/public/sandboxes/admission_test_fixture.rs @@ -163,7 +163,7 @@ impl AdmissionPoolFixture { } pub(crate) fn controller_with(substrate: KubernetesSubstrate) -> SandboxController { - SandboxController::new(Substrate::Kubernetes(Box::new(substrate))) + SandboxController::new(Substrate::from_kubernetes(substrate)) } /// A default-shape native create of the fixture template. diff --git a/crates/libs/api/src/test_support.rs b/crates/libs/api/src/test_support.rs index 3b57d6351..661fee163 100644 --- a/crates/libs/api/src/test_support.rs +++ b/crates/libs/api/src/test_support.rs @@ -59,7 +59,7 @@ fn enter_sim_worker_process_with_token_paths( /// A Direct substrate over the test database; it needs no apiserver. pub(crate) fn direct_substrate(pool: &PgPool) -> Substrate { - Substrate::Direct(Box::new(DirectSubstrate::new(pool.clone()))) + Substrate::from_direct(DirectSubstrate::new(pool.clone())) } /// An enabled cluster-admission config with no Redis URL, so the admission diff --git a/crates/libs/cluster/controller/server/src/leader_election/mod.rs b/crates/libs/cluster/controller/server/src/leader_election/mod.rs index cc9312cd9..db37b2166 100644 --- a/crates/libs/cluster/controller/server/src/leader_election/mod.rs +++ b/crates/libs/cluster/controller/server/src/leader_election/mod.rs @@ -172,6 +172,7 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; +use pplx_space_api::controllers::substrate::SubstrateCapabilities; use pplx_space_common_core::types::PositiveDuration; use tokio::sync::watch; use tokio::time::{Instant, MissedTickBehavior, interval, sleep_until, timeout}; @@ -431,6 +432,14 @@ pub struct LeaderElectionConfig { } impl LeaderElectionConfig { + /// Whether this replica contends for the leader lease before running the + /// reconcilers: only on a substrate whose + /// [`contends_for_leader_lease`](SubstrateCapabilities::contends_for_leader_lease) + /// is set, and only while the kill-switch leaves election enabled. + pub fn contends(&self, capabilities: SubstrateCapabilities) -> bool { + self.enabled && capabilities.contends_for_leader_lease + } + pub fn from_env() -> Self { let identity = election_identity(); let timings = Timings::derive( diff --git a/crates/libs/cluster/controller/server/src/leader_election/tests/mod.rs b/crates/libs/cluster/controller/server/src/leader_election/tests/mod.rs index 63de326e7..95c66c281 100644 --- a/crates/libs/cluster/controller/server/src/leader_election/tests/mod.rs +++ b/crates/libs/cluster/controller/server/src/leader_election/tests/mod.rs @@ -17,7 +17,38 @@ use fixtures::{ expect_lease_unreachable, marker_path, test_config, wait_for_marker, }; -use super::Campaign; +use super::{Campaign, LeaderElectionConfig}; + +/// A replica contends for the lease only on a substrate that runs the +/// controller as several replicas, and the kill-switch turns contention off +/// everywhere. Direct is a single local process and runs the reconcilers +/// unconditionally. +#[test] +fn contention_follows_the_substrate_and_the_kill_switch() { + use pplx_space_api::controllers::substrate::SubstrateCapabilities; + + for (substrate, capabilities, enabled, expected) in [ + ("direct", SubstrateCapabilities::DIRECT, true, false), + ("direct", SubstrateCapabilities::DIRECT, false, false), + ("kubernetes", SubstrateCapabilities::KUBERNETES, true, true), + ( + "kubernetes", + SubstrateCapabilities::KUBERNETES, + false, + false, + ), + ] { + let config = LeaderElectionConfig { + enabled, + ..test_config("replica-a") + }; + assert_eq!( + config.contends(capabilities), + expected, + "{substrate} substrate, election enabled: {enabled}" + ); + } +} /// A free lease is won immediately; the leadership token stays live while held, /// and `resign` releases the lease. diff --git a/crates/libs/cluster/controller/server/src/lib.rs b/crates/libs/cluster/controller/server/src/lib.rs index 6cfbd0e73..0d9ad3e91 100644 --- a/crates/libs/cluster/controller/server/src/lib.rs +++ b/crates/libs/cluster/controller/server/src/lib.rs @@ -6,7 +6,7 @@ //! a deployment that needs private wiring can supply its own `main` against //! this library without copying any of the service. -use pplx_space_api::controllers::substrate::Substrate; +use pplx_space_api::controllers::substrate::{Substrate, SubstrateAdapter}; use pplx_space_api::egress_profiles::{EgressProfiles, EgressProfilesConfigError}; use pplx_space_api::tls_interception_exemptions::{ TlsInterceptionExemptions, TlsInterceptionExemptionsConfigError, @@ -133,76 +133,72 @@ pub async fn run( .await .map_err(RunClusterControllerError::Substrate)?; - // Contend for cluster-controller leadership — but only on the Kubernetes - // substrate (an apiserver to host the `Lease`, and a multi-replica - // Deployment). Only the lease holder runs the reconciler set: two concurrent - // runners would double-provision the warm pool and double-drive every CAS - // loop. The Direct substrate is a single local process, so it runs the - // reconcilers unconditionally; likewise the kill-switch (which then assumes - // a single replica). + // Contend for cluster-controller leadership — but only on a substrate that + // `contends_for_leader_lease` (Kubernetes: an apiserver to host the + // `Lease`, and a multi-replica Deployment). Only the lease holder runs the + // reconciler set: two concurrent runners would double-provision the warm + // pool and double-drive every CAS loop. The Direct substrate is a single + // local process, so it runs the reconcilers unconditionally; likewise the + // kill-switch (which then assumes a single replica). let le_config = leader_election::LeaderElectionConfig::from_env(); - // Bound out here, not inside the match arm that creates it: the readiness - // marker's writer task lives only as long as a handle does, and an arm-scoped - // binding would stop it the moment the arm ended. Underscore-prefixed rather - // than `let _ =`, which would drop it immediately. + // Bound out here, not inside the branch that creates it: the readiness + // marker's writer task lives only as long as a handle does, and a + // branch-scoped binding would stop it the moment the branch ended. + // Underscore-prefixed rather than `let _ =`, which would drop it + // immediately. let mut _readiness = None; let authority: leader_election::ReconcilerAuthority = - match substrate.as_kubernetes() { - Some(_) if le_config.enabled => { - // Renewals and standby polls are the idle steady state, so the - // lease gets a client that records spans only for failures. - // The renewal in flight at shutdown is abandoned, not failed. - let lease_client = - pplx_space_common_svc::kube_client::failure_traced_kube_client_from_env( - shutdown_token.clone(), - ) - .await - .map_err(RunClusterControllerError::LeaseClient)?; - let store = std::sync::Arc::new(leader_election::KubeLeaseStore::new( - lease_client, - &config.lease_namespace, - )); - tracing::info!( - identity = %le_config.identity, - liveness_budget_secs = le_config - .lease_failure_grace - .map(|g| g.get().as_secs_f64()), - "{SERVICE_NAME} contending for leadership" - ); - // `resolve` owns the outcome-to-action mapping so it can be - // tested; the arms below only carry it out. The liveness exit - // must be non-zero: retrying in-process won't recover a wedged - // client, and a NotReady-forever standby that nothing restarts - // is the gap this closes. The error log that precedes it already - // carries the identity, the measured window and the budget, so - // this only has to be the exit. - match leader_election::campaign(store, le_config, shutdown_token.clone()) - .await - .resolve() - { - leader_election::CampaignAction::ExitZero => { - tracing::info!( - "{SERVICE_NAME} asked to shut down before winning leadership" - ); - return Ok(()); - } - leader_election::CampaignAction::ExitNonZero => { - return Err(RunClusterControllerError::LeaseUnreachable); - } - leader_election::CampaignAction::Lead(leadership) => { - tracing::info!("{SERVICE_NAME} elected leader; starting reconcilers"); - leader_election::ReconcilerAuthority::Leader(leadership) - } + if le_config.contends(substrate.capabilities()) { + // Renewals and standby polls are the idle steady state, so the + // lease gets a client that records spans only for failures. + // The renewal in flight at shutdown is abandoned, not failed. + let lease_client = + pplx_space_common_svc::kube_client::failure_traced_kube_client_from_env( + shutdown_token.clone(), + ) + .await + .map_err(RunClusterControllerError::LeaseClient)?; + let store = std::sync::Arc::new(leader_election::KubeLeaseStore::new( + lease_client, + &config.lease_namespace, + )); + tracing::info!( + identity = %le_config.identity, + liveness_budget_secs = le_config + .lease_failure_grace + .map(|g| g.get().as_secs_f64()), + "{SERVICE_NAME} contending for leadership" + ); + // `resolve` owns the outcome-to-action mapping so it can be + // tested; the arms below only carry it out. The liveness exit + // must be non-zero: retrying in-process won't recover a wedged + // client, and a NotReady-forever standby that nothing restarts + // is the gap this closes. The error log that precedes it already + // carries the identity, the measured window and the budget, so + // this only has to be the exit. + match leader_election::campaign(store, le_config, shutdown_token.clone()) + .await + .resolve() + { + leader_election::CampaignAction::ExitZero => { + tracing::info!("{SERVICE_NAME} asked to shut down before winning leadership"); + return Ok(()); + } + leader_election::CampaignAction::ExitNonZero => { + return Err(RunClusterControllerError::LeaseUnreachable); + } + leader_election::CampaignAction::Lead(leadership) => { + tracing::info!("{SERVICE_NAME} elected leader; starting reconcilers"); + leader_election::ReconcilerAuthority::Leader(leadership) } } - _ => { - tracing::info!( - "{SERVICE_NAME} running reconcilers without leader election \ + } else { + tracing::info!( + "{SERVICE_NAME} running reconcilers without leader election \ (direct substrate or kill-switch); this assumes a single replica" - ); - _readiness = Some(leader_election::signal_ready(&le_config)); - leader_election::ReconcilerAuthority::ElectionDisabled - } + ); + _readiness = Some(leader_election::signal_ready(&le_config)); + leader_election::ReconcilerAuthority::ElectionDisabled }; // Drive the reconcilers off the leadership token (cancelled on process 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 f57a78edd..b27266a2d 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 @@ -694,9 +694,9 @@ async fn native_executor_requeues_create_when_cluster_admission_rejects() -> Tes }; // 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( + let substrate = Substrate::from_direct( pplx_space_api::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )); + ); let executor = NativePublicOperationExecutor::with_cluster_admission( db.pool.clone(), substrate.clone(), @@ -807,9 +807,9 @@ async fn native_executor_persists_operator_policy_on_update_and_template_reset() .execute(&db.pool) .await?; // Direct is only a placeholder here: this test never reaches the substrate. - let substrate = Substrate::Direct(Box::new( + let substrate = Substrate::from_direct( pplx_space_api::controllers::substrate::direct::DirectSubstrate::new(db.pool.clone()), - )); + ); let executor = NativePublicOperationExecutor::with_cluster_admission( db.pool.clone(), substrate.clone(), diff --git a/crates/libs/cluster/controller/server/src/reconcilers.rs b/crates/libs/cluster/controller/server/src/reconcilers.rs index 05b0dc282..2de2d8764 100644 --- a/crates/libs/cluster/controller/server/src/reconcilers.rs +++ b/crates/libs/cluster/controller/server/src/reconcilers.rs @@ -377,7 +377,7 @@ pub async fn run_reconcilers( // already marked failed but still carrying active execution rows. Runs for // both substrates because a stuck row outside any drain context still // wedges future operations on the row, and the sink only needs DB access + - // `SubstrateAdapter` methods, which the `Substrate` enum exposes + // `SubstrateAdapter` methods, which `Substrate` exposes // uniformly. Transient lifecycle aging remains default-disabled, // while failed-sandbox cleanup runs unless // `SPACE_FAILED_SANDBOX_CLEANUP_ENABLED=false`. 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 b69c3af5e..04577ba54 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 @@ -52,9 +52,7 @@ fn age() -> PositiveDuration { fn direct_sink(pool: &PgPool) -> SubstrateSweepSink { SubstrateSweepSink::new( pool.clone(), - Arc::new(Substrate::Direct(Box::new(DirectSubstrate::new( - pool.clone(), - )))), + Arc::new(Substrate::from_direct(DirectSubstrate::new(pool.clone()))), ) } diff --git a/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/store.rs b/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/store.rs index 27dfeae01..149aca7ab 100644 --- a/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/store.rs +++ b/crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/store.rs @@ -1,4 +1,4 @@ -//! Production [`SweepSink`] backed by Postgres and the `Substrate` enum. +//! Production [`SweepSink`] backed by Postgres and the substrate. use pplx_space_common_core::sql; use std::collections::HashMap; 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 6ed3a0989..b0d6d421c 100644 --- a/crates/libs/cluster/controller/server/src/warm_pool/tests.rs +++ b/crates/libs/cluster/controller/server/src/warm_pool/tests.rs @@ -156,7 +156,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)))), + substrate: Arc::new(Substrate::from_direct(DirectSubstrate::new(postgres))), namespace: "space".to_string(), pod_config: pool_pod_config(), warming_slot_timeout: DEFAULT_WARMING_SLOT_TIMEOUT, @@ -208,7 +208,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)))), + substrate: Arc::new(Substrate::from_direct(DirectSubstrate::new(postgres))), namespace: "space".to_string(), pod_config: pool_pod_config(), warming_slot_timeout: DEFAULT_WARMING_SLOT_TIMEOUT, @@ -884,9 +884,9 @@ 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( + substrate: Arc::new(Substrate::from_direct(DirectSubstrate::new( db.pool.clone(), - )))), + ))), namespace: "space".to_string(), pod_config: pool_pod_config(), warming_slot_timeout: DEFAULT_WARMING_SLOT_TIMEOUT, @@ -978,9 +978,9 @@ 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( + substrate: Arc::new(Substrate::from_direct(DirectSubstrate::new( db.pool.clone(), - )))), + ))), namespace: "space".to_string(), pod_config: pool_pod_config(), warming_slot_timeout: DEFAULT_WARMING_SLOT_TIMEOUT, @@ -1337,9 +1337,9 @@ 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( + substrate: Arc::new(Substrate::from_direct(DirectSubstrate::new( db.pool.clone(), - )))), + ))), namespace: "space".to_string(), pod_config: pool_pod_config(), warming_slot_timeout: DEFAULT_WARMING_SLOT_TIMEOUT,