fix(supervisor): fence refused launches and remove legacy mode

Signed-off-by: Drew Newberry <anewberry@nvidia.com>

Author: drew

.agents/skills/helm-dev-environment/SKILL.md

@@ -101,7 +101,7 @@ OPENSHELL_E2E_KUBE_IMAGE_PULL_SECRET=e2e-regcred \
 
 The shared supervisor is always a pool of `supervisor.replicas` members
 (default 1). With the gateway topology the chart renders the gateway's
-`[openshell.gateway.supervisor_pool]` table (`enabled = true`, plus
+`[openshell.gateway.supervisor_pool]` tuning table (only
 `recovery_timeout_secs` from `supervisor.recoveryTimeoutSeconds`, default 60;
 the driver's `shared_supervisor` table stays unchanged), a
 RollingUpdate Deployment (`maxSurge: 1`, `maxUnavailable: 0`) with preferred

crates/openshell-server/README.md

@@ -20,10 +20,11 @@ authorize the new one.
 
 ## Shared supervisor pool
 
-With `[openshell.gateway.supervisor_pool] enabled = true`, the `SharedSupervision`
-the gateway hands its compute driver coordinates a pool of shared supervisor
-hosts; the driver itself is unchanged. The gateway refuses a pool when no
-in-process driver uses the shared supervisor. `supervisor_pool` keeps the
+The `SharedSupervision` the gateway hands a compute driver that asks for it
+always coordinates a pool of shared supervisor hosts, at any member count; the
+driver itself is unchanged. `[openshell.gateway.supervisor_pool]` only tunes it
+(`recovery_timeout_secs`), and the gateway refuses the table when no in-process
+driver uses the shared supervisor. `supervisor_pool` keeps the
 pool's durable state and reconciles it every two seconds on every gateway
 replica, independent of which replica runs the compute driver's watch:
 
@@ -52,18 +53,29 @@ replica, independent of which replica runs the compute driver's watch:
   atomic store writes and holds no lock: it reserves the route with a row named
   by the route and identified by a fresh writer token, stores the material, has
   a `supervisor_instance_material_candidate` row own the handle, and publishes
-  by advancing exactly its reservation's version. Collection deletes a
-  candidate's material only once its recording no longer holds the reservation,
-  after which that recording can never publish; a reservation that does not
-  publish within two minutes is fenced by removing it. A withdrawal removes the
-  material and then the record, or fences a reservation; an ended generation
+  by advancing exactly its reservation's version. A launch publishes its
+  record as `Admitting`, which is not desired (nothing places, charges, or
+  grants it) and keeps its candidate; a driver recovery's backfill publishes
+  directly. Collection deletes a candidate's material only once its recording
+  no longer holds its reserved or admitting row, after which that recording
+  can never publish or be admitted; a reservation or admitting record older
+  than two minutes (a stalled, failed, or crashed launch) is fenced by
+  removing it at its version, and its material is then collected. A
+  withdrawal removes the material and then the record, or fences a reserved or
+  admitting row; an ended generation
   (sandbox gone, replaced, stopped, or deleting) is withdrawn by the sweep; and
   the compute driver's recovery backfills instances it runs without a record and
   withdraws records published before its listing that it no longer lists.
-- **Placement and ownership.** A launch records its owner, the live member the
-  consistent-hash ring prefers (`gateway_ring`), right after the instance; a
-  member at its 256-instance capacity is skipped, and with no live member with
-  room the launch is refused as a capacity refusal. One
+- **Placement and ownership.** A launch admits its instance on the live member
+  the consistent-hash ring prefers (`gateway_ring`): one transaction creates
+  the owner record, charges the member's capacity, and publishes the
+  `Admitting` record, conditioned on that record's version, so an owner always
+  has its published instance and a removed record never gains one. A member at
+  its 256-instance capacity is skipped. With no live member with room the
+  launch is refused as a capacity refusal: it removes the admitting record it
+  created, only at the version it created (a concurrent launch of the same
+  generation that admitted it first wins), and then that record's candidate
+  and material; a removal that fails is left to fencing. One
   `supervisor_instance_owner` record per instance names the member, its
   process, and the ownership term (the sandbox's credential epoch). Records
   change only by compare-and-swap, and a live owner keeps its instances.

crates/openshell-server/src/config_file.rs

@@ -192,7 +192,7 @@ pub struct GatewayFileSection {
     pub otlp: Option<OtlpConfig>,
     #[serde(default)]
     pub ocsf_log: Option<OcsfLogConfig>,
-    /// Run the shared supervisor as a high-availability pool of replicas.
+    /// Tuning of the shared supervisor pool.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub supervisor_pool: Option<SupervisorPoolConfig>,
 
@@ -205,19 +205,17 @@ pub struct GatewayFileSection {
     pub database_url: Option<String>,
 }
 
-/// `[openshell.gateway.supervisor_pool]`: run the shared supervisor as a pool.
+/// `[openshell.gateway.supervisor_pool]`: optional tuning of the shared
+/// supervisor pool.
 ///
-/// The gateway coordinates the shared supervisor's replicas for high
-/// availability: it places each sandbox instance on a live member by
+/// The shared supervisor always runs as a pool that the gateway coordinates
+/// for high availability: it places each sandbox instance on a live member by
 /// consistent hash, records its owner durably, and moves the instances of a
 /// failed or leaving member to another member, which adopts their running
-/// workloads.
+/// workloads. Without this table the defaults apply.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct SupervisorPoolConfig {
-    /// Coordinate the shared supervisor's replicas as a pool.
-    #[serde(default)]
-    pub enabled: bool,
     /// Seconds a sandbox runtime keeps its workload frozen after losing its
     /// supervisor before terminating it, from 10 to 300. The pool's
     /// supervisors offer it to each runtime; unset offers
@@ -226,17 +224,6 @@ pub struct SupervisorPoolConfig {
     pub recovery_timeout_secs: Option<u32>,
 }
 
-impl SupervisorPoolConfig {
-    /// The recovery deadline the pool offers, when the pool is enabled.
-    #[must_use]
-    pub fn enabled_recovery_timeout_secs(&self) -> Option<u32> {
-        self.enabled.then(|| {
-            self.recovery_timeout_secs
-                .unwrap_or(openshell_supervisor_control::POOL_RECOVERY_TIMEOUT_SECS)
-        })
-    }
-}
-
 /// Gateway listener TLS fields accepted in `gateway.toml`.
 ///
 /// Client-certificate handshake policy is derived from the presence of a
@@ -1779,24 +1766,14 @@ database_lock_max_connections = 0
     }
 
     #[test]
-    fn parses_the_supervisor_pool() {
+    fn parses_the_supervisor_pool_tuning() {
         let pool = |table: &str| {
             let tmp = write_tmp(&format!("[openshell.gateway.supervisor_pool]\n{table}"));
             load(tmp.path()).map(|file| file.openshell.gateway.supervisor_pool)
         };
-        // Enabled, a pool offers its default recovery deadline unless
-        // configured; disabled, it offers none.
-        let enabled = pool("enabled = true\n").unwrap().unwrap();
-        assert_eq!(
-            enabled.enabled_recovery_timeout_secs(),
-            Some(openshell_supervisor_control::POOL_RECOVERY_TIMEOUT_SECS)
-        );
-        let configured = pool("enabled = true\nrecovery_timeout_secs = 90\n")
-            .unwrap()
-            .unwrap();
-        assert_eq!(configured.enabled_recovery_timeout_secs(), Some(90));
-        let disabled = pool("recovery_timeout_secs = 90\n").unwrap().unwrap();
-        assert_eq!(disabled.enabled_recovery_timeout_secs(), None);
+        let configured = pool("recovery_timeout_secs = 90\n").unwrap().unwrap();
+        assert_eq!(configured.recovery_timeout_secs, Some(90));
+        assert_eq!(pool("").unwrap().unwrap().recovery_timeout_secs, None);
         let tmp = write_tmp("[openshell.gateway]\n");
         assert!(
             load(tmp.path())
@@ -1806,22 +1783,23 @@ database_lock_max_connections = 0
                 .supervisor_pool
                 .is_none()
         );
-        // Recovery stays bounded, and unknown keys are refused.
+        // Recovery stays bounded, and unknown keys, including the removed
+        // `enabled` switch, are refused.
         for out_of_range in ["9", "301"] {
             assert!(matches!(
-                pool(&format!(
-                    "enabled = true\nrecovery_timeout_secs = {out_of_range}\n"
-                )),
+                pool(&format!("recovery_timeout_secs = {out_of_range}\n")),
                 Err(ConfigFileError::InvalidValue {
                     field: "openshell.gateway.supervisor_pool.recovery_timeout_secs",
                     ..
                 })
             ));
         }
-        assert!(matches!(
-            pool("enabled = true\nhigh_availability = true\n"),
-            Err(ConfigFileError::Parse { .. })
-        ));
+        for unknown in ["enabled = true\n", "high_availability = true\n"] {
+            assert!(
+                matches!(pool(unknown), Err(ConfigFileError::Parse { .. })),
+                "{unknown}"
+            );
+        }
     }
 
     #[test]

crates/openshell-server/src/lib.rs

@@ -821,17 +821,15 @@ pub(crate) async fn run_server(
             ttl_secs = ttl.as_secs(),
             "shared supervisor assignment stream enabled; minting supervisor tokens"
         );
-        if controller.is_pool() {
-            info!("shared supervisor pool enabled; reconciling pool membership and ownership");
-            supervisor_pool::SupervisorPool::new(
-                state.store.clone(),
-                controller.clone(),
-                authority.clone(),
-                state.credentials.clone(),
-                state.replica_id.clone(),
-            )
-            .start(shutdown_rx.clone());
-        }
+        info!("shared supervisor pool enabled; reconciling pool membership and ownership");
+        supervisor_pool::SupervisorPool::new(
+            state.store.clone(),
+            controller.clone(),
+            authority.clone(),
+            state.credentials.clone(),
+            state.replica_id.clone(),
+        )
+        .start(shutdown_rx.clone());
         spawn_supervisor_token_refresh(authority, ttl, controller.clone(), shutdown_rx.clone());
     }
 
@@ -1708,25 +1706,25 @@ pub struct ComputeDriverBuildContext<'a> {
     config: ComputeDriverConfigContext<'a>,
     shutdown_rx: watch::Receiver<bool>,
     shared_supervision: &'a OnceLock<openshell_supervisor_control::SharedSupervision>,
-    /// Recovery deadline of a configured supervisor pool.
-    supervisor_pool: Option<u32>,
+    /// Recovery deadline from `[openshell.gateway.supervisor_pool]`, when
+    /// configured.
+    supervisor_recovery_timeout_secs: Option<u32>,
 }
 
 impl ComputeDriverBuildContext<'_> {
-    /// The gateway's coordination of a shared supervisor host, for an
+    /// The gateway's coordination of the shared supervisor pool, for an
     /// in-process driver whose configuration places new generations on it.
-    /// The gateway owns the desired set, serves it to the host over the
-    /// private assignment stream, mints the supervisor token, and publishes
-    /// each renewal for the driver to deliver. With
-    /// `[openshell.gateway.supervisor_pool]` enabled it coordinates a pool of
-    /// hosts. Created on first request; every call returns the same
-    /// coordination.
+    /// The gateway owns the desired set and ownership durably, serves each
+    /// member over the private assignment stream, mints the supervisor
+    /// token, and publishes each renewal for the driver to deliver.
+    /// `[openshell.gateway.supervisor_pool]` only tunes it. Created on first
+    /// request; every call returns the same coordination.
     pub fn shared_supervision(&self) -> openshell_supervisor_control::SharedSupervision {
         self.shared_supervision
             .get_or_init(|| {
                 let supervision = openshell_supervisor_control::SharedSupervision::new();
-                if let Some(recovery_timeout_secs) = self.supervisor_pool {
-                    supervision.enable_pool(recovery_timeout_secs);
+                if let Some(recovery_timeout_secs) = self.supervisor_recovery_timeout_secs {
+                    supervision.set_recovery_timeout_secs(recovery_timeout_secs);
                 }
                 supervision
             })
@@ -1850,8 +1848,7 @@ async fn build_compute_runtime(
     info!(driver = %driver.name(), "Using compute driver");
     let supervisor_pool = driver_startup
         .file
-        .and_then(|file| file.openshell.gateway.supervisor_pool)
-        .and_then(|pool| pool.enabled_recovery_timeout_secs());
+        .and_then(|file| file.openshell.gateway.supervisor_pool);
     let runtime = match driver {
         ConfiguredComputeDriver::Registered(registration) => {
             let shared_supervision = OnceLock::new();
@@ -1865,15 +1862,18 @@ async fn build_compute_runtime(
                 },
                 shutdown_rx,
                 shared_supervision: &shared_supervision,
-                supervisor_pool,
+                supervisor_recovery_timeout_secs: supervisor_pool
+                    .and_then(|pool| pool.recovery_timeout_secs),
             };
             let instance = registration.factory.build(build_context).await?;
             let supervisor_host = shared_supervision
                 .into_inner()
                 .map(|supervision| supervision.controller().clone());
             if supervisor_pool.is_some() && supervisor_host.is_none() {
                 return Err(Error::config(
-                    "openshell.gateway.supervisor_pool requires a compute driver that hosts                      sandboxes on the gateway's shared supervisor (Kubernetes                      supervisor_topology = \"gateway\")",
+                    "openshell.gateway.supervisor_pool tunes the shared supervisor pool, but no \
+                     compute driver hosts sandboxes on it; set the Kubernetes driver's \
+                     supervisor_topology = \"gateway\" or remove the table",
                 ));
             }
             match instance {
@@ -1916,8 +1916,8 @@ async fn build_compute_runtime(
         }
         ConfiguredComputeDriver::Remote { .. } if supervisor_pool.is_some() => {
             return Err(Error::config(
-                "openshell.gateway.supervisor_pool requires an in-process compute driver that \
-                 hosts sandboxes on the gateway's shared supervisor",
+                "openshell.gateway.supervisor_pool tunes the shared supervisor pool, which \
+                 only an in-process compute driver can use; remove the table",
             ));
         }
         ConfiguredComputeDriver::Remote { name } => {
@@ -2343,10 +2343,16 @@ mod tests {
     struct TestComputeDriverFactory;
 
     #[test]
-    fn a_configured_supervisor_pool_makes_the_drivers_shared_supervision_a_pool() {
+    fn shared_supervision_is_always_a_pool_tuned_by_its_table() {
         let endpoint_overrides = std::collections::BTreeMap::new();
         let (_shutdown, shutdown_rx) = watch::channel(false);
-        for supervisor_pool in [None, Some(75)] {
+        for (configured, offered) in [
+            (
+                None,
+                openshell_supervisor_control::POOL_RECOVERY_TIMEOUT_SECS,
+            ),
+            (Some(75), 75),
+        ] {
             let shared = std::sync::OnceLock::new();
             let context = super::ComputeDriverBuildContext {
                 config: super::ComputeDriverConfigContext {
@@ -2364,19 +2370,21 @@ mod tests {
                 },
                 shutdown_rx: shutdown_rx.clone(),
                 shared_supervision: &shared,
-                supervisor_pool,
+                supervisor_recovery_timeout_secs: configured,
             };
-            // The driver asks for shared supervision as before; the gateway's
-            // configuration alone decides whether it coordinates a pool.
+            // The driver asks for shared supervision as before; it is the
+            // pool, with the configured recovery deadline or the default.
             let supervision = context.shared_supervision();
-            assert_eq!(supervision.is_pool(), supervisor_pool.is_some());
             assert_eq!(
                 supervision.controller().pool_recovery_timeout_secs(),
-                supervisor_pool
+                offered
             );
             assert_eq!(
-                context.shared_supervision().is_pool(),
-                supervisor_pool.is_some()
+                context
+                    .shared_supervision()
+                    .controller()
+                    .pool_recovery_timeout_secs(),
+                offered
             );
             assert!(shared.get().is_some());
         }