From 2f45cba650b8877c640a04047d7a8bce728ede8a Mon Sep 17 00:00:00 2001 From: Taylor Mutch Date: Mon, 26 Oct 2026 15:29:32 -0700 Subject: [PATCH 1/4] 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 3c52ebb127..600c6154ea 100644 --- a/crates/openshell-server/README.md +++ b/crates/openshell-server/README.md @@ -295,3 +295,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 69208348de..abebf78e0b 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; @@ -527,6 +528,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 140f372347..bf11c4f1fa 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -322,6 +322,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, @@ -666,6 +668,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 8d39d6bbaf..75980201a4 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -1929,9 +1929,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, @@ -1952,3 +1952,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 23d58efa25..529aba1710 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -1698,8 +1698,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 @@ -1711,7 +1711,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 2fd722666e58c8165d41dbcaef49e2611296ede4 Mon Sep 17 00:00:00 2001 From: Taylor Mutch Date: Mon, 2 Nov 2026 15:59:11 -0800 Subject: [PATCH 2/4] 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 600c6154ea..7c39fd947d 100644 --- a/crates/openshell-server/README.md +++ b/crates/openshell-server/README.md @@ -298,6 +298,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 75980201a4..12d28bce56 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -1953,6 +1953,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 529aba1710..20caa213a2 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -1713,6 +1713,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 fcb9be2936b2cb900ea83e93d45009eeeca37d9a Mon Sep 17 00:00:00 2001 From: Taylor Mutch Date: Wed, 4 Nov 2026 14:12:52 -0800 Subject: [PATCH 3/4] 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 From 591548a4503f6e69eb7e0f22bae083a03d9d359c Mon Sep 17 00:00:00 2001 From: Taylor Mutch Date: Mon, 9 Nov 2026 10:42:40 -0800 Subject: [PATCH 4/4] fix(discovery): separate and bound ingress discovery service 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 | 8 +- .../sandbox_name_fallback_integration.rs | 15 - crates/openshell-sdk/src/client.rs | 19 +- crates/openshell-sdk/src/raw.rs | 9 + crates/openshell-sdk/tests/client_mock.rs | 99 +- crates/openshell-server/README.md | 23 +- crates/openshell-server/src/grpc/mod.rs | 37 +- .../src/grpc/service_advertisement.rs | 176 +-- .../src/grpc/service_discovery.rs | 1002 ++++++++++++++--- crates/openshell-server/src/lib.rs | 10 +- crates/openshell-server/src/multiplex.rs | 75 +- crates/openshell-server/src/reflection.rs | 6 +- .../openshell-server/src/service_routing.rs | 18 +- crates/openshell-server/src/storage_proto.rs | 11 +- crates/openshell-server/tests/common/mod.rs | 14 - .../tests/supervisor_relay_integration.rs | 14 - docs/how-it-works/gateways/configuration.mdx | 17 +- proto/openshell.proto | 40 +- sdk/typescript/src/raw.ts | 2 +- skills/debug-openshell-cluster/SKILL.md | 10 + 24 files changed, 1243 insertions(+), 422 deletions(-) diff --git a/crates/openshell-cli/tests/ensure_providers_integration.rs b/crates/openshell-cli/tests/ensure_providers_integration.rs index 29be137f18..be3e4f2e49 100644 --- a/crates/openshell-cli/tests/ensure_providers_integration.rs +++ b/crates/openshell-cli/tests/ensure_providers_integration.rs @@ -80,21 +80,6 @@ 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 f2b569de96..1017712035 100644 --- a/crates/openshell-cli/tests/mtls_integration.rs +++ b/crates/openshell-cli/tests/mtls_integration.rs @@ -32,21 +32,6 @@ 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 f14805f17d..89247ba718 100644 --- a/crates/openshell-cli/tests/provider_commands_integration.rs +++ b/crates/openshell-cli/tests/provider_commands_integration.rs @@ -235,21 +235,6 @@ 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 bc8197872d..587910fbaf 100644 --- a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs +++ b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs @@ -102,21 +102,6 @@ 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 4238254ed4..772a26e58e 100644 --- a/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs +++ b/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs @@ -163,13 +163,7 @@ 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 c218a7fcb7..497f3ce778 100644 --- a/crates/openshell-cli/tests/sandbox_name_fallback_integration.rs +++ b/crates/openshell-cli/tests/sandbox_name_fallback_integration.rs @@ -47,21 +47,6 @@ 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/src/client.rs b/crates/openshell-sdk/src/client.rs index 8241ac49b5..da415535d9 100644 --- a/crates/openshell-sdk/src/client.rs +++ b/crates/openshell-sdk/src/client.rs @@ -12,7 +12,7 @@ use crate::auth::{BearerSlot, EdgeAuthInterceptor, bearer_metadata}; use crate::config::{AuthConfig, ClientConfig}; use crate::error::{Result, SdkError}; use crate::pagination::{Page, Pager}; -use crate::raw::{AuthedGrpcClient, AuthedSandboxUpdatesClient}; +use crate::raw::{AuthedGrpcClient, AuthedSandboxUpdatesClient, AuthedServiceDiscoveryClient}; use crate::refresh::{RefreshedToken, TokenSource}; use crate::routing::OwnerCache; use crate::types::{ @@ -149,6 +149,23 @@ impl OpenShellClient { Ok(self.raw_sandbox_updates()) } + /// Authenticated client for the separate ingress discovery service. + /// Shares the channel and live auth slot; like [`Self::raw_grpc`], does not + /// refresh or retry credentials by itself. + pub fn raw_service_discovery(&self) -> AuthedServiceDiscoveryClient { + proto::service_discovery_client::ServiceDiscoveryClient::with_interceptor( + self.channel.clone(), + self.interceptor.clone(), + ) + } + + /// Like [`Self::raw_service_discovery`], refreshing bearer credentials first. + /// Reactive retry remains the raw caller's responsibility. + pub async fn raw_service_discovery_fresh(&self) -> Result { + self.ensure_fresh().await?; + Ok(self.raw_service_discovery()) + } + /// Force an OIDC refresh and write the new token into the live bearer /// slot, regardless of expiry. Returns `true` when a refresher is wired /// and a fresh token was minted, `false` for static auth. Use after a diff --git a/crates/openshell-sdk/src/raw.rs b/crates/openshell-sdk/src/raw.rs index 15e26d52cc..4bfe7676a7 100644 --- a/crates/openshell-sdk/src/raw.rs +++ b/crates/openshell-sdk/src/raw.rs @@ -21,6 +21,7 @@ pub use openshell_core::proto; pub use openshell_core::proto::open_shell_client::OpenShellClient as GrpcClient; pub use openshell_core::proto::sandbox_updates_client::SandboxUpdatesClient; +pub use openshell_core::proto::service_discovery_client::ServiceDiscoveryClient; pub use openshell_core::proto::{ CreateSandboxRequest, CreateSandboxTemplateRequest, CreateWorkspaceRequest, DeleteSandboxRequest, DeleteSandboxTemplateRequest, DeleteWorkspaceRequest, ExecSandboxRequest, @@ -53,3 +54,11 @@ pub type AuthedSandboxUpdatesClient = SandboxUpdatesClient< crate::EdgeAuthInterceptor, >, >; + +/// Application ingress discovery using the SDK transport and live auth slot. +pub type AuthedServiceDiscoveryClient = ServiceDiscoveryClient< + tonic::service::interceptor::InterceptedService< + tonic::transport::Channel, + crate::EdgeAuthInterceptor, + >, +>; diff --git a/crates/openshell-sdk/tests/client_mock.rs b/crates/openshell-sdk/tests/client_mock.rs index 3e8cee1579..f98a180eea 100644 --- a/crates/openshell-sdk/tests/client_mock.rs +++ b/crates/openshell-sdk/tests/client_mock.rs @@ -245,20 +245,6 @@ 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, @@ -1209,6 +1195,35 @@ impl SandboxUpdates for TestOpenShell { } } +#[tonic::async_trait] +impl proto::service_discovery_server::ServiceDiscovery for TestOpenShell { + async fn get_service_route_snapshot( + &self, + request: tonic::Request, + ) -> Result, Status> { + if let Some(expected) = &self.state.require_bearer + && request + .metadata() + .get("authorization") + .and_then(|v| v.to_str().ok()) + != Some(expected.as_str()) + { + return Err(Status::unauthenticated("invalid or missing bearer token")); + } + Ok(Response::new(proto::GetServiceRouteSnapshotResponse { + snapshot: Some(proto::ServiceRouteSnapshot::default()), + })) + } + type WatchServiceRoutesStream = + tokio_stream::wrappers::ReceiverStream>; + async fn watch_service_routes( + &self, + _: tonic::Request, + ) -> Result, Status> { + Err(Status::unimplemented("unused")) + } +} + /// Spin up the mock gateway, return its endpoint URL. async fn start_mock(state: Arc) -> String { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -1218,11 +1233,15 @@ async fn start_mock(state: Arc) -> String { let updates = SandboxUpdatesServer::new(TestOpenShell { state: state.clone(), }); + let discovery = proto::service_discovery_server::ServiceDiscoveryServer::new(TestOpenShell { + state: state.clone(), + }); let svc = OpenShellServer::new(TestOpenShell { state }); tokio::spawn(async move { let _ = tonic::transport::Server::builder() .add_service(svc) .add_service(updates) + .add_service(discovery) .serve_with_incoming(stream) .await; }); @@ -2471,6 +2490,58 @@ async fn raw_sandbox_updates_shares_transport_and_live_auth_slot() { assert_eq!(*state.last_update.lock().await, Some(request)); } +#[tokio::test] +async fn raw_discovery_shares_transport_and_refreshes_the_live_auth_slot() { + let state = Arc::new(MockState { + require_bearer: Some("Bearer fresh-token".into()), + ..Default::default() + }); + let endpoint = start_mock(state).await; + let calls = Arc::new(AtomicU32::new(0)); + let mut config = ClientConfig::new(&endpoint); + config.auth = Some(AuthConfig::Oidc { + token: "stale-token".into(), + expires_at: Some( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() + + 5, + ), + refresh: Some(Arc::new(OneShotRefresher { + calls: calls.clone(), + })), + }); + let client = OpenShellClient::connect(config).await.unwrap(); + let mut discovery = client.raw_service_discovery(); + let request = proto::GetServiceRouteSnapshotRequest { + workspace_scope: Some(proto::workspace_selector("team")), + }; + assert_eq!( + discovery + .get_service_route_snapshot(request.clone()) + .await + .unwrap_err() + .code(), + tonic::Code::Unauthenticated + ); + assert_eq!(calls.load(Ordering::SeqCst), 0); + client + .raw_service_discovery_fresh() + .await + .unwrap() + .get_service_route_snapshot(request.clone()) + .await + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + discovery.get_service_route_snapshot(request).await.unwrap(); + client + .raw_grpc() + .health(proto::HealthRequest {}) + .await + .unwrap(); +} + #[tokio::test] async fn watch_logs_forwards_logs_and_warnings() { let state = Arc::new(MockState { diff --git a/crates/openshell-server/README.md b/crates/openshell-server/README.md index 7c39fd947d..d8b27b05a8 100644 --- a/crates/openshell-server/README.md +++ b/crates/openshell-server/README.md @@ -302,7 +302,7 @@ 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 +`openshell.v1.ServiceDiscovery.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. @@ -315,11 +315,18 @@ 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 +Collection validates each installation once; lease renewals during a scan do not +invalidate the view. This is not a cross-resource transaction. Reads are bounded +to 1,000 records, 4,096 routes, 4 MiB, and ten seconds. A per-workspace cache shares +one collection for a one-second refresh interval, retains at most 16 workspaces, +and never returns expired routes. At most four collections and 16 unary snapshot +requests run concurrently per gateway. Unary admission permits four requests per +principal and eight per workspace. Overflow returns `RESOURCE_EXHAUSTED`; storage +failure or timeout returns `UNAVAILABLE`; malformed records return `INTERNAL`. +Watches recheck bearer authentication and workspace authorization each second, +use a one-message buffer, and end after one minute or a five-second blocked send. +Each gateway allows 32 watches, at most four per principal and 16 per workspace. +Cache reuse adds at most one refresh interval to collection freshness; consumers +must independently honor the advertised lease expiry. +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 abebf78e0b..b006c7268e 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -14,7 +14,7 @@ pub mod supervisor_host; pub use sandbox::mint_persisted_authentication; mod service; pub mod service_advertisement; -mod service_discovery; +pub mod service_discovery; mod validation; pub mod workspace; @@ -528,22 +528,6 @@ 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, @@ -1452,3 +1436,22 @@ mod tests { assert_eq!(public.protocol_minor, 2); } } + +#[tonic::async_trait] +impl openshell_core::proto::service_discovery_server::ServiceDiscovery for OpenShellService { + 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 + } +} diff --git a/crates/openshell-server/src/grpc/service_advertisement.rs b/crates/openshell-server/src/grpc/service_advertisement.rs index cc4c8af52c..3e953c1120 100644 --- a/crates/openshell-server/src/grpc/service_advertisement.rs +++ b/crates/openshell-server/src/grpc/service_advertisement.rs @@ -468,12 +468,19 @@ async fn accept_inner( /// Discovery must call this before publishing each persisted installation. /// Freshness failures are absence; storage/decoding failures are errors. pub async fn is_current(state: &ServerState, record: &Installation) -> Result { + Ok(current_routes(state, record).await?.is_some()) +} + +pub(super) async fn current_routes( + state: &ServerState, + record: &Installation, +) -> Result>, Status> { if !state.config.supervisor_app_listener.enabled || record.expires_at_ms <= current_time_ms() || record.service_domain != state.config.supervisor_app_listener.domain || record.routes.is_empty() { - return Ok(false); + return Ok(None); } let Some(sandbox) = state .store @@ -481,7 +488,7 @@ pub async fn is_current(state: &ServerState, record: &Installation) -> Result Result Result Result Result Result, mut shutdown: tokio::sync::watch:: } #[cfg(test)] -pub mod tests { +mod tests { + use super::test_support::installed_fixture; use super::*; fn test_lease_authority() -> LeaseAuthority { @@ -1072,6 +1082,82 @@ pub mod tests { ); } + #[tokio::test] + async fn accepted_installs_use_trusted_endpoints_and_withdraw_on_declaration_or_owner_change() { + let (state, owner, mut report) = installed_fixture().await; + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .unwrap(); + let stored = state + .store + .get(OBJECT_TYPE, "service-installation:sandbox") + .await + .unwrap() + .unwrap(); + let installation: Installation = serde_json::from_slice(&stored.payload).unwrap(); + assert_eq!(installation.host, "10.0.0.7"); + assert_eq!(installation.port, 9443); + assert!(is_current(&state, &installation).await.unwrap()); + report.lease_sequence += 1; + report.routes[0].resource_version += 1; + assert!( + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .is_err() + ); + let mut endpoint = state + .store + .get_message::("endpoint") + .await + .unwrap() + .unwrap(); + endpoint.target_port = 9090; + state.store.put_message(&endpoint).await.unwrap(); + assert!(!is_current(&state, &installation).await.unwrap()); + let index = SupervisorOwnerIndex::new(state.store.clone(), OWNER_TTL); + index + .publish( + "sandbox", + "replacement", + "instance", + 2, + "other-replica", + "local://other", + ) + .await + .unwrap(); + assert!( + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .is_err() + ); + } + + #[test] + fn reports_cannot_broaden_or_partially_install_declarations() { + let declared = route(); + let mut report = SupervisorServiceAdvertisement { + routes: vec![declared.clone()], + ..Default::default() + }; + validate_routes(&report, std::slice::from_ref(&declared)).unwrap(); + report.routes[0].target_port = 9090; + assert!(validate_routes(&report, std::slice::from_ref(&declared)).is_err()); + report.routes = vec![declared.clone(), declared.clone()]; + assert!(validate_routes(&report, std::slice::from_ref(&declared)).is_err()); + report.routes = vec![declared.clone()]; + let mut second = declared.clone(); + second.endpoint_id = "second".into(); + second.name = "other".into(); + assert!(validate_routes(&report, &[declared.clone(), second]).is_err()); + report.routes.clear(); + validate_routes(&report, &[declared]).unwrap(); + } +} + +#[cfg(test)] +pub(super) mod test_support { + use super::*; pub async fn installed_fixture() -> (Arc, OwnerGuard, SupervisorServiceAdvertisement) { use openshell_core::proto::datamodel::v1::ObjectMeta; @@ -1167,76 +1253,4 @@ pub mod tests { }; (state, owner, report) } - - #[tokio::test] - async fn accepted_installs_use_trusted_endpoints_and_withdraw_on_declaration_or_owner_change() { - let (state, owner, mut report) = installed_fixture().await; - accept(&state, &owner, &report, current_time_ms() + 10_000) - .await - .unwrap(); - let stored = state - .store - .get(OBJECT_TYPE, "service-installation:sandbox") - .await - .unwrap() - .unwrap(); - let installation: Installation = serde_json::from_slice(&stored.payload).unwrap(); - assert_eq!(installation.host, "10.0.0.7"); - assert_eq!(installation.port, 9443); - assert!(is_current(&state, &installation).await.unwrap()); - report.lease_sequence += 1; - report.routes[0].resource_version += 1; - assert!( - accept(&state, &owner, &report, current_time_ms() + 10_000) - .await - .is_err() - ); - let mut endpoint = state - .store - .get_message::("endpoint") - .await - .unwrap() - .unwrap(); - endpoint.target_port = 9090; - state.store.put_message(&endpoint).await.unwrap(); - assert!(!is_current(&state, &installation).await.unwrap()); - let index = SupervisorOwnerIndex::new(state.store.clone(), OWNER_TTL); - index - .publish( - "sandbox", - "replacement", - "instance", - 2, - "other-replica", - "local://other", - ) - .await - .unwrap(); - assert!( - accept(&state, &owner, &report, current_time_ms() + 10_000) - .await - .is_err() - ); - } - - #[test] - fn reports_cannot_broaden_or_partially_install_declarations() { - let declared = route(); - let mut report = SupervisorServiceAdvertisement { - routes: vec![declared.clone()], - ..Default::default() - }; - validate_routes(&report, std::slice::from_ref(&declared)).unwrap(); - report.routes[0].target_port = 9090; - assert!(validate_routes(&report, std::slice::from_ref(&declared)).is_err()); - report.routes = vec![declared.clone(), declared.clone()]; - assert!(validate_routes(&report, std::slice::from_ref(&declared)).is_err()); - report.routes = vec![declared.clone()]; - let mut second = declared.clone(); - second.endpoint_id = "second".into(); - second.name = "other".into(); - assert!(validate_routes(&report, &[declared.clone(), second]).is_err()); - report.routes.clear(); - validate_routes(&report, &[declared]).unwrap(); - } } diff --git a/crates/openshell-server/src/grpc/service_discovery.rs b/crates/openshell-server/src/grpc/service_discovery.rs index eacb348f6e..18eafaf89f 100644 --- a/crates/openshell-server/src/grpc/service_discovery.rs +++ b/crates/openshell-server/src/grpc/service_discovery.rs @@ -8,15 +8,15 @@ use std::time::Duration; use openshell_core::proto::{ DiscoveredServiceRoute, GetServiceRouteSnapshotRequest, GetServiceRouteSnapshotResponse, - Sandbox, SandboxServiceRoute, ServiceRouteEndpoint, ServiceRouteSnapshot, - WatchServiceRoutesRequest, WatchServiceRoutesResponse, + Sandbox, 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 super::service_advertisement::{Installation, OBJECT_TYPE, current_routes}; use crate::ServerState; use crate::auth::principal::Principal; use crate::auth::workspace_authz::{ @@ -33,6 +33,193 @@ const REFRESH_INTERVAL: Duration = Duration::from_secs(1); const SEND_TIMEOUT: Duration = Duration::from_secs(5); pub type RouteWatchStream = ReceiverStream>; +// Per-process limits. Key maps retain only live permits; callers cannot grow +// them indefinitely by varying subjects or workspace names. +const MAX_UNARY: usize = 16; +const MAX_COLLECTIONS: usize = 4; +const MAX_CACHED_WORKSPACES: usize = 16; +const WATCHES_PER_PRINCIPAL: usize = 4; +const WATCHES_PER_WORKSPACE: usize = 16; +type CachedView = Option<(tokio::time::Instant, Result)>; +type CacheSlot = Arc>; + +#[derive(Debug)] +pub struct DiscoveryState { + unary: Arc, + collections: Arc, + shares: std::sync::Mutex< + std::collections::HashMap>, + >, + cache: tokio::sync::Mutex>, +} + +impl Default for DiscoveryState { + fn default() -> Self { + Self { + unary: Arc::new(tokio::sync::Semaphore::new(MAX_UNARY)), + collections: Arc::new(tokio::sync::Semaphore::new(MAX_COLLECTIONS)), + shares: std::sync::Mutex::default(), + cache: tokio::sync::Mutex::default(), + } + } +} + +fn principal_key(principal: &Principal) -> Result { + match principal { + Principal::User(user) => Ok(format!( + "user:{:?}:{}", + user.identity.provider, user.identity.subject + )), + Principal::Operator(operator) => Ok(format!("operator:{}", operator.certificate_sha256)), + _ => Err(Status::permission_denied( + "service discovery requires a user/API identity", + )), + } +} + +impl DiscoveryState { + fn share( + &self, + key: String, + limit: usize, + ) -> Result { + let mut shares = self + .shares + .lock() + .map_err(|_| Status::internal("discovery limits unavailable"))?; + shares.retain(|_, value| value.strong_count() > 0); + let semaphore = shares + .get(&key) + .and_then(std::sync::Weak::upgrade) + .unwrap_or_else(|| { + let semaphore = Arc::new(tokio::sync::Semaphore::new(limit)); + shares.insert(key, Arc::downgrade(&semaphore)); + semaphore + }); + semaphore + .try_acquire_owned() + .map_err(|_| Status::resource_exhausted("discovery caller or workspace limit exceeded")) + } + + pub(crate) fn admit( + &self, + principal: &Principal, + workspaces: &[String], + watch: bool, + ) -> Result, Status> { + let kind = if watch { "watch" } else { "snapshot" }; + let mut permits = Vec::new(); + if !watch { + permits.push( + self.unary.clone().try_acquire_owned().map_err(|_| { + Status::resource_exhausted("too many discovery snapshot requests") + })?, + ); + } + permits.push(self.share( + format!("{kind}:{}", principal_key(principal)?), + WATCHES_PER_PRINCIPAL, + )?); + for workspace in workspaces { + permits.push(self.share( + format!("{kind}:workspace:{workspace}"), + if watch { WATCHES_PER_WORKSPACE } else { 8 }, + )?); + } + Ok(permits) + } + + async fn slot(&self, workspace: &str) -> Result { + let mut cache = self.cache.lock().await; + let now = tokio::time::Instant::now(); + if let Some((used, slot)) = cache.get_mut(workspace) { + *used = now; + return Ok(slot.clone()); + } + if cache.len() >= MAX_CACHED_WORKSPACES { + let oldest = cache + .iter() + .filter(|(_, (_, slot))| Arc::strong_count(slot) == 1) + .min_by_key(|(_, (used, _))| *used) + .map(|(name, _)| name.clone()); + if let Some(oldest) = oldest { + cache.remove(&oldest); + } else { + return Err(Status::resource_exhausted("discovery cache busy")); + } + } + let slot = Arc::new(tokio::sync::Mutex::new(None)); + cache.insert(workspace.to_string(), (now, slot.clone())); + Ok(slot) + } +} + +async fn cached_snapshot( + state: &ServerState, + workspace: &str, +) -> Result { + tokio::time::timeout(SNAPSHOT_TIMEOUT, async { + let slot = state.discovery.slot(workspace).await?; + let mut cached = slot.lock().await; + if let Some((collected, result)) = cached.as_ref() + && collected.elapsed() < REFRESH_INTERVAL + { + // A cached route must never outlive the independently enforced lease. + let snapshot = result.clone()?; + return if snapshot.routes.iter().all(route_is_live) { + Ok(snapshot) + } else { + finish_snapshot(snapshot.routes) + }; + } + let _permit = state + .discovery + .collections + .clone() + .try_acquire_owned() + .map_err(|_| Status::resource_exhausted("too many discovery collections"))?; + let result = snapshot(state, workspace).await; + *cached = Some((tokio::time::Instant::now(), result.clone())); + result + }) + .await + .map_err(|_| Status::unavailable("service discovery collection timed out"))? +} + +/// Captured only by discovery streams; credentials are never logged or persisted. +#[derive(Clone)] +pub struct BearerRecheck { + pub(crate) chain: crate::auth::authenticator::AuthenticatorChain, + pub(crate) headers: http::HeaderMap, + pub(crate) policy: Option, + pub(crate) path: String, +} + +impl BearerRecheck { + pub(crate) async fn check(&self, original: &Principal) -> Result { + use crate::auth::authenticator::Authentication; + let outcome = self + .chain + .authenticate(&self.headers, &self.path) + .await + .map_err(|e| e.status)?; + let Authentication::Accepted(principal) = outcome else { + return Err(Status::unauthenticated( + "discovery credential is no longer valid", + )); + }; + if principal_key(&principal)? != principal_key(original)? { + return Err(Status::unauthenticated("discovery identity changed")); + } + if let (Some(policy), Principal::User(user)) = (&self.policy, &principal) { + policy + .check(&user.identity, &self.path) + .map_err(|e| e.status)?; + } + Ok(principal) + } +} + async fn authorize( state: &ServerState, principal: &Principal, @@ -70,12 +257,20 @@ pub async fn get_snapshot( state: &Arc, request: Request, ) -> Result, Status> { - let principal = super::extract_principal(&request)?; + let mut 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?; + let _permits = state + .discovery + .admit(&principal, &[workspace.to_string()], false)?; + let snapshot = cached_snapshot(state, workspace).await?; + if let Some(recheck) = request.extensions().get::() { + principal = recheck.check(&principal).await?; + } + authorize(state, &principal, workspace).await?; Ok(Response::new(GetServiceRouteSnapshotResponse { - snapshot: Some(snapshot(state, workspace).await?), + snapshot: Some(snapshot), })) } @@ -83,17 +278,25 @@ pub async fn watch( state: &Arc, request: Request, ) -> Result, Status> { - let principal = super::extract_principal(&request)?; + let mut 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?; + let shares = state + .discovery + .admit(&principal, std::slice::from_ref(&workspace), true)?; + let recheck = request.extensions().get::().cloned(); 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 initial = cached_snapshot(state, &workspace).await?; + if let Some(recheck) = &recheck { + principal = recheck.check(&principal).await?; + } + authorize(state, &principal, &workspace).await?; let mut revision = initial.revision.clone(); let (tx, rx) = tokio::sync::mpsc::channel(1); tx.try_send(Ok(WatchServiceRoutesResponse { @@ -103,45 +306,50 @@ pub async fn watch( 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), - }) + let _shares = shares; + let run = async { + 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, + _ = interval.tick() => {} + } + let refresh = async { + authorize(&state, &principal, &workspace).await?; + let snapshot = cached_snapshot(&state, &workspace).await?; + if let Some(recheck) = &recheck { + principal = recheck.check(&principal).await?; + } + authorize(&state, &principal, &workspace).await?; + Ok::<_, Status>(snapshot) + }; + let result = tokio::select! { + () = tx.closed() => 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, + result = tokio::time::timeout(SEND_TIMEOUT, tx.send(message)) => result, + }; + if failed || !matches!(sent, Ok(Ok(()))) { + return; } - 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; } - } + }; + let _ = tokio::time::timeout(WATCH_LIFETIME, run).await; }); Ok(Response::new(ReceiverStream::new(rx))) } @@ -184,124 +392,122 @@ pub async fn snapshot( } 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))) - { + // A bounded per-installation validated view, not a cross-resource transaction. + // Lease renewals must not invalidate an otherwise current installation. + 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"))?; + let Some(installed_routes) = current_routes(state, &installation).await? else { 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", - )); + 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")); } - snapshot.revision = hex::encode( - openshell_crypto::sha256(&bytes) - .map_err(|_| Status::internal("snapshot fingerprint failed"))?, - ); - return Ok(snapshot); + for (bytes, route) in installation.routes.iter().zip(installed_routes) { + let label = crate::service_routing::service_route_readable_label( + workspace, + sandbox.object_name(), + &route.name, + ); + let hostname = format!("{label}.{}", installation.service_domain); + let version = installation_version(&installation, bytes)?; + 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: 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", + )); + } + } + } + finish_snapshot(routes) +} + +fn route_is_live(route: &DiscoveredServiceRoute) -> bool { + route + .expires_at + .as_ref() + .and_then(|t| openshell_core::time::timestamp_to_millis(t).ok()) + .is_some_and(|t| t > current_time_ms()) +} + +fn finish_snapshot( + mut routes: Vec, +) -> Result { + if routes.len() > MAX_ROUTES { + return Err(Status::resource_exhausted( + "too many discovered service routes", + )); } - Err(Status::unavailable( - "service installations changed during collection; retry snapshot", + routes.retain(route_is_live); + 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"))?, + ); + Ok(snapshot) +} + +fn installation_version(installation: &Installation, bytes: &[u8]) -> Result { + 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"))?; + Ok(hex::encode(version)) } #[cfg(test)] mod tests { use super::*; - use crate::grpc::service_advertisement::{accept, tests::installed_fixture}; + use crate::grpc::service_advertisement::{accept, test_support::installed_fixture}; use crate::grpc::test_support::authed_request; use tokio_stream::StreamExt as _; @@ -357,6 +563,20 @@ mod tests { assert_eq!(snapshot_a.routes.len(), 1); let route = &snapshot_a.routes[0]; assert_eq!(route.hostname, "default--demo--web.services.example.com"); + let mut routing_config = state.config.clone(); + routing_config.service_routing.base_domains = vec!["services.example.com".into()]; + let served = crate::service_routing::endpoint_url( + &routing_config, + &route.workspace, + &route.sandbox, + &route.name, + ) + .unwrap(); + assert_eq!( + url::Url::parse(&served).unwrap().host_str(), + Some(route.hostname.as_str()) + ); + 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( @@ -625,4 +845,482 @@ mod tests { tonic::Code::Unavailable ); } + fn user(subject: &str) -> Principal { + let mut request = authed_request(()); + let Principal::User(mut user) = request.extensions_mut().remove::().unwrap() + else { + panic!("user") + }; + user.identity.subject = subject.into(); + Principal::User(user) + } + + fn discovered() -> DiscoveredServiceRoute { + DiscoveredServiceRoute { + workspace: "default".into(), + sandbox: "demo".into(), + name: "web".into(), + hostname: "default--demo--web.services.example.com".into(), + expires_at: Some( + openshell_core::time::timestamp_from_millis(current_time_ms() + 60_000).unwrap(), + ), + ..Default::default() + } + } + + #[test] + fn route_and_wire_size_limits_reject_partial_views() { + assert_eq!( + finish_snapshot(vec![discovered(); MAX_ROUTES]) + .unwrap() + .routes + .len(), + MAX_ROUTES + ); + assert_eq!( + finish_snapshot(vec![discovered(); MAX_ROUTES + 1]) + .unwrap_err() + .code(), + tonic::Code::ResourceExhausted + ); + let mut route = discovered(); + route.hostname = "a".repeat(MAX_BYTES); + assert_eq!( + finish_snapshot(vec![route]).unwrap_err().code(), + tonic::Code::ResourceExhausted + ); + } + + #[tokio::test] + async fn identity_fingerprint_changes_for_every_identity_input_but_not_renewal() { + let (state, owner, report) = installed_fixture().await; + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .unwrap(); + let object = state + .store + .get(OBJECT_TYPE, "service-installation:sandbox") + .await + .unwrap() + .unwrap(); + let installation: Installation = serde_json::from_slice(&object.payload).unwrap(); + let bytes = &installation.routes[0]; + let original = installation_version(&installation, bytes).unwrap(); + let mutations: &[fn(&mut Installation)] = &[ + |i| i.session_id.push('x'), + |i| i.supervisor_instance_id.push('x'), + |i| i.connection_epoch += 1, + |i| i.ownership_epoch += 1, + |i| i.pool_route.push('x'), + |i| i.runtime_identity.push('x'), + |i| i.configuration_instance_id.push('x'), + |i| i.config_revision += 1, + |i| i.service_domain.push('x'), + |i| i.host.push('x'), + |i| i.port += 1, + ]; + for (index, mutate) in mutations.iter().enumerate() { + let mut changed = installation.clone(); + mutate(&mut changed); + assert_ne!( + installation_version(&changed, bytes).unwrap(), + original, + "input {index}" + ); + } + let mut changed_bytes = bytes.clone(); + changed_bytes.push(0); + assert_ne!( + installation_version(&installation, &changed_bytes).unwrap(), + original + ); + let mut renewed = installation.clone(); + renewed.sequence += 1; + renewed.expires_at_ms += 1000; + assert_eq!(installation_version(&renewed, bytes).unwrap(), original); + } + + #[tokio::test] + async fn snapshot_cache_shares_collection_and_refreshes_after_a_tick() { + let (state, owner, report) = installed_fixture().await; + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .unwrap(); + let first = cached_snapshot(&state, "default").await.unwrap(); + assert_eq!(first.routes.len(), 1); + let busy = state + .discovery + .collections + .clone() + .acquire_many_owned(u32::try_from(MAX_COLLECTIONS).unwrap()) + .await + .unwrap(); + let (a, b) = tokio::join!( + cached_snapshot(&state, "default"), + cached_snapshot(&state, "default") + ); + assert_eq!(a.unwrap(), first); + assert_eq!(b.unwrap(), first); + assert_eq!( + cached_snapshot(&state, "other").await.unwrap_err().code(), + tonic::Code::ResourceExhausted + ); + drop(busy); + state + .store + .delete(OBJECT_TYPE, "service-installation:sandbox") + .await + .unwrap(); + tokio::time::sleep(REFRESH_INTERVAL).await; + assert!( + cached_snapshot(&state, "default") + .await + .unwrap() + .routes + .is_empty() + ); + } + + #[tokio::test] + async fn expired_cached_routes_are_never_returned() { + let (state, _, _) = installed_fixture().await; + let slot = state.discovery.slot("default").await.unwrap(); + let mut route = discovered(); + route.expires_at = + Some(openshell_core::time::timestamp_from_millis(current_time_ms() - 1).unwrap()); + *slot.lock().await = Some(( + tokio::time::Instant::now(), + Ok(ServiceRouteSnapshot { + routes: vec![route], + revision: "old".into(), + }), + )); + assert!( + cached_snapshot(&state, "default") + .await + .unwrap() + .routes + .is_empty() + ); + } + + #[test] + fn admission_bounds_each_principal_workspace_and_unary_pool_and_releases_capacity() { + let state = DiscoveryState::default(); + let a = vec!["a".into()]; + let b = vec!["b".into()]; + let mut watches = Vec::new(); + for n in 0..4 { + for _ in 0..4 { + watches.push(state.admit(&user(&format!("u{n}")), &a, true).unwrap()); + } + } + assert_eq!( + state.admit(&user("u0"), &b, true).unwrap_err().code(), + tonic::Code::ResourceExhausted + ); + assert_eq!( + state.admit(&user("other"), &a, true).unwrap_err().code(), + tonic::Code::ResourceExhausted + ); + assert!(state.admit(&user("other"), &b, true).is_ok()); + drop(watches); + assert!(state.admit(&user("u0"), &a, true).is_ok()); + let mut unary = Vec::new(); + for n in 0..MAX_UNARY { + unary.push( + state + .admit(&user(&format!("u{n}")), &[format!("w{n}")], false) + .unwrap(), + ); + } + assert_eq!( + state.admit(&user("extra"), &b, false).unwrap_err().code(), + tonic::Code::ResourceExhausted + ); + drop(unary); + assert!(state.admit(&user("extra"), &b, false).is_ok()); + assert!(state.shares.lock().unwrap().len() <= 4); + } + + #[tokio::test] + async fn collection_survives_concurrent_lease_renewals() { + use crate::persistence::WriteCondition; + let (state, owner, report) = installed_fixture().await; + accept(&state, &owner, &report, current_time_ms() + 60_000) + .await + .unwrap(); + let record = state + .store + .get(OBJECT_TYPE, "service-installation:sandbox") + .await + .unwrap() + .unwrap(); + // A scan of 50 installation records remains valid while their lease + // resource versions change. Every record uses the live fixture's authority. + for n in 0..50 { + state + .store + .put_if( + OBJECT_TYPE, + &format!("renewal-{n:03}"), + &format!("renewal-{n:03}"), + "default", + &record.payload, + None, + WriteCondition::MustCreate, + ) + .await + .unwrap(); + } + let (stop, mut stopped) = tokio::sync::watch::channel(false); + let writes = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let renew_state = state.clone(); + let count = writes.clone(); + let renew = tokio::spawn(async move { + loop { + for n in 0..50 { + let id = format!("renewal-{n:03}"); + let record = renew_state + .store + .get(OBJECT_TYPE, &id) + .await + .unwrap() + .unwrap(); + renew_state + .store + .put_if( + OBJECT_TYPE, + &id, + &id, + "default", + &record.payload, + None, + WriteCondition::MatchResourceVersion(record.resource_version), + ) + .await + .unwrap(); + count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + tokio::select! { _=stopped.changed()=>break, ()=tokio::time::sleep(Duration::from_millis(1))=>{} } + } + }); + for _ in 0..3 { + assert_eq!(snapshot(&state, "default").await.unwrap().routes.len(), 51); + } + stop.send(true).unwrap(); + renew.await.unwrap(); + assert!(writes.load(std::sync::atomic::Ordering::Relaxed) > 50); + } + + #[tokio::test] + async fn watch_ends_when_its_bearer_expires_or_is_revoked() { + use crate::auth::authenticator::{ + AuthFailureReason, AuthMechanism, AuthRejection, Authentication, Authenticator, + AuthenticatorChain, + }; + struct Credential { + valid: Arc, + principal: Principal, + reason: AuthFailureReason, + } + #[async_trait::async_trait] + impl Authenticator for Credential { + async fn authenticate( + &self, + _: &http::HeaderMap, + _: &str, + ) -> Result { + if self.valid.load(std::sync::atomic::Ordering::Relaxed) { + Ok(Authentication::Accepted(self.principal.clone())) + } else { + Err(AuthRejection::new( + AuthMechanism::Oidc, + self.reason, + Status::unauthenticated("expired or revoked"), + )) + } + } + } + for reason in [ + AuthFailureReason::ExpiredCredential, + AuthFailureReason::StaleIdentity, + ] { + let (state, _, _) = installed_fixture().await; + let mut req = authed_request(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + }); + let principal = req.extensions().get::().unwrap().clone(); + let valid = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let recheck = BearerRecheck { + chain: AuthenticatorChain::new(vec![Arc::new(Credential { + valid: valid.clone(), + principal: principal.clone(), + reason, + })]), + headers: http::HeaderMap::new(), + policy: None, + path: "/openshell.v1.ServiceDiscovery/WatchServiceRoutes".into(), + }; + recheck.check(&principal).await.unwrap(); + req.extensions_mut().insert(recheck); + let mut stream = watch(&state, req).await.unwrap().into_inner(); + assert!(stream.next().await.unwrap().is_ok()); + valid.store(false, std::sync::atomic::Ordering::Relaxed); + let error = tokio::time::timeout(Duration::from_secs(3), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::Unauthenticated); + assert!(stream.next().await.is_none()); + assert_eq!(state.discovery_watches.available_permits(), 32); + assert!( + state + .discovery + .admit(&principal, &["default".into()], true) + .is_ok() + ); + } + } + + #[tokio::test] + async fn record_count_limit_is_enforced_before_decoding() { + let (state, _, _) = installed_fixture().await; + for n in 0..=MAX_RECORDS { + let id = format!("record-{n}"); + state + .store + .put_if( + OBJECT_TYPE, + &id, + &id, + "default", + b"{}", + None, + crate::persistence::WriteCondition::MustCreate, + ) + .await + .unwrap(); + } + assert_eq!( + snapshot(&state, "default").await.unwrap_err().code(), + tonic::Code::ResourceExhausted + ); + } + + #[tokio::test] + async fn separate_discovery_service_serves_generated_clients_without_management_stubs() { + use openshell_core::proto::{ + service_discovery_client::ServiceDiscoveryClient, + service_discovery_server::ServiceDiscoveryServer, + }; + let (state, owner, report) = installed_fixture().await; + accept(&state, &owner, &report, current_time_ms() + 10_000) + .await + .unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let service = ServiceDiscoveryServer::with_interceptor( + super::super::OpenShellService::new(state), + |mut req: Request<()>| { + req.extensions_mut().insert(user("client")); + Ok(req) + }, + ); + let (stop, stopped) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(service) + .serve_with_incoming_shutdown( + tokio_stream::wrappers::TcpListenerStream::new(listener), + async { + let _ = stopped.await; + }, + ) + .await + .unwrap(); + }); + let mut client = ServiceDiscoveryClient::connect(format!("http://{address}")) + .await + .unwrap(); + let response = client + .get_service_route_snapshot(request()) + .await + .unwrap() + .into_inner() + .snapshot + .unwrap(); + assert_eq!( + response.routes[0].hostname, + "default--demo--web.services.example.com" + ); + let mut stream = client + .watch_service_routes(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + }) + .await + .unwrap() + .into_inner(); + assert_eq!( + stream.message().await.unwrap().unwrap().snapshot.unwrap(), + response + ); + drop(stream); + drop(client); + stop.send(()).unwrap(); + server.await.unwrap(); + let pool = + prost_reflect::DescriptorPool::decode(openshell_core::FILE_DESCRIPTOR_SET).unwrap(); + assert!( + pool.get_service_by_name("openshell.v1.OpenShell") + .unwrap() + .methods() + .all(|method| !matches!( + method.name(), + "GetServiceRouteSnapshot" | "WatchServiceRoutes" + )) + ); + } + #[tokio::test] + async fn watch_reauthorizes_the_refreshed_identity_instead_of_retaining_old_admin_roles() { + use crate::auth::authenticator::{ + AuthRejection, Authentication, Authenticator, AuthenticatorChain, + }; + struct MutableIdentity(Arc>); + #[async_trait::async_trait] + impl Authenticator for MutableIdentity { + async fn authenticate( + &self, + _: &http::HeaderMap, + _: &str, + ) -> Result { + Ok(Authentication::Accepted(self.0.lock().unwrap().clone())) + } + } + let (state, _, _) = installed_fixture().await; + let mut req = authed_request(WatchServiceRoutesRequest { + workspace_scope: request().workspace_scope, + }); + let identity = Arc::new(std::sync::Mutex::new( + req.extensions().get::().unwrap().clone(), + )); + req.extensions_mut().insert(BearerRecheck { + chain: AuthenticatorChain::new(vec![Arc::new(MutableIdentity(identity.clone()))]), + headers: http::HeaderMap::new(), + policy: None, + path: "/openshell.v1.ServiceDiscovery/WatchServiceRoutes".into(), + }); + let mut stream = watch(&state, req).await.unwrap().into_inner(); + assert!(stream.next().await.unwrap().is_ok()); + if let Principal::User(user) = &mut *identity.lock().unwrap() { + user.identity.roles = vec!["openshell-user".into()]; + } + let error = tokio::time::timeout(Duration::from_secs(3), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::PermissionDenied); + } } diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index bf11c4f1fa..db682128bc 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -322,8 +322,9 @@ pub(crate) struct ServerStartupConfig { /// Server state shared across handlers. #[derive(Debug)] pub struct ServerState { + pub(crate) discovery: grpc::service_discovery::DiscoveryState, /// Process-local bound on active discovery streams. - pub discovery_watches: Arc, + pub(crate) discovery_watches: Arc, /// Server configuration. pub config: Config, @@ -668,6 +669,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: grpc::service_discovery::DiscoveryState::default(), discovery_watches: Arc::new(tokio::sync::Semaphore::new(32)), extension_mint_limiter: auth::extension_mint_limit::ExtensionMintLimiter::default(), middleware_registry: Arc::new(MiddlewareRegistry::default()), @@ -3176,7 +3178,11 @@ mod tests { .iter() .map(|service| service.name.as_str()) .collect::>(), - ["openshell.v1.OpenShell", "openshell.v1.SandboxUpdates"] + [ + "openshell.v1.OpenShell", + "openshell.v1.SandboxUpdates", + "openshell.v1.ServiceDiscovery" + ] ); stop_listener(shutdown, handle).await; diff --git a/crates/openshell-server/src/multiplex.rs b/crates/openshell-server/src/multiplex.rs index 438b7d4bc6..eec5c72dcf 100644 --- a/crates/openshell-server/src/multiplex.rs +++ b/crates/openshell-server/src/multiplex.rs @@ -19,6 +19,7 @@ use hyper_util::{ use metrics::{counter, histogram}; use openshell_core::proto::open_shell_server::OpenShellServer; use openshell_core::proto::sandbox_updates_server::SandboxUpdatesServer; +use openshell_core::proto::service_discovery_server::ServiceDiscoveryServer; use openshell_core::proto::supervisor_host::v1::supervisor_host_server::SupervisorHostServer; use openshell_core::{ Config, @@ -313,6 +314,13 @@ impl MultiplexService { self.state.gateway_interceptors.clone(), Some(self.state.clone()), ); + let discovery = ServiceDiscoveryServer::new(OpenShellService::new(self.state.clone())) + .max_decoding_message_size(MAX_GRPC_DECODE_SIZE); + let discovery = GatewayInterceptorGrpcService::new( + discovery, + self.state.gateway_interceptors.clone(), + Some(self.state.clone()), + ); let authz_policy = self.state.config.oidc.as_ref().map(|oidc| AuthzPolicy { admin_role: oidc.admin_role.clone(), user_role: oidc.user_role.clone(), @@ -320,7 +328,9 @@ impl MultiplexService { }); let authenticator_chain = build_authenticator_chain(&self.state); let grpc_service = AuthGrpcRouter::with_peer_identity( - GrpcRouter::new(openshell, reflection).with_sandbox_updates(sandbox_updates), + GrpcRouter::new(openshell, reflection) + .with_sandbox_updates(sandbox_updates) + .with_discovery(discovery), authenticator_chain, authz_policy, self.state @@ -997,10 +1007,11 @@ where /// Combined gRPC service for public gateway services and reflection. #[derive(Clone)] -pub struct GrpcRouter { +pub struct GrpcRouter { openshell: N, reflection: R, sandbox_updates: Option, + discovery: Option, } impl GrpcRouter { @@ -1009,6 +1020,7 @@ impl GrpcRouter { openshell, reflection, sandbox_updates: None, + discovery: None, } } @@ -1017,13 +1029,25 @@ impl GrpcRouter { openshell: self.openshell, reflection: self.reflection, sandbox_updates: Some(sandbox_updates), + discovery: self.discovery, + } + } +} + +impl GrpcRouter { + fn with_discovery(self, discovery: T) -> GrpcRouter { + GrpcRouter { + openshell: self.openshell, + reflection: self.reflection, + sandbox_updates: self.sandbox_updates, + discovery: Some(discovery), } } } pub const REFLECTION_PATH_PREFIX: &str = "/grpc.reflection.v1."; -impl tower::Service> for GrpcRouter +impl tower::Service> for GrpcRouter where N: tower::Service> + Clone + Send + 'static, N::Response: Send, @@ -1039,6 +1063,11 @@ where + Send + 'static, U::Future: Send, + D: tower::Service, Response = N::Response, Error = N::Error> + + Clone + + Send + + 'static, + D::Future: Send, B: Send + 'static, { type Response = N::Response; @@ -1060,6 +1089,13 @@ where && let Some(mut svc) = self.sandbox_updates.clone() { Box::pin(async move { svc.ready().await?.call(req).await }) + } else if req + .uri() + .path() + .starts_with("/openshell.v1.ServiceDiscovery/") + && let Some(mut svc) = self.discovery.clone() + { + Box::pin(async move { svc.ready().await?.call(req).await }) } else { let mut svc = self.openshell.clone(); Box::pin(async move { svc.ready().await?.call(req).await }) @@ -1276,7 +1312,7 @@ where ))); } Principal::Operator(operator) - } else if let Some(chain) = chain { + } else if let Some(chain) = &chain { match chain.authenticate(req.headers(), &path).await { Ok(Authentication::Accepted(p)) => p, Ok(outcome) => match (mtls_auth_enabled, peer_identity) { @@ -1397,6 +1433,18 @@ where return Ok(status_response(status)); } + if path.starts_with("/openshell.v1.ServiceDiscovery/") + && req.headers().contains_key(http::header::AUTHORIZATION) + && let Some(chain) = &chain + { + let recheck = crate::grpc::service_discovery::BearerRecheck { + chain: chain.clone(), + headers: req.headers().clone(), + policy: authz_policy.clone(), + path: path.clone(), + }; + req.extensions_mut().insert(recheck); + } req.extensions_mut().insert(principal); let call = inner.ready().await?.call(req); audit::with_request(path, origin, call).await @@ -2943,11 +2991,14 @@ mod tests { calls: calls.clone(), }; let mut router = GrpcRouter::new(service("openshell"), service("reflection")) - .with_sandbox_updates(service("sandbox_updates")); + .with_sandbox_updates(service("sandbox_updates")) + .with_discovery(service("discovery")); for path in [ "/openshell.v1.OpenShell/Health", "/openshell.v1.SandboxUpdates/UpdateSandbox", + "/openshell.v1.ServiceDiscovery/GetServiceRouteSnapshot", + "/openshell.v1.ServiceDiscovery/WatchServiceRoutes", "/grpc.reflection.v1.ServerReflection/ServerReflectionInfo", ] { router @@ -2963,7 +3014,13 @@ mod tests { assert_eq!( *calls.lock().unwrap(), - vec!["openshell", "sandbox_updates", "reflection"] + vec![ + "openshell", + "sandbox_updates", + "discovery", + "discovery", + "reflection" + ] ); } @@ -3062,7 +3119,11 @@ mod tests { assert_eq!( names, - vec!["openshell.v1.OpenShell", "openshell.v1.SandboxUpdates"] + vec![ + "openshell.v1.OpenShell", + "openshell.v1.SandboxUpdates", + "openshell.v1.ServiceDiscovery" + ] ); let descriptor_response = responses.message().await.unwrap().unwrap(); diff --git a/crates/openshell-server/src/reflection.rs b/crates/openshell-server/src/reflection.rs index d8eb12482e..463a937b1d 100644 --- a/crates/openshell-server/src/reflection.rs +++ b/crates/openshell-server/src/reflection.rs @@ -28,7 +28,11 @@ use openshell_core::Config; use crate::multiplex::GrpcRateLimiter; const REFLECTED_PROTO_ROOTS: &[&str] = &["openshell.proto"]; -const ADVERTISED_SERVICES: &[&str] = &["openshell.v1.OpenShell", "openshell.v1.SandboxUpdates"]; +const ADVERTISED_SERVICES: &[&str] = &[ + "openshell.v1.OpenShell", + "openshell.v1.SandboxUpdates", + "openshell.v1.ServiceDiscovery", +]; pub type GatewayReflectionServer = ServerReflectionServer; diff --git a/crates/openshell-server/src/service_routing.rs b/crates/openshell-server/src/service_routing.rs index ece2add1e0..4d8c4a2ace 100644 --- a/crates/openshell-server/src/service_routing.rs +++ b/crates/openshell-server/src/service_routing.rs @@ -200,11 +200,10 @@ fn endpoint_host( let base_domain = config.base_domains.first()?; Some(discriminator.map_or_else( || { - if service.is_empty() { - format!("{workspace}--{sandbox}.{base_domain}") - } else { - format!("{workspace}--{sandbox}--{service}.{base_domain}") - } + format!( + "{}.{base_domain}", + service_route_readable_label(workspace, sandbox, service) + ) }, |gateway_name| { format!( @@ -221,6 +220,15 @@ fn endpoint_host( )) } +/// Readable DNS label shared by discovery and gateway routing. +pub fn service_route_readable_label(workspace: &str, sandbox: &str, service: &str) -> String { + if service.is_empty() { + format!("{workspace}--{sandbox}") + } else { + format!("{workspace}--{sandbox}--{service}") + } +} + /// Returns the stable, one-label DNS name for a service route. pub fn service_route_hash_label( gateway_name: &str, diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index 8613a60f91..f4a31e52ed 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -177,8 +177,9 @@ mod tests { // selections absent; storage.v1 remains frozen. Public field 3 retains its // image wire identity, now with explicit presence. Compute configuration // updates use a separate capability/service and add one classified RPC. + // Discovery RPCs move to ServiceDiscovery; message and durable closures are unchanged. const PUBLIC_RPC_SCHEMA_SHA256: &str = - "1fce709ed9a1c91e307c46c5fe68be470cbe88ca70ea0087e771c973a9052308"; + "daa132037ee47c7a46b08dfab9bab379961bc06ff3e9ebbb329809b724f2a4f6"; const DURABLE_SCHEMA_SHA256: &str = "cb15f8f79a855064a43fe8c6cd908bf07eca6604e2bf8ab7d2de0ef33e97da4e"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = @@ -596,8 +597,10 @@ mod tests { } if !matches!( (package, service_name), - ("openshell.v1", "OpenShell" | "SandboxUpdates") - | ("openshell.inference.v1", "Inference") + ( + "openshell.v1", + "OpenShell" | "SandboxUpdates" | "ServiceDiscovery" + ) | ("openshell.inference.v1", "Inference") ) { continue; } @@ -672,7 +675,7 @@ mod tests { .iter() .filter(|method| method.starts_with("openshell.v1.OpenShell/")) .count(), - 82 + PROVIDER_READINESS_RPC_SIGNATURES.len() + 80 + PROVIDER_READINESS_RPC_SIGNATURES.len() + PEER_OWNER_RPC_SIGNATURES.len() + PEER_CONFIG_RPC_SIGNATURES.len() ); diff --git a/crates/openshell-server/tests/common/mod.rs b/crates/openshell-server/tests/common/mod.rs index a0e4f5f4ff..dd91664d39 100644 --- a/crates/openshell-server/tests/common/mod.rs +++ b/crates/openshell-server/tests/common/mod.rs @@ -52,20 +52,6 @@ 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 f858ebf163..ceb2fa6bf8 100644 --- a/crates/openshell-server/tests/supervisor_relay_integration.rs +++ b/crates/openshell-server/tests/supervisor_relay_integration.rs @@ -50,20 +50,6 @@ 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 12d28bce56..18a179cf45 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -1957,7 +1957,8 @@ 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 +Discovery clients use the separate `openshell.v1.ServiceDiscovery` gRPC service. +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 @@ -1965,3 +1966,17 @@ 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. + +Discovery limits apply per gateway process: 1,000 installation records, 4,096 +routes, and 4 MiB per snapshot, with a ten-second collection timeout. Up to four +collections and 16 unary requests run concurrently; unary requests are limited +to four per principal and eight per workspace. Watches are limited to 32 total, +four per principal, and 16 per workspace. Capacity and size limits return +`RESOURCE_EXHAUSTED`; storage failures and timeouts return `UNAVAILABLE`; corrupt +records return `INTERNAL`. Disabling the app listener returns `FAILED_PRECONDITION`. + +Consumers share a one-second collection cache, bounded to 16 workspaces. Cache reuse adds at most one refresh interval to collection freshness; +cached routes never outlive their advertised expiry. Watches recheck bearer +credentials and workspace access every second and terminate if either is lost. +OIDC token revocation follows the configured verifier's semantics; offline JWT +verification cannot detect issuer-side revocation before token expiry. diff --git a/proto/openshell.proto b/proto/openshell.proto index 17d86452d9..198990a364 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -251,24 +251,6 @@ 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) = { @@ -876,6 +858,28 @@ service OpenShell { } } +// Application ingress consumers discover routes independently of sandbox management. +service ServiceDiscovery { + // 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" + }; + } + +} + // Optional sandbox mutations are separate from OpenShell so existing generated // OpenShell server implementations remain source-compatible. service SandboxUpdates { diff --git a/sdk/typescript/src/raw.ts b/sdk/typescript/src/raw.ts index 067a6558b9..3a45c49f9b 100644 --- a/sdk/typescript/src/raw.ts +++ b/sdk/typescript/src/raw.ts @@ -4,7 +4,7 @@ export * from './gen/datamodel_pb.js'; export * from './gen/google/rpc/error_details_pb.js'; // Advanced surface: the full generated protobuf types (messages, enums, and the -// OpenShell and SandboxUpdates service descriptors) for callers using the raw escape hatch on +// OpenShell, SandboxUpdates, and ServiceDiscovery service descriptors) for callers using the raw escape hatch on // OpenShellClient / SandboxClient (`.raw` and `.transport`). These are the // uncurated wire types; import them from '@nvidia/openshell-sdk/raw'. The // curated entry point stays free of generated types so its surface does not diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 20caa213a2..8acc9d48a7 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -1728,3 +1728,13 @@ verify workload identity/memory survives and new connections reach the replaceme Shared serving leases last at most fifteen seconds from issuance, independently of configuration renewal; include membership failure detection when measuring a partition's stale-owner interval. Storage failures prevent renewal. + + +Application ingress consumers call the separate `openshell.v1.ServiceDiscovery` +service, rather than `OpenShell`. Discovery snapshots share a one-second cache +per workspace (16 cached workspaces, four concurrent collections). The gateway +admits 16 unary requests (four per principal, eight per workspace), and 32 watches +(four per principal, 16 per workspace). `RESOURCE_EXHAUSTED` means capacity or +record/route/byte limits; `UNAVAILABLE` means storage or collection timeout. +Watches recheck bearer credentials and membership every second. Allow one refresh +interval for withdrawals; route lease expiry remains independently enforced.