Skip to content

Commit 5afa86a

Browse files
committed
feat(supervisor): record leased application route installations
Signed-off-by: Taylor Mutch <taylormutch@gmail.com>
1 parent a816fa1 commit 5afa86a

33 files changed

Lines changed: 2349 additions & 468 deletions

File tree

‎.agents/skills/helm-dev-environment/SKILL.md‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,6 +195,17 @@ every member one at a time. The PID must survive each case, while an open
195195
replacement confirmed within the recovery timeout, so the runtime terminated
196196
the workload.
197197

198+
For RFC 0017 app-listener validation, enable the listener and supply the app TLS
199+
Secret on the shared Deployment, plus a separate NetworkPolicy allowing the test
200+
client/proxy to reach its app port. Runtime-port access does not grant app access.
201+
Discovery returns the acknowledged owner's Pod IP and app port; no headless app
202+
Service is needed. Exercise HTTP/WebSocket reconnection through graceful downscale
203+
and abrupt loss, preserving workload generation/PID/in-memory state and the public
204+
hostname. Verify stale-owner rejection independently of the replacement's success.
205+
Shared serving leases expire within fifteen seconds of issuance and do not renew
206+
from configuration alone. Do not claim seamless socket migration or infer external
207+
ingress readiness from an advertisement.
208+
198209
### 2. Deploy OpenShell
199210

200211
**Iterative dev** (rebuilds on file changes, recommended during active development):

‎crates/openshell-core/src/app_listener.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,3 +69,9 @@ mod tests {
6969
assert!(config.validate().is_err());
7070
}
7171
}
72+
73+
/// A synchronous view of installed ingress, evaluated on the sandbox instance
74+
/// runtime. The requested lease must fit entirely within serving freshness.
75+
pub type InstallationSource = std::sync::Arc<
76+
dyn Fn(std::time::Duration) -> crate::proto::SupervisorServiceAdvertisement + Send + Sync,
77+
>;

‎crates/openshell-driver-docker/src/lib.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3346,6 +3346,7 @@ impl ComputeDriver for DockerComputeDriver {
33463346
}
33473347

33483348
Ok(Response::new(GetSandboxResponse {
3349+
supervisor_app_endpoint: None,
33493350
sandbox: Some(sandbox),
33503351
}))
33513352
}

‎crates/openshell-driver-kubernetes/src/driver.rs‎

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2026,6 +2026,201 @@ impl KubernetesComputeDriver {
20262026
Ok(())
20272027
}
20282028

