trending/OpenShell
1
0
Fork 0
mirror of https://github.com/NVIDIA/OpenShell.git synced 2026-10-30 07:33:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(supervisor): gate launch on streamed admission

Browse Source 
Signed-off-by: Piotr Mlocek <pmlocek@nvidia.com>
This commit is contained in:
Piotr Mlocek
2026-09-18 15:29:35 -07:00
parent bc616a8a08
commit d6090bc36e
8 changed files with 674 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files
architecture/sandbox.md
+31 -16
View File
@@ -659,7 +659,18 @@ sequenceDiagram
GW->>DB: Build fresh authoritative bootstrap
GW->>SUP: SessionAccepted(bootstrap)
SUP->>SUP: Initialize from bootstrap
SUP->>GW: ConfigBootstrapResult
SUP->>GW: ConfigBootstrapResult(component outcomes, admission)
GW->>DB: Persist admission for the delivered generation
GW->>SUP: ConfigurationAdmission(durable state)
alt Configuration accepted
SUP->>SUP: Launch workload
else Configuration rejected
GW->>SUP: ConfigUpdate(repaired snapshot)
SUP->>GW: ConfigUpdateResult(component outcome, admission)
GW->>DB: Persist accepted repair
GW->>SUP: ConfigurationAdmission(accepted)
SUP->>SUP: Launch workload on the same supervisor
end
```
The gateway can send complete component replacements on the accepted stream
@@ -682,21 +693,25 @@ without changing publishers. Provider payloads can contain
credentials, so the gateway does not persist or render complete stream
messages in logs.
The supervisor currently parses and ignores stream-delivered configuration.
Polling remains the only path that changes runtime state and repairs dropped or
unavailable delivery. The gateway serializes construction per sandbox and
component, and coalesces repeated mutations into the latest full snapshot. An
enqueue result means only that the local stream queue accepted the message. A
bounded scope fanout scheduler coalesces repeated workspace and global changes,
and semaphores sized from the database pool bound delivery workers and snapshot
builds. Fanout waits for worker capacity before admitting each recipient, so a
fleet-wide change cannot create a fleet-sized task backlog or saturate the store
and credential backends. Snapshot construction has a deadline that starts once
a build holds a permit, and the gateway rejects encoded stream messages that
approach the transport decoder limit. A later migration will apply these
payloads directly and acknowledge their exact revisions before removing
supervisor polling. At that point, the gateway will require a valid bootstrap
before marking a session ready.
Current-protocol supervisors apply stream-delivered configuration and report
both component outcomes and admission for the exact gateway-authored generation.
The gateway validates generation identity, persists admission, and returns that
durable state on the stream. The supervisor holds the workload boundary until
it receives an accepted acknowledgement. Rejection leaves the stream and
supervisor alive so a later complete replacement can repair the generation and
release the same workload. Compatibility protocol revisions continue using
polling.
The gateway serializes construction per sandbox and component, and coalesces
repeated mutations into the latest full snapshot. An enqueue result means only
that the local stream queue accepted the message. A bounded scope fanout
scheduler coalesces repeated workspace and global changes, and semaphores sized
from the database pool bound delivery workers and snapshot builds. Fanout waits
for worker capacity before admitting each recipient, so a fleet-wide change
cannot create a fleet-sized task backlog or saturate the store and credential
backends. Snapshot construction has a deadline that starts once a build holds a
permit, and the gateway rejects encoded stream messages that approach the
transport decoder limit.
## Configuration Admission
crates/openshell-server/src/compute/mod.rs
+41 -24
View File
@@ -3358,7 +3358,7 @@ impl ComputeRuntime {
.await
}
pub async fn supervisor_session_initialized(
pub async fn supervisor_session_admission(
&self,
sandbox_id: &str,
instance_id: &str,
@@ -3475,7 +3475,10 @@ impl ComputeRuntime {
status.exit_code = None;
if let Some(admission) = admission {
status.configuration_admission = Some(admission.clone());
status.configuration_activated = Some(true);
status.configuration_activated = Some(
admission.state
== i32::from(openshell_core::proto::ConfigurationAdmissionState::Accepted),
);
}
sandbox.set_phase(SandboxPhase::Ready as i32);
} else {
@@ -10507,45 +10510,59 @@ mod tests {
}
#[tokio::test]
async fn supervisor_session_initialized_accepts_streamed_configuration_atomically() {
async fn supervisor_session_admission_preserves_repair_on_same_instance() {
use openshell_core::proto::{ConfigurationAdmissionState, SandboxConfigurationAdmission};
let runtime = test_runtime(Arc::new(TestDriver::default())).await;
let mut sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning);
sandbox
.status
.get_or_insert_default()
.configuration_admission = Some(SandboxConfigurationAdmission {
instance_id: "pending-fence".into(),
state: ConfigurationAdmissionState::Pending.into(),
..Default::default()
});
let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning);
runtime.store.put_message(&sandbox).await.unwrap();
let admission = SandboxConfigurationAdmission {
instance_id: "test-generation".into(),
state: ConfigurationAdmissionState::Accepted.into(),
let mut admission = SandboxConfigurationAdmission {
instance_id: "configuration-1".into(),
state: ConfigurationAdmissionState::Rejected.into(),
policy_version: 7,
policy_hash: "policy-hash".into(),
config_revision: 11,
provider_env_revision: 13,
error: String::new(),
error: "invalid image policy".into(),
};
runtime
.supervisor_session_initialized("sb-1", "test-generation", &admission)
.supervisor_session_admission("sb-1", "supervisor-1", &admission)
.await
.unwrap();
let stored = runtime
let rejected = runtime
.store
.get_message::<Sandbox>("sb-1")
.await
.unwrap()
.unwrap();
let status = stored.status.as_ref().unwrap();
assert_eq!(status.configuration_admission.as_ref(), Some(&admission));
assert_eq!(status.configuration_activated, Some(true));
assert_eq!(stored.phase(), SandboxPhase::Ready as i32);
assert_eq!(rejected.phase(), SandboxPhase::Provisioning as i32);
assert_eq!(
rejected.status.as_ref().unwrap().configuration_activated,
Some(false)
);
admission.state = ConfigurationAdmissionState::Accepted.into();
admission.error.clear();
runtime
.supervisor_session_admission("sb-1", "supervisor-1", &admission)
.await
.unwrap();
let repaired = runtime
.store
.get_message::<Sandbox>("sb-1")
.await
.unwrap()
.unwrap();
assert_eq!(repaired.phase(), SandboxPhase::Ready as i32);
assert_eq!(
repaired.status.as_ref().unwrap().main_process_instance_id,
"supervisor-1"
);
assert_eq!(
repaired.status.as_ref().unwrap().configuration_activated,
Some(true)
);
}
#[tokio::test]
crates/openshell-server/src/storage_proto.rs
+1 -1
View File
@@ -162,7 +162,7 @@ mod tests {
const STORAGE_V1_SCHEMA_SHA256: &str =
"d68401809d8cea445c35233ef32412bbd041cb2ac5acaf368a0d0bf74d2ddf17";
const PUBLIC_RPC_SCHEMA_SHA256: &str =
"6c9b49049f0de12aecb68621b139553e35d6fa8fae2e35e4002c75e5e1bdbe60";
"ab978d653fc8bff9f7dac633dd651b015fc73aecf3704d1a43f3b32e0143167f";
const DURABLE_SCHEMA_SHA256: &str =
"965a8a09fa80168906a4f9cc6169d3623b98f5281c367695f2f2898230915dd0";
const PUBLIC_DURABLE_OVERLAP_SHA256: &str =
crates/openshell-server/src/supervisor_session.rs
+265 -60
View File
@@ -21,9 +21,10 @@ use openshell_core::proto::{
ConfigApplyOutcome, ConfigBootstrap, ConfigComponent, ConfigComponentApplyResult,
ConfigSnapshotRevision, ConfigUpdate, ConfigUpdateResult, GatewayMessage, PolicySource,
ProviderReadinessObservation, RelayFrame, RelayInit, RelayOpen, ReportMainProcessExitRequest,
ReportMainProcessExitResponse, Sandbox, SandboxPhase, SessionAccepted, SessionRejected,
SshRelayTarget, StartupConfigCandidate, SupervisorMessage, config_snapshot_revision,
config_update, gateway_message, relay_open, startup_config_prepared, supervisor_message,
ReportMainProcessExitResponse, Sandbox, SandboxConfigurationAdmission, SandboxPhase,
SessionAccepted, SessionRejected, SshRelayTarget, StartupConfigCandidate, SupervisorMessage,
config_snapshot_revision, config_update, gateway_message, relay_open, startup_config_prepared,
supervisor_message,
};
use openshell_core::proto::{
LEGACY_SUPERVISOR_PROTOCOL_REVISION, PREVIOUS_SUPERVISOR_PROTOCOL_REVISION,
@@ -124,9 +125,83 @@ struct InFlightConfigUpdate {
update_id: String,
component_sequence: u64,
revision: ConfigSnapshotRevision,
admission: Option<SandboxConfigurationAdmission>,
sent_at: Instant,
}
struct CompletedConfigUpdate {
outcome: ConfigApplyOutcome,
admission: Option<SandboxConfigurationAdmission>,
}
fn expected_configuration_admission(
snapshot: &openshell_core::proto::SandboxConfigSnapshot,
) -> SandboxConfigurationAdmission {
use openshell_core::proto::ConfigurationAdmissionState;
SandboxConfigurationAdmission {
instance_id: snapshot.configuration_instance_id.clone(),
state: if snapshot.configuration_admitted {
ConfigurationAdmissionState::Accepted.into()
} else {
ConfigurationAdmissionState::Rejected.into()
},
policy_version: snapshot.version,
policy_hash: snapshot.policy_hash.clone(),
config_revision: snapshot.config_revision,
provider_env_revision: snapshot.provider_env_revision,
error: snapshot.configuration_error.clone(),
}
}
fn validate_configuration_admission(
reported: Option<&SandboxConfigurationAdmission>,
expected: &SandboxConfigurationAdmission,
outcome: ConfigApplyOutcome,
) -> Result<SandboxConfigurationAdmission, Status> {
use openshell_core::proto::ConfigurationAdmissionState;
let reported = reported.ok_or_else(|| {
Status::invalid_argument("sandbox configuration admission result is required")
})?;
if reported.instance_id != expected.instance_id
|| reported.policy_version != expected.policy_version
|| reported.policy_hash != expected.policy_hash
|| reported.config_revision != expected.config_revision
|| reported.provider_env_revision != expected.provider_env_revision
{
return Err(Status::invalid_argument(
"configuration admission does not match the delivered generation",
));
}
let expected_state = ConfigurationAdmissionState::try_from(expected.state).unwrap_or_default();
let reported_state = ConfigurationAdmissionState::try_from(reported.state).unwrap_or_default();
if expected_state == ConfigurationAdmissionState::Rejected {
if reported_state != ConfigurationAdmissionState::Rejected {
return Err(Status::invalid_argument(
"rejected gateway configuration was reported as accepted",
));
}
return Ok(expected.clone());
}
if outcome_acknowledges_revision(outcome)
&& reported_state == ConfigurationAdmissionState::Accepted
{
let mut admission = expected.clone();
admission.error.clear();
return Ok(admission);
}
if reported_state == ConfigurationAdmissionState::Rejected {
let mut admission = expected.clone();
admission.state = ConfigurationAdmissionState::Rejected.into();
admission.error = "effective configuration could not be activated".to_string();
return Ok(admission);
}
Err(Status::invalid_argument(
"configuration admission is inconsistent with the apply result",
))
}
/// Idempotency state for tool server endpoint-status reports from one live supervisor.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct EndpointReportCursor {
@@ -184,8 +259,9 @@ fn build_config_update(
) -> (GatewayMessage, InFlightConfigUpdate) {
state.sequence = state.sequence.saturating_add(1);
let component_sequence = state.sequence;
let (component, revision) = match message {
let (component, revision, admission) = match message {
SupervisorConfigMessage::SandboxConfig(snapshot) => {
let admission = expected_configuration_admission(&snapshot);
let revision = ConfigSnapshotRevision {
component: Some(config_snapshot_revision::Component::SandboxConfig(
openshell_core::proto::SandboxConfigRevision {
@@ -197,7 +273,11 @@ fn build_config_update(
},
)),
};
(config_update::Component::SandboxConfig(*snapshot), revision)
(
config_update::Component::SandboxConfig(*snapshot),
revision,
Some(admission),
)
}
SupervisorConfigMessage::ProviderEnvironment(snapshot) => {
let revision = ConfigSnapshotRevision {
@@ -208,6 +288,7 @@ fn build_config_update(
(
config_update::Component::ProviderEnvironment(snapshot),
revision,
None,
)
}
};
@@ -224,6 +305,7 @@ fn build_config_update(
update_id,
component_sequence,
revision,
admission,
sent_at: Instant::now(),
},
)
@@ -495,7 +577,7 @@ impl SupervisorSessionRegistry {
sandbox_id: &str,
session_id: &str,
result: &ConfigUpdateResult,
) -> Result<(), Status> {
) -> Result<CompletedConfigUpdate, Status> {
let component = result
.result
.as_ref()
@@ -533,6 +615,10 @@ impl SupervisorSessionRegistry {
));
}
let outcome = validate_component_apply_result(component_result, &in_flight.revision)?;
let completed = CompletedConfigUpdate {
outcome,
admission: in_flight.admission.clone(),
};
if outcome_acknowledges_revision(outcome) {
delivery_state.last_acknowledged_revision = Some(in_flight.revision.clone());
}
@@ -541,7 +627,7 @@ impl SupervisorSessionRegistry {
if delivery_state.last_acknowledged_revision.as_ref()
== Some(&config_message_revision(&pending))
{
return Ok(());
return Ok(completed);
}
let (message, next) = build_config_update(delivery_state, pending);
match session.tx.try_send(message) {
@@ -551,7 +637,7 @@ impl SupervisorSessionRegistry {
}
}
}
Ok(())
Ok(completed)
}
/// Record a successfully persisted bootstrap result in the live session's
@@ -1288,18 +1374,10 @@ async fn accept_supervisor_session(
outbound_tx: mpsc::Sender<GatewayMessage>,
mut inbound: tonic::Streaming<SupervisorMessage>,
) -> Result<(), Status> {
let bootstrap_admission = bootstrap.as_ref().and_then(|bootstrap| {
let snapshot = bootstrap.sandbox_config.as_ref()?;
Some(openshell_core::proto::SandboxConfigurationAdmission {
instance_id: instance_id.clone(),
state: openshell_core::proto::ConfigurationAdmissionState::Accepted.into(),
policy_version: snapshot.version,
policy_hash: snapshot.policy_hash.clone(),
config_revision: snapshot.config_revision,
provider_env_revision: snapshot.provider_env_revision,
error: String::new(),
})
});
let expected_bootstrap_admission = bootstrap
.as_ref()
.and_then(|bootstrap| bootstrap.sandbox_config.as_ref())
.map(expected_configuration_admission);
let expected_bootstrap_revisions = bootstrap
.as_ref()
.map(bootstrap_revision_fence)
@@ -1389,7 +1467,7 @@ async fn accept_supervisor_session(
}
if !stream_applies_config
&& !mark_supervisor_initialized(&state, &sandbox_id, &session_id, &instance_id, None).await
&& !mark_supervisor_initialized(&state, &sandbox_id, &session_id, &instance_id).await
{
state
.supervisor_sessions
@@ -1436,7 +1514,7 @@ async fn accept_supervisor_session(
&instance_id,
stream_applies_config,
&expected_bootstrap_revisions,
bootstrap_admission.as_ref(),
expected_bootstrap_admission.as_ref(),
&session_tx,
&mut inbound,
shutdown_rx,
@@ -1888,7 +1966,7 @@ async fn run_session_loop(
instance_id: &str,
stream_applies_config: bool,
expected_bootstrap_revisions: &[(ConfigComponent, ConfigSnapshotRevision)],
bootstrap_admission: Option<&openshell_core::proto::SandboxConfigurationAdmission>,
expected_bootstrap_admission: Option<&SandboxConfigurationAdmission>,
tx: &mpsc::Sender<GatewayMessage>,
inbound: &mut tonic::Streaming<SupervisorMessage>,
mut shutdown_rx: oneshot::Receiver<()>,
@@ -1910,7 +1988,7 @@ async fn run_session_loop(
msg = inbound.message() => {
match msg {
Ok(Some(msg)) => {
let bootstrap_succeeded = match msg.payload.as_ref() {
let bootstrap_result = match msg.payload.as_ref() {
Some(supervisor_message::Payload::ConfigBootstrapResult(result))
if stream_applies_config =>
{
@@ -1918,7 +1996,7 @@ async fn run_session_loop(
&result.results,
expected_bootstrap_revisions,
) {
Ok(succeeded) => Some(succeeded),
Ok(succeeded) => Some((succeeded, result.admission.clone())),
Err(error) => {
warn!(
sandbox_id,
@@ -1936,29 +2014,45 @@ async fn run_session_loop(
state,
sandbox_id,
session_id,
instance_id,
stream_applies_config,
tx,
msg,
).await;
match bootstrap_succeeded {
Some(true) => {
if !mark_supervisor_initialized(
if let Some((succeeded, reported_admission)) = bootstrap_result {
bootstrap_complete = true;
let Some(expected_admission) = expected_bootstrap_admission else {
warn!(sandbox_id, session_id, "supervisor bootstrap omitted expected admission");
break;
};
let outcome = if succeeded {
ConfigApplyOutcome::Applied
} else {
ConfigApplyOutcome::Unsupported
};
let admission = match validate_configuration_admission(
reported_admission.as_ref(),
expected_admission,
outcome,
) {
Ok(admission) => admission,
Err(error) => {
warn!(sandbox_id, session_id, error = %error, "invalid supervisor bootstrap admission");
break;
}
};
if !persist_and_ack_admission(
state,
sandbox_id,
session_id,
instance_id,
bootstrap_admission,
&admission,
tx,
)
.await
{
break;
}
bootstrap_complete = true;
}
Some(false) => {
warn!(sandbox_id, session_id, "supervisor configuration bootstrap failed");
break;
}
None => {}
}
}
Ok(None) => {
@@ -2010,7 +2104,9 @@ async fn handle_supervisor_message(
state: &Arc<ServerState>,
sandbox_id: &str,
session_id: &str,
instance_id: &str,
stream_applies_config: bool,
tx: &mpsc::Sender<GatewayMessage>,
msg: SupervisorMessage,
) {
match msg.payload {
@@ -2049,18 +2145,21 @@ async fn handle_supervisor_message(
);
}
Some(supervisor_message::Payload::ConfigUpdateResult(result)) => {
if let Err(error) = state
let completed = match state
.supervisor_sessions
.complete_config_update(sandbox_id, session_id, &result)
{
debug!(
sandbox_id,
session_id,
error = %error,
"ignored unmatched supervisor configuration result"
);
return;
}
Ok(completed) => completed,
Err(error) => {
debug!(
sandbox_id,
session_id,
error = %error,
"ignored unmatched supervisor configuration result"
);
return;
}
};
if let Some(result) = result.result.as_ref()
&& let Err(error) = record_component_apply_result(state, sandbox_id, result).await
{
@@ -2074,6 +2173,29 @@ async fn handle_supervisor_message(
error = %error,
"failed to persist supervisor configuration result"
);
return;
}
if let Some(expected_admission) = completed.admission.as_ref() {
let admission = match validate_configuration_admission(
result.admission.as_ref(),
expected_admission,
completed.outcome,
) {
Ok(admission) => admission,
Err(error) => {
warn!(sandbox_id, session_id, error = %error, "invalid supervisor configuration admission");
return;
}
};
let _ = persist_and_ack_admission(
state,
sandbox_id,
session_id,
instance_id,
&admission,
tx,
)
.await;
}
}
Some(supervisor_message::Payload::ConfigBootstrapResult(result)) => {
@@ -2190,7 +2312,6 @@ async fn mark_supervisor_initialized(
sandbox_id: &str,
session_id: &str,
instance_id: &str,
admission: Option<&openshell_core::proto::SandboxConfigurationAdmission>,
) -> bool {
if !state
.supervisor_sessions
@@ -2198,18 +2319,11 @@ async fn mark_supervisor_initialized(
{
return false;
}
let result = if let Some(admission) = admission {
state
.compute
.supervisor_session_initialized(sandbox_id, instance_id, admission)
.await
} else {
state
.compute
.supervisor_session_connected(sandbox_id, instance_id)
.await
};
if let Err(err) = result {
if let Err(err) = state
.compute
.supervisor_session_connected(sandbox_id, instance_id)
.await
{
warn!(
sandbox_id,
session_id,
@@ -2223,6 +2337,43 @@ async fn mark_supervisor_initialized(
}
}
async fn persist_and_ack_admission(
state: &Arc<ServerState>,
sandbox_id: &str,
session_id: &str,
instance_id: &str,
admission: &SandboxConfigurationAdmission,
tx: &mpsc::Sender<GatewayMessage>,
) -> bool {
if !state
.supervisor_sessions
.is_current_session(sandbox_id, session_id)
{
return false;
}
if let Err(error) = state
.compute
.supervisor_session_admission(sandbox_id, instance_id, admission)
.await
{
warn!(sandbox_id, session_id, error = %error, "failed to persist supervisor configuration admission");
return false;
}
if tx
.send(GatewayMessage {
payload: Some(gateway_message::Payload::ConfigurationAdmission(
admission.clone(),
)),
})
.await
.is_err()
{
return false;
}
state.telemetry.sandbox_session_connected(sandbox_id);
true
}
async fn record_component_apply_result(
state: &Arc<ServerState>,
sandbox_id: &str,
@@ -2387,7 +2538,24 @@ mod tests {
})),
});
for original in std::iter::once(bootstrap).chain(updates) {
let admission = SandboxConfigurationAdmission {
instance_id: "configuration-1".into(),
state: openshell_core::proto::ConfigurationAdmissionState::Accepted.into(),
policy_version: 3,
policy_hash: "policy-hash".into(),
config_revision: 7,
provider_env_revision: 5,
error: String::new(),
};
let admission_ack = GatewayMessage {
payload: Some(gateway_message::Payload::ConfigurationAdmission(
admission.clone(),
)),
};
for original in std::iter::once(bootstrap)
.chain(updates)
.chain(std::iter::once(admission_ack))
{
let decoded = GatewayMessage::decode(original.encode_to_vec().as_slice()).unwrap();
assert_eq!(decoded, original);
}
@@ -2420,6 +2588,7 @@ mod tests {
outcome: ConfigApplyOutcome::Applied.into(),
failure: None,
}),
admission: Some(admission),
},
)),
};
@@ -2434,6 +2603,38 @@ mod tests {
assert!(validate_protocol_revision("sb-1", LEGACY_SUPERVISOR_PROTOCOL_REVISION).is_ok());
}
#[test]
fn configuration_admission_preserves_gateway_rejection_and_accepts_repair() {
use openshell_core::proto::ConfigurationAdmissionState;
let mut expected = SandboxConfigurationAdmission {
instance_id: "configuration-1".into(),
state: ConfigurationAdmissionState::Rejected.into(),
policy_version: 3,
policy_hash: "policy-hash".into(),
config_revision: 7,
provider_env_revision: 5,
error: "invalid image policy".into(),
};
let rejected = validate_configuration_admission(
Some(&expected),
&expected,
ConfigApplyOutcome::Applied,
)
.unwrap();
assert_eq!(rejected, expected);
expected.state = ConfigurationAdmissionState::Accepted.into();
expected.error.clear();
let accepted = validate_configuration_admission(
Some(&expected),
&expected,
ConfigApplyOutcome::Applied,
)
.unwrap();
assert_eq!(accepted, expected);
}
#[test]
fn bootstrap_requires_every_delivered_component_to_succeed() {
let sandbox_revision = ConfigSnapshotRevision {
@@ -2518,7 +2719,7 @@ mod tests {
state.supervisor_sessions.register(
"sb-bootstrap-ack".into(),
"session-1".into(),
tx,
tx.clone(),
shutdown_tx,
);
let snapshot = ProviderEnvironmentSnapshot {
@@ -2540,11 +2741,14 @@ mod tests {
&state,
"sb-bootstrap-ack",
"session-1",
"instance-1",
true,
&tx,
SupervisorMessage {
payload: Some(supervisor_message::Payload::ConfigBootstrapResult(
ConfigBootstrapResult {
results: vec![result],
admission: None,
},
)),
},
@@ -3183,6 +3387,7 @@ mod tests {
outcome: ConfigApplyOutcome::Applied.into(),
failure: None,
}),
admission: None,
},
)
.unwrap();
crates/openshell-supervisor-process/src/delegated.rs
+151 -119
View File
@@ -9,7 +9,8 @@ use std::time::Duration;
use miette::Result;
use openshell_isolation_interface::contract::{
BoundaryExec, BoundaryLoopbackConnector, BoundaryProcess,
BackendError, BoundaryDuplexStream, BoundaryExec, BoundaryLoopbackConnector, BoundaryProcess,
LoopbackTarget,
};
use openshell_ocsf::{ActivityId, AppLifecycleBuilder, SeverityId, StatusId, ocsf_emit};
@@ -27,6 +28,105 @@ pub struct BoundaryAccess {
main_session: Option<Arc<crate::main_session::MainSession>>,
}
#[derive(Default)]
struct DeferredLoopbackConnector {
connector: tokio::sync::RwLock<Option<Arc<dyn BoundaryLoopbackConnector>>>,
ready: tokio::sync::Notify,
}
impl DeferredLoopbackConnector {
async fn install(&self, connector: Arc<dyn BoundaryLoopbackConnector>) {
*self.connector.write().await = Some(connector);
self.ready.notify_waiters();
}
}
#[async_trait::async_trait]
impl BoundaryLoopbackConnector for DeferredLoopbackConnector {
async fn connect(&self, target: LoopbackTarget) -> Result<BoundaryDuplexStream, BackendError> {
loop {
let notified = self.ready.notified();
let connector = self.connector.read().await.clone();
if let Some(connector) = connector {
return connector.connect(target).await;
}
notified.await;
}
}
}
/// A supervisor stream that has reported configuration admission while the
/// boundary remains held in its confirmed, pre-workload state.
pub struct PrestartedSupervisorSession {
terminating: Arc<AtomicBool>,
task: Option<tokio::task::JoinHandle<()>>,
readiness: tokio::sync::watch::Receiver<bool>,
loopback: Arc<DeferredLoopbackConnector>,
}
impl Drop for PrestartedSupervisorSession {
fn drop(&mut self) {
if let Some(task) = self.task.take() {
self.terminating.store(true, Ordering::Release);
task.abort();
}
}
}
/// Resume a prepared supervisor stream and report its bootstrap result before
/// workload activation. Relay requests wait until the running boundary installs
/// its loopback connector.
pub async fn start_prepared_supervisor_session(
prepared: crate::supervisor_session::PreparedSupervisorSession,
bootstrap_result: Option<openshell_core::proto::ConfigBootstrapResult>,
ssh_socket_path: Option<&str>,
config_apply_tx: tokio::sync::mpsc::Sender<crate::supervisor_session::ConfigApplyRequest>,
supervisor_session_updates: Option<tokio::sync::watch::Sender<Option<String>>>,
) -> Result<PrestartedSupervisorSession> {
let terminating = Arc::new(AtomicBool::new(false));
let loopback = Arc::new(DeferredLoopbackConnector::default());
let target = ssh_socket_path.map_or_else(
|| std::path::PathBuf::from(openshell_core::container_paths::SSH_SOCKET_PATH),
std::path::PathBuf::from,
);
let (task, mut readiness) = crate::supervisor_session::spawn_prepared(
prepared,
bootstrap_result,
target,
loopback.clone(),
None,
terminating.clone(),
config_apply_tx,
supervisor_session_updates,
);
let ready = tokio::time::timeout(
crate::supervisor_session::SESSION_PREPARE_TIMEOUT,
readiness.wait_for(|ready| *ready),
)
.await
.map(|result| result.map(|_| ()));
match ready {
Ok(Ok(())) => Ok(PrestartedSupervisorSession {
terminating,
task: Some(task),
readiness,
loopback,
}),
Ok(Err(_)) => {
task.abort();
Err(miette::miette!(
"prepared supervisor session ended before bootstrap acknowledgement"
))
}
Err(_) => {
task.abort();
Err(miette::miette!(
"prepared supervisor session did not receive accepted configuration admission before the provisioning deadline"
))
}
}
}
impl BoundaryAccess {
/// Stable supervisor instance ID used for lifecycle reporting.
#[must_use]
@@ -88,59 +188,22 @@ pub async fn start_boundary_access(
port_forward: Arc<dyn BoundaryLoopbackConnector>,
agent: Arc<dyn BoundaryProcess>,
supervisor_session_updates: Option<tokio::sync::watch::Sender<Option<String>>>,
prepared_supervisor_session: Option<crate::supervisor_session::PreparedSupervisorSession>,
prepared_bootstrap_result: Option<openshell_core::proto::ConfigBootstrapResult>,
mut prestarted_supervisor_session: Option<PrestartedSupervisorSession>,
config_apply_tx: Option<
tokio::sync::mpsc::Sender<crate::supervisor_session::ConfigApplyRequest>,
>,
) -> Result<BoundaryAccess> {
let terminating = Arc::new(AtomicBool::new(false));
if let Some(prestarted) = prestarted_supervisor_session.as_ref() {
prestarted.loopback.install(port_forward.clone()).await;
}
let terminating = prestarted_supervisor_session.as_ref().map_or_else(
|| Arc::new(AtomicBool::new(false)),
|prestarted| prestarted.terminating.clone(),
);
let Some(ssh_socket_path) = ssh_socket_path.map(std::path::PathBuf::from) else {
let (session_task, session_readiness) = match (prepared_supervisor_session, config_apply_tx)
{
(Some(prepared), Some(config_apply_tx)) => {
// Configuration admission is carried by the supervisor stream,
// even when this sandbox does not expose SSH. Keep the stream
// alive and use the canonical socket path only as the target for
// any unexpected relay request; no listener is started here.
let (task, mut accepted) = crate::supervisor_session::spawn_prepared(
prepared,
prepared_bootstrap_result,
std::path::PathBuf::from(openshell_core::container_paths::SSH_SOCKET_PATH),
port_forward,
None,
terminating.clone(),
config_apply_tx,
supervisor_session_updates,
);
let accepted_result = tokio::time::timeout(
Duration::from_secs(10),
accepted.wait_for(|ready| *ready),
)
.await
.map(|result| result.map(|_| ()));
match accepted_result {
Ok(Ok(())) => (Some(task), Some(accepted)),
Ok(Err(_)) => {
task.abort();
return Err(miette::miette!(
"prepared supervisor session ended before bootstrap acknowledgement"
));
}
Err(_) => {
task.abort();
return Err(miette::miette!(
"prepared supervisor session did not become ready within 10 seconds"
));
}
}
}
(Some(_), None) => {
return Err(miette::miette!(
"prepared supervisor session requires a configuration apply channel"
));
}
(None, _) => (None, None),
let (session_task, session_readiness) = match prestarted_supervisor_session.as_mut() {
Some(prestarted) => (prestarted.task.take(), Some(prestarted.readiness.clone())),
None => (None, None),
};
return Ok(BoundaryAccess {
instance_id,
@@ -205,78 +268,47 @@ pub async fn start_boundary_access(
}
}
let (session_task, session_readiness) = match (
prepared_supervisor_session,
config_apply_tx.clone(),
openshell_endpoint,
sandbox_id,
) {
(Some(prepared), Some(config_apply_tx), _, _) => {
let (task, mut accepted) = crate::supervisor_session::spawn_prepared(
prepared,
prepared_bootstrap_result,
ssh_socket_path,
port_forward,
None,
terminating.clone(),
config_apply_tx,
supervisor_session_updates,
);
let accepted_result =
tokio::time::timeout(Duration::from_secs(10), accepted.wait_for(|ready| *ready))
.await
.map(|result| result.map(|_| ()));
match accepted_result {
Ok(Ok(())) => (Some(task), Some(accepted)),
Ok(Err(_)) => {
task.abort();
return Err(miette::miette!(
"prepared supervisor session ended before bootstrap acknowledgement"
));
}
Err(_) => {
task.abort();
return Err(miette::miette!(
"prepared supervisor session did not become ready within 10 seconds"
));
let (session_task, session_readiness) = match prestarted_supervisor_session.as_mut() {
Some(prestarted) => (prestarted.task.take(), Some(prestarted.readiness.clone())),
None => match (openshell_endpoint, sandbox_id) {
(Some(endpoint), Some(id)) => {
let (task, mut accepted) = crate::supervisor_session::spawn_with_readiness(
endpoint.to_string(),
id.to_string(),
ssh_socket_path,
port_forward,
None,
terminating.clone(),
crate::supervisor_session::SessionRuntimeContext {
instance_id: instance_id.clone(),
session_id_updates: supervisor_session_updates,
config_apply_tx,
},
);
let accepted_result = tokio::time::timeout(
Duration::from_secs(10),
accepted.wait_for(|ready| *ready),
)
.await
.map(|result| result.map(|_| ()));
match accepted_result {
Ok(Ok(())) => (Some(task), Some(accepted)),
Ok(Err(_)) => {
task.abort();
return Err(miette::miette!(
"supervisor session ended before gateway acceptance"
));
}
Err(_) => {
task.abort();
return Err(miette::miette!(
"gateway did not accept supervisor session within 10 seconds"
));
}
}
}
}
(None, _, Some(endpoint), Some(id)) => {
let (task, mut accepted) = crate::supervisor_session::spawn_with_readiness(
endpoint.to_string(),
id.to_string(),
ssh_socket_path,
port_forward,
None,
terminating.clone(),
crate::supervisor_session::SessionRuntimeContext {
instance_id: instance_id.clone(),
session_id_updates: supervisor_session_updates,
config_apply_tx,
},
);
let accepted_result =
tokio::time::timeout(Duration::from_secs(10), accepted.wait_for(|ready| *ready))
.await
.map(|result| result.map(|_| ()));
match accepted_result {
Ok(Ok(())) => (Some(task), Some(accepted)),
Ok(Err(_)) => {
task.abort();
return Err(miette::miette!(
"supervisor session ended before gateway acceptance"
));
}
Err(_) => {
task.abort();
return Err(miette::miette!(
"gateway did not accept supervisor session within 10 seconds"
));
}
}
}
_ => (None, None),
_ => (None, None),
},
};
Ok(BoundaryAccess {
crates/openshell-supervisor-process/src/supervisor_session.rs
+15 -2
View File
@@ -47,7 +47,7 @@ const MAX_BACKOFF: Duration = Duration::from_secs(30);
const CONFIG_APPLY_TIMEOUT: Duration = Duration::from_mins(1);
// The gateway may hold the initial stream open while an operator repairs an
// invalid image policy. Keep this above the five-minute provisioning window.
const SESSION_PREPARE_TIMEOUT: Duration = Duration::from_mins(6);
pub const SESSION_PREPARE_TIMEOUT: Duration = Duration::from_mins(6);
type StartupPolicyPreparer =
Box<dyn FnOnce(SandboxPolicy) -> Result<Option<SandboxPolicy>, String> + Send>;
@@ -734,7 +734,9 @@ async fn run_prepared_session(
.map_err(|_| "failed to queue configuration bootstrap result")?;
}
let config_sequences = Arc::new(Mutex::new(ConfigSequenceWatermarks::default()));
config.ready_tx.send_replace(true);
if prepared.protocol_revision != SUPERVISOR_PROTOCOL_REVISION {
config.ready_tx.send_replace(true);
}
// Main loop: receive gateway messages + send heartbeats.
let mut heartbeat_interval =
@@ -762,6 +764,7 @@ async fn run_prepared_session(
terminating: &config.terminating,
config_apply_tx: config.config_apply_tx.as_ref(),
config_sequences: &config_sequences,
ready_tx: &config.ready_tx,
};
handle_gateway_message(
&msg,
@@ -824,6 +827,7 @@ async fn apply_bootstrap(
)
})
.collect(),
admission: None,
};
};
let (response, receiver) = tokio::sync::oneshot::channel();
@@ -837,12 +841,14 @@ async fn apply_bootstrap(
{
return ConfigBootstrapResult {
results: Vec::new(),
admission: None,
};
}
match tokio::time::timeout(CONFIG_APPLY_TIMEOUT, receiver).await {
Ok(Ok(result)) => result,
_ => ConfigBootstrapResult {
results: Vec::new(),
admission: None,
},
}
}
@@ -930,6 +936,7 @@ struct GatewayMessageContext<'a> {
terminating: &'a Arc<AtomicBool>,
config_apply_tx: Option<&'a mpsc::Sender<ConfigApplyRequest>>,
config_sequences: &'a Arc<Mutex<ConfigSequenceWatermarks>>,
ready_tx: &'a watch::Sender<bool>,
}
fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext<'_>) {
@@ -937,6 +944,11 @@ fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext<
Some(gateway_message::Payload::Heartbeat(_)) => {
// Gateway heartbeat — nothing to do.
}
Some(gateway_message::Payload::ConfigurationAdmission(admission)) => {
let accepted = admission.state
== i32::from(openshell_core::proto::ConfigurationAdmissionState::Accepted);
context.ready_tx.send_replace(accepted);
}
Some(gateway_message::Payload::ConfigUpdate(update)) => {
let update = update.clone();
let tx = context.tx.clone();
@@ -971,6 +983,7 @@ fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext<
result: Some(failed_component_result(
component, revision, outcome, code, message,
)),
admission: None,
};
let result = if invalid_update {
fallback(
crates/openshell-supervisor/src/lib.rs
+162 -45
View File
@@ -809,6 +809,7 @@ pub async fn run_sandbox(
let mut prepared_bootstrap_result = stream_bootstrap.as_ref().map(|bootstrap| {
use openshell_core::proto::{ConfigApplyOutcome, ConfigBootstrapResult, ConfigComponent};
let mut results = Vec::with_capacity(2);
let mut admission = None;
if let Some(snapshot) = bootstrap.provider_environment.as_ref() {
let revision = provider_config_revision(snapshot.provider_env_revision);
let outcome = if provider_bootstrap_degraded {
@@ -827,22 +828,31 @@ pub async fn run_sandbox(
if let Some(snapshot) = bootstrap.sandbox_config.as_ref() {
let settings: openshell_core::grpc_client::SettingsPollResult = snapshot.clone().into();
let revision = sandbox_config_revision(&settings);
let outcome = if loaded_policy_origin.allows_gateway_policy_reload() {
let outcome = if !snapshot.configuration_admitted {
ConfigApplyOutcome::FailedClosed
} else if loaded_policy_origin.allows_gateway_policy_reload() {
ConfigApplyOutcome::Applied
} else {
ConfigApplyOutcome::RetainedLocalOverride
};
let applied_revision =
(outcome != ConfigApplyOutcome::RetainedLocalOverride).then_some(revision.clone());
matches!(outcome, ConfigApplyOutcome::Applied).then_some(revision.clone());
results.push(config_apply_result(
ConfigComponent::SandboxConfig,
revision,
applied_revision,
outcome,
None,
(!snapshot.configuration_admitted).then(|| {
(
"configuration_rejected",
snapshot.configuration_error.clone(),
true,
)
}),
));
admission = Some(streamed_configuration_admission(snapshot, true));
}
ConfigBootstrapResult { results }
ConfigBootstrapResult { results, admission }
});
if credential_gating_unavailable(
@@ -1131,8 +1141,12 @@ pub async fn run_sandbox(
let poll_pid = entrypoint_pid.clone();
let poll_provider_credentials = provider_credentials.clone();
let poll_policy_local = networking.as_ref().map(|n| n.policy_local_ctx.clone());
let poll_endpoint_policy = loaded_policy_origin
.allows_gateway_policy_reload()
let initial_stream_configuration_admitted = stream_bootstrap
.as_ref()
.and_then(|bootstrap| bootstrap.sandbox_config.as_ref())
.is_none_or(|snapshot| snapshot.configuration_admitted);
let poll_endpoint_policy = (loaded_policy_origin.allows_gateway_policy_reload()
&& initial_stream_configuration_admitted)
.then(|| retained_proto.clone())
.flatten();
let poll_interval_secs: u64 = std::env::var("OPENSHELL_POLICY_POLL_INTERVAL_SECS")
@@ -1172,9 +1186,10 @@ pub async fn run_sandbox(
supervisor_session_id: supervisor_session_id.clone(),
};
let wait_for_workload = !uses_stream_configuration;
tokio::spawn(async move {
let mut workload_started = workload_started_rx;
if workload_started.wait_for(|started| *started).await.is_err() {
if wait_for_workload && workload_started.wait_for(|started| *started).await.is_err() {
return;
}
if let Err(e) = run_policy_poll_loop(poll_ctx).await {
@@ -1203,6 +1218,24 @@ pub async fn run_sandbox(
};
tokio::pin!(proxy_exited);
// Report the stream-delivered bootstrap result while the boundary remains
// confirmed but held. Docker and other admission-aware drivers do not
// release the workload until the gateway accepts this result, so delaying
// the stream until after `start_agent` would deadlock activation.
let prestarted_supervisor_session = match prepared_supervisor_session.take() {
Some(prepared) => Some(
openshell_supervisor_process::delegated::start_prepared_supervisor_session(
prepared,
prepared_bootstrap_result.take(),
ssh_socket_path.as_deref(),
config_apply_tx.clone(),
Some(supervisor_session_updates.clone()),
)
.await?,
),
None => None,
};
let (confirmed, backend_name) = remote_ready;
let exit_code = {
let running = confirmed
@@ -1226,8 +1259,7 @@ pub async fn run_sandbox(
running.loopback_connector(),
agent.clone(),
Some(supervisor_session_updates),
prepared_supervisor_session,
prepared_bootstrap_result.take(),
prestarted_supervisor_session,
Some(config_apply_tx),
)
.await?;
@@ -2499,17 +2531,32 @@ async fn load_policy(
}
}
let configuration_rejected = stream_bootstrap && !snapshot.configuration_admitted;
let mut loaded_policy_revision =
policy_bound_to_snapshot.then(|| LoadedPolicyRevision::from_snapshot(&snapshot));
if configuration_rejected {
ocsf_emit!(
ConfigStateChangeBuilder::new(ocsf_ctx())
.severity(SeverityId::High)
.status(StatusId::Failure)
.state(StateId::Other, "configuration_error")
.message(snapshot.configuration_error.clone())
.build()
);
proto_policy = openshell_policy::restrictive_default_policy();
}
// Build OPA engine from baked-in rules + typed proto data.
// In cluster mode, proxy networking is always enabled so OPA is
// always required for allow/deny decisions.
// The initial load uses pid=0 (no symlink resolution) because the
// container hasn't started yet. After the entrypoint spawns, the
// engine is rebuilt with the real PID for symlink resolution.
info!("Creating OPA engine from proto policy data");
let mut has_last_valid_policy = true;
if !configuration_rejected {
info!("Creating OPA engine from proto policy data");
}
let mut has_last_valid_policy = !configuration_rejected;
let engine = match OpaEngine::from_proto(&proto_policy) {
Ok(engine) => Arc::new(engine),
Err(e) => {
@@ -2550,13 +2597,18 @@ async fn load_policy(
install_builtin_middleware_registry(&engine).await?;
let middleware_services = snapshot.supervisor_middleware_services.clone();
let middleware_services = if configuration_rejected {
Vec::new()
} else {
snapshot.supervisor_middleware_services.clone()
};
let middleware_registry_status = if middleware_services.is_empty() {
MiddlewareRegistryStatus::Synchronized
} else if let Err(error) = grpc_retry("Middleware connect", || {
let middleware_services = middleware_services.clone();
let extension_credentials = extension_credentials.clone();
let extension_authentication_enabled = snapshot.extension_authentication_enabled;
let extension_authentication_enabled =
snapshot.extension_authentication_enabled && !configuration_rejected;
async move {
let credentials = if extension_authentication_enabled {
openshell_core::grpc_client::CachedOpenShellClient::connect_with_credentials(
@@ -2624,8 +2676,8 @@ async fn load_policy(
revision: loaded_policy_revision,
has_last_valid_policy,
},
agent_proposals_enabled_from_settings(&snapshot.settings),
snapshot.extension_authentication_enabled,
!configuration_rejected && agent_proposals_enabled_from_settings(&snapshot.settings),
snapshot.extension_authentication_enabled && !configuration_rejected,
));
}
@@ -3950,6 +4002,48 @@ fn config_apply_result(
}
}
fn config_apply_result_activates(
result: &openshell_core::proto::ConfigComponentApplyResult,
) -> bool {
use openshell_core::proto::ConfigApplyOutcome;
matches!(
ConfigApplyOutcome::try_from(result.outcome),
Ok(ConfigApplyOutcome::Applied
| ConfigApplyOutcome::IgnoredDuplicate
| ConfigApplyOutcome::RetainedLocalOverride
| ConfigApplyOutcome::Degraded)
)
}
fn streamed_configuration_admission(
snapshot: &openshell_core::proto::SandboxConfigSnapshot,
runtime_applied: bool,
) -> openshell_core::proto::SandboxConfigurationAdmission {
use openshell_core::proto::{ConfigurationAdmissionState, SandboxConfigurationAdmission};
let accepted = snapshot.configuration_admitted && runtime_applied;
SandboxConfigurationAdmission {
instance_id: snapshot.configuration_instance_id.clone(),
state: if accepted {
ConfigurationAdmissionState::Accepted.into()
} else {
ConfigurationAdmissionState::Rejected.into()
},
policy_version: snapshot.version,
policy_hash: snapshot.policy_hash.clone(),
config_revision: snapshot.config_revision,
provider_env_revision: snapshot.provider_env_revision,
error: if accepted {
String::new()
} else if snapshot.configuration_error.is_empty() {
"effective configuration could not be activated".to_string()
} else {
snapshot.configuration_error.clone()
},
}
}
#[allow(clippy::too_many_arguments)]
async fn apply_stream_config_request<C: PolicyGatewayClient>(
ctx: &PolicyPollLoopContext,
@@ -3980,6 +4074,7 @@ async fn apply_stream_config_request<C: PolicyGatewayClient>(
response,
} => {
let mut results = Vec::with_capacity(2);
let mut admission = None;
if let Some(snapshot) = bootstrap.provider_environment {
let result =
apply_stream_provider_snapshot(ctx, snapshot, current_provider_env_revision);
@@ -3999,32 +4094,38 @@ async fn apply_stream_config_request<C: PolicyGatewayClient>(
results.push(result);
}
if let Some(snapshot) = bootstrap.sandbox_config {
results.push(
apply_stream_sandbox_snapshot(
ctx,
client,
snapshot.into(),
current_config_revision,
current_stream_sandbox_revision,
current_policy_version,
current_policy_hash,
current_endpoint_policy,
current_middleware_services,
current_extension_authentication_enabled,
middleware_registry_status,
current_settings,
reloads_gateway_policy,
has_last_valid_policy,
)
.await,
);
let expected_admission = snapshot.clone();
let result = apply_stream_sandbox_snapshot(
ctx,
client,
snapshot.into(),
current_config_revision,
current_stream_sandbox_revision,
current_policy_version,
current_policy_hash,
current_endpoint_policy,
current_middleware_services,
current_extension_authentication_enabled,
middleware_registry_status,
current_settings,
reloads_gateway_policy,
has_last_valid_policy,
)
.await;
admission = Some(streamed_configuration_admission(
&expected_admission,
config_apply_result_activates(&result),
));
results.push(result);
}
let _ = response.send(ConfigBootstrapResult { results });
let _ = response.send(ConfigBootstrapResult { results, admission });
}
ConfigApplyRequest::Update { update, response } => {
let mut admission = None;
let result = match update.component {
Some(config_update::Component::SandboxConfig(snapshot)) => {
apply_stream_sandbox_snapshot(
let expected_admission = snapshot.clone();
let result = apply_stream_sandbox_snapshot(
ctx,
client,
snapshot.into(),
@@ -4040,7 +4141,12 @@ async fn apply_stream_config_request<C: PolicyGatewayClient>(
reloads_gateway_policy,
has_last_valid_policy,
)
.await
.await;
admission = Some(streamed_configuration_admission(
&expected_admission,
config_apply_result_activates(&result),
));
result
}
Some(config_update::Component::ProviderEnvironment(snapshot)) => {
let result = apply_stream_provider_snapshot(
@@ -4079,6 +4185,7 @@ async fn apply_stream_config_request<C: PolicyGatewayClient>(
update_id: update.update_id,
component_sequence: update.component_sequence,
result: Some(result),
admission,
});
}
}
@@ -4443,24 +4550,29 @@ async fn run_policy_poll_loop_with_client<C: PolicyGatewayClient>(
.map_or(0, |snapshot| snapshot.config_revision);
let mut current_stream_sandbox_revision = initial_stream_snapshot
.as_ref()
.filter(|snapshot| snapshot.configuration_admitted)
.map(sandbox_config_revision)
.filter(|_| ctx.loaded_policy_origin.allows_gateway_policy_reload());
let mut current_provider_env_revision: u64 = ctx.provider_credentials.snapshot().revision;
let mut current_policy_version: u32 = initial_stream_snapshot
.as_ref()
.filter(|snapshot| snapshot.configuration_admitted)
.map_or(0, |snapshot| snapshot.version);
let mut current_policy_hash = initial_stream_snapshot
.as_ref()
.filter(|snapshot| snapshot.configuration_admitted)
.map_or_else(String::new, |snapshot| snapshot.policy_hash.clone());
let mut current_policy_generation = None;
let mut current_endpoint_policy = ctx.endpoint_policy.clone();
let mut current_middleware_services = initial_stream_snapshot
.as_ref()
.filter(|snapshot| snapshot.configuration_admitted)
.map_or_else(Vec::new, |snapshot| {
snapshot.supervisor_middleware_services.clone()
});
let mut current_extension_authentication_enabled = initial_stream_snapshot
.as_ref()
.filter(|snapshot| snapshot.configuration_admitted)
.map_or(ctx.extension_authentication_enabled, |snapshot| {
snapshot.extension_authentication_enabled
});
@@ -4470,6 +4582,7 @@ async fn run_policy_poll_loop_with_client<C: PolicyGatewayClient>(
openshell_core::proto::EffectiveSetting,
> = initial_stream_snapshot
.as_ref()
.filter(|snapshot| snapshot.configuration_admitted)
.map_or_else(std::collections::HashMap::new, |snapshot| {
snapshot.settings.clone()
});
@@ -4494,7 +4607,9 @@ async fn run_policy_poll_loop_with_client<C: PolicyGatewayClient>(
// reconciled below instead of being recorded as already applied.
if let Some(snapshot) = initial_stream_snapshot.as_ref() {
let _ = ctx.workspace_tx.send(snapshot.workspace.clone());
if let Some(generation) = initial_generation {
if snapshot.configuration_admitted
&& let Some(generation) = initial_generation
{
ctx.provider_readiness.policy_activated(
&EnvironmentIdentity::from_settings(snapshot),
snapshot.config_revision,
@@ -4503,13 +4618,15 @@ async fn run_policy_poll_loop_with_client<C: PolicyGatewayClient>(
current_policy_generation = Some(generation);
}
apply_ocsf_json_setting(&ctx.ocsf_enabled, &snapshot.settings);
endpoint_status::reset(
ctx.endpoint_observation_tx.as_ref(),
current_endpoint_policy.as_ref(),
&current_policy_hash,
ctx.provider_credentials.snapshot().revision,
)
.await;
if snapshot.configuration_admitted {
endpoint_status::reset(
ctx.endpoint_observation_tx.as_ref(),
current_endpoint_policy.as_ref(),
&current_policy_hash,
ctx.provider_credentials.snapshot().revision,
)
.await;
}
} else {
match client.poll_settings(&ctx.sandbox).await {
Ok(result) => {
proto/openshell.proto
+8
View File
@@ -2875,6 +2875,9 @@ message GatewayMessage {
RelayClose relay_close = 5;
ConfigUpdate config_update = 6;
StartupConfigCandidate startup_config_candidate = 7;
// Durable gateway acknowledgement of streamed configuration admission.
// The supervisor must not launch the workload until this is ACCEPTED.
SandboxConfigurationAdmission configuration_admission = 8;
}
}
@@ -2994,11 +2997,16 @@ message ConfigUpdateResult {
// Echoes ConfigUpdate.component_sequence.
uint64 component_sequence = 2;
ConfigComponentApplyResult result = 3;
// Admission state for a sandbox-configuration update. Omitted for provider
// environment updates and compatibility supervisors.
SandboxConfigurationAdmission admission = 4;
}
// Aggregate application result for the SessionAccepted bootstrap.
message ConfigBootstrapResult {
repeated ConfigComponentApplyResult results = 1;
// Admission state of the complete gateway-authored bootstrap generation.
SandboxConfigurationAdmission admission = 2;
}
// Gateway rejects the supervisor session.