From 57eb16594b12b171fe413d83eedf573a6e4dcfa4 Mon Sep 17 00:00:00 2001 From: Taylor Mutch Date: Mon, 26 Oct 2026 15:29:32 -0700 Subject: [PATCH 1/3] feat(gateway): expose workspace service route discovery Signed-off-by: Taylor Mutch --- .../tests/ensure_providers_integration.rs | 15 + .../openshell-cli/tests/mtls_integration.rs | 15 + .../tests/provider_commands_integration.rs | 15 + .../sandbox_create_lifecycle_integration.rs | 15 + .../sandbox_exec_streaming_integration.rs | 7 + .../sandbox_name_fallback_integration.rs | 15 + crates/openshell-sdk/tests/client_mock.rs | 14 + crates/openshell-server/README.md | 24 + crates/openshell-server/src/grpc/mod.rs | 17 + .../src/grpc/service_advertisement.rs | 3 +- .../src/grpc/service_discovery.rs | 586 ++++++++++++++++++ crates/openshell-server/src/lib.rs | 3 + crates/openshell-server/src/storage_proto.rs | 10 +- crates/openshell-server/tests/common/mod.rs | 14 + .../tests/supervisor_relay_integration.rs | 14 + docs/how-it-works/gateways/configuration.mdx | 15 +- proto/openshell.proto | 62 ++ skills/debug-openshell-cluster/SKILL.md | 13 +- 18 files changed, 845 insertions(+), 12 deletions(-) create mode 100644 crates/openshell-server/src/grpc/service_discovery.rs diff --git a/crates/openshell-cli/tests/ensure_providers_integration.rs b/crates/openshell-cli/tests/ensure_providers_integration.rs index be3e4f2e49..29be137f18 100644 --- a/crates/openshell-cli/tests/ensure_providers_integration.rs +++ b/crates/openshell-cli/tests/ensure_providers_integration.rs @@ -80,6 +80,21 @@ impl TestOpenShell { #[tonic::async_trait] impl OpenShell for TestOpenShell { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = tokio_stream::wrappers::ReceiverStream< + Result, + >; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/crates/openshell-cli/tests/mtls_integration.rs b/crates/openshell-cli/tests/mtls_integration.rs index 1017712035..f2b569de96 100644 --- a/crates/openshell-cli/tests/mtls_integration.rs +++ b/crates/openshell-cli/tests/mtls_integration.rs @@ -32,6 +32,21 @@ struct TestOpenShell; #[tonic::async_trait] impl OpenShell for TestOpenShell { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = tokio_stream::wrappers::ReceiverStream< + Result, + >; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/crates/openshell-cli/tests/provider_commands_integration.rs b/crates/openshell-cli/tests/provider_commands_integration.rs index 89247ba718..f14805f17d 100644 --- a/crates/openshell-cli/tests/provider_commands_integration.rs +++ b/crates/openshell-cli/tests/provider_commands_integration.rs @@ -235,6 +235,21 @@ impl TestOpenShell { #[tonic::async_trait] impl OpenShell for TestOpenShell { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = tokio_stream::wrappers::ReceiverStream< + Result, + >; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs index 587910fbaf..bc8197872d 100644 --- a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs +++ b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs @@ -102,6 +102,21 @@ struct TestOpenShell { #[tonic::async_trait] impl OpenShell for TestOpenShell { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = tokio_stream::wrappers::ReceiverStream< + Result, + >; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs b/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs index 31e67baf0f..4238254ed4 100644 --- a/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs +++ b/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs @@ -163,6 +163,13 @@ macro_rules! mock_gateway { ) => { #[tonic::async_trait] impl OpenShell for MockGateway { + async fn get_service_route_snapshot(&self, _: Request) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = tokio_stream::wrappers::ReceiverStream>; + async fn watch_service_routes(&self, _: Request) -> Result, Status> { + Err(Status::unimplemented("unused")) + } $(async fn $method(&self, _: Request<$request>) -> Result, Status> { Err(Status::unimplemented("unused test RPC")) })* diff --git a/crates/openshell-cli/tests/sandbox_name_fallback_integration.rs b/crates/openshell-cli/tests/sandbox_name_fallback_integration.rs index 497f3ce778..c218a7fcb7 100644 --- a/crates/openshell-cli/tests/sandbox_name_fallback_integration.rs +++ b/crates/openshell-cli/tests/sandbox_name_fallback_integration.rs @@ -47,6 +47,21 @@ struct TestOpenShell { #[tonic::async_trait] impl OpenShell for TestOpenShell { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = tokio_stream::wrappers::ReceiverStream< + Result, + >; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/crates/openshell-sdk/tests/client_mock.rs b/crates/openshell-sdk/tests/client_mock.rs index 1f3f791a08..3e8cee1579 100644 --- a/crates/openshell-sdk/tests/client_mock.rs +++ b/crates/openshell-sdk/tests/client_mock.rs @@ -245,6 +245,20 @@ fn workload_template_proto(name: &str, workspace: &str) -> proto::SandboxWorkloa #[tonic::async_trait] impl OpenShell for TestOpenShell { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = + tokio_stream::wrappers::ReceiverStream>; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/crates/openshell-server/README.md b/crates/openshell-server/README.md index 97569a44e1..56a2c27a7b 100644 --- a/crates/openshell-server/README.md +++ b/crates/openshell-server/README.md @@ -261,3 +261,27 @@ timeouts; bounded record attempts prevent a malformed or stalled record from starving later cleanup. Failed attempts are revisited on the next scan. Discovery consumers must not treat these records as application health or external ingress readiness. + +## Workspace service discovery + +`GetServiceRouteSnapshot` and `WatchServiceRoutes` expose accepted installations +for one explicit active workspace. They require `sandbox:read` and workspace +user access; omitted/all-workspaces selectors and sandbox principals are rejected. +Dedicated discovery machine grants are separate work. + +Each successful response replaces the complete view. Consumers preserve the +returned hostname as TLS SNI and HTTP authority when dialing the supervisor +endpoint, remove absent routes, and expire cached entries at `expires_at` even +when discovery is unavailable. An error is not an empty replacement. The opaque +snapshot revision includes lease renewals; `installation_version` remains stable +across those renewals and changes when installation identity changes. These +records establish neither application health nor ingress readiness. + +Collection checks installation resource versions across two scans, retrying once +if changed. This is a verified view, not a transaction across configuration and +owner resources. Reads are bounded to 1,000 installation records, 4,096 routes, +4 MiB, and ten seconds. Overflow, corruption, or storage failure returns an error. +Watches refresh and reauthorize each second, use a one-message buffer, and end +after one minute or a five-second blocked send. Each gateway process accepts at +most 32 watches. Reconnect and relist on stream termination; revisions are opaque +equality tokens, not resumable cursors or an ordering across gateway replicas. diff --git a/crates/openshell-server/src/grpc/mod.rs b/crates/openshell-server/src/grpc/mod.rs index 2efcb9873e..1c19fc51a1 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -14,6 +14,7 @@ pub mod supervisor_host; pub use sandbox::mint_persisted_authentication; mod service; pub mod service_advertisement; +mod service_discovery; mod validation; pub mod workspace; @@ -521,6 +522,22 @@ impl OpenShell for OpenShellService { service::handle_list_services(&self.state, request).await } + async fn get_service_route_snapshot( + &self, + request: Request, + ) -> Result, Status> { + service_discovery::get_snapshot(&self.state, request).await + } + + type WatchServiceRoutesStream = service_discovery::RouteWatchStream; + + async fn watch_service_routes( + &self, + request: Request, + ) -> Result, Status> { + service_discovery::watch(&self.state, request).await + } + async fn delete_service( &self, request: Request, diff --git a/crates/openshell-server/src/grpc/service_advertisement.rs b/crates/openshell-server/src/grpc/service_advertisement.rs index dff8257fcf..cc4c8af52c 100644 --- a/crates/openshell-server/src/grpc/service_advertisement.rs +++ b/crates/openshell-server/src/grpc/service_advertisement.rs @@ -615,7 +615,7 @@ pub fn spawn_cleanup(state: Arc, mut shutdown: tokio::sync::watch:: } #[cfg(test)] -mod tests { +pub mod tests { use super::*; fn test_lease_authority() -> LeaseAuthority { @@ -1103,6 +1103,7 @@ mod tests { port: 9443, domain: "services.example.com".into(), }; + Arc::get_mut(&mut state).unwrap().admin_role = "openshell-admin".into(); let sandbox = Sandbox { metadata: Some(ObjectMeta { id: "sandbox".into(), diff --git a/crates/openshell-server/src/grpc/service_discovery.rs b/crates/openshell-server/src/grpc/service_discovery.rs new file mode 100644 index 0000000000..d552833d51 --- /dev/null +++ b/crates/openshell-server/src/grpc/service_discovery.rs @@ -0,0 +1,586 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Bounded, full-replacement workspace views of accepted app installations. + +use std::sync::Arc; +use std::time::Duration; + +use openshell_core::proto::{ + DiscoveredServiceRoute, GetServiceRouteSnapshotRequest, GetServiceRouteSnapshotResponse, + Sandbox, SandboxServiceRoute, ServiceRouteEndpoint, ServiceRouteSnapshot, + WatchServiceRoutesRequest, WatchServiceRoutesResponse, +}; +use openshell_core::{ObjectName, ObjectWorkspace}; +use prost::Message as _; +use tokio_stream::wrappers::ReceiverStream; +use tonic::{Request, Response, Status}; + +use super::service_advertisement::{Installation, OBJECT_TYPE, is_current}; +use crate::ServerState; +use crate::auth::principal::Principal; +use crate::auth::workspace_authz::{ + MinWorkspaceRole, authorize_workspace, selected_workspace_name, +}; +use crate::persistence::{ObjectCursor, ObjectRecord, current_time_ms}; + +const MAX_RECORDS: usize = 1000; +const MAX_ROUTES: usize = 4096; +const MAX_BYTES: usize = 4 * 1024 * 1024; +const SNAPSHOT_TIMEOUT: Duration = Duration::from_secs(10); +const WATCH_LIFETIME: Duration = Duration::from_mins(1); +const REFRESH_INTERVAL: Duration = Duration::from_secs(1); +const SEND_TIMEOUT: Duration = Duration::from_secs(5); +pub type RouteWatchStream = ReceiverStream>; + +async fn authorize( + state: &ServerState, + principal: &Principal, + workspace: &str, +) -> Result<(), Status> { + if !matches!(principal, Principal::User(_) | Principal::Operator(_)) { + return Err(Status::permission_denied( + "service discovery requires a user/API identity", + )); + } + authorize_workspace( + &state.store, + &state.admin_role, + principal, + workspace, + MinWorkspaceRole::User, + ) + .await?; + super::workspace::resolve_workspace(state.store.as_ref(), workspace) + .await? + .ensure_active()?; + Ok(()) +} + +pub async fn get_snapshot( + state: &Arc, + request: Request, +) -> Result, Status> { + let principal = super::extract_principal(&request)?; + let workspace = selected_workspace_name(request.get_ref().workspace_scope.as_ref())?; + authorize(state, &principal, workspace).await?; + Ok(Response::new(GetServiceRouteSnapshotResponse { + snapshot: Some(snapshot(state, workspace).await?), + })) +} + +pub async fn watch( + state: &Arc, + request: Request, +) -> Result, Status> { + let principal = super::extract_principal(&request)?; + let workspace = + selected_workspace_name(request.get_ref().workspace_scope.as_ref())?.to_string(); + authorize(state, &principal, &workspace).await?; + let permit = state + .discovery_watches + .clone() + .try_acquire_owned() + .map_err(|_| Status::resource_exhausted("too many service discovery watches"))?; + let initial = snapshot(state, &workspace).await?; + let mut revision = initial.revision.clone(); + let (tx, rx) = tokio::sync::mpsc::channel(1); + tx.try_send(Ok(WatchServiceRoutesResponse { + snapshot: Some(initial), + })) + .map_err(|_| Status::cancelled("discovery client closed"))?; + let state = state.clone(); + tokio::spawn(async move { + let _permit = permit; + let deadline = tokio::time::Instant::now() + WATCH_LIFETIME; + let mut interval = tokio::time::interval(REFRESH_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + loop { + tokio::select! { + () = tx.closed() => return, + () = tokio::time::sleep_until(deadline) => return, + _ = interval.tick() => {} + } + let refresh = async { + authorize(&state, &principal, &workspace).await?; + snapshot(&state, &workspace).await + }; + let result = tokio::select! { + () = tx.closed() => return, + () = tokio::time::sleep_until(deadline) => return, + result = refresh => result, + }; + let message = match result { + Ok(next) if next.revision == revision => continue, + Ok(next) => { + revision.clone_from(&next.revision); + Ok(WatchServiceRoutesResponse { + snapshot: Some(next), + }) + } + Err(error) => Err(error), + }; + let failed = message.is_err(); + let sent = tokio::select! { + () = tx.closed() => return, + () = tokio::time::sleep_until(deadline) => return, + result = tokio::time::timeout(SEND_TIMEOUT, tx.send(message)) => result, + }; + if failed || !matches!(sent, Ok(Ok(()))) { + return; + } + } + }); + Ok(Response::new(ReceiverStream::new(rx))) +} + +async fn records(state: &ServerState, workspace: &str) -> Result, Status> { + let mut records = Vec::new(); + let mut cursor: Option = None; + let mut bytes = 0usize; + loop { + let page = state + .store + .list_after(OBJECT_TYPE, workspace, cursor.as_ref(), 128) + .await + .map_err(|_| Status::unavailable("service installation storage unavailable"))?; + for record in &page { + bytes = bytes.saturating_add(record.payload.len()); + if bytes > MAX_BYTES || records.len() + page.len() > MAX_RECORDS { + return Err(Status::resource_exhausted( + "service discovery collection limit exceeded", + )); + } + } + let full = page.len() == 128; + cursor = page.last().map(ObjectCursor::from); + records.extend(page); + if !full { + break; + } + } + Ok(records) +} + +pub async fn snapshot( + state: &ServerState, + workspace: &str, +) -> Result { + tokio::time::timeout(SNAPSHOT_TIMEOUT, collect(state, workspace)) + .await + .map_err(|_| Status::unavailable("service discovery collection timed out"))? +} + +async fn collect(state: &ServerState, workspace: &str) -> Result { + // Verify the installation collection did not change across the scan. + // This is a bounded verified view, not a cross-resource transaction. + for _ in 0..2 { + let before = records(state, workspace).await?; + let mut routes = Vec::new(); + for object in &before { + let installation: Installation = serde_json::from_slice(&object.payload) + .map_err(|_| Status::internal("invalid service installation record"))?; + if !is_current(state, &installation).await? { + continue; + } + let sandbox = state + .store + .get_message::(&installation.sandbox_id) + .await + .map_err(|_| Status::unavailable("sandbox storage unavailable"))? + .ok_or_else(|| Status::unavailable("sandbox changed during discovery"))?; + if sandbox.object_workspace() != workspace { + return Err(Status::internal("installation workspace mismatch")); + } + for bytes in &installation.routes { + let route = SandboxServiceRoute::decode(bytes.as_slice()) + .map_err(|_| Status::internal("invalid installed service route"))?; + let label = if route.name.is_empty() { + format!("{workspace}--{}", sandbox.object_name()) + } else { + format!("{workspace}--{}--{}", sandbox.object_name(), route.name) + }; + if label.len() > 63 { + return Err(Status::failed_precondition( + "service hostname exceeds DNS label limit", + )); + } + let hostname = format!("{label}.{}", installation.service_domain); + let identity = serde_json::to_vec(&( + "openshell-installation-v1", + &installation.session_id, + &installation.supervisor_instance_id, + installation.connection_epoch, + installation.ownership_epoch, + &installation.pool_route, + &installation.runtime_identity, + &installation.configuration_instance_id, + installation.config_revision, + &installation.service_domain, + &installation.host, + installation.port, + bytes, + )) + .map_err(|_| Status::internal("installation identity encoding failed"))?; + let version = openshell_crypto::sha256(&identity) + .map_err(|_| Status::internal("installation fingerprint failed"))?; + routes.push(DiscoveredServiceRoute { + workspace: workspace.to_string(), + sandbox: sandbox.object_name().to_string(), + name: route.name, + hostname, + endpoint: Some(ServiceRouteEndpoint { + host: installation.host.clone(), + port: installation.port, + }), + installation_version: hex::encode(version), + declaration_revision: route.resource_version, + expires_at: Some( + openshell_core::time::timestamp_from_millis(installation.expires_at_ms) + .map_err(|_| Status::internal("invalid installation expiry"))?, + ), + }); + if routes.len() > MAX_ROUTES { + return Err(Status::resource_exhausted( + "too many discovered service routes", + )); + } + } + } + let after = records(state, workspace).await?; + if before + .iter() + .map(|r| (&r.id, r.resource_version)) + .ne(after.iter().map(|r| (&r.id, r.resource_version))) + { + continue; + } + routes.retain(|route| { + route + .expires_at + .as_ref() + .and_then(|t| openshell_core::time::timestamp_to_millis(t).ok()) + .is_some_and(|t| t > current_time_ms()) + }); + routes.sort_by(|a, b| { + (&a.workspace, &a.sandbox, &a.name).cmp(&(&b.workspace, &b.sandbox, &b.name)) + }); + let mut snapshot = ServiceRouteSnapshot { + routes, + revision: String::new(), + }; + let bytes = snapshot.encode_to_vec(); + if bytes.len() > MAX_BYTES - 128 { + return Err(Status::resource_exhausted( + "service discovery response too large", + )); + } + snapshot.revision = hex::encode( + openshell_crypto::sha256(&bytes) + .map_err(|_| Status::internal("snapshot fingerprint failed"))?, + ); + return Ok(snapshot); + } + Err(Status::unavailable( + "service installations changed during collection; retry snapshot", + )) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::grpc::service_advertisement::{accept, tests::installed_fixture}; + use crate::grpc::test_support::authed_request; + use tokio_stream::StreamExt as _; + + fn request() -> GetServiceRouteSnapshotRequest { + GetServiceRouteSnapshotRequest { + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + } + } + + #[tokio::test] + async fn snapshot_is_shared_between_gateway_states_and_rejects_expired_or_replaced_owners() { + let (state, owner, report) = installed_fixture().await; + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .unwrap(); + let snapshot_a = get_snapshot(&state, authed_request(request())) + .await + .unwrap() + .into_inner() + .snapshot + .unwrap(); + assert_eq!(snapshot_a.routes.len(), 1); + let route = &snapshot_a.routes[0]; + assert_eq!(route.hostname, "default--demo--web.services.example.com"); + assert_eq!(route.endpoint.as_ref().unwrap().port, 9443); + let compute = crate::compute::new_test_runtime(state.store.clone()).await; + let mut replica = ServerState::new( + state.config.clone(), + state.store.clone(), + compute, + crate::sandbox_index::SandboxIndex::new(), + crate::sandbox_watch::SandboxWatchBus::new(), + crate::tracing_bus::TracingLogBus::new(), + Arc::new(crate::supervisor_session::SupervisorSessionRegistry::new()), + None, + ); + replica.admin_role = state.admin_role.clone(); + replica.provider_profile_sources = state.provider_profile_sources.clone(); + let replica = Arc::new(replica); + let snapshot_b = get_snapshot(&replica, authed_request(request())) + .await + .unwrap() + .into_inner() + .snapshot + .unwrap(); + assert_eq!(snapshot_a, snapshot_b); + crate::supervisor_owner::SupervisorOwnerIndex::new( + state.store.clone(), + crate::supervisor_owner::OWNER_TTL, + ) + .publish( + "sandbox", + "replacement", + "instance", + 2, + "other-replica", + "local://other", + ) + .await + .unwrap(); + assert!(snapshot(&state, "default").await.unwrap().routes.is_empty()); + assert!( + snapshot(&replica, "default") + .await + .unwrap() + .routes + .is_empty() + ); + } + + #[tokio::test] + async fn watch_reports_lease_only_renewal_then_withdrawal_and_releases_its_slot() { + let (state, owner, mut report) = installed_fixture().await; + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .unwrap(); + let mut stream = watch( + &state, + authed_request(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + }), + ) + .await + .unwrap() + .into_inner(); + let first = stream.next().await.unwrap().unwrap().snapshot.unwrap(); + assert_eq!(state.discovery_watches.available_permits(), 31); + report.lease_sequence = 2; + accept(&state, &owner, &report, current_time_ms() + 14_000) + .await + .unwrap(); + let renewed = tokio::time::timeout(Duration::from_secs(3), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap() + .snapshot + .unwrap(); + assert_ne!(first.revision, renewed.revision); + assert_eq!( + first.routes[0].installation_version, + renewed.routes[0].installation_version + ); + assert_ne!(first.routes[0].expires_at, renewed.routes[0].expires_at); + report.lease_sequence = 3; + report.routes.clear(); + accept(&state, &owner, &report, current_time_ms() + 14_000) + .await + .unwrap(); + let withdrawn = tokio::time::timeout(Duration::from_secs(3), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap() + .snapshot + .unwrap(); + assert!(withdrawn.routes.is_empty()); + drop(stream); + tokio::time::timeout(Duration::from_secs(2), async { + while state.discovery_watches.available_permits() != 32 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } + + #[tokio::test] + async fn explicit_workspace_and_membership_are_required_and_watch_capacity_is_bounded() { + let (state, _, _) = installed_fixture().await; + assert_eq!( + get_snapshot( + &state, + authed_request(GetServiceRouteSnapshotRequest::default()) + ) + .await + .unwrap_err() + .code(), + tonic::Code::InvalidArgument + ); + assert!( + get_snapshot( + &state, + authed_request(GetServiceRouteSnapshotRequest { + workspace_scope: Some(openshell_core::proto::all_workspaces_selector()) + }) + ) + .await + .is_err() + ); + let mut outsider = authed_request(request()); + if let Some(Principal::User(user)) = outsider.extensions_mut().get_mut::() { + user.identity.roles = vec!["openshell-user".into()]; + user.identity.subject = "outsider".into(); + } + assert_eq!( + get_snapshot(&state, outsider).await.unwrap_err().code(), + tonic::Code::PermissionDenied + ); + let mut anonymous = Request::new(request()); + anonymous.extensions_mut().insert(Principal::Anonymous); + assert_eq!( + get_snapshot(&state, anonymous).await.unwrap_err().code(), + tonic::Code::PermissionDenied + ); + let _all = state + .discovery_watches + .clone() + .acquire_many_owned(32) + .await + .unwrap(); + let error = watch( + &state, + authed_request(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + }), + ) + .await + .expect_err("capacity denied"); + assert_eq!(error.code(), tonic::Code::ResourceExhausted); + } + + #[tokio::test] + async fn watch_terminates_when_workspace_membership_is_revoked() { + use openshell_core::proto::{ObjectMeta, WorkspaceMember, WorkspaceRole}; + let (state, _, _) = installed_fixture().await; + let member = WorkspaceMember { + metadata: Some(ObjectMeta { + id: "discovery-member".into(), + name: "discovery-reader".into(), + workspace: "default".into(), + ..Default::default() + }), + principal_subject: "discovery-reader".into(), + role: WorkspaceRole::User.into(), + }; + state.store.put_message(&member).await.unwrap(); + let mut request = authed_request(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + }); + if let Some(Principal::User(user)) = request.extensions_mut().get_mut::() { + user.identity.roles = vec!["openshell-user".into()]; + user.identity.subject = "discovery-reader".into(); + } + let mut stream = watch(&state, request).await.unwrap().into_inner(); + assert!(stream.next().await.unwrap().is_ok()); + state + .store + .delete("workspace_member", "discovery-member") + .await + .unwrap(); + let error = tokio::time::timeout(Duration::from_secs(3), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::PermissionDenied); + assert!(stream.next().await.is_none()); + assert_eq!(state.discovery_watches.available_permits(), 32); + } + + #[tokio::test] + async fn quiet_watch_ends_at_lifetime_and_releases_capacity() { + let (state, _, _) = installed_fixture().await; + let mut stream = watch( + &state, + authed_request(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + }), + ) + .await + .unwrap() + .into_inner(); + assert!(stream.next().await.unwrap().is_ok()); + tokio::task::yield_now().await; + tokio::time::pause(); + tokio::time::advance(WATCH_LIFETIME).await; + assert!( + tokio::time::timeout(WATCH_LIFETIME + REFRESH_INTERVAL, stream.next()) + .await + .unwrap() + .is_none() + ); + assert_eq!(state.discovery_watches.available_permits(), 32); + } + + #[tokio::test] + async fn collection_overflow_corruption_and_storage_failure_are_errors_not_empty_views() { + let (state, _, _) = installed_fixture().await; + state + .store + .put_if( + OBJECT_TYPE, + "broken", + "broken", + "default", + b"not-json", + None, + crate::persistence::WriteCondition::MustCreate, + ) + .await + .unwrap(); + assert_eq!( + snapshot(&state, "default").await.unwrap_err().code(), + tonic::Code::Internal + ); + state.store.delete(OBJECT_TYPE, "broken").await.unwrap(); + let bytes = vec![b'x'; MAX_BYTES + 1]; + state + .store + .put_if( + OBJECT_TYPE, + "oversized", + "oversized", + "default", + &bytes, + None, + crate::persistence::WriteCondition::MustCreate, + ) + .await + .unwrap(); + assert_eq!( + snapshot(&state, "default").await.unwrap_err().code(), + tonic::Code::ResourceExhausted + ); + state.store.delete(OBJECT_TYPE, "oversized").await.unwrap(); + state.store.close().await; + assert_eq!( + snapshot(&state, "default").await.unwrap_err().code(), + tonic::Code::Unavailable + ); + } +} diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index b68ba0d89c..c34aaa7890 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -319,6 +319,8 @@ pub(crate) struct ServerStartupConfig { /// Server state shared across handlers. #[derive(Debug)] pub struct ServerState { + /// Process-local bound on active discovery streams. + pub discovery_watches: Arc, /// Server configuration. pub config: Config, @@ -658,6 +660,7 @@ impl ServerState { service_route_index, gateway_ring: Arc::new(std::sync::RwLock::new(gateway_ring::GatewayRing::default())), gateway_peers: Arc::new(std::sync::RwLock::new(HashMap::new())), + discovery_watches: Arc::new(tokio::sync::Semaphore::new(32)), extension_mint_limiter: auth::extension_mint_limit::ExtensionMintLimiter::default(), middleware_registry: Arc::new(MiddlewareRegistry::default()), oidc_cache, diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index d8da420871..8613a60f91 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -178,7 +178,7 @@ mod tests { // image wire identity, now with explicit presence. Compute configuration // updates use a separate capability/service and add one classified RPC. const PUBLIC_RPC_SCHEMA_SHA256: &str = - "3b4c2af6cf22e7ed1d84279ff865acb0db7199717bee9bc041caef791334104d"; + "1fce709ed9a1c91e307c46c5fe68be470cbe88ca70ea0087e771c973a9052308"; const DURABLE_SCHEMA_SHA256: &str = "cb15f8f79a855064a43fe8c6cd908bf07eca6604e2bf8ab7d2de0ef33e97da4e"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = @@ -652,7 +652,7 @@ mod tests { } assert_eq!( compiled_method_count, - 109 + PROVIDER_READINESS_RPC_SIGNATURES.len() + 111 + PROVIDER_READINESS_RPC_SIGNATURES.len() + PEER_OWNER_RPC_SIGNATURES.len() + PEER_CONFIG_RPC_SIGNATURES.len() + isolation_methods.len() @@ -662,7 +662,7 @@ mod tests { ); assert_eq!( methods.len(), - 81 + PROVIDER_READINESS_RPC_SIGNATURES.len() + 83 + PROVIDER_READINESS_RPC_SIGNATURES.len() + PEER_OWNER_RPC_SIGNATURES.len() + PEER_CONFIG_RPC_SIGNATURES.len(), "inventory every public gateway RPC" @@ -672,7 +672,7 @@ mod tests { .iter() .filter(|method| method.starts_with("openshell.v1.OpenShell/")) .count(), - 80 + PROVIDER_READINESS_RPC_SIGNATURES.len() + 82 + PROVIDER_READINESS_RPC_SIGNATURES.len() + PEER_OWNER_RPC_SIGNATURES.len() + PEER_CONFIG_RPC_SIGNATURES.len() ); @@ -723,7 +723,7 @@ mod tests { overlap_hash.as_str(), ), ( - (337, 31), + (344, 31), (100, 22), (86, 22), PUBLIC_RPC_SCHEMA_SHA256, diff --git a/crates/openshell-server/tests/common/mod.rs b/crates/openshell-server/tests/common/mod.rs index dd91664d39..a0e4f5f4ff 100644 --- a/crates/openshell-server/tests/common/mod.rs +++ b/crates/openshell-server/tests/common/mod.rs @@ -52,6 +52,20 @@ pub struct TestOpenShell; #[tonic::async_trait] impl OpenShell for TestOpenShell { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = + ReceiverStream>; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/crates/openshell-server/tests/supervisor_relay_integration.rs b/crates/openshell-server/tests/supervisor_relay_integration.rs index ceb2fa6bf8..f858ebf163 100644 --- a/crates/openshell-server/tests/supervisor_relay_integration.rs +++ b/crates/openshell-server/tests/supervisor_relay_integration.rs @@ -50,6 +50,20 @@ struct RelayGateway { #[tonic::async_trait] impl OpenShell for RelayGateway { + async fn get_service_route_snapshot( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } + type WatchServiceRoutesStream = + ReceiverStream>; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } async fn peer_notify_config_update( &self, _request: tonic::Request, diff --git a/docs/how-it-works/gateways/configuration.mdx b/docs/how-it-works/gateways/configuration.mdx index 175836ccb4..402c4f4cd1 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -1928,9 +1928,9 @@ Expose services with the existing commands. For direct listener testing, an unnamed service on sandbox `demo` in workspace `default` uses SNI and Host `default--demo.services.example.com`; a service named `web` uses `default--demo--web.services.example.com`. Configure DNS or your ingress to reach -the owning supervisor while preserving that hostname. Automatic discovery and -application URL publication are separate work; current CLI URLs still identify -the existing gateway path. +the owning supervisor while preserving that hostname. The discovery API supplies accepted supervisor endpoints. Application URL +publication is separate work; current CLI URLs still identify the existing +gateway path. Enabled supervisors report installed service routes through their authenticated gateway sessions. The gateway accepts only current configuration and ownership, @@ -1951,3 +1951,12 @@ application streams even if configuration continues renewing. Under a partition, allow for pool membership failure detection before the last serving lease expires. Storage errors prevent lease renewal. Discovery expiry and serving expiry are separate from external ingress readiness. + +Discovery clients can call `GetServiceRouteSnapshot` or `WatchServiceRoutes` with +one explicit workspace selector, `sandbox:read`, and workspace user access. Each +response replaces the workspace route view and includes the hostname, supervisor +TLS endpoint, installation version, and expiry. Preserve the hostname as TLS SNI +and HTTP authority when routing directly to the endpoint. Remove absent routes +and expire cached entries even during discovery errors. An API error does not +mean the workspace has no routes. Watches end after one minute; reconnect and +relist when they end. Discovery-only machine credentials are separate work. diff --git a/proto/openshell.proto b/proto/openshell.proto index b40ec68263..17d86452d9 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -251,6 +251,24 @@ service OpenShell { }; } + // Read installed, unexpired application routes in one explicit workspace. + rpc GetServiceRouteSnapshot(GetServiceRouteSnapshotRequest) returns (GetServiceRouteSnapshotResponse) { + option (openshell.options.v1.authorization) = { + auth_mode: "bearer" + scope: "sandbox:read" + workspace_role: "user" + }; + } + + // Full replacement views; reauthorize during the bounded watch lifetime. + rpc WatchServiceRoutes(WatchServiceRoutesRequest) returns (stream WatchServiceRoutesResponse) { + option (openshell.options.v1.authorization) = { + auth_mode: "bearer" + scope: "sandbox:read" + workspace_role: "user" + }; + } + // Delete one sandbox HTTP service endpoint. rpc DeleteService(DeleteServiceRequest) returns (DeleteServiceResponse) { option (openshell.options.v1.authorization) = { @@ -4245,3 +4263,47 @@ message SupervisorServiceAdvertisement { // Domain actually installed by the process listener, not the gateway URL. string service_domain = 5; } + +// One explicit active workspace is required. Omitted and all-workspaces +// selectors are rejected; discovery-only machine grants are separate work. +message GetServiceRouteSnapshotRequest { + openshell.datamodel.v1.WorkspaceSelector workspace_scope = 1; +} +message GetServiceRouteSnapshotResponse { + ServiceRouteSnapshot snapshot = 1; +} +// Same workspace authorization as GetServiceRouteSnapshot. A watch sends an +// initial replacement and later replacements, including lease-only renewal. +// Clients reconnect/relist when the bounded stream ends and expire cached +// routes during discovery errors; errors are not empty replacements. +message WatchServiceRoutesRequest { + openshell.datamodel.v1.WorkspaceSelector workspace_scope = 1; +} +message WatchServiceRoutesResponse { + ServiceRouteSnapshot snapshot = 1; +} +message ServiceRouteSnapshot { + repeated DiscoveredServiceRoute routes = 1; + // Opaque equality token for this complete view, including lease expiries. + // No ordering across replicas or server restarts is implied. + string revision = 2; +} +// Trusted supervisor TLS endpoint. Preserve DiscoveredServiceRoute.hostname +// as SNI and HTTP authority; this endpoint never addresses a workload port. +message ServiceRouteEndpoint { + string host = 1; + uint32 port = 2; +} +// Accepted installation evidence, not application health or ingress readiness. +message DiscoveredServiceRoute { + string workspace = 1; + string sandbox = 2; + // Canonical service name; empty identifies the existing unnamed service. + string name = 3; + string hostname = 4; + ServiceRouteEndpoint endpoint = 5; + // Changes on owner/backend/declaration replacement, but not lease renewal. + string installation_version = 6; + uint64 declaration_revision = 7; + google.protobuf.Timestamp expires_at = 8; +} diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index ec8c9dee6f..36c4abf57e 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -1688,8 +1688,8 @@ renew a 30-second lease; rejected configuration or expired freshness closes routes. When the listener is enabled, polling delays are capped at five seconds even if the configured interval is longer; failed polls do not renew the lease. App traffic uses application/edge authentication independently of gateway -user authentication. This listener stage does not add service discovery or -change the public URLs printed by the CLI. +user authentication. The discovery API publishes accepted listener installations; public URLs +printed by the CLI continue to use the gateway path. When enabled application routes are not published, check the authenticated supervisor session and installed configuration before checking ingress. The @@ -1701,7 +1701,14 @@ compute driver once per challenge/report exchange, with fresh resolution for the next challenge. Acceptance still checks durable ownership and shared member readiness; ambiguous shared ownership must not publish an address. -For shared pools, compare the discovered endpoint to the +For discovery troubleshooting, request `GetServiceRouteSnapshot` for an explicit +active workspace using an identity with `sandbox:read` and workspace user access. +`WatchServiceRoutes` rechecks access every second and ends after one minute; +clients must reconnect/relist. Compare installation versions separately from +snapshot revisions: lease renewal changes expiry and the snapshot revision while +preserving installation identity. Treat unavailable/corrupt/over-limit results as +errors, and expire cached routes independently. Do not turn an error into an +empty replacement. For shared pools, compare the discovered endpoint to the acknowledged owner's Pod IP and configured app port, not its runtime port or a load-balanced supervisor Service. Publication requires a current ownership term, live member, and acknowledged ready report. During takeover old streams may close; From 9a4abc91e65394d14fd59670b362cf86748dc9cd Mon Sep 17 00:00:00 2001 From: Taylor Mutch Date: Mon, 2 Nov 2026 15:59:11 -0800 Subject: [PATCH 2/3] fix(gateway): gate service discovery on the app listener Signed-off-by: Taylor Mutch --- crates/openshell-server/README.md | 4 ++ .../src/grpc/service_discovery.rs | 43 +++++++++++++++++++ docs/how-it-works/gateways/configuration.mdx | 4 ++ skills/debug-openshell-cluster/SKILL.md | 2 + 4 files changed, 53 insertions(+) diff --git a/crates/openshell-server/README.md b/crates/openshell-server/README.md index 56a2c27a7b..0a37f8d496 100644 --- a/crates/openshell-server/README.md +++ b/crates/openshell-server/README.md @@ -264,6 +264,10 @@ records as application health or external ingress readiness. ## Workspace service discovery +Both discovery RPCs require `supervisor_app_listener.enabled = true`. When the +listener is disabled, they return `FAILED_PRECONDITION` without collecting routes +or starting a watch. + `GetServiceRouteSnapshot` and `WatchServiceRoutes` expose accepted installations for one explicit active workspace. They require `sandbox:read` and workspace user access; omitted/all-workspaces selectors and sandbox principals are rejected. diff --git a/crates/openshell-server/src/grpc/service_discovery.rs b/crates/openshell-server/src/grpc/service_discovery.rs index d552833d51..e3b0f58ff7 100644 --- a/crates/openshell-server/src/grpc/service_discovery.rs +++ b/crates/openshell-server/src/grpc/service_discovery.rs @@ -57,11 +57,21 @@ async fn authorize( Ok(()) } +fn require_enabled(state: &ServerState) -> Result<(), Status> { + if !state.config.supervisor_app_listener.enabled { + return Err(Status::failed_precondition( + "service discovery requires the supervisor app listener to be enabled", + )); + } + Ok(()) +} + pub async fn get_snapshot( state: &Arc, request: Request, ) -> Result, Status> { let principal = super::extract_principal(&request)?; + require_enabled(state)?; let workspace = selected_workspace_name(request.get_ref().workspace_scope.as_ref())?; authorize(state, &principal, workspace).await?; Ok(Response::new(GetServiceRouteSnapshotResponse { @@ -74,6 +84,7 @@ pub async fn watch( request: Request, ) -> Result, Status> { let principal = super::extract_principal(&request)?; + require_enabled(state)?; let workspace = selected_workspace_name(request.get_ref().workspace_scope.as_ref())?.to_string(); authorize(state, &principal, &workspace).await?; @@ -300,6 +311,38 @@ mod tests { } } + #[tokio::test] + async fn disabled_listener_rejects_discovery_without_allocating_a_watch() { + let (mut state, owner, report) = installed_fixture().await; + // Persist a route first: disabling must reject discovery even with stored evidence. + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .unwrap(); + Arc::get_mut(&mut state) + .unwrap() + .config + .supervisor_app_listener + .enabled = false; + let permits = state.discovery_watches.available_permits(); + let error = get_snapshot(&state, authed_request(request())) + .await + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::FailedPrecondition); + assert!(error.message().contains("app listener")); + let error = watch( + &state, + authed_request(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + ..Default::default() + }), + ) + .await + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::FailedPrecondition); + assert!(error.message().contains("app listener")); + assert_eq!(state.discovery_watches.available_permits(), permits); + } + #[tokio::test] async fn snapshot_is_shared_between_gateway_states_and_rejects_expired_or_replaced_owners() { let (state, owner, report) = installed_fixture().await; diff --git a/docs/how-it-works/gateways/configuration.mdx b/docs/how-it-works/gateways/configuration.mdx index 402c4f4cd1..66f9848ebc 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -1952,6 +1952,10 @@ allow for pool membership failure detection before the last serving lease expire Storage errors prevent lease renewal. Discovery expiry and serving expiry are separate from external ingress readiness. +Both discovery RPCs require `supervisor_app_listener.enabled = true`. When the +listener is disabled, they return `FAILED_PRECONDITION` without collecting routes +or starting a watch. + Discovery clients can call `GetServiceRouteSnapshot` or `WatchServiceRoutes` with one explicit workspace selector, `sandbox:read`, and workspace user access. Each response replaces the workspace route view and includes the hostname, supervisor diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 36c4abf57e..21ec64595e 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -1703,6 +1703,8 @@ readiness; ambiguous shared ownership must not publish an address. For discovery troubleshooting, request `GetServiceRouteSnapshot` for an explicit active workspace using an identity with `sandbox:read` and workspace user access. +Both discovery RPCs return `FAILED_PRECONDITION` when +`supervisor_app_listener.enabled` is false; enable the listener before discovering routes. `WatchServiceRoutes` rechecks access every second and ends after one minute; clients must reconnect/relist. Compare installation versions separately from snapshot revisions: lease renewal changes expiry and the snapshot revision while From 217848261bc15e0350a7f285c90448e0b66d098b Mon Sep 17 00:00:00 2001 From: Taylor Mutch Date: Wed, 4 Nov 2026 14:12:52 -0800 Subject: [PATCH 3/3] test(discovery): remove redundant request defaults before reader extension Signed-off-by: Taylor Mutch --- crates/openshell-server/src/grpc/service_discovery.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/openshell-server/src/grpc/service_discovery.rs b/crates/openshell-server/src/grpc/service_discovery.rs index e3b0f58ff7..eacb348f6e 100644 --- a/crates/openshell-server/src/grpc/service_discovery.rs +++ b/crates/openshell-server/src/grpc/service_discovery.rs @@ -333,7 +333,6 @@ mod tests { &state, authed_request(WatchServiceRoutesRequest { workspace_scope: request().workspace_scope, - ..Default::default() }), ) .await