2029+
/// Resolve only a unique, live driver-owned supervisor. Multi-member pool
2030+
/// resolution must use the HA assignment/member API rather than guess.
2031+
pub async fn supervisor_app_endpoint(
2032+
&self,
2033+
sandbox_id: &str,
2034+
) -> Result<
2035+
Option<openshell_core::proto::compute::v1::SupervisorAppEndpoint>,
2036+
KubernetesDriverError,
2037+
> {
2038+
if !self.config.sandbox_runtime.app_listener.common.enabled {
2039+
return Ok(None);
2040+
}
2041+
let api = self
2042+
.supported_sandbox_api_for_lookup(self.client.clone())
2043+
.await
2044+
.map_err(KubernetesDriverError::Message)?;
2045+
let objects = api
2046+
.api
2047+
.list(&ListParams::default().labels(&self.sandbox_lookup_selector(sandbox_id)))
2048+
.await
2049+
.map_err(KubernetesDriverError::from_kube)?
2050+
.items;
2051+
let [object]: [DynamicObject; 1] = match objects.try_into() {
2052+
Ok(objects) => objects,
2053+
Err(_) => return Ok(None),
2054+
};
2055+
let namespace = object.metadata.namespace.as_deref().ok_or_else(|| {
2056+
KubernetesDriverError::Precondition("sandbox namespace missing".into())
2057+
})?;
2058+
let uid = object
2059+
.metadata
2060+
.uid
2061+
.as_deref()
2062+
.ok_or_else(|| KubernetesDriverError::Precondition("sandbox UID missing".into()))?;
2063+
let placement = supervisor_placement(&object)?;
2064+
let mut pool_route = String::new();
2065+
let mut ownership_epoch = 0;
2066+
let mut member_address = None;
2067+
let pods = if let SupervisorPlacement::Shared { route } = &placement {
2068+
let Some(shared) = &self.shared_supervisor else {
2069+
return Ok(None);
2070+
};
2071+
let controller = shared.supervision().controller();
2072+
let Some(view) = controller.pool_view() else {
2073+
return Ok(None);
2074+
};
2075+
let Some(owner) = view
2076+
.owners
2077+
.get(route)
2078+
.filter(|_| view.desired.contains(route))
2079+
else {
2080+
return Ok(None);
2081+
};
2082+
let Some(member) = view.members.iter().find(|member| {
2083+
member.member_id == owner.member_id
2084+
&& member.process_id == owner.process_id
2085+
&& !member.leaving
2086+
}) else {
2087+
return Ok(None);
2088+
};
2089+
if !matches!(controller.observed(route),
2090+
openshell_supervisor_control::ObservedInstance::Reported(status)
2091+
if status.sandbox_id == sandbox_id && status.ready
2092+
&& status.phase == openshell_supervisor_control::proto::SandboxInstancePhase::Supervising as i32)
2093+
{
2094+
return Ok(None);
2095+
}
2096+
let address = member
2097+
.runtime_address
2098+
.parse::<std::net::SocketAddr>()
2099+
.map_err(|_| {
2100+
KubernetesDriverError::Precondition(
2101+
"pool member runtime address is not an IP endpoint".into(),
2102+
)
2103+
})?;
2104+
let pod =
2105+
Api::<Pod>::namespaced(self.client.clone(), &shared.config().service_namespace)
2106+
.get_opt(&owner.member_id)
2107+
.await
2108+
.map_err(KubernetesDriverError::from_kube)?;
2109+
let Some(pod) = pod else {
2110+
return Ok(None);
2111+
};
2112+
let service =
2113+
Api::<Service>::namespaced(self.client.clone(), &shared.config().service_namespace)
2114+
.get(&shared.config().service_name)
2115+
.await
2116+
.map_err(KubernetesDriverError::from_kube)?;
2117+
let Some(selector) = service
2118+
.spec
2119+
.as_ref()
2120+
.and_then(|spec| spec.selector.as_ref())
2121+
.filter(|selector| !selector.is_empty())
2122+
else {
2123+
return Ok(None);
2124+
};
2125+
if !selector.iter().all(|(key, value)| {
2126+
pod.metadata
2127+
.labels
2128+
.as_ref()
2129+
.and_then(|labels| labels.get(key))
2130+
== Some(value)
2131+
}) {
2132+
return Ok(None);
2133+
}
2134+
pool_route = route.clone();
2135+
ownership_epoch = owner.auth_epoch;
2136+
member_address = Some(address.ip());
2137+
vec![pod]
2138+
} else {
2139+
let selector =
2140+
format!("openshell.ai/sandbox-id={sandbox_id},openshell.ai/component=supervisor");
2141+
Api::<Pod>::namespaced(self.client.clone(), namespace)
2142+
.list(&ListParams::default().labels(&selector))
2143+
.await
2144+
.map_err(KubernetesDriverError::from_kube)?
2145+
.items
2146+
};
2147+
// Includes terminating members in the ambiguity check; never guess a
2148+
// replacement's ownership during a rollout.
2149+
let [pod]: [Pod; 1] = match pods.try_into() {
2150+
Ok(pods) => pods,
2151+
Err(_) => return Ok(None),
2152+
};
2153+
if pod.metadata.deletion_timestamp.is_some() {
2154+
return Ok(None);
2155+
}
2156+
let app_port = i32::from(self.config.sandbox_runtime.app_listener.common.port);
2157+
if !pod
2158+
.spec
2159+
.as_ref()
2160+
.and_then(|spec| {
2161+
spec.containers
2162+
.iter()
2163+
.find(|container| container.name == "supervisor")
2164+
})
2165+
.is_some_and(|container| {
2166+
container.ports.as_ref().is_some_and(|ports| {
2167+
ports.iter().any(|port| {
2168+
port.name.as_deref() == Some("app") && port.container_port == app_port
2169+
})
2170+
}) && container.env.as_ref().is_some_and(|env| {
2171+
env.iter().any(|entry| {
2172+
entry.name == "OPENSHELL_APP_LISTENER_ENABLED"
2173+
&& entry.value.as_deref() == Some("true")
2174+
})
2175+
})
2176+
})
2177+
{
2178+
return Ok(None);
2179+
}
2180+
let Some(pod_uid) = pod.metadata.uid.as_deref() else {
2181+
return Ok(None);
2182+
};
2183+
if matches!(placement, SupervisorPlacement::Dedicated)
2184+
&& !pod
2185+
.metadata
2186+
.owner_references
2187+
.as_ref()
2188+
.is_some_and(|owners| owners.iter().any(|o| o.uid == uid && o.kind == "Sandbox"))
2189+
{
2190+
return Ok(None);
2191+
}
2192+
let Some(status) = pod.status else {
2193+
return Ok(None);
2194+
};
2195+
if status.phase.as_deref() != Some("Running") {
2196+
return Ok(None);
2197+
}
2198+
let Some(host) = status
2199+
.pod_ip
2200+
.filter(|ip| ip.parse::<std::net::IpAddr>().is_ok())
2201+
else {
2202+
return Ok(None);
2203+
};
2204+
if member_address
2205+
.is_some_and(|address| host.parse::<std::net::IpAddr>().ok() != Some(address))
2206+
{
2207+
return Ok(None);
2208+
}
2209+
let component = match placement {
2210+
SupervisorPlacement::Dedicated => pod_uid.to_string(),
2211+
SupervisorPlacement::Shared { route } => route,
2212+
};
2213+
Ok(Some(
2214+
openshell_core::proto::compute::v1::SupervisorAppEndpoint {
2215+
host,
2216+
port: u32::from(self.config.sandbox_runtime.app_listener.common.port),
2217+
runtime_identity: kubernetes_runtime_identity(namespace, uid, &component),
2218+
pool_route,
2219+
ownership_epoch,
2220+
},
2221+
))
2222+
}
2223+
20292224
pub async fn get_sandbox(&self, sandbox_id: &str) -> Result<Option<Sandbox>, String> {
20302225
info!(
20312226
sandbox_id = %sandbox_id,

‎crates/openshell-driver-kubernetes/src/grpc.rs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,11 @@ impl ComputeDriver for ComputeDriverService {
124124
.map_err(Status::internal)?
125125
.ok_or_else(|| Status::not_found("sandbox not found"))?;
126126
Ok(Response::new(GetSandboxResponse {
127+
supervisor_app_endpoint: self
128+
.driver
129+
.supervisor_app_endpoint(&request.sandbox_id)
130+
.await
131+
.map_err(|e| Status::unavailable(e.to_string()))?,
127132
sandbox: Some(sandbox),
128133
}))
129134
})

‎crates/openshell-driver-mxc/src/grpc.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@ impl ComputeDriver for ComputeDriverService {
8989
));
9090
}
9191
Ok(Response::new(GetSandboxResponse {
92+
supervisor_app_endpoint: None,
9293
sandbox: Some(sandbox),
9394
}))
9495
}

