diff --git a/architecture/sandbox.md b/architecture/sandbox.md index 56b7d635d..daba7bbac 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -476,7 +476,9 @@ quickly. ## Supervisor Configuration Delivery The gateway and supervisor must implement the same internal supervisor protocol -revision. The gateway includes a configuration bootstrap when it accepts a +revision. Peers built before the handshake existed report revision zero and are +accepted for one release with a warning and a counter, because sandboxes keep +their supervisor binary until they are recreated. The gateway includes a configuration bootstrap when it accepts a `ConnectSupervisor` session and can send complete component replacements on the same stream after policy, settings, or provider state changes. Bootstrap construction does not gate the session while polling remains diff --git a/crates/openshell-core/src/proto/mod.rs b/crates/openshell-core/src/proto/mod.rs index d80325e65..2ab1e155d 100644 --- a/crates/openshell-core/src/proto/mod.rs +++ b/crates/openshell-core/src/proto/mod.rs @@ -101,3 +101,11 @@ pub fn all_workspaces_selector() -> WorkspaceSelector { /// Bump this when either peer can no longer honor the previous stream /// semantics. pub const SUPERVISOR_PROTOCOL_REVISION: u32 = 1; + +/// Revision implied by peers built before the handshake existed. Proto3 leaves +/// the field unset, so such peers report zero. +/// +/// Sandboxes keep their supervisor binary until they are recreated, so a +/// gateway upgrade must keep serving them for one release. Remove this +/// allowance once every supported release sends an explicit revision. +pub const LEGACY_SUPERVISOR_PROTOCOL_REVISION: u32 = 0; diff --git a/crates/openshell-server/src/config_delivery.rs b/crates/openshell-server/src/config_delivery.rs index e70bd4d8b..b9c749527 100644 --- a/crates/openshell-server/src/config_delivery.rs +++ b/crates/openshell-server/src/config_delivery.rs @@ -541,14 +541,8 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; use super::*; - use crate::grpc::{OpenShellService, test_support::test_server_state}; - use openshell_core::proto::{ - GatewayMessage, ObjectMeta, SandboxSpec, SupervisorHello, SupervisorMessage, - gateway_message, open_shell_client::OpenShellClient, open_shell_server::OpenShellServer, - supervisor_message, - }; - use tokio::sync::mpsc; - use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream}; + use crate::grpc::test_support::{connect_supervisor_stream, test_server_state}; + use openshell_core::proto::{GatewayMessage, ObjectMeta, SandboxSpec, gateway_message}; fn key(sandbox_id: &str, component: ConfigComponentKind) -> DeliveryKey { DeliveryKey { @@ -727,35 +721,15 @@ mod tests { .await .unwrap(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let address = listener.local_addr().unwrap(); - let server = tokio::spawn( - tonic::transport::Server::builder() - .add_service(OpenShellServer::new(OpenShellService::new(Arc::clone( - &state, - )))) - .serve_with_incoming(TcpListenerStream::new(listener)), - ); - let mut client = OpenShellClient::connect(format!("http://{address}")) - .await - .unwrap(); - let (tx, rx) = mpsc::channel(4); - tx.send(SupervisorMessage { - payload: Some(supervisor_message::Payload::Hello(SupervisorHello { - sandbox_id: "sandbox".into(), - instance_id: "instance".into(), - protocol_revision: openshell_core::proto::SUPERVISOR_PROTOCOL_REVISION, - })), - }) + let mut harness = connect_supervisor_stream( + &state, + "sandbox", + openshell_core::proto::SUPERVISOR_PROTOCOL_REVISION, + ) .await .unwrap(); - let mut stream = client - .connect_supervisor(ReceiverStream::new(rx)) - .await - .unwrap() - .into_inner(); - let first = tokio::time::timeout(Duration::from_secs(5), stream.message()) + let first = tokio::time::timeout(Duration::from_secs(5), harness.inbound.message()) .await .unwrap() .unwrap() @@ -766,7 +740,7 @@ mod tests { )); publish_sandbox_components(&state, "sandbox", ConfigComponents::SANDBOX_CONFIG); - let update = tokio::time::timeout(Duration::from_secs(5), stream.message()) + let update = tokio::time::timeout(Duration::from_secs(5), harness.inbound.message()) .await .unwrap() .unwrap() @@ -777,9 +751,6 @@ mod tests { payload: Some(gateway_message::Payload::ConfigUpdate(_)) } )); - - drop(tx); - server.abort(); } #[test] diff --git a/crates/openshell-server/src/grpc/mod.rs b/crates/openshell-server/src/grpc/mod.rs index 95bd6eaa3..15bbcd13b 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -818,8 +818,74 @@ pub mod test_support { use crate::supervisor_session::SupervisorSessionRegistry; use crate::tracing_bus::TracingLogBus; use openshell_core::Config; + use openshell_core::proto::open_shell_client::OpenShellClient; + use openshell_core::proto::open_shell_server::OpenShellServer; + use openshell_core::proto::{ + GatewayMessage, SupervisorHello, SupervisorMessage, supervisor_message, + }; + use tokio::sync::mpsc; + use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream}; use tonic::Request; + /// A live `ConnectSupervisor` stream against an in-process gateway. + pub struct SupervisorStreamHarness { + server: tokio::task::JoinHandle>, + /// Held so the supervisor side of the stream stays open. + _outbound: mpsc::Sender, + pub inbound: tonic::Streaming, + } + + impl Drop for SupervisorStreamHarness { + fn drop(&mut self) { + self.server.abort(); + } + } + + /// Serve `state` on loopback and open a supervisor stream whose hello + /// carries the given protocol revision. Returns the gRPC status when the + /// gateway rejects the stream before accepting it. + pub async fn connect_supervisor_stream( + state: &Arc, + sandbox_id: &str, + protocol_revision: u32, + ) -> Result { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(OpenShellServer::new(super::OpenShellService::new( + Arc::clone(state), + ))) + .serve_with_incoming(TcpListenerStream::new(listener)), + ); + let mut client = OpenShellClient::connect(format!("http://{address}")) + .await + .unwrap(); + let (outbound, rx) = mpsc::channel(4); + outbound + .send(SupervisorMessage { + payload: Some(supervisor_message::Payload::Hello(SupervisorHello { + sandbox_id: sandbox_id.into(), + instance_id: "instance".into(), + protocol_revision, + })), + }) + .await + .unwrap(); + let inbound = match client.connect_supervisor(ReceiverStream::new(rx)).await { + Ok(response) => response.into_inner(), + Err(status) => { + server.abort(); + return Err(status); + } + }; + Ok(SupervisorStreamHarness { + server, + _outbound: outbound, + inbound, + }) + } + /// Wrap a proto message in a `Request` with a dev principal injected. /// /// The dev principal matches the unauthenticated dev user: subject diff --git a/crates/openshell-server/src/supervisor_session.rs b/crates/openshell-server/src/supervisor_session.rs index c5668cc41..5cb66ac16 100644 --- a/crates/openshell-server/src/supervisor_session.rs +++ b/crates/openshell-server/src/supervisor_session.rs @@ -15,12 +15,12 @@ use tonic::{Request, Response, Status}; use tracing::{debug, info, warn}; use uuid::Uuid; -use openshell_core::proto::SUPERVISOR_PROTOCOL_REVISION; use openshell_core::proto::{ ConfigUpdate, GatewayMessage, RelayFrame, RelayInit, RelayOpen, ReportMainProcessExitRequest, ReportMainProcessExitResponse, Sandbox, SandboxPhase, SessionAccepted, SshRelayTarget, SupervisorMessage, config_update, gateway_message, relay_open, supervisor_message, }; +use openshell_core::proto::{LEGACY_SUPERVISOR_PROTOCOL_REVISION, SUPERVISOR_PROTOCOL_REVISION}; use openshell_core::transport_errors::is_expected_transport_close_status; use crate::ServerState; @@ -810,7 +810,7 @@ pub async fn handle_connect_supervisor( if sandbox_id.is_empty() { return Err(Status::invalid_argument("sandbox_id is required")); } - validate_protocol_revision(hello.protocol_revision)?; + validate_protocol_revision(&sandbox_id, hello.protocol_revision)?; if let Some(principal) = principal.as_ref() { crate::auth::guard::ensure_sandbox_principal_scope(principal, &sandbox_id)?; } @@ -960,13 +960,20 @@ pub async fn handle_connect_supervisor( Ok(Response::new(stream)) } -fn validate_protocol_revision(supervisor_revision: u32) -> Result<(), Status> { - if supervisor_revision == SUPERVISOR_PROTOCOL_REVISION { - Ok(()) - } else { - Err(Status::failed_precondition(format!( - "supervisor protocol revision mismatch: gateway requires {SUPERVISOR_PROTOCOL_REVISION}, supervisor offered {supervisor_revision}" - ))) +fn validate_protocol_revision(sandbox_id: &str, supervisor_revision: u32) -> Result<(), Status> { + match supervisor_revision { + SUPERVISOR_PROTOCOL_REVISION => Ok(()), + LEGACY_SUPERVISOR_PROTOCOL_REVISION => { + counter!("openshell_supervisor_protocol_legacy_sessions_total").increment(1); + warn!( + sandbox_id = %sandbox_id, + "supervisor session: supervisor predates the protocol handshake; recreate the sandbox before the next gateway upgrade" + ); + Ok(()) + } + other => Err(Status::failed_precondition(format!( + "supervisor protocol revision mismatch: gateway requires {SUPERVISOR_PROTOCOL_REVISION}, supervisor offered {other}" + ))), } } @@ -1246,9 +1253,77 @@ mod tests { } #[test] - fn supervisor_protocol_revision_must_match_exactly() { - assert!(validate_protocol_revision(SUPERVISOR_PROTOCOL_REVISION).is_ok()); - let error = validate_protocol_revision(SUPERVISOR_PROTOCOL_REVISION + 1).unwrap_err(); + fn supervisor_protocol_revision_accepts_current_and_legacy_peers() { + assert!(validate_protocol_revision("sb-1", SUPERVISOR_PROTOCOL_REVISION).is_ok()); + assert!(validate_protocol_revision("sb-1", LEGACY_SUPERVISOR_PROTOCOL_REVISION).is_ok()); + } + + async fn state_with_sandbox(sandbox_id: &str) -> Arc { + let state = crate::grpc::test_support::test_server_state().await; + state + .store + .put_message(&sandbox_record(sandbox_id, sandbox_id)) + .await + .unwrap(); + state + } + + async fn first_gateway_message( + harness: &mut crate::grpc::test_support::SupervisorStreamHarness, + ) -> GatewayMessage { + tokio::time::timeout(Duration::from_secs(5), harness.inbound.message()) + .await + .expect("gateway response before timeout") + .expect("stream open") + .expect("gateway message") + } + + #[tokio::test] + async fn legacy_supervisor_without_protocol_revision_is_accepted() { + let state = state_with_sandbox("sb-legacy").await; + let mut harness = crate::grpc::test_support::connect_supervisor_stream( + &state, + "sb-legacy", + LEGACY_SUPERVISOR_PROTOCOL_REVISION, + ) + .await + .expect("legacy supervisor must connect"); + + let Some(gateway_message::Payload::SessionAccepted(accepted)) = + first_gateway_message(&mut harness).await.payload + else { + panic!("expected SessionAccepted"); + }; + assert_eq!(accepted.protocol_revision, SUPERVISOR_PROTOCOL_REVISION); + assert!( + state + .supervisor_sessions + .is_current_session("sb-legacy", &accepted.session_id) + ); + } + + #[tokio::test] + async fn unknown_supervisor_protocol_revision_is_rejected() { + let state = state_with_sandbox("sb-future").await; + let Err(status) = crate::grpc::test_support::connect_supervisor_stream( + &state, + "sb-future", + SUPERVISOR_PROTOCOL_REVISION + 1, + ) + .await + else { + panic!("mismatched revision must be rejected"); + }; + + assert_eq!(status.code(), tonic::Code::FailedPrecondition); + assert!(status.message().contains("revision mismatch")); + assert!(state.supervisor_sessions.connected_sandbox_ids().is_empty()); + } + + #[test] + fn supervisor_protocol_revision_rejects_unknown_peers() { + let error = + validate_protocol_revision("sb-1", SUPERVISOR_PROTOCOL_REVISION + 1).unwrap_err(); assert_eq!(error.code(), tonic::Code::FailedPrecondition); assert!(error.message().contains("revision mismatch")); } diff --git a/crates/openshell-supervisor-process/src/supervisor_session.rs b/crates/openshell-supervisor-process/src/supervisor_session.rs index 9b6882f84..959d7dd0d 100644 --- a/crates/openshell-supervisor-process/src/supervisor_session.rs +++ b/crates/openshell-supervisor-process/src/supervisor_session.rs @@ -17,13 +17,13 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; -use openshell_core::proto::SUPERVISOR_PROTOCOL_REVISION; use openshell_core::proto::open_shell_client::OpenShellClient; use openshell_core::proto::{ FinalizeMainProcessExitRequest, GatewayMessage, RelayFrame, RelayInit, RelayOpen, RelayOpenResult, ReportMainProcessExitRequest, SupervisorHeartbeat, SupervisorHello, SupervisorMessage, TcpRelayTarget, gateway_message, relay_open, supervisor_message, }; +use openshell_core::proto::{LEGACY_SUPERVISOR_PROTOCOL_REVISION, SUPERVISOR_PROTOCOL_REVISION}; use openshell_ocsf::{ ActivityId, ConnectionInfo, Endpoint, EventContext, NetworkActivityBuilder, OcsfEvent, SeverityId, StatusId, ocsf_emit, @@ -447,13 +447,18 @@ async fn run_single_session( fn validate_gateway_protocol_revision( gateway_revision: u32, ) -> Result<(), Box> { - if gateway_revision == SUPERVISOR_PROTOCOL_REVISION { - Ok(()) - } else { - Err(format!( - "supervisor protocol revision mismatch: supervisor requires {SUPERVISOR_PROTOCOL_REVISION}, gateway offered {gateway_revision}" + match gateway_revision { + SUPERVISOR_PROTOCOL_REVISION => Ok(()), + LEGACY_SUPERVISOR_PROTOCOL_REVISION => { + warn!( + "supervisor session: gateway predates the protocol handshake; upgrade the gateway before pinning newer supervisor images" + ); + Ok(()) + } + other => Err(format!( + "supervisor protocol revision mismatch: supervisor requires {SUPERVISOR_PROTOCOL_REVISION}, gateway offered {other}" ) - .into()) + .into()), } } @@ -863,8 +868,13 @@ mod target_tests { use super::*; #[test] - fn gateway_protocol_revision_must_match_exactly() { + fn gateway_protocol_revision_accepts_current_and_legacy_peers() { assert!(validate_gateway_protocol_revision(SUPERVISOR_PROTOCOL_REVISION).is_ok()); + assert!(validate_gateway_protocol_revision(LEGACY_SUPERVISOR_PROTOCOL_REVISION).is_ok()); + } + + #[test] + fn gateway_protocol_revision_rejects_unknown_peers() { let error = validate_gateway_protocol_revision(SUPERVISOR_PROTOCOL_REVISION + 1) .expect_err("version skew must be rejected"); assert!(error.to_string().contains("revision mismatch")); diff --git a/docs/reference/gateway-config.mdx b/docs/reference/gateway-config.mdx index 8aeb2281e..6a0be10c8 100644 --- a/docs/reference/gateway-config.mdx +++ b/docs/reference/gateway-config.mdx @@ -108,6 +108,8 @@ disable_tls = false default_image = "ghcr.io/nvidia/openshell-community/sandboxes/base:latest" # Defaults to the gateway version. Custom builds must match the gateway's # internal supervisor protocol revision; mismatched peers reject the session. +# Supervisors from releases before the handshake existed still connect, with +# a gateway warning, until the next release. # supervisor_image = "ghcr.io/nvidia/openshell/supervisor:" client_tls_secret_name = "openshell-client-tls" service_account_name = "openshell-sandbox" diff --git a/proto/openshell.proto b/proto/openshell.proto index 9e1a3e7f9..a63d33ac4 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -2517,6 +2517,7 @@ message SupervisorHello { // Supervisor instance ID (e.g. boot id or process epoch). string instance_id = 2; // Exact internal stream protocol revision implemented by this supervisor. + // Zero identifies a supervisor built before the handshake existed. uint32 protocol_revision = 3; } diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 4f730587c..ec6f3790f 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -16,11 +16,11 @@ The target deployment flow is: 1. Operator starts or deploys the gateway with system packages, systemd, or Helm. The CLI does not start, stop, or destroy gateway services. 2. Operator configures the compute driver. 3. Operator provides the CLI and supervisor authentication material required by the deployment mode: edge or OIDC user auth, optional CLI mTLS, and gateway-minted sandbox JWTs. - -If supervisor sessions fail with a protocol revision mismatch, check that custom supervisor images match the gateway release. Gateway and supervisor require the same internal protocol revision; authentication success does not make mismatched versions compatible. See the published [gateway configuration reference](https://docs.nvidia.com/openshell/latest/reference/gateway-config). 4. The CLI registers a reachable gateway endpoint with `openshell gateway add`. 5. The gateway creates sandboxes through the selected compute driver. +If supervisor sessions fail with a protocol revision mismatch, check that custom supervisor images match the gateway release. Gateway and supervisor require the same internal protocol revision; authentication success does not make mismatched versions compatible. Supervisors that predate the handshake still connect for one release. The gateway logs a warning for each such session and counts them in the `openshell_supervisor_protocol_legacy_sessions_total` metric, so recreate those sandboxes before the next gateway upgrade. See the published [gateway configuration reference](https://docs.nvidia.com/openshell/latest/reference/gateway-config.md). + The `openshell-gateway` composition crate explicitly installs its compiled Docker, Podman, Kubernetes, and VM registrations at startup; `openshell-server` does not link compute-driver crates. Custom gateway binaries may include a