Skip to content

Commit 7200e68

Browse files
pimlockpimlock
committed
feat(config): deliver gateway configuration over the supervisor session
Supervisors poll the gateway for configuration every 10 seconds. Add an opt-in push path: with config_delivery_mode = "push", the gateway delivers sandbox configuration and provider environments over the existing ConnectSupervisor session, and supervisors apply and acknowledge them. Poll stays the default, and older supervisors and gateways keep polling. The gateway builds a streamed bootstrap that the supervisor applies before starting the workload, then runs a per-session delivery task that coalesces publications, takes FIFO build permits sized from the database pool, retries failed components with backoff, and reconciles periodically. Both components build from one shared input load, and the bootstrap checks that they agree on provider revision, attachments, and effective policy. In HA, replicas notify the session owner, which rebuilds from shared state. Builds also get cheaper in poll mode: interceptor provider-profile sources are served from a snapshot refreshed every 10 seconds, and provider environment builds read refresh states once and load providers concurrently. The supervisor applies polled and streamed configuration through one runtime, now in its own module, and stops polling once a push session is applying configuration. New metrics cover builds, deliveries, apply results, bootstraps, and peer notifications. Part of #1731. Signed-off-by: Piotr Mlocek <[email protected]>
1 parent 6a6401b commit 7200e68

60 files changed

