diff --git a/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs b/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs index 31e67baf0f..772a26e58e 100644 --- a/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs +++ b/crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs @@ -163,6 +163,7 @@ macro_rules! mock_gateway { ) => { #[tonic::async_trait] impl OpenShell for MockGateway { + $(async fn $method(&self, _: Request<$request>) -> Result, Status> { Err(Status::unimplemented("unused test RPC")) })* 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 1f3f791a08..f98a180eea 100644 --- a/crates/openshell-sdk/tests/client_mock.rs +++ b/crates/openshell-sdk/tests/client_mock.rs @@ -1195,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(); @@ -1204,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; }); @@ -2457,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 3c52ebb127..d8b27b05a8 100644 --- a/crates/openshell-server/README.md +++ b/crates/openshell-server/README.md @@ -295,3 +295,38 @@ 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 + +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. + +`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. + +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 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 69208348de..b006c7268e 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; +pub mod service_discovery; mod validation; pub mod workspace; @@ -1435,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 dff8257fcf..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)] mod tests { + use super::test_support::installed_fixture; use super::*; fn test_lease_authority() -> LeaseAuthority { @@ -1072,6 +1082,82 @@ 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; @@ -1103,6 +1189,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(), @@ -1166,76 +1253,4 @@ 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 new file mode 100644 index 0000000000..18eafaf89f --- /dev/null +++ b/crates/openshell-server/src/grpc/service_discovery.rs @@ -0,0 +1,1326 @@ +// 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, 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, current_routes}; +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>; + +// 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, + 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(()) +} + +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 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), + })) +} + +pub async fn watch( + state: &Arc, + request: Request, +) -> Result, Status> { + 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 = 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 { + snapshot: Some(initial), + })) + .map_err(|_| Status::cancelled("discovery client closed"))?; + let state = state.clone(); + tokio::spawn(async move { + let _permit = permit; + 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; + } + } + }; + let _ = tokio::time::timeout(WATCH_LIFETIME, run).await; + }); + 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 { + // 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; + }; + 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, 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", + )); + } + 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, test_support::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 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, + }), + ) + .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; + 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"); + 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( + 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 + ); + } + 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 140f372347..db682128bc 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -322,6 +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(crate) discovery_watches: Arc, /// Server configuration. pub config: Config, @@ -666,6 +669,8 @@ 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()), oidc_cache, @@ -3173,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 d8da420871..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 = - "3b4c2af6cf22e7ed1d84279ff865acb0db7199717bee9bc041caef791334104d"; + "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; } @@ -652,7 +655,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 +665,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" @@ -723,7 +726,7 @@ mod tests { overlap_hash.as_str(), ), ( - (337, 31), + (344, 31), (100, 22), (86, 22), PUBLIC_RPC_SCHEMA_SHA256, diff --git a/docs/how-it-works/gateways/configuration.mdx b/docs/how-it-works/gateways/configuration.mdx index 8d39d6bbaf..18a179cf45 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,31 @@ 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. + +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 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 +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 b40ec68263..198990a364 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -858,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 { @@ -4245,3 +4267,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/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 23d58efa25..8acc9d48a7 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,16 @@ 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. +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 +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; @@ -1719,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.