‎crates/openshell-driver-podman/src/grpc.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ impl ComputeDriver for ComputeDriverService {
119119
.map_err(Status::from)?
120120
.ok_or_else(|| Status::not_found("sandbox not found"))?;
121121
Ok(Response::new(GetSandboxResponse {
122+
supervisor_app_endpoint: None,
122123
sandbox: Some(sandbox),
123124
}))
124125
})

‎crates/openshell-driver-vm/src/driver.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4939,6 +4939,7 @@ impl ComputeDriver for VmDriver {
49394939
}
49404940

49414941
Ok(Response::new(GetSandboxResponse {
4942+
supervisor_app_endpoint: None,
49424943
sandbox: Some(sandbox),
49434944
}))
49444945
}

‎crates/openshell-server/README.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -189,3 +189,16 @@ runs under `mise run test:rust:postgres` (see `TESTING.md`) and still accepts
189189
`OPENSHELL_REFRESH_TEST_DATABASE_URL` pointing to a disposable PostgreSQL database.
190190
It exercises independent sessions without the local mutex. Do not point this
191191
fixture at an existing deployment database.
192+
193+
## Application route installation records
194+
195+
Supervisor installation reports arrive only through the authenticated sandbox
196+
session. The gateway issues bounded, single-use lease challenges and validates
197+
reports under the sandbox mutation guard. Accepted records contain current
198+
service declarations, configuration/runtime identity, session fencing, a
199+
trusted driver-resolved app endpoint, and gateway lease expiry. The generic
200+
store key is prefixed `service-installation:`; records are ephemeral and do not
201+
change durable sandbox protobufs. Reads revalidate owner, configuration,
202+
generation, and expiry. Cleanup scans bounded pages and uses conditional deletes
203+
so concurrent renewals are preserved. Discovery consumers must not treat these
204+
records as application health or external ingress readiness.

0 commit comments

Comments
 (0)