Lines changed: 20910 additions & 6388 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.github/workflows/e2e-docker-test.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ on:
2929
required: false
3030
type: string
3131
default: >-
32-
[{"suite":"python","cmd":"mise run --no-deps --skip-deps e2e:python","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-python","cmd":"mise run --no-deps --skip-deps e2e:oidc-python:docker","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-pkce-docker","cmd":"mise run --no-deps --skip-deps e2e:oidc-pkce:docker","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"rust-docker","cmd":"mise run --no-deps --skip-deps e2e:rust","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"oci-image","cmd":"mise run --no-deps --skip-deps e2e:oci-image","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"mcp","cmd":"mise run --no-deps --skip-deps e2e:mcp","apt_packages":"","python_proto":false,"mcp":true}]
32+
[{"suite":"python","cmd":"mise run --no-deps --skip-deps e2e:python","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-python","cmd":"mise run --no-deps --skip-deps e2e:oidc-python:docker","apt_packages":"","python_proto":true,"mcp":false},{"suite":"oidc-pkce-docker","cmd":"mise run --no-deps --skip-deps e2e:oidc-pkce:docker","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"rust-docker","cmd":"mise run --no-deps --skip-deps e2e:rust","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"rust-docker-push","cmd":"mise run --no-deps --skip-deps e2e:rust:push","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"oci-image","cmd":"mise run --no-deps --skip-deps e2e:oci-image","apt_packages":"openssh-client","python_proto":false,"mcp":false},{"suite":"mcp","cmd":"mise run --no-deps --skip-deps e2e:mcp","apt_packages":"","python_proto":false,"mcp":true}]
3333
3434
permissions:
3535
actions: read

‎TESTING.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -224,6 +224,7 @@ Suites:
224224
the selected deployment.
225225
- Docker suite (`--features e2e-docker`) - includes Docker-only coverage such as Dockerfile image builds, Docker preflight checks, and managed Docker gateway start.
226226
- Docker GPU suite (`--features e2e-docker-gpu`) - Docker suite plus GPU sandbox smoke coverage.
227+
- Configuration push suite (`--features e2e-config-push`) - CLI conformance and the policy tests against a Docker gateway with `config_delivery_mode = "push"`. Run it with `mise run e2e:rust:push`.
227228
- VM suite (`--features e2e-vm`) - runs e2e tests on a VM.
228229
- Kubernetes credential-driver suite (`--features e2e-kubernetes-credential-drivers`) - targeted Kubernetes Secrets and Vault provider credential storage coverage.
229230

‎crates/openshell-cli/tests/ensure_providers_integration.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,13 @@ impl TestOpenShell {
7979

8080
#[tonic::async_trait]
8181
impl OpenShell for TestOpenShell {
82+
async fn peer_notify_config_update(
83+
&self,
84+
_request: tonic::Request<openshell_core::proto::PeerNotifyConfigUpdateRequest>,
85+
) -> Result<Response<openshell_core::proto::PeerNotifyConfigUpdateResponse>, Status> {
86+
Err(Status::unimplemented("not used by this test server"))
87+
}
88+
8289
async fn peer_report_provider_readiness(
8390
&self,
8491
_request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

‎crates/openshell-cli/tests/mtls_integration.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,13 @@ struct TestOpenShell;
3232

3333
#[tonic::async_trait]
3434
impl OpenShell for TestOpenShell {
35+
async fn peer_notify_config_update(
36+
&self,
37+
_request: tonic::Request<openshell_core::proto::PeerNotifyConfigUpdateRequest>,
38+
) -> Result<Response<openshell_core::proto::PeerNotifyConfigUpdateResponse>, Status> {
39+
Err(Status::unimplemented("not used by this test server"))
40+
}
41+
3542
async fn peer_report_provider_readiness(
3643
&self,
3744
_request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

‎crates/openshell-cli/tests/provider_commands_integration.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,13 @@ impl TestOpenShell {
233233

234234
#[tonic::async_trait]
235235
impl OpenShell for TestOpenShell {
236+
async fn peer_notify_config_update(
237+
&self,
238+
_request: tonic::Request<openshell_core::proto::PeerNotifyConfigUpdateRequest>,
239+
) -> Result<Response<openshell_core::proto::PeerNotifyConfigUpdateResponse>, Status> {
240+
Err(Status::unimplemented("not used by this test server"))
241+
}
242+
236243
async fn peer_report_provider_readiness(
237244
&self,
238245
_request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

‎crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,13 @@ struct TestOpenShell {
101101

102102
#[tonic::async_trait]
103103
impl OpenShell for TestOpenShell {
104+
async fn peer_notify_config_update(
105+
&self,
106+
_request: tonic::Request<openshell_core::proto::PeerNotifyConfigUpdateRequest>,
107+
) -> Result<Response<openshell_core::proto::PeerNotifyConfigUpdateResponse>, Status> {
108+
Err(Status::unimplemented("not used by this test server"))
109+
}
110+
104111
async fn peer_report_provider_readiness(
105112
&self,
106113
_request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

‎crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -275,6 +275,7 @@ mock_gateway! {
275275
peer_report_provider_readiness(proto::ReportProviderReadinessRequest) -> proto::ReportProviderReadinessResponse;
276276
peer_report_endpoint_status(proto::ReportEndpointStatusRequest) -> proto::ReportEndpointStatusResponse;
277277
peer_get_sandbox_provider_status(proto::GetSandboxProviderStatusRequest) -> proto::GetSandboxProviderStatusResponse;
278+
peer_notify_config_update(proto::PeerNotifyConfigUpdateRequest) -> proto::PeerNotifyConfigUpdateResponse;
278279
submit_policy_analysis(proto::SubmitPolicyAnalysisRequest) -> proto::SubmitPolicyAnalysisResponse;
279280
get_draft_policy(proto::GetDraftPolicyRequest) -> proto::GetDraftPolicyResponse;
280281
approve_draft_chunk(proto::ApproveDraftChunkRequest) -> proto::ApproveDraftChunkResponse;

‎crates/openshell-cli/tests/sandbox_name_fallback_integration.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,13 @@ struct TestOpenShell {
4747

4848
#[tonic::async_trait]
4949
impl OpenShell for TestOpenShell {
50+
async fn peer_notify_config_update(
51+
&self,
52+
_request: tonic::Request<openshell_core::proto::PeerNotifyConfigUpdateRequest>,
53+
) -> Result<Response<openshell_core::proto::PeerNotifyConfigUpdateResponse>, Status> {
54+
Err(Status::unimplemented("not used by this test server"))
55+
}
56+
5057
async fn peer_report_provider_readiness(
5158
&self,
5259
_request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

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

Lines changed: 49 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,33 @@ pub const DEFAULT_DOCKER_NETWORK_NAME: &str = "openshell-docker";
4646
/// Default domain used for browser-facing sandbox service URLs.
4747
pub const DEFAULT_SERVICE_ROUTING_DOMAIN: &str = "openshell.localhost";
4848

49+
/// Gateway delivery path for supervisor configuration.
50+
///
51+
/// `Push` sends configuration over the supervisor session to supervisors that
52+
/// apply streamed configuration, which then stop polling. Other supervisors
53+
/// keep polling.
54+
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
55+
#[serde(rename_all = "snake_case")]
56+
pub enum ConfigDeliveryMode {
57+
#[default]
58+
Poll,
59+
Push,
60+
}
61+
62+
impl FromStr for ConfigDeliveryMode {
63+
type Err = String;
64+
65+
fn from_str(value: &str) -> Result<Self, Self::Err> {
66+
match value {
67+
"poll" => Ok(Self::Poll),
68+
"push" => Ok(Self::Push),
69+
_ => Err(format!(
70+
"invalid config delivery mode '{value}'; expected poll or push"
71+
)),
72+
}
73+
}
74+
}
75+
4976
/// Gateway posture when a sandbox rejects a candidate policy generation.
5077
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
5178
#[serde(rename_all = "snake_case")]
@@ -209,6 +236,9 @@ pub struct Config {
209236
/// Security posture for rejected sandbox policy generations.
210237
pub policy_validation_failure_mode: PolicyValidationFailureMode,
211238

239+
/// Whether supervisors poll for configuration or receive it over their session.
240+
pub config_delivery_mode: ConfigDeliveryMode,
241+
212242
/// TLS configuration. When `None`, the server listens on plaintext HTTP.
213243
pub tls: Option<TlsConfig>,
214244

@@ -883,6 +913,7 @@ impl Config {
883913
metrics_bind_address: None,
884914
log_level: default_log_level(),
885915
policy_validation_failure_mode: PolicyValidationFailureMode::default(),
916+
config_delivery_mode: ConfigDeliveryMode::default(),
886917
tls,
887918
oidc: None,
888919
auth: GatewayAuthConfig::default(),
@@ -1149,10 +1180,11 @@ const fn default_ssh_session_ttl_secs() -> u64 {
11491180
#[cfg(test)]
11501181
mod tests {
11511182
use super::{
1152-
AppArmorProfile, Config, DEFAULT_SERVICE_ROUTING_DOMAIN, GatewayInterceptorBindingPolicy,
1153-
GatewayInterceptorConfig, GatewayInterceptorFailurePolicy, GatewayJwtConfig,
1154-
GatewayProviderProfileSourceConfig, ImagePullPolicy, PolicyValidationFailureMode,
1155-
UpstreamProxyConfig, default_sandbox_pids_limit, normalize_compute_driver_name,
1183+
AppArmorProfile, Config, ConfigDeliveryMode, DEFAULT_SERVICE_ROUTING_DOMAIN,
1184+
GatewayInterceptorBindingPolicy, GatewayInterceptorConfig, GatewayInterceptorFailurePolicy,
1185+
GatewayJwtConfig, GatewayProviderProfileSourceConfig, ImagePullPolicy,
1186+
PolicyValidationFailureMode, UpstreamProxyConfig, default_sandbox_pids_limit,
1187+
normalize_compute_driver_name,
11561188
};
11571189
use std::net::SocketAddr;
11581190
use std::time::Duration;
@@ -1172,6 +1204,19 @@ mod tests {
11721204
assert!("keep_old".parse::<PolicyValidationFailureMode>().is_err());
11731205
}
11741206

1207+
#[test]
1208+
fn config_delivery_defaults_to_poll_and_rejects_unknown_modes() {
1209+
assert_eq!(
1210+
Config::new(None).config_delivery_mode,
1211+
ConfigDeliveryMode::Poll
1212+
);
1213+
assert_eq!(
1214+
"push".parse::<ConfigDeliveryMode>().unwrap(),
1215+
ConfigDeliveryMode::Push
1216+
);
1217+
assert!("enabled".parse::<ConfigDeliveryMode>().is_err());
1218+
}
1219+
11751220
#[test]
11761221
fn compute_driver_name_normalization_accepts_builtin_and_custom_names() {
11771222
assert_eq!(normalize_compute_driver_name(" VM ").unwrap(), "vm");

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

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1437,6 +1437,7 @@ pub struct SettingsPollResult {
14371437
pub version: u32,
14381438
pub policy_hash: String,
14391439
pub config_revision: u64,
1440+
pub settings_revision: u64,
14401441
pub policy_source: PolicySource,
14411442
/// Effective settings keyed by name.
14421443
pub settings: HashMap<String, crate::proto::EffectiveSetting>,
@@ -1463,6 +1464,7 @@ fn settings_poll_result(inner: crate::proto::GetSandboxConfigResponse) -> Settin
14631464
version: inner.version,
14641465
policy_hash: inner.policy_hash,
14651466
config_revision: inner.config_revision,
1467+
settings_revision: inner.settings_revision,
14661468
policy_source: PolicySource::try_from(inner.policy_source)
14671469
.unwrap_or(PolicySource::Unspecified),
14681470
settings: inner.settings,
@@ -1479,6 +1481,32 @@ fn settings_poll_result(inner: crate::proto::GetSandboxConfigResponse) -> Settin
14791481
}
14801482
}
14811483

1484+
/// Delivered snapshots go through the polling projection, so both paths
1485+
/// apply the same defaults and validation.
1486+
impl From<crate::proto::SandboxConfigSnapshot> for SettingsPollResult {
1487+
fn from(snapshot: crate::proto::SandboxConfigSnapshot) -> Self {
1488+
settings_poll_result(crate::proto::GetSandboxConfigResponse {
1489+
policy: snapshot.policy,
1490+
version: snapshot.version,
1491+
policy_hash: snapshot.policy_hash,
1492+
settings: snapshot.settings,
1493+
config_revision: snapshot.config_revision,
1494+
policy_source: snapshot.policy_source,
1495+
global_policy_version: snapshot.global_policy_version,
1496+
provider_env_revision: snapshot.provider_env_revision,
1497+
supervisor_middleware_services: snapshot.supervisor_middleware_services,
1498+
workspace: snapshot.workspace,
1499+
policy_validation_failure_mode: snapshot.policy_validation_failure_mode,
1500+
extension_authentication_enabled: snapshot.extension_authentication_enabled,
1501+
configuration_admitted: snapshot.configuration_admitted,
1502+
provider_attachment_epoch: snapshot.provider_attachment_epoch,
1503+
configuration_instance_id: snapshot.configuration_instance_id,
1504+
configuration_error: snapshot.configuration_error,
1505+
settings_revision: snapshot.settings_revision,
1506+
})
1507+
}
1508+
}
1509+
14821510
#[cfg(test)]
14831511
mod settings_poll_tests {
14841512
use super::{learned_workspace_selector, settings_poll_result};
@@ -1521,6 +1549,25 @@ mod settings_poll_tests {
15211549
assert!(!legacy.extension_authentication_enabled);
15221550
}
15231551

1552+
#[test]
1553+
fn delivered_sandbox_snapshot_uses_the_polling_projection() {
1554+
let result = super::SettingsPollResult::from(crate::proto::SandboxConfigSnapshot {
1555+
version: 7,
1556+
policy_hash: "hash-7".to_string(),
1557+
config_revision: 42,
1558+
provider_env_revision: 9,
1559+
workspace: "workspace-a".to_string(),
1560+
extension_authentication_enabled: true,
1561+
..Default::default()
1562+
});
1563+
assert_eq!(result.version, 7);
1564+
assert_eq!(result.policy_hash, "hash-7");
1565+
assert_eq!(result.config_revision, 42);
1566+
assert_eq!(result.provider_env_revision, 9);
1567+
assert_eq!(result.workspace, "workspace-a");
1568+
assert!(result.extension_authentication_enabled);
1569+
}
1570+
15241571
#[test]
15251572
fn workspace_selector_is_omitted_until_bootstrap_learns_the_workspace() {
15261573
assert!(learned_workspace_selector(None).is_none());
@@ -1555,6 +1602,140 @@ pub struct ProviderEnvironmentResult {
15551602
pub stable_placeholder_environment_keys: Vec<String>,
15561603
}
15571604

1605+
/// Delivered snapshots go through the polling projection, so both paths
1606+
/// reject invalid expiration times and withhold on unknown readiness reasons.
1607+
impl TryFrom<crate::proto::ProviderEnvironmentSnapshot> for ProviderEnvironmentResult {
1608+
type Error = miette::Report;
1609+
1610+
fn try_from(snapshot: crate::proto::ProviderEnvironmentSnapshot) -> Result<Self> {
1611+
use crate::proto::ProviderEnvironmentValueClassification;
1612+
1613+
let mut response = GetSandboxProviderEnvironmentResponse {
1614+
provider_env_revision: snapshot.provider_env_revision,
1615+
dynamic_credentials: snapshot.dynamic_credentials,
1616+
provider_attachment_epoch: snapshot.provider_attachment_epoch,
1617+
policy_hash: snapshot.policy_hash,
1618+
readiness_reason: snapshot.readiness_reason,
1619+
files: snapshot.files,
1620+
// Supervisor session payloads are version-locked with the gateway,
1621+
// so every delivered snapshot carries stable-placeholder metadata.
1622+
supports_stable_placeholder_environment_keys: true,
1623+
..Default::default()
1624+
};
1625+
for value in snapshot.values {
1626+
if let Some(expiration_time) = value.expiration_time {
1627+
response
1628+
.credential_expiration_times
1629+
.insert(value.name.clone(), expiration_time);
1630+
}
1631+
match ProviderEnvironmentValueClassification::try_from(value.classification)
1632+
.unwrap_or_default()
1633+
{
1634+
ProviderEnvironmentValueClassification::NonSecret => {
1635+
response
1636+
.non_secret_environment_keys
1637+
.push(value.name.clone());
1638+
}
1639+
ProviderEnvironmentValueClassification::StaticCredential => {
1640+
if value.stable_placeholder {
1641+
response
1642+
.stable_placeholder_environment_keys
1643+
.push(value.name.clone());
1644+
}
1645+
if let Some(binding) = value.static_credential_binding {
1646+
response
1647+
.static_credential_bindings
1648+
.insert(value.name.clone(), binding);
1649+
}
1650+
}
1651+
ProviderEnvironmentValueClassification::Unspecified => {}
1652+
}
1653+
response.environment.insert(value.name, value.value);
1654+
}
1655+
provider_environment_result(response, false)
1656+
}
1657+
}
1658+
1659+
#[cfg(test)]
1660+
mod provider_snapshot_tests {
1661+
use super::ProviderEnvironmentResult;
1662+
use crate::proto::{
1663+
ProviderEnvironmentSnapshot, ProviderEnvironmentValue,
1664+
ProviderEnvironmentValueClassification, ProviderReadinessReason, StaticCredentialBinding,
1665+
};
1666+
1667+
#[test]
1668+
fn delivered_provider_snapshot_preserves_secret_classification_metadata() {
1669+
let result = ProviderEnvironmentResult::try_from(ProviderEnvironmentSnapshot {
1670+
provider_env_revision: 11,
1671+
values: vec![
1672+
ProviderEnvironmentValue {
1673+
name: "REGION".to_string(),
1674+
value: "west".to_string(),
1675+
classification: ProviderEnvironmentValueClassification::NonSecret.into(),
1676+
..Default::default()
1677+
},
1678+
ProviderEnvironmentValue {
1679+
name: "TOKEN".to_string(),
1680+
value: "redacted".to_string(),
1681+
classification: ProviderEnvironmentValueClassification::StaticCredential.into(),
1682+
static_credential_binding: Some(StaticCredentialBinding::default()),
1683+
expiration_time: Some(prost_types::Timestamp {
1684+
seconds: 1_900_000_000,
1685+
nanos: 123_000_000,
1686+
}),
1687+
stable_placeholder: true,
1688+
},
1689+
],
1690+
..Default::default()
1691+
})
1692+
.expect("valid provider snapshot");
1693+
assert_eq!(result.provider_env_revision, 11);
1694+
assert_eq!(result.environment.len(), 2);
1695+
assert_eq!(result.non_secret_environment_keys, ["REGION"]);
1696+
assert!(result.static_credential_bindings.contains_key("TOKEN"));
1697+
assert_eq!(result.stable_placeholder_environment_keys, ["TOKEN"]);
1698+
assert_eq!(
1699+
result.credential_expires_at_ms.get("TOKEN"),
1700+
Some(&1_900_000_000_123)
1701+
);
1702+
assert_eq!(
1703+
result.readiness_reason,
1704+
ProviderReadinessReason::Unspecified
1705+
);
1706+
}
1707+
1708+
#[test]
1709+
fn delivered_provider_snapshot_withholds_unknown_readiness_reason() {
1710+
let result = ProviderEnvironmentResult::try_from(ProviderEnvironmentSnapshot {
1711+
readiness_reason: i32::MAX,
1712+
..Default::default()
1713+
})
1714+
.expect("valid provider snapshot");
1715+
assert_eq!(
1716+
result.readiness_reason,
1717+
ProviderReadinessReason::CredentialsWithheld
1718+
);
1719+
}
1720+
1721+
#[test]
1722+
fn delivered_provider_snapshot_rejects_invalid_credential_expiration() {
1723+
let result = ProviderEnvironmentResult::try_from(ProviderEnvironmentSnapshot {
1724+
values: vec![ProviderEnvironmentValue {
1725+
name: "TOKEN".to_string(),
1726+
value: "redacted".to_string(),
1727+
expiration_time: Some(prost_types::Timestamp {
1728+
seconds: 1_900_000_000,
1729+
nanos: -1,
1730+
}),
1731+
..Default::default()
1732+
}],
1733+
..Default::default()
1734+
});
1735+
assert!(result.is_err());
1736+
}
1737+
}
1738+
15581739
pub struct ProviderSubjectTokenExchangeResult {
15591740
pub access_token: String,
15601741
pub expires_in: i64,

0 commit comments

Comments
 (0)