From 9aed067420921c789484482a56391745c04521ff Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Thu, 22 Oct 2026 15:20:44 -0700 Subject: [PATCH 1/3] fix(supervisor): follow the sandbox reconciler leader Signed-off-by: Drew Newberry --- .agents/skills/helm-dev-environment/SKILL.md | 11 +- crates/openshell-server/src/compute/lease.rs | 98 +++++- crates/openshell-server/src/compute/mod.rs | 127 +++++++- crates/openshell-server/src/lib.rs | 64 ++-- .../openshell-supervisor-control/Cargo.toml | 3 + crates/openshell-supervisor-control/README.md | 32 ++ .../src/controller.rs | 302 +++++++++++++++++- .../openshell-supervisor-control/src/lib.rs | 3 +- skills/debug-openshell-cluster/SKILL.md | 17 + 9 files changed, 606 insertions(+), 51 deletions(-) diff --git a/.agents/skills/helm-dev-environment/SKILL.md b/.agents/skills/helm-dev-environment/SKILL.md index e5c1e6b1b3..209d2ab707 100644 --- a/.agents/skills/helm-dev-environment/SKILL.md +++ b/.agents/skills/helm-dev-environment/SKILL.md @@ -71,7 +71,7 @@ gatewayConfig: The gateway chart creates a single-replica shared supervisor Deployment, Service, NetworkPolicy, and an empty `-supervisor-token` Secret in the release namespace, regardless of shared/managed/operator workspace mode. -The gateway mints a supervisor token from its sandbox-JWT signing key, writes +The gateway publisher mints a supervisor token from its sandbox-JWT signing key, writes it into that Secret (the gateway Role may `get`/`patch` only that Secret), and renews it at half its lifetime; the supervisor mounts the Secret and presents the token on the private `openshell.supervisor.host.v1` assignment stream, @@ -96,6 +96,15 @@ OPENSHELL_E2E_KUBE_IMAGE_PULL_SECRET=e2e-regcred \ mise run e2e:kubernetes:shared-supervisor ``` +With replicated gateway storage, the existing sandbox reconciler lease holder +also owns the shared supervisor assignment stream and token renewal. Standbys +refuse `Attach` with `UNAVAILABLE`; the supervisor retries through a fresh +connection, so its configured gateway endpoint must reach all eligible replicas +without sticky routing to a standby. A new leader recovers durable assignments +before publishing. Instances keep running during reconnection. Sandbox creates +through a standby may wait for driver reconciliation and the leader observations +carried by the next lease renewal; expired observations mean unknown supervision. + `mise run e2e:kubernetes:shared-supervisor-spiffe` installs pinned SPIRE charts (`ci/values-spire-delegation-stack.yaml`, removed again on exit) and the `openshell.gateway.experimental_spiffe_delegation` overlay diff --git a/crates/openshell-server/src/compute/lease.rs b/crates/openshell-server/src/compute/lease.rs index fbf039ec81..999f55556b 100644 --- a/crates/openshell-server/src/compute/lease.rs +++ b/crates/openshell-server/src/compute/lease.rs @@ -9,14 +9,20 @@ //! //! The lease payload is a small JSON blob — no protobuf definition needed. //! CAS via `put_if` / `delete_if` provides cross-replica safety; the lease -//! is an optimization to reduce contention, not a correctness mechanism. +//! also fences the shared supervisor's assignment publisher. Each holder +//! stops publishing at a conservative local deadline, including when storage +//! is unavailable. Credential-free supervisor observations ride the same CAS +//! so foreground requests on standby replicas can read the leader's view. use crate::persistence::{PersistenceError, Store, WriteCondition}; use openshell_core::time::now_ms; +use openshell_supervisor_control::SharedObservation; use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; use std::sync::Arc; use std::time::Duration; use thiserror::Error; +use tokio::time::Instant; const LEASE_OBJECT_TYPE: &str = "reconciler_lease"; const LEASE_SINGLETON_ID: &str = "singleton"; @@ -42,6 +48,8 @@ pub enum LeaseError { struct LeasePayload { holder: String, acquired_at_ms: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + supervisor_observations: Option>, } #[derive(Debug, Clone)] @@ -51,15 +59,23 @@ pub struct LeaseRecord { pub acquired_at_ms: i64, pub resource_version: u64, pub updated_at_ms: i64, + pub supervisor_observations: Option>, } #[derive(Debug)] pub struct LeaseGuard { resource_version: u64, acquired_at_ms: i64, + /// Conservative local bound: measured before the CAS, so delayed storage + /// responses never extend publication beyond this lease's lifetime. + deadline: Instant, } impl LeaseGuard { + pub fn deadline(&self) -> Instant { + self.deadline + } + #[allow(dead_code)] pub fn resource_version(&self) -> u64 { self.resource_version @@ -87,10 +103,12 @@ impl ReconcilerLease { /// Attempt to create the lease record. Succeeds only if no lease exists. pub async fn try_acquire(&self) -> Result { + let deadline = Instant::now() + self.ttl; let acquired_at_ms = now_ms(); let payload = LeasePayload { holder: self.replica_id.clone(), acquired_at_ms, + supervisor_observations: None, }; let payload_bytes = serde_json::to_vec(&payload).map_err(|e| PersistenceError::Encode(e.to_string()))?; @@ -111,6 +129,7 @@ impl ReconcilerLease { Ok(result) => Ok(LeaseGuard { resource_version: result.resource_version, acquired_at_ms, + deadline, }), Err(PersistenceError::UniqueViolation { .. }) => Err(LeaseError::AlreadyHeld), Err(e) => Err(LeaseError::Store(e)), @@ -127,10 +146,12 @@ impl ReconcilerLease { return Err(LeaseError::AlreadyHeld); } + let deadline = Instant::now() + self.ttl; let acquired_at_ms = now_ms(); let payload = LeasePayload { holder: self.replica_id.clone(), acquired_at_ms, + supervisor_observations: None, }; let payload_bytes = serde_json::to_vec(&payload).map_err(|e| PersistenceError::Encode(e.to_string()))?; @@ -151,6 +172,7 @@ impl ReconcilerLease { Ok(result) => Ok(LeaseGuard { resource_version: result.resource_version, acquired_at_ms, + deadline, }), Err(PersistenceError::Conflict { .. }) => Err(LeaseError::Conflict), Err(e) => Err(LeaseError::Store(e)), @@ -170,9 +192,21 @@ impl ReconcilerLease { /// Renew the lease by CAS-writing the same payload to bump /// `updated_at_ms` and `resource_version`. pub async fn renew(&self, guard: &mut LeaseGuard) -> Result<(), LeaseError> { + self.renew_with_observations(guard, None).await + } + + /// Publish credential-free supervisor observations with the same CAS that + /// renews ownership. A former owner cannot overwrite a new owner's view. + pub async fn renew_with_observations( + &self, + guard: &mut LeaseGuard, + supervisor_observations: Option>, + ) -> Result<(), LeaseError> { + let deadline = Instant::now() + self.ttl; let payload = LeasePayload { holder: self.replica_id.clone(), acquired_at_ms: guard.acquired_at_ms, + supervisor_observations, }; let payload_bytes = serde_json::to_vec(&payload).map_err(|e| PersistenceError::Encode(e.to_string()))?; @@ -192,6 +226,7 @@ impl ReconcilerLease { { Ok(result) => { guard.resource_version = result.resource_version; + guard.deadline = deadline; Ok(()) } Err(PersistenceError::Conflict { .. }) => Err(LeaseError::Conflict), @@ -236,10 +271,20 @@ impl ReconcilerLease { acquired_at_ms: payload.acquired_at_ms, resource_version: record.resource_version, updated_at_ms: record.updated_at_ms, + supervisor_observations: payload.supervisor_observations, })) } } +/// Remaining freshness of a lease observation. Future timestamps never grant +/// more than one TTL; readers also bound the cached view with a local timer. +pub fn remaining_ttl(updated_at_ms: i64, ttl: Duration) -> Duration { + let age_ms = now_ms().saturating_sub(updated_at_ms).max(0); + ttl.saturating_sub(Duration::from_millis( + u64::try_from(age_ms).unwrap_or(u64::MAX), + )) +} + /// Derive a stable replica identity for lease ownership. /// /// Managed workloads commonly receive a stable runtime identity through @@ -291,6 +336,57 @@ mod tests { assert_eq!(record.holder, "replica-1"); } + #[tokio::test] + async fn supervisor_observations_share_lease_cas_and_clear_on_takeover() { + let store = test_store().await; + // A zero TTL lets the test deterministically steal without wall-clock sleeps. + let first = lease(store.clone(), "replica-1", Duration::ZERO); + let second = lease(store, "replica-2", LEASE_TTL); + let mut guard = first.try_acquire().await.unwrap(); + let observations = BTreeMap::from([("rt-a".to_string(), SharedObservation::Pending)]); + first + .renew_with_observations(&mut guard, Some(observations.clone())) + .await + .unwrap(); + assert_eq!( + first.read().await.unwrap().unwrap().supervisor_observations, + Some(observations.clone()) + ); + // Simulate expiration using the same TTL-zero contender as existing + // lease tests. Acquisition clears every previous owner's observation. + let stealer = lease(second.store.clone(), "replica-2", Duration::ZERO); + let mut replacement = stealer.acquire_or_steal().await.unwrap(); + assert!( + second + .read() + .await + .unwrap() + .unwrap() + .supervisor_observations + .is_none() + ); + assert!(matches!( + first + .renew_with_observations(&mut guard, Some(observations.clone())) + .await, + Err(LeaseError::Conflict) + )); + second + .renew_with_observations(&mut replacement, Some(observations.clone())) + .await + .unwrap(); + assert_eq!( + first.read().await.unwrap().unwrap().supervisor_observations, + Some(observations) + ); + assert!(matches!( + first.release(guard).await, + Err(LeaseError::Conflict) + )); + second.release(replacement).await.unwrap(); + assert!(second.read().await.unwrap().is_none()); + } + #[tokio::test] async fn acquire_fails_when_lease_held() { let store = test_store().await; diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 1ec4cfd522..e0e5007035 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -884,6 +884,11 @@ impl ComputeRuntime { mut self, supervisor_host: Option, ) -> Self { + if !self.store.is_single_replica() + && let Some(controller) = &supervisor_host + { + controller.require_leadership(); + } self.supervisor_host = supervisor_host; self } @@ -4184,6 +4189,9 @@ impl ComputeRuntime { match lease.acquire_or_steal().await { Ok(guard) => { info!(replica = %lease.replica_id(), "acquired reconciler lease"); + if let Some(controller) = &self.supervisor_host { + controller.start_publishing(guard.deadline()); + } self.run_as_holder(&lease, guard, &mut shutdown_rx, startup_rx.clone()) .await; } @@ -4193,6 +4201,29 @@ impl ComputeRuntime { error = %e, "reconciler lease acquisition attempt failed" ); + if let Some(controller) = &self.supervisor_host { + // Unknown on every failed/expired read. Never turn stale + // readiness into a terminal absence for a running sandbox. + let observations = match tokio::time::timeout( + LEASE_ACQUIRE_INTERVAL, + lease.read(), + ) + .await + { + Ok(Ok(Some(record))) => { + let remaining = + lease::remaining_ttl(record.updated_at_ms, LEASE_TTL); + record + .supervisor_observations + .filter(|_| !remaining.is_zero()) + .map(|observations| { + (tokio::time::Instant::now() + remaining, observations) + }) + } + _ => None, + }; + controller.mirror_observations(observations); + } tokio::select! { () = tokio::time::sleep(LEASE_ACQUIRE_INTERVAL) => {} _ = shutdown_rx.changed() => { @@ -4250,17 +4281,36 @@ impl ComputeRuntime { loop { tokio::select! { + () = tokio::time::sleep_until(guard.deadline()) => { + warn!("reconciler lease expired locally — releasing holder role"); + break; + } () = tokio::time::sleep(LEASE_RENEWAL_INTERVAL) => { - match lease.renew(&mut guard).await { - Ok(()) => { + let previous_deadline = guard.deadline(); + let renewed = tokio::time::timeout_at(previous_deadline, async { + if let Some(controller) = &self.supervisor_host { + lease.renew_with_observations(&mut guard, controller.shared_observations()).await + } else { + lease.renew(&mut guard).await + } + }).await; + match renewed { + Ok(Ok(())) => { + if tokio::time::Instant::now() >= previous_deadline { + warn!("reconciler renewal completed after local expiry — releasing holder role"); + break; + } + if let Some(controller) = &self.supervisor_host { + controller.renew_publishing(guard.deadline()); + } debug!(replica = %lease.replica_id(), "renewed reconciler lease"); } - Err(e) => { - warn!( - replica = %lease.replica_id(), - error = %e, - "reconciler lease renewal failed — releasing holder role" - ); + Ok(Err(e)) => { + warn!(error = %e, "reconciler lease renewal failed — releasing holder role"); + break; + } + Err(_) => { + warn!("reconciler lease renewal timed out — releasing holder role"); break; } } @@ -4268,6 +4318,9 @@ impl ComputeRuntime { _ = shutdown_rx.changed() => { if *shutdown_rx.borrow() { info!(replica = %lease.replica_id(), "shutdown — releasing reconciler lease"); + if let Some(controller) = &self.supervisor_host { + controller.stop_publishing(); + } if let Err(e) = lease.release(guard).await { warn!(error = %e, "failed to release reconciler lease on shutdown"); } @@ -4282,6 +4335,9 @@ impl ComputeRuntime { } } + if let Some(controller) = &self.supervisor_host { + controller.stop_publishing(); + } let _ = cancel_tx.send(true); let _ = watch_handle.await; let _ = reconcile_handle.await; @@ -8997,6 +9053,61 @@ mod tests { } } + #[tokio::test] + async fn shared_supervisor_follows_reconciler_leadership_between_replicas() { + use openshell_supervisor_control::SupervisorHostController; + let first_driver = ControlledDriver::new(); + let second_driver = ControlledDriver::new(); + let mut first = test_runtime(first_driver.clone()).await; + let mut second = test_runtime(second_driver.clone()).await; + first.replica_id = "first".to_string(); + second.replica_id = "second".to_string(); + second.store = first.store.clone(); + let first_controller = SupervisorHostController::new(); + let second_controller = SupervisorHostController::new(); + first_controller.require_leadership(); + second_controller.require_leadership(); + first.supervisor_host = Some(first_controller.clone()); + second.supervisor_host = Some(second_controller.clone()); + let (first_stop, first_shutdown) = watch::channel(false); + let (second_stop, second_shutdown) = watch::channel(false); + let (_startup, started) = watch::channel(true); + let first_task = + tokio::spawn(Arc::new(first).lease_coordinator(first_shutdown, started.clone())); + tokio::time::timeout( + Duration::from_secs(5), + first_driver.watch_started.notified(), + ) + .await + .unwrap(); + assert!(first_controller.is_publisher()); + let second_task = + tokio::spawn(Arc::new(second).lease_coordinator(second_shutdown, started)); + assert!(!second_controller.is_publisher()); + first_stop.send_replace(true); + tokio::time::timeout(Duration::from_secs(5), first_task) + .await + .unwrap() + .unwrap(); + assert!(!first_controller.is_publisher()); + tokio::time::timeout( + Duration::from_secs(10), + second_driver.watch_started.notified(), + ) + .await + .unwrap(); + assert!(second_controller.is_publisher()); + // Every new term requires durable recovery, including an explicit + // empty result. An uninitialized controller must not stop workloads. + assert!(!second_controller.is_initialized()); + second_stop.send_replace(true); + tokio::time::timeout(Duration::from_secs(5), second_task) + .await + .unwrap() + .unwrap(); + assert!(!second_controller.is_publisher()); + } + #[tokio::test] async fn create_runtime_inputs_reach_driver_without_persisting_effective_policy_or_secrets() { let driver = Arc::new(TestDriver::default()); diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 79d45b8dcd..d262496697 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -253,33 +253,46 @@ fn spawn_supervisor_token_refresh( mut shutdown_rx: watch::Receiver, ) { tokio::spawn(async move { + let mut next_mint = tokio::time::Instant::now(); + let mut credential = controller.credential(); loop { - let delay = match authority.mint_supervisor_token(ttl) { - Ok(minted) => { - let remaining = minted.expires_at.saturating_sub( - i64::try_from( - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_or(0, |elapsed| elapsed.as_secs()), - ) - .unwrap_or(i64::MAX), - ); - controller.publish_credential( - openshell_supervisor_control::SupervisorCredential { - token: minted.token, - expires_at: minted.expires_at, - }, - ); - Duration::from_secs(u64::try_from(remaining).unwrap_or(0) / 2) - .max(Duration::from_secs(1)) - } - Err(error) => { - warn!(%error, "failed to mint the supervisor token; retrying"); - Duration::from_secs(10) - } - }; + if credential.has_changed().unwrap_or(false) && credential.borrow_and_update().is_none() + { + next_mint = tokio::time::Instant::now(); + } + if !controller.is_publisher() { + // Mint promptly when this replica acquires the existing lease; + // standbys never compete to overwrite the supervisor's token. + next_mint = tokio::time::Instant::now(); + } else if tokio::time::Instant::now() >= next_mint { + let delay = match authority.mint_supervisor_token(ttl) { + Ok(minted) => { + let remaining = minted.expires_at.saturating_sub( + i64::try_from( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |elapsed| elapsed.as_secs()), + ) + .unwrap_or(i64::MAX), + ); + controller.publish_credential( + openshell_supervisor_control::SupervisorCredential { + token: minted.token, + expires_at: minted.expires_at, + }, + ); + Duration::from_secs(u64::try_from(remaining).unwrap_or(0) / 2) + .max(Duration::from_secs(1)) + } + Err(error) => { + warn!(%error, "failed to mint the supervisor token; retrying"); + Duration::from_secs(10) + } + }; + next_mint = tokio::time::Instant::now() + delay; + } tokio::select! { - () = tokio::time::sleep(delay) => {} + () = tokio::time::sleep(Duration::from_secs(1)) => {} changed = shutdown_rx.changed() => { if changed.is_err() || *shutdown_rx.borrow() { return; @@ -289,6 +302,7 @@ fn spawn_supervisor_token_refresh( } }); } + pub use auth::audit::ConnectionSource; pub use multiplex::{MultiplexService, MultiplexedService}; pub use persistence::Store; diff --git a/crates/openshell-supervisor-control/Cargo.toml b/crates/openshell-supervisor-control/Cargo.toml index b61564c2a5..962609cda6 100644 --- a/crates/openshell-supervisor-control/Cargo.toml +++ b/crates/openshell-supervisor-control/Cargo.toml @@ -23,5 +23,8 @@ tokio-stream = { workspace = true } tonic = { workspace = true } tracing = { workspace = true } +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } + [lints] workspace = true diff --git a/crates/openshell-supervisor-control/README.md b/crates/openshell-supervisor-control/README.md index 124caa5080..9fc194d859 100644 --- a/crates/openshell-supervisor-control/README.md +++ b/crates/openshell-supervisor-control/README.md @@ -63,3 +63,35 @@ its own infrastructure evidence alongside the coordinator's verdict, and delivers the token into the Secret the supervisor Pod mounts. The supervisor (`openshell-supervisor`) consumes only the registration and status types; it owns admission, per-instance runtime, routing, and cleanup. + +## Gateway leadership + +With replicated gateway storage, the existing sandbox reconciler lease also +owns supervisor assignment publication. A standby refuses `Attach` with +`UNAVAILABLE`; the supervisor opens a fresh connection to the gateway endpoint +on retry. That endpoint must reach all eligible gateway replicas, without +session affinity pinning the supervisor to a standby. Single-replica storage +keeps a local publisher and does not acquire a database lease. + +Acquiring the lease starts a new term with an uninitialized desired set. The +compute driver's initial watch recovers the complete set from durable records +before any snapshot is sent. A new term invalidates an older recovery pass. +Losing or releasing the lease closes the assignment stream before worker +cleanup, and a conservative monotonic deadline stops publication even when a +renewal cannot finish. Hosted instances keep running while reconnecting. +Only the publisher mints supervisor tokens; the existing delivery path carries +them to the supervisor. + +Foreground sandbox operations still run on every gateway replica. Their launch +material and lifecycle changes are durable, so the leader's existing driver +reconciliation discovers creates and withdrawals from other replicas. The +leader includes credential-free, generation-routed instance observations in +the same compare-and-swap write that renews the reconciler lease. Standbys use +that view for readiness; they never publish assignments from it. Missing, +unreadable, or expired observations mean unknown supervision, not an absent +instance. Observations are eventually consistent with lease renewals (10 +seconds) and standby polls (5 seconds); driver reconciliation can add up to its +normal interval before a remote lifecycle change reaches the supervisor. + +This coordinates gateway ownership of one shared supervisor. It does not add +supervisor HA, scheduling, autoscaling, or recovery from supervisor process loss. diff --git a/crates/openshell-supervisor-control/src/controller.rs b/crates/openshell-supervisor-control/src/controller.rs index 2d0aa885d9..c59783e188 100644 --- a/crates/openshell-supervisor-control/src/controller.rs +++ b/crates/openshell-supervisor-control/src/controller.rs @@ -29,11 +29,14 @@ use std::collections::BTreeMap; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use serde::{Deserialize, Serialize}; use tokio::sync::{mpsc, watch}; +use tokio::time::Instant; use tokio_stream::StreamExt as _; use tokio_stream::wrappers::ReceiverStream; use tonic::{Status, Streaming}; +use crate::SandboxInstanceStatusReport; use crate::proto::{ AssignmentSnapshot, HostAccepted, HostGatewayHeartbeat, HostGatewayMessage, HostHello, HostMessage, HostStatus, SandboxInstanceAssignment, SandboxInstanceStatus, @@ -98,6 +101,33 @@ pub enum ObservedInstance { Reported(SandboxInstanceStatus), } +/// Credential-free observations carried by the gateway's existing reconciler +/// lease. Routes are generation-specific; an omitted route is unknown, never +/// evidence that a newly created instance is absent. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum SharedObservation { + Pending, + Absent, + Reported(SandboxInstanceStatusReport), +} + +#[derive(Clone, Copy)] +enum Publisher { + Standalone, + Standby, + Lease(Instant), +} + +impl Publisher { + fn active(self) -> bool { + match self { + Self::Standalone => true, + Self::Standby => false, + Self::Lease(deadline) => Instant::now() < deadline, + } + } +} + struct Assigned { assignment: SandboxInstanceAssignment, /// Snapshot revision that first carried this exact assignment. @@ -163,6 +193,10 @@ struct Observed { } struct Inner { + publisher: Mutex, + /// Standbys read only lease-fenced observations, never another replica's + /// desired state or credentials. The local deadline also bounds cached reads. + mirror: Mutex)>>, desired: Mutex, /// Latest supervisor token the gateway minted; `None` until the first. credential: watch::Sender>, @@ -216,6 +250,8 @@ impl SupervisorHostController { pub fn new() -> Self { Self { inner: Arc::new(Inner { + publisher: Mutex::new(Publisher::Standalone), + mirror: Mutex::new(None), desired: Mutex::new(Desired::default()), credential: watch::Sender::new(None), revision: watch::Sender::new(0), @@ -224,9 +260,102 @@ impl SupervisorHostController { } } + /// Require the existing gateway reconciler lease before serving a host. + /// Call before exposing this controller on a replicated gateway. + pub fn require_leadership(&self) { + self.stop_publishing(); + } + + /// Start a new reconciler term. Discard the previous replica-local view: + /// only a fresh complete recovery may initialize this term's assignments. + /// Bumping the revision fences recovery work begun in an earlier term. + pub fn start_publishing(&self, deadline: Instant) { + let mut publisher = lock(&self.inner.publisher); + let mut desired = lock(&self.inner.desired); + let revision = desired.revision + 1; + *desired = Desired { + revision, + ..Desired::default() + }; + *lock(&self.inner.mirror) = None; + Self::disconnect(&mut lock(&self.inner.observed)); + *publisher = Publisher::Lease(deadline); + self.inner.revision.send_replace(revision); + } + + /// Extend publication only after the existing lease was renewed. + pub fn renew_publishing(&self, deadline: Instant) { + let mut publisher = lock(&self.inner.publisher); + if matches!(*publisher, Publisher::Lease(_)) && publisher.active() { + *publisher = Publisher::Lease(deadline); + } + } + + /// Revoke publication before releasing the lease or waiting for worker + /// cleanup. Existing sandbox instances survive their stream closing. + pub fn stop_publishing(&self) { + let mut publisher = lock(&self.inner.publisher); + *publisher = Publisher::Standby; + Self::disconnect(&mut lock(&self.inner.observed)); + self.inner.credential.send_replace(None); + } + + fn disconnect(observed: &mut Observed) { + if let Some(session) = observed.session.take() { + session.superseded.send_replace(true); + } + observed.status = None; + } + + #[must_use] + pub fn is_publisher(&self) -> bool { + lock(&self.inner.publisher).active() + } + + /// Snapshot observations for the next CAS renewal of the reconciler + /// lease. Contains no launch material or bearer tokens. + #[must_use] + pub fn shared_observations(&self) -> Option> { + let publisher = lock(&self.inner.publisher); + if !publisher.active() { + return None; + } + let desired = lock(&self.inner.desired); + let observed = lock(&self.inner.observed); + observed.session.as_ref()?; + Some( + desired + .assignments + .keys() + .map(|route| { + let value = match Self::observe(&desired, &observed, route) { + ObservedInstance::Reported(status) => SharedObservation::Reported( + SandboxInstanceStatusReport::from_proto(&status), + ), + ObservedInstance::Absent => SharedObservation::Absent, + _ => SharedObservation::Pending, + }; + (route.clone(), value) + }) + .collect(), + ) + } + + /// Replace the standby's view from a fresh read of the existing lease. + /// Failed reads, absent owners, and expired leases clear it immediately. + pub fn mirror_observations( + &self, + observations: Option<(Instant, BTreeMap)>, + ) { + *lock(&self.inner.mirror) = observations; + } + /// Publish a freshly minted supervisor token for delivery to the host. pub fn publish_credential(&self, credential: SupervisorCredential) { - self.inner.credential.send_replace(Some(credential)); + let publisher = lock(&self.inner.publisher); + if publisher.active() { + self.inner.credential.send_replace(Some(credential)); + } } /// The current supervisor token and every renewal after it. @@ -274,7 +403,7 @@ impl SupervisorHostController { assignments: impl IntoIterator, ) -> Result { let mut desired = lock(&self.inner.desired); - if desired.initialized && desired.revision != baseline { + if desired.revision != baseline { return Err(AssignmentError::Conflict { baseline, current: desired.revision, @@ -375,8 +504,30 @@ impl SupervisorHostController { /// What the connected host reports for `route`. #[must_use] pub fn observed(&self, route: &str) -> ObservedInstance { - let desired = lock(&self.inner.desired); - let observed = lock(&self.inner.observed); + let publisher = lock(&self.inner.publisher); + if !publisher.active() { + let mirror = lock(&self.inner.mirror); + return match mirror + .as_ref() + .filter(|(deadline, _)| Instant::now() < *deadline) + .and_then(|(_, observations)| observations.get(route)) + { + Some(SharedObservation::Reported(status)) => { + ObservedInstance::Reported(status.to_proto()) + } + Some(SharedObservation::Absent) => ObservedInstance::Absent, + Some(SharedObservation::Pending) => ObservedInstance::Pending, + None => ObservedInstance::NoHost, + }; + } + Self::observe( + &lock(&self.inner.desired), + &lock(&self.inner.observed), + route, + ) + } + + fn observe(desired: &Desired, observed: &Observed, route: &str) -> ObservedInstance { if observed.session.is_none() { return ObservedInstance::NoHost; } @@ -405,6 +556,11 @@ impl SupervisorHostController { identity: HostIdentity, mut inbound: Streaming, ) -> Result>, Status> { + if !self.is_publisher() { + return Err(Status::unavailable( + "reconnect to the gateway reconciler leader", + )); + } let hello = match tokio::time::timeout(HELLO_TIMEOUT, inbound.next()).await { Ok(Some(Ok(HostMessage { payload: Some(host_message::Payload::Hello(hello)), @@ -463,6 +619,12 @@ impl SupervisorHostController { identity: HostIdentity, hello: &HostHello, ) -> Result<(u64, watch::Receiver), Status> { + let publisher = lock(&self.inner.publisher); + if !publisher.active() { + return Err(Status::unavailable( + "reconnect to the gateway reconciler leader", + )); + } let mut observed = lock(&self.inner.observed); if let Some(current) = &observed.session && current.instance_id == hello.instance_id @@ -575,7 +737,7 @@ impl SupervisorHostController { tokio::pin!(credential_expiry); let mut revision = self.inner.revision.subscribe(); let mut sent_revision = 0; - let mut last_heard = tokio::time::Instant::now(); + let mut last_heard = Instant::now(); let mut heartbeat = tokio::time::interval(HOST_HEARTBEAT_INTERVAL); heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); // A send that cannot complete within the bound ends the session, so a @@ -594,11 +756,22 @@ impl SupervisorHostController { } }; loop { + let publisher = *lock(&self.inner.publisher); + if !publisher.active() || *superseded.borrow() { + return; + } + let lease_expiry = async move { + if let Publisher::Lease(deadline) = publisher { + tokio::time::sleep_until(deadline).await; + } else { + std::future::pending::<()>().await; + } + }; + tokio::pin!(lease_expiry); let current = *revision.borrow_and_update(); - if current > sent_revision { - let Some(snapshot) = self.snapshot() else { - continue; - }; + if current > sent_revision + && let Some(snapshot) = self.snapshot() + { tracing::debug!( session = session_id, revision = snapshot.revision, @@ -606,13 +779,19 @@ impl SupervisorHostController { "sending supervisor host assignment snapshot" ); let revision_sent = snapshot.revision; - if !send(host_gateway_message::Payload::Assignments(snapshot)).await { - return; + tokio::select! { + biased; + _ = superseded.changed() => return, + () = &mut lease_expiry => return, + delivered = send(host_gateway_message::Payload::Assignments(snapshot)) => { + if !delivered { return; } + } } self.record_sent(session_id, revision_sent); sent_revision = current; } tokio::select! { + () = &mut lease_expiry => {}, changed = revision.changed() => { if changed.is_err() { return; @@ -620,13 +799,13 @@ impl SupervisorHostController { } message = inbound.next() => match message { Some(Ok(HostMessage { payload: Some(host_message::Payload::Status(status)) })) => { - last_heard = tokio::time::Instant::now(); + last_heard = Instant::now(); if let Err(reason) = self.record_status(session_id, status) { tracing::warn!(session = session_id, reason, "dropped supervisor host status report"); } } Some(Ok(HostMessage { payload: Some(host_message::Payload::Heartbeat(_)) })) => { - last_heard = tokio::time::Instant::now(); + last_heard = Instant::now(); } Some(Ok(_)) => { let _ = tx @@ -650,8 +829,13 @@ impl SupervisorHostController { tracing::warn!(session = session_id, "supervisor host stream went silent; closing it"); return; } - if !send(host_gateway_message::Payload::Heartbeat(HostGatewayHeartbeat {})).await { - return; + tokio::select! { + biased; + _ = superseded.changed() => return, + () = &mut lease_expiry => return, + delivered = send(host_gateway_message::Payload::Heartbeat(HostGatewayHeartbeat {})) => { + if !delivered { return; } + } } } _ = superseded.changed() => return, @@ -786,6 +970,94 @@ mod tests { assert_eq!(controller.assign(assignment("rt-0", b"x")), Ok(())); } + #[tokio::test(start_paused = true)] + async fn leadership_fences_old_sessions_recovery_and_cached_observations() { + let controller = SupervisorHostController::new(); + controller + .replace_assignments(0, [assignment("rt-old", b"old")]) + .unwrap(); + let old_baseline = controller.revision(); + controller.require_leadership(); + let identity = HostIdentity { + subject: "openshell-supervisor:test".to_string(), + token_id: "token".to_string(), + expires_at: i64::MAX, + }; + let hello = HostHello { + instance_id: "boot".to_string(), + connection_epoch: 1, + }; + assert_eq!( + controller + .begin_session(identity.clone(), &hello) + .unwrap_err() + .code(), + tonic::Code::Unavailable + ); + controller.start_publishing(Instant::now() + Duration::from_secs(30)); + assert!(!controller.is_initialized()); + assert!(controller.assignment("rt-old").is_none()); + assert!(controller.replace_assignments(old_baseline, []).is_err()); + // No authoritative empty snapshot before the new term's full recovery. + assert!(controller.snapshot().is_none()); + controller + .replace_assignments(controller.revision(), [assignment("rt-new", b"new")]) + .unwrap(); + let (session, revoked) = controller.begin_session(identity.clone(), &hello).unwrap(); + controller.record_sent(session, controller.revision()); + controller + .record_status(session, status(controller.revision(), &["rt-new"])) + .unwrap(); + + let standby = SupervisorHostController::new(); + standby.require_leadership(); + standby.mirror_observations(Some(( + Instant::now() + Duration::from_secs(30), + controller.shared_observations().unwrap(), + ))); + assert!( + matches!(standby.observed("rt-new"), ObservedInstance::Reported(status) if status.ready) + ); + assert_eq!( + standby.observed("rt-created-on-standby"), + ObservedInstance::NoHost + ); + + // Renewing extends the local bound, but a cached standby view expires + // unless it too reads a fresh lease. Expiry is unknown, never absent. + tokio::time::advance(Duration::from_secs(20)).await; + controller.renew_publishing(Instant::now() + Duration::from_secs(30)); + tokio::time::advance(Duration::from_secs(11)).await; + assert!(controller.is_publisher()); + assert_eq!(standby.observed("rt-new"), ObservedInstance::NoHost); + controller.stop_publishing(); + assert!(*revoked.borrow()); + assert!(controller.record_status(session, status(0, &[])).is_err()); + assert_eq!(controller.observed("rt-new"), ObservedInstance::NoHost); + assert_eq!( + controller + .begin_session(identity.clone(), &hello) + .unwrap_err() + .code(), + tonic::Code::Unavailable + ); + + controller.start_publishing(Instant::now() + Duration::from_secs(30)); + assert!(!controller.is_initialized()); + assert!(controller.snapshot().is_none()); + tokio::time::advance(Duration::from_secs(31)).await; + // A late renewal cannot revive an expired term. + controller.renew_publishing(Instant::now() + Duration::from_secs(30)); + assert!(!controller.is_publisher()); + assert_eq!( + controller + .begin_session(identity, &hello) + .unwrap_err() + .code(), + tonic::Code::Unavailable + ); + } + #[test] fn sessions_fence_stale_streams_epochs_and_acks() { let controller = SupervisorHostController::new(); diff --git a/crates/openshell-supervisor-control/src/lib.rs b/crates/openshell-supervisor-control/src/lib.rs index 4401ff3d84..077c8b6594 100644 --- a/crates/openshell-supervisor-control/src/lib.rs +++ b/crates/openshell-supervisor-control/src/lib.rs @@ -38,7 +38,8 @@ pub mod proto { pub use controller::{ AssignmentError, HOST_HEARTBEAT_INTERVAL, HostIdentity, MAX_ASSIGNED_INSTANCES, - MAX_REPORTED_INSTANCES, ObservedInstance, SupervisorCredential, SupervisorHostController, + MAX_REPORTED_INSTANCES, ObservedInstance, SharedObservation, SupervisorCredential, + SupervisorHostController, }; pub use coordinator::{ InstanceSupervision, LaunchError, RecoverableInstance, RecoveryBaseline, RecoveryOutcome, diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 40351c998e..ed319575dc 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -687,6 +687,23 @@ returns `UNAVAILABLE` with "lock sandbox SSH identity failed", and neither `openshell_server_mutation_lock_timeouts_total` nor `openshell_server_mutation_lock_errors_total` counts it. +With `supervisor_topology = "gateway"` and replicated gateway storage, the +shared supervisor's assignment stream belongs to the existing sandbox +reconciler lease holder. Look for `acquired reconciler lease` and `supervisor +host attached` on the same gateway Pod. A standby returns `UNAVAILABLE` with +`reconnect to the gateway reconciler leader`; the supervisor retries using a +fresh connection. Its gateway endpoint must balance across eligible replicas; +remove session affinity that pins retries to a standby. Hosted workloads +continue running during reconnection. After takeover, the new leader recovers +durable assignments before sending its first snapshot. + +Creates through a standby can take an extra driver reconciliation interval +before the leader assigns their instances. The standby reads readiness from +observations carried by the reconciler lease, refreshed on renewal (10 +seconds) and polled every 5 seconds. An expired or unreadable lease means +unknown supervision, not instance failure. Check gateway storage connectivity +and lease-renewal logs when readiness lags across all sandboxes. + For multi-replica gateway installs, supervisor and client session traffic may be served by a non-owner gateway replica and relayed to the current supervisor owner over the internal `PeerRelay` RPC. Check the headless peer Service, From 5c3fcb344c8d989ee0ca54258668c5a4aab030ba Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Thu, 22 Oct 2026 15:22:03 -0700 Subject: [PATCH 2/3] fix(supervisor): keep undiscovered assignments pending Signed-off-by: Drew Newberry --- crates/openshell-supervisor-control/README.md | 5 ++- .../src/controller.rs | 43 ++++++++++++------- 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/crates/openshell-supervisor-control/README.md b/crates/openshell-supervisor-control/README.md index 9fc194d859..eb4bfc1895 100644 --- a/crates/openshell-supervisor-control/README.md +++ b/crates/openshell-supervisor-control/README.md @@ -89,7 +89,10 @@ leader includes credential-free, generation-routed instance observations in the same compare-and-swap write that renews the reconciler lease. Standbys use that view for readiness; they never publish assignments from it. Missing, unreadable, or expired observations mean unknown supervision, not an absent -instance. Observations are eventually consistent with lease renewals (10 +instance. A route the leader has not recovered yet stays pending, even if +its connected host has already acknowledged a snapshot without that route. +Only an acknowledged assignment can establish that an instance is absent. +Observations are eventually consistent with lease renewals (10 seconds) and standby polls (5 seconds); driver reconciliation can add up to its normal interval before a remote lifecycle change reaches the supervisor. diff --git a/crates/openshell-supervisor-control/src/controller.rs b/crates/openshell-supervisor-control/src/controller.rs index c59783e188..1b4415d763 100644 --- a/crates/openshell-supervisor-control/src/controller.rs +++ b/crates/openshell-supervisor-control/src/controller.rs @@ -91,11 +91,11 @@ pub struct SupervisorCredential { pub enum ObservedInstance { /// No host session is connected. NoHost, - /// A host is connected but has not reconciled to the snapshot that - /// assigned this route. + /// A host is connected, but this gateway has not recovered the route's + /// assignment yet or the host has not reconciled the snapshot assigning it. Pending, - /// The host reconciled to that snapshot (or the route is not assigned) - /// and does not host the route. + /// The host reconciled to the snapshot assigning this route and does + /// not host it. Absent, /// The host's latest report of the instance. Reported(SandboxInstanceStatus), @@ -200,7 +200,7 @@ struct Inner { desired: Mutex, /// Latest supervisor token the gateway minted; `None` until the first. credential: watch::Sender>, - /// Current desired revision; 0 while uninitialized. + /// Monotonic local revision, also bumped at the start of a new term. revision: watch::Sender, observed: Mutex, } @@ -371,8 +371,8 @@ impl SupervisorHostController { lock(&self.inner.desired).initialized } - /// Current desired revision: 0 while uninitialized. Read it before - /// deriving a full list for [`Self::replace_assignments`]; the + /// Current desired revision, including the fence for this leadership + /// term. Read it before deriving a full list for [`Self::replace_assignments`]; the /// coordinator does so through /// [`SharedSupervision::begin_recovery`](crate::SharedSupervision::begin_recovery). #[must_use] @@ -393,8 +393,8 @@ impl SupervisorHostController { /// durable records while the revision was `baseline`. /// /// Initializes the controller; edits made before initialization are - /// applied on top of the list. After initialization the replacement is - /// compare-and-replace: when the revision moved past `baseline` while + /// applied on top of the list. The replacement is compare-and-replace: + /// when the revision moved past `baseline` while /// the list was being derived, nothing changes and the conflict is /// returned so the caller derives the list again. pub fn replace_assignments( @@ -531,6 +531,16 @@ impl SupervisorHostController { if observed.session.is_none() { return ObservedInstance::NoHost; } + let Some(assigned) = desired + .assignments + .get(route) + .filter(|_| desired.initialized) + else { + // A create can have reached another replica before this leader's + // recovery discovers its durable assignment. Never suspend that + // workload based on a snapshot that did not assign it. + return ObservedInstance::Pending; + }; let Some(status) = &observed.status else { return ObservedInstance::Pending; }; @@ -541,11 +551,10 @@ impl SupervisorHostController { { return ObservedInstance::Reported(instance.clone()); } - match desired.assignments.get(route) { - Some(assigned) if assigned.revision > status.applied_revision => { - ObservedInstance::Pending - } - _ => ObservedInstance::Absent, + if assigned.revision > status.applied_revision { + ObservedInstance::Pending + } else { + ObservedInstance::Absent } } @@ -1000,10 +1009,12 @@ mod tests { assert!(controller.replace_assignments(old_baseline, []).is_err()); // No authoritative empty snapshot before the new term's full recovery. assert!(controller.snapshot().is_none()); + let (session, revoked) = controller.begin_session(identity.clone(), &hello).unwrap(); + controller.record_status(session, status(0, &[])).unwrap(); + assert_eq!(controller.observed("rt-new"), ObservedInstance::Pending); controller .replace_assignments(controller.revision(), [assignment("rt-new", b"new")]) .unwrap(); - let (session, revoked) = controller.begin_session(identity.clone(), &hello).unwrap(); controller.record_sent(session, controller.revision()); controller .record_status(session, status(controller.revision(), &["rt-new"])) @@ -1108,7 +1119,7 @@ mod tests { // A newer assignment is pending until the host reconciles past it. controller.assign(assignment("rt-b", b"b")).unwrap(); assert_eq!(controller.observed("rt-b"), ObservedInstance::Pending); - assert_eq!(controller.observed("rt-unknown"), ObservedInstance::Absent); + assert_eq!(controller.observed("rt-unknown"), ObservedInstance::Pending); controller.record_sent(current, 2); controller.record_status(current, status(2, &[])).unwrap(); assert_eq!(controller.observed("rt-b"), ObservedInstance::Absent); From c4897f9c966859a81f617161b98b6b3d4060c7f4 Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Thu, 22 Oct 2026 17:00:43 -0700 Subject: [PATCH 3/3] fix(supervisor): fence recovery and retry standby discovery Signed-off-by: Drew Newberry --- crates/openshell-supervisor-control/README.md | 59 ++- .../src/controller.rs | 389 +++++++++++++++--- .../src/coordinator.rs | 95 ++++- .../openshell-supervisor-control/src/lib.rs | 7 +- crates/openshell-supervisor/README.md | 3 +- .../openshell-supervisor/src/shared_host.rs | 96 ++++- 6 files changed, 544 insertions(+), 105 deletions(-) diff --git a/crates/openshell-supervisor-control/README.md b/crates/openshell-supervisor-control/README.md index eb4bfc1895..4bdf2e184d 100644 --- a/crates/openshell-supervisor-control/README.md +++ b/crates/openshell-supervisor-control/README.md @@ -23,7 +23,15 @@ re-implementing the control contracts. delayed older hello is refused, and reports from superseded streams, acks of revisions never sent, regressing acks, and oversized reports are dropped. A stream ends when the token that opened it expires; hosted - instances keep running. At most 256 instances are assigned. + instances keep running. At most 256 instances are assigned. Selection + considers only candidates still in the durable runnable records: a + recovered list beyond capacity keeps the candidates that are currently + assigned or that the host reports running, then fills the rest in route + order and leaves the excess unassigned. After a takeover, an oversized + list keeps the previous set until the host's initial report shows what it + runs, so the bounded set never stops a running instance. + After initialization every withdrawal advances the revision, so a + recovery listed before it cannot re-admit the withdrawn route. - `registration`: `SandboxInstanceRegistration` is the JSON launch material one assignment carries and the driver persists with the generation. It grants the supervisor exactly that generation's gateway and Sandbox Protocol @@ -35,17 +43,20 @@ re-implementing the control contracts. - `coordinator`: `SharedSupervision` is what the gateway constructs and hands to the compute driver. `launch` validates the driver's registration, persists the encoded launch material through the driver's operation, and - only then assigns the instance; a host at capacity is refused before the - workload runs. `begin_recovery` captures + only then assigns the instance; the publisher refuses a create beyond + capacity before the workload runs. `begin_recovery` captures the revision before the driver lists its durable records and `complete_recovery` reuses current assignments, reads the rest through the driver's operation, binds each to its recorded route and sandbox, and keeps - the previous set when any read fails; a list that raced a lifecycle edit is - refused and derived again. `supervision` and `wait_for_ready` interpret the - host's observations in one place: a disconnected host is unknown, a host - that has not reconciled to the assignment yet is degraded, and only a host - that reconciled past it without hosting the instance, hosts it for another - sandbox, or reports it ended makes it unavailable. + the previous set when any read fails or material is malformed; a list that + raced a lifecycle edit or the end of the term is refused and derived again. + A listed generation whose material is gone, or that exceeds capacity, is + excluded for that recovery. `supervision` and `wait_for_ready` interpret + the host's observations in one place: a disconnected host is unknown, a + host that has not reconciled to the assignment or exclusion yet is + degraded, and only a host that reconciled past it without hosting the + instance, hosts it for another sandbox, or reports it ended makes it + unavailable. - `credential`: `deliver_credentials` delivers the current supervisor token and every renewal through the driver's operation, retrying a failed delivery until it succeeds or a newer token supersedes it. @@ -61,37 +72,49 @@ its lifetime, and gives the coordination to the driver. The driver back launch material, provisions the workload's path to the supervisor, checks its own infrastructure evidence alongside the coordinator's verdict, and delivers the token into the Secret the supervisor Pod mounts. The supervisor -(`openshell-supervisor`) consumes only the registration and status types; it -owns admission, per-instance runtime, routing, and cleanup. +(`openshell-supervisor`) consumes only the registration and status types and +the private standby refusal marker (`is_standby_refusal`); it owns admission, +per-instance runtime, routing, and cleanup. ## Gateway leadership With replicated gateway storage, the existing sandbox reconciler lease also owns supervisor assignment publication. A standby refuses `Attach` with -`UNAVAILABLE`; the supervisor opens a fresh connection to the gateway endpoint -on retry. That endpoint must reach all eligible gateway replicas, without +`UNAVAILABLE` and the `x-openshell-supervisor-host: standby` metadata marker; +the supervisor retries after a short jittered delay on a fresh connection to +the gateway endpoint, and keeps its exponential backoff for every other +failure. That endpoint must reach all eligible gateway replicas, without session affinity pinning the supervisor to a standby. Single-replica storage keeps a local publisher and does not acquire a database lease. Acquiring the lease starts a new term with an uninitialized desired set. The compute driver's initial watch recovers the complete set from durable records before any snapshot is sent. A new term invalidates an older recovery pass. -Losing or releasing the lease closes the assignment stream before worker -cleanup, and a conservative monotonic deadline stops publication even when a -renewal cannot finish. Hosted instances keep running while reconnecting. +Losing or releasing the lease closes the assignment stream and discards the +term's desired state before worker cleanup, and a conservative monotonic +deadline stops publication even when a renewal cannot finish; an expired or +former publisher edits nothing. Hosted instances keep running while +reconnecting. Only the publisher mints supervisor tokens; the existing delivery path carries them to the supervisor. Foreground sandbox operations still run on every gateway replica. Their launch material and lifecycle changes are durable, so the leader's existing driver -reconciliation discovers creates and withdrawals from other replicas. The +reconciliation discovers creates and withdrawals from other replicas; a +standby keeps no assignments of its own. Admission through a standby is +therefore eventual: a create beyond capacity is not refused up front but left +unassigned by the leader's recovery, and its supervision then becomes +unavailable. The leader includes credential-free, generation-routed instance observations in the same compare-and-swap write that renews the reconciler lease. Standbys use that view for readiness; they never publish assignments from it. Missing, unreadable, or expired observations mean unknown supervision, not an absent instance. A route the leader has not recovered yet stays pending, even if its connected host has already acknowledged a snapshot without that route. -Only an acknowledged assignment can establish that an instance is absent. +Only an acknowledged assignment, or an acknowledged snapshot omitting a route +the leader's latest complete recovery excluded, can establish that an +instance is absent. Exclusions belong to the term and are cleared when the +route is reassigned, withdrawn, or recovered with its material again. Observations are eventually consistent with lease renewals (10 seconds) and standby polls (5 seconds); driver reconciliation can add up to its normal interval before a remote lifecycle change reaches the supervisor. diff --git a/crates/openshell-supervisor-control/src/controller.rs b/crates/openshell-supervisor-control/src/controller.rs index 1b4415d763..b24a33e27b 100644 --- a/crates/openshell-supervisor-control/src/controller.rs +++ b/crates/openshell-supervisor-control/src/controller.rs @@ -20,12 +20,16 @@ //! before the records were listed, so a list that raced a lifecycle edit //! never resurrects a withdrawn instance or discards a new one. //! +//! Only the publisher owns desired state. A replica that does not publish +//! keeps none: its lifecycle edits are durable records that the publisher's +//! next recovery discovers, so admission on a replicated gateway is eventual. +//! //! Exactly one host serves a gateway: a new attach from a newer connection of //! the same host process, or from a new host process, supersedes the current //! session; reports and acks from a superseded stream are dropped, and a //! delayed hello from an older connection of the current process is refused. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -54,10 +58,16 @@ const HELLO_TIMEOUT: Duration = Duration::from_secs(10); const SEND_TIMEOUT: Duration = Duration::from_secs(30); /// Gateway-to-host messages queued before the stream applies backpressure. const OUTBOUND_QUEUE: usize = 16; -/// Most sandbox instances one shared supervisor host runs. The gateway -/// refuses to assign beyond it, and the host refuses snapshots beyond it, so -/// a sandbox is never created that its host could not accept. +/// Most sandbox instances one shared supervisor host runs. The publishing +/// gateway refuses to assign beyond it, recovery leaves any excess +/// unassigned, and the host refuses snapshots beyond it. pub const MAX_ASSIGNED_INSTANCES: usize = 256; +/// Metadata key marking a standby replica's `UNAVAILABLE` refusal of `Attach`. +/// +/// Paired with [`STANDBY_REFUSAL_VALUE`], it lets the host retry leader +/// discovery promptly rather than backing off as it would from an outage. +pub const STANDBY_REFUSAL_METADATA: &str = "x-openshell-supervisor-host"; +pub const STANDBY_REFUSAL_VALUE: &str = "standby"; /// Most instances one host status report may describe: every assignment plus /// instances still being removed. pub const MAX_REPORTED_INSTANCES: usize = 4 * MAX_ASSIGNED_INSTANCES; @@ -92,10 +102,12 @@ pub enum ObservedInstance { /// No host session is connected. NoHost, /// A host is connected, but this gateway has not recovered the route's - /// assignment yet or the host has not reconciled the snapshot assigning it. + /// assignment yet or the host has not reconciled the snapshot assigning + /// or excluding it. Pending, /// The host reconciled to the snapshot assigning this route and does - /// not host it. + /// not host it, or to one omitting a route the latest complete recovery + /// left unassigned. Absent, /// The host's latest report of the instance. Reported(SandboxInstanceStatus), @@ -141,6 +153,10 @@ struct Desired { assignments: BTreeMap, /// Edits made before initialization, applied over the first full list. pending: BTreeMap>, + /// Routes the latest complete recovery listed but left unassigned (no + /// launch material, or beyond capacity), each with the first revision + /// whose snapshot omits it. Assigning or withdrawing a route clears it. + excluded: BTreeMap, } impl Desired { @@ -237,6 +253,35 @@ pub enum AssignmentError { /// The change would assign more instances than one host runs. #[error("the shared supervisor hosts at most {MAX_ASSIGNED_INSTANCES} sandbox instances")] Capacity, + /// This replica does not publish assignments, so a full resync begun in + /// an ended term is discarded. + #[error("this gateway replica does not publish shared supervisor assignments")] + Standby, +} + +/// The refusal a replica that does not publish gives an attaching host. +fn standby_refusal() -> Status { + let mut metadata = tonic::metadata::MetadataMap::new(); + metadata.insert( + STANDBY_REFUSAL_METADATA, + tonic::metadata::MetadataValue::from_static(STANDBY_REFUSAL_VALUE), + ); + Status::with_metadata( + tonic::Code::Unavailable, + "reconnect to the gateway reconciler leader", + metadata, + ) +} + +/// Whether `status` is a standby replica's refusal of `Attach`, as opposed to +/// a transport or gateway failure. +#[must_use] +pub fn is_standby_refusal(status: &Status) -> bool { + status.code() == tonic::Code::Unavailable + && status + .metadata() + .get(STANDBY_REFUSAL_METADATA) + .is_some_and(|value| value.to_str().ok() == Some(STANDBY_REFUSAL_VALUE)) } fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { @@ -293,11 +338,21 @@ impl SupervisorHostController { /// Revoke publication before releasing the lease or waiting for worker /// cleanup. Existing sandbox instances survive their stream closing. + /// The term's desired state is discarded and its revision fenced, so a + /// standby holds none and a recovery begun in the term cannot + /// reinitialize it. pub fn stop_publishing(&self) { let mut publisher = lock(&self.inner.publisher); *publisher = Publisher::Standby; + let mut desired = lock(&self.inner.desired); + let revision = desired.revision + 1; + *desired = Desired { + revision, + ..Desired::default() + }; Self::disconnect(&mut lock(&self.inner.observed)); self.inner.credential.send_replace(None); + self.inner.revision.send_replace(revision); } fn disconnect(observed: &mut Observed) { @@ -327,6 +382,7 @@ impl SupervisorHostController { desired .assignments .keys() + .chain(desired.excluded.keys()) .map(|route| { let value = match Self::observe(&desired, &observed, route) { ObservedInstance::Reported(status) => SharedObservation::Reported( @@ -390,19 +446,32 @@ impl SupervisorHostController { } /// Replace the whole desired set with `assignments`, derived from - /// durable records while the revision was `baseline`. + /// durable records while the revision was `baseline`; `excluded` names + /// the listed routes that are deliberately left unassigned. /// /// Initializes the controller; edits made before initialization are /// applied on top of the list. The replacement is compare-and-replace: /// when the revision moved past `baseline` while /// the list was being derived, nothing changes and the conflict is - /// returned so the caller derives the list again. + /// returned so the caller derives the list again. Only the publisher + /// replaces its set. + /// + /// Only listed candidates are selected. A list beyond the host's + /// capacity keeps those currently assigned or that the host reports + /// running, then fills the rest in route order; the remainder is + /// excluded like `excluded`. Until this term's host has sent its initial + /// report it is refused, since any cut could stop a running instance. pub fn replace_assignments( &self, baseline: u64, assignments: impl IntoIterator, + excluded: impl IntoIterator, ) -> Result { + let publisher = lock(&self.inner.publisher); let mut desired = lock(&self.inner.desired); + if !publisher.active() { + return Err(AssignmentError::Standby); + } if desired.revision != baseline { return Err(AssignmentError::Conflict { baseline, @@ -425,8 +494,42 @@ impl SupervisorHostController { } } } + let mut excluded = excluded + .into_iter() + .filter(|route| !resolved.contains_key(route) && !desired.pending.contains_key(route)) + .collect::>(); if resolved.len() > MAX_ASSIGNED_INSTANCES { - return Err(AssignmentError::Capacity); + let observed = lock(&self.inner.observed); + let Some(status) = &observed.status else { + return Err(AssignmentError::Capacity); + }; + let running = status + .instances + .iter() + .filter(|instance| { + !SandboxInstanceStatusReport::from_proto(instance) + .phase + .is_terminal() + }) + .map(|instance| instance.route.as_str()) + .collect::>(); + // A stable sort keeps route order within each tier. + let mut ranked = resolved.keys().cloned().collect::>(); + ranked.sort_by_key(|route| { + ( + !desired.assignments.contains_key(route), + !running.contains(route.as_str()), + ) + }); + let overflow = ranked.split_off(MAX_ASSIGNED_INSTANCES); + tracing::warn!( + unassigned = overflow.len(), + "shared sandbox instances exceed the supervisor's capacity; leaving the excess unassigned" + ); + for route in overflow { + resolved.remove(&route); + excluded.insert(route); + } } desired.pending.clear(); let next = desired.revision + 1; @@ -444,14 +547,33 @@ impl SupervisorHostController { desired.revision = next; self.inner.revision.send_replace(next); } - Ok(desired.revision) + // A route that stays excluded keeps the revision that first omitted + // it, so an unchanged recovery needs no new snapshot or ack. + let revision = desired.revision; + let previous = std::mem::take(&mut desired.excluded); + desired.excluded = excluded + .into_iter() + .map(|route| { + let since = previous.get(&route).copied().unwrap_or(revision); + (route, since) + }) + .collect(); + Ok(revision) } /// Assign one instance. Before initialization the edit is kept until the /// first full list arrives. Refused when it would exceed the host's /// capacity, so the caller fails its create before anything runs. + /// + /// A replica that does not publish keeps nothing: the persisted launch + /// material is the assignment, and the publisher's recovery admits it or + /// leaves it unassigned beyond capacity. pub fn assign(&self, assignment: SandboxInstanceAssignment) -> Result<(), AssignmentError> { + let publisher = lock(&self.inner.publisher); let mut desired = lock(&self.inner.desired); + if !publisher.active() { + return Ok(()); + } if !desired.initialized { let pending_additions = desired .pending @@ -473,6 +595,7 @@ impl SupervisorHostController { { return Err(AssignmentError::Capacity); } + desired.excluded.remove(&assignment.route); let next = desired.revision + 1; if desired.put(assignment, next) { desired.revision = next; @@ -481,18 +604,28 @@ impl SupervisorHostController { Ok(()) } - /// Withdraw one instance so the host stops it. + /// Withdraw one instance so the host stops it. A replica that does not + /// publish keeps nothing; the publisher's recovery discovers the stop. + /// + /// After initialization every withdrawal advances the revision, even for + /// a route that is excluded or not yet discovered, so a recovery listed + /// before it conflicts instead of admitting the withdrawn generation. No + /// tombstone is kept: the next recovery lists the durable records again. pub fn unassign(&self, route: &str) { + let publisher = lock(&self.inner.publisher); let mut desired = lock(&self.inner.desired); + if !publisher.active() { + return; + } if !desired.initialized { desired.pending.insert(route.to_string(), None); return; } - if desired.assignments.remove(route).is_some() { - desired.revision += 1; - let revision = desired.revision; - self.inner.revision.send_replace(revision); - } + desired.excluded.remove(route); + desired.assignments.remove(route); + desired.revision += 1; + let revision = desired.revision; + self.inner.revision.send_replace(revision); } /// Whether a host session is currently connected. @@ -531,15 +664,22 @@ impl SupervisorHostController { if observed.session.is_none() { return ObservedInstance::NoHost; } - let Some(assigned) = desired + let since = match desired .assignments .get(route) .filter(|_| desired.initialized) - else { - // A create can have reached another replica before this leader's - // recovery discovers its durable assignment. Never suspend that - // workload based on a snapshot that did not assign it. - return ObservedInstance::Pending; + { + Some(assigned) => assigned.revision, + // The latest complete recovery listed the route and left it + // unassigned; the host's ack of a snapshot omitting it is a verdict. + None => match desired.excluded.get(route) { + Some(since) => *since, + // A create can have reached another replica before this + // leader's recovery discovers its durable assignment. Never + // suspend that workload based on a snapshot that did not + // assign it. + None => return ObservedInstance::Pending, + }, }; let Some(status) = &observed.status else { return ObservedInstance::Pending; @@ -551,7 +691,7 @@ impl SupervisorHostController { { return ObservedInstance::Reported(instance.clone()); } - if assigned.revision > status.applied_revision { + if since > status.applied_revision { ObservedInstance::Pending } else { ObservedInstance::Absent @@ -566,9 +706,7 @@ impl SupervisorHostController { mut inbound: Streaming, ) -> Result>, Status> { if !self.is_publisher() { - return Err(Status::unavailable( - "reconnect to the gateway reconciler leader", - )); + return Err(standby_refusal()); } let hello = match tokio::time::timeout(HELLO_TIMEOUT, inbound.next()).await { Ok(Some(Ok(HostMessage { @@ -623,16 +761,14 @@ impl SupervisorHostController { /// Start a session for `hello`, superseding the current one. A hello from /// the current host process with a connection epoch that is not newer is /// a delayed or replayed connection and is refused. - fn begin_session( + pub(crate) fn begin_session( &self, identity: HostIdentity, hello: &HostHello, ) -> Result<(u64, watch::Receiver), Status> { let publisher = lock(&self.inner.publisher); if !publisher.active() { - return Err(Status::unavailable( - "reconnect to the gateway reconciler leader", - )); + return Err(standby_refusal()); } let mut observed = lock(&self.inner.observed); if let Some(current) = &observed.session @@ -695,7 +831,7 @@ impl SupervisorHostController { } /// Note that `revision` was sent on session `id`. - fn record_sent(&self, id: u64, revision: u64) { + pub(crate) fn record_sent(&self, id: u64, revision: u64) { let mut observed = lock(&self.inner.observed); if let Some(session) = observed.session.as_mut().filter(|session| session.id == id) { session.sent_revision = session.sent_revision.max(revision); @@ -705,7 +841,7 @@ impl SupervisorHostController { /// Record a report from session `id`. Reports from other sessions, acks /// of revisions never sent on this session, regressing acks, and /// oversized reports are dropped. - fn record_status(&self, id: u64, status: HostStatus) -> Result<(), &'static str> { + pub(crate) fn record_status(&self, id: u64, status: HostStatus) -> Result<(), &'static str> { let mut observed = lock(&self.inner.observed); let Some(session) = observed.session.as_ref().filter(|session| session.id == id) else { return Err("report from a superseded supervisor host session"); @@ -918,6 +1054,7 @@ mod tests { assignment("rt-stopped-before-list", b"old"), assignment("rt-listed", b"b"), ], + [], ), Ok(1) ); @@ -941,6 +1078,7 @@ mod tests { assignment("rt-created-before-list", b"a"), assignment("rt-listed", b"b"), ], + [], ), Ok(1) ); @@ -958,6 +1096,7 @@ mod tests { assignment("rt-created-before-list", b"a"), assignment("rt-listed", b"b"), ], + [], ), Err(AssignmentError::Conflict { baseline: 1, @@ -965,13 +1104,13 @@ mod tests { }) ); assert!(controller.assignment("rt-listed").is_none()); - assert_eq!(controller.replace_assignments(2, []), Ok(3)); + assert_eq!(controller.replace_assignments(2, [], []), Ok(3)); assert!(controller.snapshot().unwrap().instances.is_empty()); // The gateway refuses the assignment its host could not accept. let full = (0..MAX_ASSIGNED_INSTANCES).map(|index| assignment(&format!("rt-{index}"), b"x")); - assert_eq!(controller.replace_assignments(3, full), Ok(4)); + assert_eq!(controller.replace_assignments(3, full, []), Ok(4)); assert_eq!( controller.assign(assignment("rt-one-too-many", b"x")), Err(AssignmentError::Capacity) @@ -983,10 +1122,20 @@ mod tests { async fn leadership_fences_old_sessions_recovery_and_cached_observations() { let controller = SupervisorHostController::new(); controller - .replace_assignments(0, [assignment("rt-old", b"old")]) + .replace_assignments(0, [assignment("rt-old", b"old")], []) .unwrap(); let old_baseline = controller.revision(); controller.require_leadership(); + assert!(!controller.is_initialized()); + // A standby keeps no desired state, so creates and deletes beyond the + // host's capacity are left to the leader's durable recovery. + for index in 0..=MAX_ASSIGNED_INSTANCES { + controller + .assign(assignment(&format!("rt-standby-{index}"), b"s")) + .unwrap(); + controller.unassign(&format!("rt-standby-deleted-{index}")); + } + assert!(lock(&controller.inner.desired).pending.is_empty()); let identity = HostIdentity { subject: "openshell-supervisor:test".to_string(), token_id: "token".to_string(), @@ -996,24 +1145,30 @@ mod tests { instance_id: "boot".to_string(), connection_epoch: 1, }; - assert_eq!( - controller - .begin_session(identity.clone(), &hello) - .unwrap_err() - .code(), - tonic::Code::Unavailable - ); + // The refusal is marked, and the marker survives gRPC trailers. + let refusal = controller + .begin_session(identity.clone(), &hello) + .unwrap_err(); + let mut trailers = tonic::codegen::http::HeaderMap::new(); + refusal.add_header(&mut trailers).unwrap(); + assert!(is_standby_refusal( + &Status::from_header_map(&trailers).unwrap() + )); controller.start_publishing(Instant::now() + Duration::from_secs(30)); assert!(!controller.is_initialized()); assert!(controller.assignment("rt-old").is_none()); - assert!(controller.replace_assignments(old_baseline, []).is_err()); + assert!( + controller + .replace_assignments(old_baseline, [], []) + .is_err() + ); // No authoritative empty snapshot before the new term's full recovery. assert!(controller.snapshot().is_none()); let (session, revoked) = controller.begin_session(identity.clone(), &hello).unwrap(); controller.record_status(session, status(0, &[])).unwrap(); assert_eq!(controller.observed("rt-new"), ObservedInstance::Pending); controller - .replace_assignments(controller.revision(), [assignment("rt-new", b"new")]) + .replace_assignments(controller.revision(), [assignment("rt-new", b"new")], []) .unwrap(); controller.record_sent(session, controller.revision()); controller @@ -1041,39 +1196,159 @@ mod tests { tokio::time::advance(Duration::from_secs(11)).await; assert!(controller.is_publisher()); assert_eq!(standby.observed("rt-new"), ObservedInstance::NoHost); + // Demotion discards the term's desired state, and a recovery begun in + // the term cannot reinitialize the standby. + let stale_baseline = controller.revision(); controller.stop_publishing(); assert!(*revoked.borrow()); - assert!(controller.record_status(session, status(0, &[])).is_err()); - assert_eq!(controller.observed("rt-new"), ObservedInstance::NoHost); + assert!(!controller.is_initialized()); + assert!(controller.assignment("rt-new").is_none()); assert_eq!( - controller - .begin_session(identity.clone(), &hello) - .unwrap_err() - .code(), - tonic::Code::Unavailable + controller.replace_assignments(stale_baseline, [assignment("rt-new", b"new")], []), + Err(AssignmentError::Standby) ); + assert!(!controller.is_initialized()); + assert!(controller.record_status(session, status(0, &[])).is_err()); + assert_eq!(controller.observed("rt-new"), ObservedInstance::NoHost); + let refusal = controller + .begin_session(identity.clone(), &hello) + .unwrap_err(); + assert!(is_standby_refusal(&refusal)); controller.start_publishing(Instant::now() + Duration::from_secs(30)); assert!(!controller.is_initialized()); assert!(controller.snapshot().is_none()); + let expired_baseline = controller.revision(); tokio::time::advance(Duration::from_secs(31)).await; - // A late renewal cannot revive an expired term. + // A late renewal cannot revive an expired term, and an expired + // publisher edits nothing even before its demotion runs. controller.renew_publishing(Instant::now() + Duration::from_secs(30)); assert!(!controller.is_publisher()); + controller.assign(assignment("rt-late", b"l")).unwrap(); + controller.unassign("rt-late-stop"); + assert!(lock(&controller.inner.desired).pending.is_empty()); assert_eq!( + controller.replace_assignments(expired_baseline, [], []), + Err(AssignmentError::Standby) + ); + assert!(!controller.is_initialized()); + assert!(is_standby_refusal( + &controller.begin_session(identity, &hello).unwrap_err() + )); + } + + #[test] + fn recovery_exclusions_become_absent_once_acknowledged_and_bound_capacity() { + let controller = SupervisorHostController::new(); + let identity = HostIdentity { + subject: "openshell-supervisor:test".to_string(), + token_id: "token".to_string(), + expires_at: i64::MAX, + }; + let hello = HostHello { + instance_id: "boot".to_string(), + connection_epoch: 1, + }; + let (session, _) = controller.begin_session(identity, &hello).unwrap(); + let listed = || { + (0..MAX_ASSIGNED_INSTANCES + 2) + .map(|index| assignment(&format!("rt-{index:03}"), b"x")) + .collect::>() + }; + // Beyond capacity, nothing is cut until the host reports what it runs. + assert_eq!( + controller.replace_assignments(0, listed(), ["rt-gone".to_string()]), + Err(AssignmentError::Capacity) + ); + assert!(!controller.is_initialized()); + + // The running instance stays; the excess is excluded in route order. + controller + .record_status(session, status(0, &["rt-257"])) + .unwrap(); + assert_eq!( + controller.replace_assignments(0, listed(), ["rt-gone".to_string()]), + Ok(1) + ); + assert_eq!( + controller.snapshot().unwrap().instances.len(), + MAX_ASSIGNED_INSTANCES + ); + assert!(controller.assignment("rt-257").is_some()); + let excluded = ["rt-gone", "rt-255", "rt-256"]; + for route in excluded { + assert_eq!(controller.observed(route), ObservedInstance::Pending); + } + controller.record_sent(session, 1); + controller + .record_status(session, status(1, &["rt-257"])) + .unwrap(); + for route in excluded { + assert_eq!(controller.observed(route), ObservedInstance::Absent); + } + assert_eq!( + controller.observed("rt-undiscovered"), + ObservedInstance::Pending + ); + let observations = controller.shared_observations().unwrap(); + assert_eq!( + observations.get("rt-gone"), + Some(&SharedObservation::Absent) + ); + assert!(!observations.contains_key("rt-undiscovered")); + + // A later recovery keeps current assignments over a new route that + // sorts first, and an unchanged set needs no new snapshot or ack. + let revision = controller.inner.revision.subscribe(); + let mut relisted = listed(); + relisted.push(assignment("rt-0", b"x")); + assert_eq!( + controller.replace_assignments(1, relisted, ["rt-gone".to_string()]), + Ok(1) + ); + assert!(!revision.has_changed().unwrap()); + assert_eq!(controller.observed("rt-0"), ObservedInstance::Absent); + assert_eq!(controller.observed("rt-gone"), ObservedInstance::Absent); + + // Withdrawing or reassigning an excluded route clears its verdict, a + // raced recovery records nothing, and restored material assigns it. + // A withdrawal fences recovery listed before it, so a stale list + // cannot re-admit an excluded or undiscovered withdrawn route. + let stale = controller.revision(); + controller.unassign("rt-255"); + controller.unassign("rt-undiscovered"); + assert_eq!(controller.observed("rt-255"), ObservedInstance::Pending); + let stale_list = [ + assignment("rt-255", b"x"), + assignment("rt-undiscovered", b"x"), + ]; + assert!(matches!( + controller.replace_assignments(stale, stale_list, []), + Err(AssignmentError::Conflict { .. }) + )); + assert!(controller.assignment("rt-255").is_none()); + assert!(controller.assignment("rt-undiscovered").is_none()); + controller.unassign("rt-000"); + controller.assign(assignment("rt-256", b"x")).unwrap(); + assert_eq!(controller.observed("rt-256"), ObservedInstance::Pending); + assert!( controller - .begin_session(identity, &hello) - .unwrap_err() - .code(), - tonic::Code::Unavailable + .replace_assignments(1, [], ["rt-raced".to_string()]) + .is_err() ); + assert_eq!(controller.observed("rt-raced"), ObservedInstance::Pending); + controller + .replace_assignments(controller.revision(), [assignment("rt-gone", b"x")], []) + .unwrap(); + assert_eq!(controller.observed("rt-gone"), ObservedInstance::Pending); + assert_eq!(controller.observed("rt-0"), ObservedInstance::Pending); } #[test] fn sessions_fence_stale_streams_epochs_and_acks() { let controller = SupervisorHostController::new(); controller - .replace_assignments(0, [assignment("rt-a", b"a")]) + .replace_assignments(0, [assignment("rt-a", b"a")], []) .unwrap(); assert_eq!(controller.observed("rt-a"), ObservedInstance::NoHost); diff --git a/crates/openshell-supervisor-control/src/coordinator.rs b/crates/openshell-supervisor-control/src/coordinator.rs index cedbb42bc2..16b9443216 100644 --- a/crates/openshell-supervisor-control/src/coordinator.rs +++ b/crates/openshell-supervisor-control/src/coordinator.rs @@ -40,8 +40,9 @@ pub enum LaunchError { /// The driver could not persist the launch material; nothing was assigned. #[error("{0}")] Persist(E), - /// The host is at capacity. The material was persisted; the caller rolls - /// the generation back before its workload runs. + /// The publishing gateway's host is at capacity. The material was + /// persisted; the caller rolls the generation back before its workload + /// runs. #[error("{0}")] Refused(AssignmentError), } @@ -68,11 +69,12 @@ pub struct RecoverableInstance { pub enum RecoveryOutcome { /// The desired set matches the records; carries the revision in force. Published(u64), - /// The desired set changed while the records were listed; the next pass - /// derives them again. + /// The desired set changed, or this replica stopped publishing, while + /// the records were listed; the next pass derives them again. Conflict, /// Launch material could not be read or does not match its record, or the - /// set exceeds the host's capacity; the previous set stays published. + /// set exceeds the host's capacity before the host has reported what it + /// runs; the previous set stays published. Kept, } @@ -84,8 +86,9 @@ pub enum InstanceSupervision { /// The host has not reconciled to the assigning snapshot yet, or hosts /// the instance but it is still starting. Degraded, - /// The host reconciled past the assignment and does not host the - /// instance, hosts it for another sandbox, or reports it ended. + /// The host reconciled past the assignment, or past the recovery that + /// left the instance unassigned, and does not host the instance, hosts + /// it for another sandbox, or reports it ended. Unavailable, /// No host is connected, so nothing is known. Unknown, @@ -127,8 +130,12 @@ impl SharedSupervision { /// assign the instance so the host starts it. The launch material is /// persisted first so a restarted gateway recovers the assignment from the /// generation's own records. An identical assignment changes nothing. - /// Capacity is refused as [`LaunchError::Refused`], so the caller fails - /// its create before the workload runs. + /// On the publishing gateway, capacity is refused as + /// [`LaunchError::Refused`], so the caller fails its create before the + /// workload runs. On a replicated gateway's standby, admission is + /// eventual: the leader's recovery assigns the persisted instance, or + /// leaves it unassigned beyond capacity so its supervision becomes + /// unavailable. pub async fn launch( &self, registration: &SandboxInstanceRegistration, @@ -149,7 +156,8 @@ impl SharedSupervision { } /// Withdraw a generation's sandbox instance so the host stops it. Nothing - /// happens for a route that is not assigned. + /// happens for a route that is not assigned; a standby leaves the stop to + /// the leader's recovery from durable records. pub fn withdraw(&self, route: &str) { self.controller.unassign(route); } @@ -175,8 +183,10 @@ impl SharedSupervision { /// The first completed pass initializes the desired set; until then /// nothing is sent to the host. A failed or mismatched read keeps the /// previous set, because an incomplete set would stop running instances. - /// A pass that raced a lifecycle edit is refused and derived again next - /// time. + /// A pass that raced a lifecycle edit or the end of this replica's term + /// is refused and derived again next time. A candidate whose material is + /// gone, or that exceeds the host's capacity, is left unassigned and + /// becomes absent once the host acknowledges a snapshot omitting it. pub async fn complete_recovery( &self, baseline: RecoveryBaseline, @@ -189,6 +199,7 @@ impl SharedSupervision { F: Future>, E>> + Send, { let mut assignments = Vec::with_capacity(candidates.len()); + let mut excluded = Vec::new(); for candidate in &candidates { if let Some(assignment) = self.controller.assignment(&candidate.route) { assignments.push(assignment); @@ -202,6 +213,7 @@ impl SharedSupervision { route = %candidate.route, "shared sandbox instance has no launch material; it stays unassigned" ); + excluded.push(candidate.route.clone()); continue; } Err(error) => { @@ -229,10 +241,12 @@ impl SharedSupervision { } match self .controller - .replace_assignments(baseline.revision, assignments) + .replace_assignments(baseline.revision, assignments, excluded) { Ok(revision) => RecoveryOutcome::Published(revision), - Err(AssignmentError::Conflict { .. }) => RecoveryOutcome::Conflict, + Err(AssignmentError::Conflict { .. } | AssignmentError::Standby) => { + RecoveryOutcome::Conflict + } Err(error) => { tracing::warn!(%error, "could not publish the shared sandbox instances; keeping the previous assignments"); RecoveryOutcome::Kept @@ -300,7 +314,8 @@ fn recovered_assignment( #[cfg(test)] mod tests { use super::*; - use crate::proto::SandboxInstanceAssignment; + use crate::controller::HostIdentity; + use crate::proto::{HostHello, HostStatus, SandboxInstanceAssignment}; fn candidate(route: &str, sandbox_id: &str) -> RecoverableInstance<&'static str> { RecoverableInstance { @@ -369,6 +384,56 @@ mod tests { Some(assigned.clone()) ); + // The skipped record is pending until the host acknowledges the + // snapshot omitting it, then unavailable, unlike an undiscovered + // route. A later failed read is no evidence and changes nothing. + let controller = supervision.controller(); + let (session, _) = controller + .begin_session( + HostIdentity { + subject: "openshell-supervisor:test".to_string(), + token_id: "token".to_string(), + expires_at: i64::MAX, + }, + &HostHello { + instance_id: "boot".to_string(), + connection_epoch: 1, + }, + ) + .unwrap(); + let acked = |applied_revision| HostStatus { + applied_revision, + instances: Vec::new(), + }; + controller.record_status(session, acked(0)).unwrap(); + assert_eq!( + supervision.supervision("rt-gone-0123456789", "sandbox-d"), + InstanceSupervision::Degraded + ); + controller.record_sent(session, 1); + controller.record_status(session, acked(1)).unwrap(); + assert_eq!( + supervision.supervision("rt-gone-0123456789", "sandbox-d"), + InstanceSupervision::Unavailable + ); + assert_eq!( + supervision.supervision("rt-undiscovered-0123456789", "sandbox-e"), + InstanceSupervision::Degraded + ); + let baseline = supervision.begin_recovery(); + let outcome = supervision + .complete_recovery( + baseline, + vec![candidate("rt-gone-0123456789", "sandbox-d")], + |_record| async { Err::>, _>("boom") }, + ) + .await; + assert_eq!(outcome, RecoveryOutcome::Kept); + assert_eq!( + supervision.supervision("rt-gone-0123456789", "sandbox-d"), + InstanceSupervision::Unavailable + ); + // A baseline read before a lifecycle edit is refused, and the edit stands. let baseline = supervision.begin_recovery(); supervision.withdraw(&assigned.route); diff --git a/crates/openshell-supervisor-control/src/lib.rs b/crates/openshell-supervisor-control/src/lib.rs index 077c8b6594..ae62365067 100644 --- a/crates/openshell-supervisor-control/src/lib.rs +++ b/crates/openshell-supervisor-control/src/lib.rs @@ -22,7 +22,8 @@ //! //! The private wire protocol is generated in `openshell-core` and re-exported //! as [`proto`]. The supervisor process consumes the registration and status -//! types; it never links the controller or coordinator. +//! types and the private standby refusal marker ([`is_standby_refusal`]); it +//! never uses the controller's desired state or the coordinator. pub mod controller; pub mod coordinator; @@ -38,8 +39,8 @@ pub mod proto { pub use controller::{ AssignmentError, HOST_HEARTBEAT_INTERVAL, HostIdentity, MAX_ASSIGNED_INSTANCES, - MAX_REPORTED_INSTANCES, ObservedInstance, SharedObservation, SupervisorCredential, - SupervisorHostController, + MAX_REPORTED_INSTANCES, ObservedInstance, STANDBY_REFUSAL_METADATA, STANDBY_REFUSAL_VALUE, + SharedObservation, SupervisorCredential, SupervisorHostController, is_standby_refusal, }; pub use coordinator::{ InstanceSupervision, LaunchError, RecoverableInstance, RecoveryBaseline, RecoveryOutcome, diff --git a/crates/openshell-supervisor/README.md b/crates/openshell-supervisor/README.md index 9083f62d74..1e99003ff4 100644 --- a/crates/openshell-supervisor/README.md +++ b/crates/openshell-supervisor/README.md @@ -171,7 +171,8 @@ control listener: it dials the gateway and receives its assignments from it. - **Disconnects.** Losing the gateway stream never stops an instance. The host reconnects with backoff, reports what it runs, and applies the first snapshot of the new session; only a snapshot that omits an instance, or - process shutdown, stops it. + process shutdown, stops it. A standby gateway replica's marked refusal is + retried after a short jittered delay instead, to reach the leader. - **Runtime connections.** One routed listener (`--runtime-listen`) accepts every instance's runtime-initiated Sandbox Protocol connections; see `openshell-sandbox-backend` for the routing preface. diff --git a/crates/openshell-supervisor/src/shared_host.rs b/crates/openshell-supervisor/src/shared_host.rs index a0dfd133ee..51d5eb9808 100644 --- a/crates/openshell-supervisor/src/shared_host.rs +++ b/crates/openshell-supervisor/src/shared_host.rs @@ -55,6 +55,11 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const SEND_TIMEOUT: Duration = Duration::from_secs(30); const RECONNECT_BACKOFF_MIN: Duration = Duration::from_secs(1); const RECONNECT_BACKOFF_MAX: Duration = Duration::from_secs(30); +/// Delay, plus up to [`STANDBY_RETRY_JITTER`], before reconnecting after a +/// standby gateway replica refused the host, so a load-balanced endpoint +/// reaches the reconciler leader promptly without synchronized retries. +const STANDBY_RETRY_MIN: Duration = Duration::from_millis(250); +const STANDBY_RETRY_JITTER: Duration = Duration::from_millis(750); /// Host-to-gateway messages queued before the stream applies backpressure. const OUTBOUND_QUEUE: usize = 16; /// Most sandbox instances one snapshot may assign to a host; the gateway @@ -492,11 +497,38 @@ enum StreamEnd { /// The gateway rejected or closed the stream; reconnect after a backoff. /// `accepted` records whether the gateway had accepted the host. Lost { accepted: bool }, + /// A gateway replica that does not publish assignments refused the host; + /// reconnect promptly on a fresh connection to reach the leader. + Standby, +} + +/// Delay before the next connection attempt. A standby refusal waits a +/// short jittered delay and leaves the backoff alone; any other loss waits +/// out the backoff, which restarts from its minimum once the gateway had +/// accepted the host and doubles up to its cap otherwise. +fn reconnect_delay(end: &StreamEnd, backoff: &mut Duration) -> Duration { + match end { + StreamEnd::Standby => { + let jitter_ms = u64::try_from(STANDBY_RETRY_JITTER.as_millis()) + .unwrap_or(1) + .max(1); + let jitter = uuid::Uuid::new_v4().as_u64_pair().0 % jitter_ms; + STANDBY_RETRY_MIN + Duration::from_millis(jitter) + } + StreamEnd::Lost { accepted } => { + if *accepted { + *backoff = RECONNECT_BACKOFF_MIN; + } + let delay = *backoff; + *backoff = (*backoff * 2).min(RECONNECT_BACKOFF_MAX); + delay + } + } } /// Keep one assignment stream open to the gateway until `shutdown` -/// completes, reconnecting with backoff. Hosted instances run on across -/// disconnects. +/// completes, reconnecting with backoff, or promptly after a standby +/// replica's refusal. Hosted instances run on across disconnects. pub async fn attach_to_gateway( host: Arc, link: GatewayLink, @@ -512,22 +544,25 @@ pub async fn attach_to_gateway( instance_id: instance_id.clone(), connection_epoch, }; - let StreamEnd::Lost { accepted } = tokio::select! { + let end = tokio::select! { end = run_stream(&host, &link, hello) => end, () = &mut shutdown => return, }; - if accepted { - backoff = RECONNECT_BACKOFF_MIN; + let delay = reconnect_delay(&end, &mut backoff); + match end { + StreamEnd::Standby => tracing::debug!( + retry_in_ms = delay.as_millis(), + "gateway replica is a standby; reconnecting to reach the reconciler leader" + ), + StreamEnd::Lost { .. } => tracing::info!( + retry_in_secs = delay.as_secs(), + "gateway assignment stream lost; hosted sandbox instances keep running" + ), } - tracing::info!( - retry_in_secs = backoff.as_secs(), - "gateway assignment stream lost; hosted sandbox instances keep running" - ); tokio::select! { - () = tokio::time::sleep(backoff) => {} + () = tokio::time::sleep(delay) => {} () = &mut shutdown => return, } - backoff = (backoff * 2).min(RECONNECT_BACKOFF_MAX); } } @@ -580,6 +615,9 @@ async fn run_stream(host: &Arc, link: &GatewayLink, hello: HostHello let attach = client.attach(tokio_stream::wrappers::ReceiverStream::new(rx)); let mut inbound = match tokio::time::timeout(CONNECT_TIMEOUT, attach).await { Ok(Ok(response)) => response.into_inner(), + Ok(Err(status)) if openshell_supervisor_control::is_standby_refusal(&status) => { + return StreamEnd::Standby; + } Ok(Err(status)) => { tracing::warn!(%status, "gateway refused the supervisor host stream"); return lost(false); @@ -722,3 +760,39 @@ async fn run_stream(host: &Arc, link: &GatewayLink, hello: HostHello } end } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn standby_refusals_retry_promptly_and_other_losses_back_off() { + let mut backoff = RECONNECT_BACKOFF_MIN; + let lost = StreamEnd::Lost { accepted: false }; + assert_eq!(reconnect_delay(&lost, &mut backoff), Duration::from_secs(1)); + assert_eq!(reconnect_delay(&lost, &mut backoff), Duration::from_secs(2)); + + // Standby refusals stay short, jittered, and leave the outage backoff + // where it was. + for _ in 0..32 { + let delay = reconnect_delay(&StreamEnd::Standby, &mut backoff); + assert!(delay >= STANDBY_RETRY_MIN); + assert!(delay < STANDBY_RETRY_MIN + STANDBY_RETRY_JITTER); + } + assert_eq!(reconnect_delay(&lost, &mut backoff), Duration::from_secs(4)); + for _ in 0..8 { + reconnect_delay(&lost, &mut backoff); + } + assert_eq!(reconnect_delay(&lost, &mut backoff), RECONNECT_BACKOFF_MAX); + assert_eq!( + reconnect_delay(&StreamEnd::Lost { accepted: true }, &mut backoff), + RECONNECT_BACKOFF_MIN + ); + + // Only the gateway's marked refusal counts as a standby; a bare + // UNAVAILABLE is an outage. + assert!(!openshell_supervisor_control::is_standby_refusal( + &tonic::Status::unavailable("reconnect to the gateway reconciler leader") + )); + } +}