diff --git a/architecture/gateway.md b/architecture/gateway.md index 204c0dc4d4..c220306452 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -332,6 +332,11 @@ deadline. Each retry re-reads the owner record, so a supervisor reconnect or heartbeat can surface a new owner; if no fresh reachable owner appears before the deadline, the client operation fails rather than electing an owner itself. +Nothing redistributes established sessions, so after a rolling restart the last +surviving replica holds most sessions and a new replica serves none until +sandboxes reconnect. That skew decays only as sandboxes churn. Client traffic +stays correct throughout because a non-owner relays to the owner. + File upload and download use tar-over-SSH through the same relay path. A gateway pod termination drops the active SSH proxy byte stream, so the CLI retries the whole sync operation with a fresh SSH session instead of attempting mid-stream @@ -346,10 +351,12 @@ also trust the chart CA, present the chart-generated client certificate for mTLS, and verify the stable gateway Service DNS name even when connecting to a Deployment pod IP. -`WatchSandbox` uses the local update bus for same-replica writes. One shared -poller per gateway observes resource-version changes made by other replicas and -feeds that bus for all local watchers, avoiding a database poll per client -stream. +`WatchSandbox` uses the local update bus for same-replica writes. On +multi-replica backends one shared poller per gateway observes resource-version +changes made by other replicas and feeds that bus for all local watchers, +avoiding a database poll per client stream. SQLite deployments do not run the +poller because they are single-replica and the local bus already sees every +write. Mutations whose invariants span sandbox, provider-profile, policy, or provider records take a process-local mutex and a shared PostgreSQL advisory lock. The diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index ad61b683ec..0fedc25b18 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -3281,9 +3281,9 @@ impl ComputeRuntime { else { return Ok(None); }; - let age_ms = openshell_core::time::now_ms() - owner.updated_at_ms; - let ttl_ms = i64::try_from(OWNER_TTL.as_millis()).unwrap_or(i64::MAX); - Ok((age_ms < ttl_ms).then_some(owner.supervisor_instance_id)) + Ok(owner + .is_fresh(OWNER_TTL) + .then_some(owner.supervisor_instance_id)) } async fn set_supervisor_session_state( @@ -3919,8 +3919,9 @@ impl ComputeRuntime { } let sandbox = decode_sandbox_record(¤t_record)?; - let age_ms = - openshell_core::time::now_ms().saturating_sub(current_record.created_at_ms); + let age_ms = openshell_core::time::now_ms() + .saturating_sub(current_record.created_at_ms) + .max(0); if age_ms < grace_ms { return Ok(()); } diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index ddf11690d0..427041b878 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -491,6 +491,27 @@ fn derive_peer_endpoint(config: &Config) -> Option { )) } +/// Reject a plaintext peer endpoint on a gateway that serves TLS. +/// +/// Peer relay traffic carries whole supervisor sessions between replicas. The +/// chart renders a plaintext peer endpoint only when the gateway itself serves +fn validate_peer_endpoint_scheme(config: &Config, peer_endpoint: &str) -> Result<()> { + if peer_endpoint.starts_with("https://") { + return Ok(()); + } + if config.tls.is_some() { + return Err(Error::config(format!( + "gateway peer endpoint {peer_endpoint} is plaintext but this gateway serves TLS; \ + set an https:// OPENSHELL_PEER_ENDPOINT so peer relay traffic is not downgraded" + ))); + } + warn!( + peer_endpoint, + "gateway peer relay traffic is plaintext because this gateway does not serve TLS" + ); + Ok(()) +} + /// Run the `OpenShell` server. /// /// This starts a multiplexed gRPC/HTTP server on the configured bind address. @@ -752,6 +773,20 @@ pub(crate) async fn run_server( ); } + let peer_routing_expected = state.peer_endpoint.is_some() && !state.store.is_single_replica(); + if let Some(peer_endpoint) = state.peer_endpoint.as_deref() + && peer_routing_expected + { + validate_peer_endpoint_scheme(&state.config, peer_endpoint)?; + } + if state.peer_endpoint.is_none() && !state.store.is_single_replica() { + warn!( + "no gateway peer endpoint configured; this replica owns its supervisor sessions but \ + peers cannot reach it. Single-gateway deployments are unaffected; set \ + OPENSHELL_PEER_ENDPOINT on every replica when running more than one." + ); + } + if std::env::var_os("KUBERNETES_SERVICE_HOST").is_some() { let namespace = std::env::var("OPENSHELL_POD_NAMESPACE").ok(); let service_account = std::env::var("OPENSHELL_SERVICE_ACCOUNT_NAME").ok(); @@ -786,6 +821,13 @@ pub(crate) async fn run_server( "gateway peer ServiceAccount TokenReview authentication enabled" ); } + Err(err) if peer_routing_expected => { + return Err(Error::config(format!( + "in-cluster K8s client construction failed ({err}); \ + gateway peer authentication is required because \ + OPENSHELL_PEER_ENDPOINT is configured" + ))); + } Err(err) => warn!( error = %err, "in-cluster K8s client construction failed; \ @@ -793,6 +835,14 @@ pub(crate) async fn run_server( ), } } + _ if peer_routing_expected => { + return Err(Error::config( + "OPENSHELL_POD_NAMESPACE or OPENSHELL_SERVICE_ACCOUNT_NAME missing; \ + both are required for gateway peer authentication because \ + OPENSHELL_PEER_ENDPOINT is configured" + .to_string(), + )); + } _ => { debug!( "OPENSHELL_POD_NAMESPACE or OPENSHELL_SERVICE_ACCOUNT_NAME missing; \ @@ -800,6 +850,12 @@ pub(crate) async fn run_server( ); } } + } else if peer_routing_expected { + return Err(Error::config( + "OPENSHELL_PEER_ENDPOINT is configured but the gateway is not running in a \ + Kubernetes cluster, so gateway peer authentication is unavailable" + .to_string(), + )); } let state = Arc::new(state); @@ -944,12 +1000,16 @@ pub(crate) async fn run_server( } state.compute.spawn_watchers(shutdown_rx.clone()); - sandbox_watch::spawn_store_poller( - store.clone(), - state.sandbox_watch_bus.clone(), - Duration::from_secs(1), - shutdown_rx.clone(), - ); + // The poller exists to observe writes made by other replicas. Single- + // replica backends have none, so it would only add load. + if !store.is_single_replica() { + sandbox_watch::spawn_store_poller( + store.clone(), + state.sandbox_watch_bus.clone(), + sandbox_watch::DEFAULT_STORE_POLL_INTERVAL, + shutdown_rx.clone(), + ); + } ssh_sessions::spawn_session_reaper(store.clone(), Duration::from_hours(1)); supervisor_session::spawn_relay_reaper(state.clone(), Duration::from_secs(30)); provider_refresh::spawn_refresh_worker(state.clone(), Duration::from_mins(1)); @@ -1883,7 +1943,7 @@ mod tests { GatewayListenerScope, MultiplexService, ServerState, TlsAcceptor, allow_plaintext_service_http, bind_gateway_listeners, classify_initial_bytes, configured_compute_driver, extension_token_ttl, is_benign_tls_handshake_failure, - mint_gateway_extension_credential, serve_gateway_listener, + mint_gateway_extension_credential, serve_gateway_listener, validate_peer_endpoint_scheme, }; use openshell_core::{ Config, @@ -1906,6 +1966,40 @@ mod tests { tls_test_utils::generate_test_certs_with_ca, }; + fn tls_enabled_config() -> Config { + Config::new(Some(openshell_core::TlsConfig { + cert_path: "/tmp/cert.pem".into(), + key_path: "/tmp/key.pem".into(), + client_ca_path: None, + require_client_auth: false, + external_cert_path: None, + external_key_path: None, + external_server_names: Vec::new(), + })) + } + + #[test] + fn plaintext_peer_endpoint_is_rejected_on_a_tls_gateway() { + let error = validate_peer_endpoint_scheme(&tls_enabled_config(), "http://10.0.0.1:8080") + .expect_err("plaintext peer endpoint must not be accepted alongside gateway TLS"); + assert!( + error.to_string().contains("plaintext"), + "error should name the downgrade: {error}" + ); + } + + #[test] + fn https_peer_endpoint_is_accepted_on_a_tls_gateway() { + validate_peer_endpoint_scheme(&tls_enabled_config(), "https://10.0.0.1:8080").unwrap(); + } + + #[test] + fn plaintext_peer_endpoint_is_allowed_on_a_plaintext_gateway() { + let config = Config::new(None); + assert!(config.tls.is_none()); + validate_peer_endpoint_scheme(&config, "http://10.0.0.1:8080").unwrap(); + } + static DETECTION_PROBE_ORDER: LazyLock>> = LazyLock::new(|| Mutex::new(Vec::new())); diff --git a/crates/openshell-server/src/persistence/postgres.rs b/crates/openshell-server/src/persistence/postgres.rs index ef613a77bd..d79335b5f5 100644 --- a/crates/openshell-server/src/persistence/postgres.rs +++ b/crates/openshell-server/src/persistence/postgres.rs @@ -40,6 +40,11 @@ pub struct PostgresStore { // 64-bit advisory-lock key space. const CROSS_OBJECT_ADVISORY_LOCK_KEY: i64 = 0x4f50_454e_5348_4c4c; +// Bounds the wait for the cross-object lock. The holder only validates and +// writes, so a wait this long means a stuck replica; failing beats blocking +// every sandbox and provider mutation in the fleet indefinitely. +const CROSS_OBJECT_ADVISORY_LOCK_TIMEOUT: &str = "10s"; + pub(super) struct PostgresAdvisoryLockGuard { // `close_on_drop` is set before this guard is constructed. Closing the // dedicated session releases the session-level advisory lock even when a @@ -114,6 +119,11 @@ impl PostgresStore { ) -> PersistenceResult { let mut connection = self.pool.acquire().await.map_err(|e| map_db_error(&e))?; connection.close_on_drop(); + sqlx::query("SELECT set_config('lock_timeout', $1, false)") + .bind(CROSS_OBJECT_ADVISORY_LOCK_TIMEOUT) + .execute(&mut *connection) + .await + .map_err(|e| map_db_error(&e))?; sqlx::query("SELECT pg_advisory_lock($1)") .bind(CROSS_OBJECT_ADVISORY_LOCK_KEY) .execute(&mut *connection) diff --git a/crates/openshell-server/src/sandbox_watch.rs b/crates/openshell-server/src/sandbox_watch.rs index 0fbe485f77..3a62fce398 100644 --- a/crates/openshell-server/src/sandbox_watch.rs +++ b/crates/openshell-server/src/sandbox_watch.rs @@ -13,6 +13,10 @@ use tonic::Status; use crate::persistence::Store; use openshell_core::proto::Sandbox; +/// How often [`spawn_store_poller`] rechecks watched sandboxes for writes made +/// by other gateway replicas. +pub const DEFAULT_STORE_POLL_INTERVAL: Duration = Duration::from_secs(1); + /// Broadcast bus of sandbox updates keyed by sandbox id. /// /// Producers call [`SandboxWatchBus::notify`] whenever the persisted sandbox record changes. diff --git a/crates/openshell-server/src/service_routing.rs b/crates/openshell-server/src/service_routing.rs index 0b051c6269..306ba7e9ed 100644 --- a/crates/openshell-server/src/service_routing.rs +++ b/crates/openshell-server/src/service_routing.rs @@ -100,8 +100,14 @@ impl ServiceUpstreamPool { let mut inner = self.inner.lock().unwrap(); let entries = inner.idle.get_mut(key)?; entries.retain(|entry| is_reusable(entry, now)); - let index = entries.iter().position(|entry| entry.sender.is_ready())?; - Some(entries.swap_remove(index).sender) + let taken = entries + .iter() + .position(|entry| entry.sender.is_ready()) + .map(|index| entries.swap_remove(index).sender); + if entries.is_empty() { + inner.idle.remove(key); + } + taken } fn put(&self, key: &str, sender: UpstreamSender) { @@ -435,19 +441,34 @@ async fn proxy_to_endpoint( state.service_upstreams.take(&pool_key) }; + let reused = pooled.is_some(); let mut sender = match pooled { Some(sender) => sender, None => open_upstream(&state, &sandbox, &endpoint, target_port, websocket_upgrade).await?, }; let upstream = build_upstream_request(req, target_port, websocket_upgrade)?; - let mut response = sender.send_request(upstream).await.map_err(|err| { - warn!(error = %err, "sandbox service routing: upstream HTTP request failed"); - state.service_upstreams.evict(&pool_key); - let route_err = ServiceRouteError::service_unreachable(); - emit_service_relay_failure(&endpoint, target_port, route_err.reason); - route_err - })?; + let replay = reused.then(|| replayable_request(&upstream)).flatten(); + let mut response = match sender.send_request(upstream).await { + Ok(response) => response, + Err(err) => { + warn!(error = %err, "sandbox service routing: upstream HTTP request failed"); + state.service_upstreams.evict(&pool_key); + let Some(replay) = replay else { + let route_err = ServiceRouteError::service_unreachable(); + emit_service_relay_failure(&endpoint, target_port, route_err.reason); + return Err(route_err); + }; + sender = + open_upstream(&state, &sandbox, &endpoint, target_port, websocket_upgrade).await?; + sender.send_request(replay).await.map_err(|err| { + warn!(error = %err, "sandbox service routing: upstream HTTP retry failed"); + let route_err = ServiceRouteError::service_unreachable(); + emit_service_relay_failure(&endpoint, target_port, route_err.reason); + route_err + })? + } + }; if !websocket_upgrade { state.service_upstreams.put(&pool_key, sender); @@ -574,6 +595,23 @@ async fn load_endpoint( .ok_or_else(ServiceRouteError::endpoint_not_found) } +/// Copies a request that can be sent again on a fresh connection. +/// +/// A pooled connection can be closed by the sandbox between the liveness check +/// and the send. Only bodyless methods are replayable, because the original +/// body is consumed by the failed attempt. +fn replayable_request(request: &Request) -> Option> { + if !matches!(*request.method(), Method::GET | Method::HEAD) { + return None; + } + let mut replay = Request::new(Body::empty()); + *replay.method_mut() = request.method().clone(); + *replay.uri_mut() = request.uri().clone(); + *replay.version_mut() = request.version(); + *replay.headers_mut() = request.headers().clone(); + Some(replay) +} + fn build_upstream_request( req: Request, target_port: u16, @@ -1409,6 +1447,44 @@ mod tests { ); } + #[tokio::test] + async fn take_removes_an_endpoint_left_with_no_upstreams() { + let pool = ServiceUpstreamPool::default(); + let (sender, _sandbox) = test_upstream().await; + pool.put("ep-a|8080", sender); + + assert!(pool.take("ep-a|8080").is_some()); + assert!( + !pool.inner.lock().unwrap().idle.contains_key("ep-a|8080"), + "an emptied endpoint must not linger until the next sweep" + ); + } + + #[test] + fn only_bodyless_requests_are_replayable() { + for method in [Method::GET, Method::HEAD] { + let mut request = Request::new(Body::empty()); + *request.method_mut() = method.clone(); + *request.uri_mut() = "/health".parse().unwrap(); + request + .headers_mut() + .insert(header::HOST, HeaderValue::from_static("svc")); + + let replay = + replayable_request(&request).expect("bodyless method should be replayable"); + assert_eq!(*replay.method(), method); + assert_eq!(replay.uri().path(), "/health"); + assert_eq!(replay.headers().get(header::HOST).unwrap(), "svc"); + } + + let mut post = Request::new(Body::empty()); + *post.method_mut() = Method::POST; + assert!( + replayable_request(&post).is_none(), + "a request with a body cannot be replayed" + ); + } + #[tokio::test] async fn pool_does_not_hand_out_another_endpoints_upstream() { let pool = ServiceUpstreamPool::default(); diff --git a/crates/openshell-server/src/supervisor_owner.rs b/crates/openshell-server/src/supervisor_owner.rs index 19d371d722..a517bce0fa 100644 --- a/crates/openshell-server/src/supervisor_owner.rs +++ b/crates/openshell-server/src/supervisor_owner.rs @@ -7,7 +7,7 @@ use crate::persistence::{PersistenceError, Store, WriteCondition}; use openshell_core::time::now_ms; use serde::{Deserialize, Serialize}; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use thiserror::Error; const OWNER_OBJECT_TYPE: &str = "supervisor_session_owner"; @@ -28,6 +28,14 @@ pub enum OwnerError { Store(#[from] PersistenceError), } +impl OwnerError { + /// True when another replica holds ownership, as opposed to the store + /// being unreachable. + pub fn is_ownership_lost(&self) -> bool { + matches!(self, Self::AlreadyOwned | Self::Conflict) + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] struct OwnerPayload { sandbox_id: String, @@ -52,6 +60,14 @@ pub struct OwnerRecord { pub resource_version: u64, } +impl OwnerRecord { + /// True while the record's last update is within `ttl`. + pub fn is_fresh(&self, ttl: Duration) -> bool { + let ttl_ms = i64::try_from(ttl.as_millis()).unwrap_or(i64::MAX); + now_ms().saturating_sub(self.updated_at_ms).max(0) < ttl_ms + } +} + #[derive(Debug, Clone)] pub struct OwnerGuard { pub sandbox_id: String, @@ -62,6 +78,15 @@ pub struct OwnerGuard { pub owner_peer_endpoint: String, connected_at_ms: i64, resource_version: u64, + last_renewed_at: Instant, +} + +impl OwnerGuard { + /// True once renewals have failed for long enough that another replica can + /// supersede this claim, making it unsafe to keep serving the session. + pub fn claim_expired(&self, ttl: Duration) -> bool { + self.last_renewed_at.elapsed() >= ttl + } } pub struct SupervisorOwnerIndex { @@ -119,6 +144,7 @@ impl SupervisorOwnerIndex { owner_peer_endpoint: owner_peer_endpoint.to_string(), connected_at_ms, resource_version: result.resource_version, + last_renewed_at: Instant::now(), }) } @@ -143,6 +169,7 @@ impl SupervisorOwnerIndex { { Ok(result) => { guard.resource_version = result.resource_version; + guard.last_renewed_at = Instant::now(); Ok(()) } Err(OwnerError::Store(PersistenceError::Conflict { .. })) => Err(OwnerError::Conflict), @@ -234,9 +261,7 @@ fn can_supersede( connection_epoch: u64, ttl: Duration, ) -> bool { - let age_ms = now_ms() - existing.updated_at_ms; - let ttl_ms = i64::try_from(ttl.as_millis()).unwrap_or(i64::MAX); - if age_ms >= ttl_ms { + if !existing.is_fresh(ttl) { return true; } @@ -253,6 +278,64 @@ mod tests { SupervisorOwnerIndex::new(store, ttl) } + fn record_updated_at(updated_at_ms: i64) -> OwnerRecord { + OwnerRecord { + session_id: "s1".to_string(), + supervisor_instance_id: "inst".to_string(), + connection_epoch: 1, + owner_replica_id: "gw-1".to_string(), + owner_peer_endpoint: "https://gw-1".to_string(), + connected_at_ms: 0, + updated_at_ms, + resource_version: 1, + } + } + + fn owner_ttl_ms() -> i64 { + i64::try_from(OWNER_TTL.as_millis()).unwrap() + } + + #[test] + fn freshness_clamps_a_future_timestamp_instead_of_going_negative() { + let skewed = record_updated_at(now_ms() + owner_ttl_ms() * 10); + assert!(skewed.is_fresh(OWNER_TTL)); + assert!(!can_supersede(&skewed, "other-inst", 99, OWNER_TTL)); + } + + #[test] + fn only_ownership_conflicts_count_as_lost_ownership() { + assert!(OwnerError::AlreadyOwned.is_ownership_lost()); + assert!(OwnerError::Conflict.is_ownership_lost()); + assert!( + !OwnerError::Store(PersistenceError::Database("db unreachable".to_string())) + .is_ownership_lost() + ); + } + + #[tokio::test] + async fn a_claim_expires_once_renewals_stop_for_the_ttl() { + let index = test_index(OWNER_TTL).await; + let guard = index + .publish("sbx", "s1", "inst", 1, "gw-1", "http://gw-1") + .await + .unwrap(); + assert!(!guard.claim_expired(OWNER_TTL)); + assert!(guard.claim_expired(Duration::ZERO)); + } + + #[test] + fn freshness_survives_a_corrupt_timestamp() { + let corrupt = record_updated_at(i64::MIN); + assert!(!corrupt.is_fresh(OWNER_TTL)); + } + + #[test] + fn freshness_expires_past_the_ttl() { + let stale = record_updated_at(now_ms() - owner_ttl_ms() - 1); + assert!(!stale.is_fresh(OWNER_TTL)); + assert!(can_supersede(&stale, "other-inst", 1, OWNER_TTL)); + } + #[tokio::test] async fn publish_creates_owner() { let index = test_index(OWNER_TTL).await; diff --git a/crates/openshell-server/src/supervisor_session.rs b/crates/openshell-server/src/supervisor_session.rs index 9de0d23b05..21d51611ce 100644 --- a/crates/openshell-server/src/supervisor_session.rs +++ b/crates/openshell-server/src/supervisor_session.rs @@ -50,7 +50,13 @@ const PEER_TLS_KEY_FILE_ENV: &str = "OPENSHELL_PEER_TLS_KEY_FILE"; const PEER_TLS_SERVER_NAME_ENV: &str = "OPENSHELL_PEER_TLS_SERVER_NAME"; /// How long a resolved owner record is reused before rereading the store. /// Well below `OWNER_TTL` so a cache hit can never outlive the record itself. +/// Marks an owner record written by a gateway that advertises no peer endpoint. +/// Only that gateway can serve such a session, so no peer should dial it. +const LOCAL_OWNER_ENDPOINT_SCHEME: &str = "local://"; const OWNER_CACHE_TTL: Duration = Duration::from_secs(3); +/// How often the owner cache reclaims expired entries. Rate-limited so an +/// insert never scans the whole map. +const OWNER_CACHE_SWEEP_INTERVAL: Duration = Duration::from_secs(30); /// How long the projected peer `ServiceAccount` token is held in memory. /// The kubelet rotates the file hourly, so this only bounds staleness. const PEER_TOKEN_CACHE_TTL: Duration = Duration::from_mins(5); @@ -133,7 +139,13 @@ fn nonempty_env(name: &str) -> Option { pub struct PeerRouteCache { channels: Mutex>, token: Mutex>, - owners: Mutex>, + owners: Mutex, +} + +#[derive(Default)] +struct OwnerCache { + entries: HashMap, + last_sweep: Option, } /// Hand-written so the cached `ServiceAccount` token is never formatted. @@ -212,9 +224,9 @@ impl PeerRouteCache { fn cached_owner(&self, sandbox_id: &str) -> Option { let now = Instant::now(); let mut owners = self.owners.lock().unwrap(); - let entry = owners.get(sandbox_id)?; + let entry = owners.entries.get(sandbox_id)?; if entry.expires_at <= now { - owners.remove(sandbox_id); + owners.entries.remove(sandbox_id); return None; } Some(entry.record.clone()) @@ -223,8 +235,16 @@ impl PeerRouteCache { fn store_owner(&self, sandbox_id: &str, record: &crate::supervisor_owner::OwnerRecord) { let now = Instant::now(); let mut owners = self.owners.lock().unwrap(); - owners.retain(|_, entry| entry.expires_at > now); - owners.insert( + // Expiry is enforced per entry on read, so the full scan only needs to + // reclaim memory. Rate-limit it to keep inserts off an O(n) path. + if owners + .last_sweep + .is_none_or(|last| now.duration_since(last) >= OWNER_CACHE_SWEEP_INTERVAL) + { + owners.last_sweep = Some(now); + owners.entries.retain(|_, entry| entry.expires_at > now); + } + owners.entries.insert( sandbox_id.to_string(), CachedOwner { record: record.clone(), @@ -234,7 +254,7 @@ impl PeerRouteCache { } fn evict_owner(&self, sandbox_id: &str) { - self.owners.lock().unwrap().remove(sandbox_id); + self.owners.lock().unwrap().entries.remove(sandbox_id); } } @@ -1193,6 +1213,13 @@ pub async fn open_routed_relay_with_message( backoff = (backoff * 2).min(SESSION_WAIT_MAX_BACKOFF); continue; } + if owner_endpoint_is_local_only(&owner.owner_peer_endpoint) { + return Err(Status::failed_precondition(format!( + "sandbox is owned by gateway replica {} which advertises no peer endpoint; \ + set OPENSHELL_PEER_ENDPOINT on every replica to route across replicas", + owner.owner_replica_id + ))); + } match open_peer_relay( state, owner.owner_peer_endpoint.clone(), @@ -1246,9 +1273,17 @@ async fn resolve_owner( } fn owner_is_fresh(owner: &crate::supervisor_owner::OwnerRecord) -> bool { - let age_ms = openshell_core::time::now_ms() - owner.updated_at_ms; - let ttl_ms = i64::try_from(OWNER_TTL.as_millis()).unwrap_or(i64::MAX); - age_ms < ttl_ms + owner.is_fresh(OWNER_TTL) +} + +/// Endpoint recorded when this replica advertises none. +fn local_owner_endpoint(replica_id: &str) -> String { + format!("{LOCAL_OWNER_ENDPOINT_SCHEME}{replica_id}") +} + +/// True when an owner record names a gateway that no peer can dial. +fn owner_endpoint_is_local_only(endpoint: &str) -> bool { + endpoint.starts_with(LOCAL_OWNER_ENDPOINT_SCHEME) } async fn open_peer_relay( @@ -1540,17 +1575,10 @@ pub async fn handle_connect_supervisor( require_persisted_sandbox(&state.store, &sandbox_id).await?; let session_id = Uuid::new_v4().to_string(); - let owner_peer_endpoint = state.peer_endpoint.clone().unwrap_or_default(); - if !state.store.is_single_replica() && owner_peer_endpoint.is_empty() { - return Err(Status::failed_precondition( - "gateway peer endpoint is required for multi-replica supervisor ownership", - )); - } - let owner_peer_endpoint = if owner_peer_endpoint.is_empty() { - format!("local://{}", state.replica_id) - } else { - owner_peer_endpoint - }; + let owner_peer_endpoint = state.peer_endpoint.as_deref().map_or_else( + || local_owner_endpoint(&state.replica_id), + ToString::to_string, + ); let owner_index = SupervisorOwnerIndex::new(state.store.clone(), OWNER_TTL); let owner_guard = owner_index .publish( @@ -1862,14 +1890,39 @@ async fn handle_supervisor_message( match msg.payload { Some(supervisor_message::Payload::Heartbeat(_)) => { let owner_index = SupervisorOwnerIndex::new(state.store.clone(), OWNER_TTL); - if let Err(err) = owner_index.renew(owner_guard).await { - warn!( + match owner_index.renew(owner_guard).await { + Ok(()) => {} + // Only a real ownership change ends the session. A store error + // means the database did not answer, and closing on that would + // drop every session heartbeating during the outage. + Err(err) if err.is_ownership_lost() => { + warn!( + sandbox_id = %sandbox_id, + session_id = %session_id, + error = %err, + "supervisor session: ownership lost; closing session" + ); + return false; + } + // Past the TTL our record is stale, so another replica may + // already have superseded it. Close rather than serve a + // session we can no longer claim. + Err(err) if owner_guard.claim_expired(OWNER_TTL) => { + warn!( + sandbox_id = %sandbox_id, + session_id = %session_id, + error = %err, + "supervisor session: owner renewal failed past the ownership TTL; \ + closing session" + ); + return false; + } + Err(err) => warn!( sandbox_id = %sandbox_id, session_id = %session_id, error = %err, - "supervisor session: owner renewal failed; closing session" - ); - return false; + "supervisor session: owner renewal failed; retrying on next heartbeat" + ), } } Some(supervisor_message::Payload::RelayOpenResult(result)) => { @@ -2771,6 +2824,19 @@ mod tests { } } + #[test] + fn a_gateway_without_a_peer_endpoint_still_records_ownership() { + let endpoint = local_owner_endpoint("gw-0"); + assert_eq!(endpoint, "local://gw-0"); + assert!(owner_endpoint_is_local_only(&endpoint)); + } + + #[test] + fn a_dialable_owner_endpoint_is_not_local_only() { + assert!(!owner_endpoint_is_local_only("https://10.0.0.1:8080")); + assert!(!owner_endpoint_is_local_only("http://10.0.0.1:8080")); + } + #[test] fn owner_cache_returns_stored_record() { let cache = PeerRouteCache::default(); @@ -2802,7 +2868,7 @@ mod tests { #[test] fn owner_cache_drops_entries_past_their_ttl() { let cache = PeerRouteCache::default(); - cache.owners.lock().unwrap().insert( + cache.owners.lock().unwrap().entries.insert( "sbx-a".to_string(), CachedOwner { record: owner_record("replica-a"), @@ -2811,7 +2877,7 @@ mod tests { ); assert!(cache.cached_owner("sbx-a").is_none()); - assert!(!cache.owners.lock().unwrap().contains_key("sbx-a")); + assert!(!cache.owners.lock().unwrap().entries.contains_key("sbx-a")); } #[test] diff --git a/docs/reference/gateway-config.mdx b/docs/reference/gateway-config.mdx index 9940f2ee2e..30514149eb 100644 --- a/docs/reference/gateway-config.mdx +++ b/docs/reference/gateway-config.mdx @@ -504,6 +504,26 @@ args = [ ] ``` +## Gateway Peer Routing + +Multi-replica gateways relay session-bound traffic to the replica that owns a sandbox supervisor. The Helm chart sets these variables; set them manually only outside the chart. + +| Variable | Purpose | +|---|---| +| `OPENSHELL_PEER_ENDPOINT` | Address other replicas use to reach this one. Derived from the pod DNS name when unset. | +| `OPENSHELL_PEER_SERVICE_NAME` | Peer Service used to derive the endpoint. | +| `OPENSHELL_POD_NAME` / `OPENSHELL_POD_NAMESPACE` | Pod identity used for endpoint derivation and peer verification. | +| `OPENSHELL_SERVICE_ACCOUNT_NAME` | ServiceAccount that peer tokens must present. | +| `OPENSHELL_PEER_SERVICE_ACCOUNT_TOKEN_FILE` | Projected peer token path. Defaults to `/var/run/secrets/openshell-peer/token`. | +| `OPENSHELL_PEER_TOKEN_AUDIENCE` | Required token audience. Defaults to `openshell-gateway-peer`. | +| `OPENSHELL_PEER_POD_LABELS` | Comma-separated labels a calling pod must carry. | +| `OPENSHELL_PEER_TOKEN_CACHE_TTL_SECS` | How long a verified peer identity is cached. Defaults to `60`. | +| `OPENSHELL_PEER_TLS_CA_FILE` | CA bundle for peer connections. Falls back to the platform trust store. | +| `OPENSHELL_PEER_TLS_CERT_FILE` / `OPENSHELL_PEER_TLS_KEY_FILE` | Client certificate and key for peer mTLS. Set both or neither. | +| `OPENSHELL_PEER_TLS_SERVER_NAME` | Name verified on peer certificates. | + +The gateway refuses to start when a peer endpoint is configured on a multi-replica backend and peer authentication is unavailable, because the replica could not serve relays from its peers. That happens when the pod identity variables are missing, when the in-cluster Kubernetes client cannot be built, or when the gateway is not running in a cluster. A plaintext `http://` peer endpoint is also rejected when the gateway serves TLS; use it only when the gateway itself serves plaintext. + ## Driver References Each example is a complete TOML file for one compute driver. The examples repeat `[openshell]` and `[openshell.gateway]` so they stay copyable, and the driver tables list the accepted driver-specific keys. Drivers receive only their own tables, and the gateway rejects unknown gateway and driver fields.