diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index b437d859d..f1688ab9e 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -152,6 +152,77 @@ fn report_config_update_operation( } } +/// Wait for a policy revision by polling its load status. Gateways that do not +/// track completion, such as those that keep configuration polling, return no +/// completion operation for a waited update. +async fn wait_for_policy_load( + client: &mut crate::tls::GrpcClient, + name: &str, + workspace: &str, + version: u32, + timeout_secs: u64, +) -> Result { + let deadline = Instant::now() + Duration::from_secs(timeout_secs); + loop { + if Instant::now() > deadline { + eprintln!( + "{} Timeout waiting for policy version {} to load", + "✗".red().bold(), + version + ); + return Ok(POLICY_WAIT_TIMEOUT_EXIT_CODE); + } + + tokio::time::sleep(Duration::from_secs(1)).await; + + let inner = client + .get_sandbox_policy_status(GetSandboxPolicyStatusRequest { + sandbox: name.to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector( + workspace.to_string(), + )), + version, + global: false, + }) + .await + .into_diagnostic()? + .into_inner(); + let Some(rev) = &inner.revision else { + continue; + }; + match PolicyStatus::try_from(rev.status).unwrap_or(PolicyStatus::Unspecified) { + PolicyStatus::Loaded => { + eprintln!( + "{} Policy version {} loaded (active version: {})", + "✓".green().bold(), + rev.version, + inner.active_version + ); + return Ok(0); + } + PolicyStatus::Failed => { + eprintln!( + "{} Policy version {} failed to load: {}", + "✗".red().bold(), + rev.version, + rev.load_error + ); + return Ok(1); + } + PolicyStatus::Superseded => { + eprintln!( + "{} Policy version {} was superseded (active version: {})", + "⚠".yellow().bold(), + rev.version, + inner.active_version + ); + return Ok(0); + } + _ => {} + } + } +} + fn proto_timestamp_ms(timestamp: Option<&prost_types::Timestamp>) -> i64 { timestamp .and_then(|value| openshell_core::time::timestamp_to_millis(value).ok()) @@ -5562,6 +5633,10 @@ pub async fn sandbox_policy_set( return Ok(0); } + if resp.operation.is_none() { + return wait_for_policy_load(&mut client, name, workspace, resp.version, timeout_secs) + .await; + } report_config_update_operation(resp.operation.as_ref(), resp.version)?; Ok(0) } @@ -5692,6 +5767,10 @@ pub async fn sandbox_policy_update( return Ok(0); } + if response.operation.is_none() { + return wait_for_policy_load(&mut client, name, workspace, response.version, timeout_secs) + .await; + } report_config_update_operation(response.operation.as_ref(), response.version)?; Ok(0) } diff --git a/crates/openshell-server/src/config_delivery/mod.rs b/crates/openshell-server/src/config_delivery/mod.rs index 3075b1967..a738edea6 100644 --- a/crates/openshell-server/src/config_delivery/mod.rs +++ b/crates/openshell-server/src/config_delivery/mod.rs @@ -311,7 +311,7 @@ fn bootstrap_revisions_match(bootstrap: &ConfigBootstrap) -> bool { }) } -fn push_enabled(state: &ServerState) -> bool { +pub fn push_enabled(state: &ServerState) -> bool { state.config.config_delivery_mode == ConfigDeliveryMode::Push } @@ -583,9 +583,30 @@ async fn run_build(state: &Arc, ticket: &BuildTicket) -> BuildResul OwnerCheck::Current => { let disposition = state .supervisor_config_router() - .deliver(&key.sandbox_id, session_id, message, require_acknowledgement) + .deliver( + &key.sandbox_id, + session_id, + message, + require_acknowledgement, + ) .await; record_delivery(key.component, disposition.metric_label()); + // A session that keeps polling never reports a result for the + // pending operations bound to this snapshot. + if require_acknowledgement + && !state + .supervisor_sessions + .session_applies_config(&key.sandbox_id, session_id) + && let Err(error) = + crate::config_update_operation::finish_pending_untracked(state, &key.sandbox_id) + .await + { + warn!( + sandbox_id = %key.sandbox_id, + error = %error, + "failed to finish configuration operations for a polling supervisor" + ); + } false } OwnerCheck::Remote => { diff --git a/crates/openshell-server/src/config_update_operation.rs b/crates/openshell-server/src/config_update_operation.rs index c6d45337f..4fdf3c173 100644 --- a/crates/openshell-server/src/config_update_operation.rs +++ b/crates/openshell-server/src/config_update_operation.rs @@ -472,6 +472,38 @@ async fn finish( Ok(()) } +/// Completion of streamed configuration is not tracked here: the gateway keeps +/// configuration polling, or the sandbox's supervisor does not apply streamed +/// snapshots. +const UNTRACKED_COMPLETION: &str = "the sandbox supervisor polls for configuration"; + +/// Finish an operation whose completion the gateway cannot observe. +pub async fn finish_untracked(state: &ServerState, operation_id: &str) -> Result<(), Status> { + finish( + state, + operation_id, + ConfigUpdateOperationState::Inactive, + ConfigApplyOutcome::Unspecified, + UNTRACKED_COMPLETION, + ) + .await +} + +/// Finish every pending operation for a sandbox whose supervisor polls. +pub async fn finish_pending_untracked(state: &ServerState, sandbox_id: &str) -> Result<(), Status> { + let records = state + .store + .list_pending_config_operations_for_scope(sandbox_id) + .await + .map_err(|error| Status::internal(format!("list pending operations failed: {error}")))?; + for record in records { + if let Some(operation) = record.operation.as_ref() { + finish_untracked(state, &operation.operation_id).await?; + } + } + Ok(()) +} + async fn finish_if_target_matches( state: &ServerState, operation_id: &str, diff --git a/crates/openshell-server/src/grpc/mod.rs b/crates/openshell-server/src/grpc/mod.rs index 0ca3d6935..864904717 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -1020,6 +1020,14 @@ pub mod test_support { } } + /// Switch a test state to push delivery, which tracks durable completion. + pub fn enable_push_delivery(state: &mut Arc) { + Arc::get_mut(state) + .expect("test state is not shared yet") + .config + .config_delivery_mode = openshell_core::config::ConfigDeliveryMode::Push; + } + /// Configuration stream features advertised by a test supervisor hello. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum StreamFeatures { diff --git a/crates/openshell-server/src/grpc/mutation_replay/ordinary/tests.rs b/crates/openshell-server/src/grpc/mutation_replay/ordinary/tests.rs index 894b21781..300147b31 100644 --- a/crates/openshell-server/src/grpc/mutation_replay/ordinary/tests.rs +++ b/crates/openshell-server/src/grpc/mutation_replay/ordinary/tests.rs @@ -1163,7 +1163,8 @@ async fn config_completion_timeout_preserves_replayable_committed_operation() { ConfigUpdateConsistency, ConfigUpdateOperationState, Sandbox, SandboxPhase, SandboxSpec, SandboxStatus, SettingValue, UpdateConfigRequest, setting_value, }; - let (_directory, state) = protected_state().await; + let (_directory, mut state) = protected_state().await; + crate::grpc::test_support::enable_push_delivery(&mut state); state .store .put_message(&Sandbox { diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 3bdfd1840..9986575f8 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -153,6 +153,33 @@ async fn finish_config_update_operation( consistency: ConfigUpdateConsistency, timeout: std::time::Duration, ) -> Result, Status> { + // Boxed so the many call sites in the update handler stay small. + Box::pin(finish_config_update_operation_inner( + state, + operation_id, + consistency, + timeout, + )) + .await +} + +async fn finish_config_update_operation_inner( + state: &Arc, + operation_id: &str, + consistency: ConfigUpdateConsistency, + timeout: std::time::Duration, +) -> Result, Status> { + if !crate::config_delivery::push_enabled(state) { + // Polling supervisors report no apply results, so completion is not + // tracked. Respond as a commit-only update without an operation. + config_update_operation::finish_untracked(state, operation_id).await?; + let record = config_update_operation::get_record(state, operation_id) + .await? + .ok_or_else(|| Status::internal("committed update operation disappeared"))?; + let mut response = config_update_operation::response_from_record(&record)?; + response.operation = None; + return Ok(Response::new(response)); + } config_update_operation::reconcile_one(state, operation_id).await?; if consistency == ConfigUpdateConsistency::WaitForCompletion { config_update_operation::wait_for_terminal(state, operation_id, timeout).await?; @@ -13270,6 +13297,108 @@ mod tests { ); } + #[tokio::test] + async fn poll_mode_waited_update_returns_without_completion_operation() { + let state = test_server_state().await; + let sandbox = test_sandbox( + "sb-poll-wait", + "poll-wait", + ProtoSandboxPolicy::default(), + Vec::new(), + ); + state.store.put_message(&sandbox).await.unwrap(); + + // A polling supervisor reports no apply result, so the gateway answers + // a waited update as committed instead of waiting for completion. + let response = tokio::time::timeout( + std::time::Duration::from_secs(2), + handle_update_config( + &state, + with_user(Request::new(UpdateConfigRequest { + sandbox: sandbox.object_name().to_string(), + setting_key: "ocsf_json_enabled".to_string(), + setting_value: Some(SettingValue { + value: Some(setting_value::Value::BoolValue(true)), + }), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + consistency: ConfigUpdateConsistency::WaitForCompletion.into(), + idempotency_key: "poll-wait".to_string(), + ..Default::default() + })), + ), + ) + .await + .expect("poll mode must not wait for completion") + .unwrap() + .into_inner(); + assert!(response.operation.is_none()); + assert!( + state + .store + .list_pending_config_operations_for_scope(sandbox.object_id()) + .await + .unwrap() + .is_empty(), + "poll mode leaves no pending operation behind" + ); + } + + #[tokio::test] + async fn push_mode_operation_finishes_for_a_polling_supervisor() { + let mut state = test_server_state().await; + crate::grpc::test_support::enable_push_delivery(&mut state); + let sandbox = test_sandbox( + "sb-shadow-operation", + "shadow-operation", + ProtoSandboxPolicy::default(), + Vec::new(), + ); + state.store.put_message(&sandbox).await.unwrap(); + // A snapshot-only session keeps polling and never reports results. + let _session = + crate::config_delivery::register_test_push_session(&state, &sandbox, "session-1"); + crate::supervisor_owner::SupervisorOwnerIndex::new( + Arc::clone(&state.store), + crate::supervisor_owner::OWNER_TTL, + ) + .publish( + "sb-shadow-operation", + "session-1", + "test-supervisor", + 1, + &state.replica_id, + "local://test", + ) + .await + .unwrap(); + + let response = handle_update_config( + &state, + with_user(Request::new(UpdateConfigRequest { + sandbox: sandbox.object_name().to_string(), + setting_key: "ocsf_json_enabled".to_string(), + setting_value: Some(SettingValue { + value: Some(setting_value::Value::BoolValue(true)), + }), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + consistency: ConfigUpdateConsistency::WaitForCompletion.into(), + wait_timeout: openshell_core::time::duration_from_std( + std::time::Duration::from_secs(5), + ) + .ok(), + ..Default::default() + })), + ) + .await + .unwrap() + .into_inner(); + let operation = response.operation.expect("push mode tracks the operation"); + assert_eq!( + openshell_core::proto::ConfigUpdateOperationState::try_from(operation.state).unwrap(), + openshell_core::proto::ConfigUpdateOperationState::Inactive + ); + } + #[tokio::test] async fn committed_policy_update_publishes_complete_snapshot() { let mut state = test_server_state().await; @@ -23099,7 +23228,8 @@ mod tests { #[tokio::test] async fn update_config_same_policy_hash_with_new_provenance_creates_revision() { - let state = test_server_state().await; + let mut state = test_server_state().await; + crate::grpc::test_support::enable_push_delivery(&mut state); let policy = test_policy_with_rule("sandbox_only", "sandbox.example.com"); let hash = deterministic_policy_hash(&policy); let sandbox = test_sandbox("sb-same-hash", "same-hash", policy.clone(), Vec::new()); @@ -24599,7 +24729,8 @@ mod tests { #[tokio::test] async fn stopped_sandbox_setting_update_with_default_version_commits_with_annotations() { - let state = test_server_state().await; + let mut state = test_server_state().await; + crate::grpc::test_support::enable_push_delivery(&mut state); let mut sandbox = test_sandbox( "sb-inactive-operation", "inactive-operation", @@ -24704,7 +24835,8 @@ mod tests { #[tokio::test] async fn settings_operation_without_policy_history_applies_and_wakes_local_waiter() { - let state = test_server_state().await; + let mut state = test_server_state().await; + crate::grpc::test_support::enable_push_delivery(&mut state); let sandbox = test_sandbox( "sb-operation-wake", "operation-wake", @@ -24784,7 +24916,8 @@ mod tests { #[tokio::test] async fn operation_waiter_recovers_when_notification_is_missed() { - let state = test_server_state().await; + let mut state = test_server_state().await; + crate::grpc::test_support::enable_push_delivery(&mut state); let sandbox = test_sandbox( "sb-operation-poll", "operation-poll", diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index fba51e5bb..6a1588b28 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -230,7 +230,7 @@ mod tests { // SandboxProvisioning in both closures. Old rows decode false and empty; // decoding or deadline updates cannot claim an existing attempt. const PUBLIC_RPC_SCHEMA_SHA256: &str = - "f769d9effbb594fed3110470a580c623b3211b4b027ff64fdb3a64260d6d80f6"; + "ae08da01c8ce68c29a5b298555f99721de74a0325134ce01b1be8b6ef5f5516f"; const DURABLE_SCHEMA_SHA256: &str = "8fa5153f96896035f9ffc0b9ef753a74028c81738383fa8b7ab084bd7e978f97"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = @@ -715,7 +715,7 @@ mod tests { overlap_hash.as_str(), ), ( - (327, 28), + (329, 29), (94, 21), (81, 21), PUBLIC_RPC_SCHEMA_SHA256, diff --git a/crates/openshell-server/src/supervisor_session.rs b/crates/openshell-server/src/supervisor_session.rs index 36279ee10..f5890219a 100644 --- a/crates/openshell-server/src/supervisor_session.rs +++ b/crates/openshell-server/src/supervisor_session.rs @@ -1141,6 +1141,18 @@ impl SupervisorSessionRegistry { true } + /// True when this exact session applies and acknowledges streamed + /// configuration. + pub fn session_applies_config(&self, sandbox_id: &str, session_id: &str) -> bool { + self.sessions + .lock() + .unwrap() + .get(sandbox_id) + .is_some_and(|session| { + session.session_id == session_id && session.config_sequences.acknowledged + }) + } + pub fn is_current_session(&self, sandbox_id: &str, session_id: &str) -> bool { self.sessions .lock() @@ -6816,9 +6828,12 @@ mod tests { }; let message = SupervisorConfigMessage::SandboxConfig(Box::new(snapshot.clone())); assert_eq!( - state - .supervisor_sessions - .deliver_config(sandbox_id, "session-1", message.clone(), false), + state.supervisor_sessions.deliver_config( + sandbox_id, + "session-1", + message.clone(), + false + ), DeliveryDisposition::Queued ); let Some(gateway_message::Payload::ConfigUpdate(update)) = @@ -6911,9 +6926,12 @@ mod tests { }; let message = SupervisorConfigMessage::SandboxConfig(Box::new(snapshot.clone())); assert_eq!( - state - .supervisor_sessions - .deliver_config(sandbox_id, "session-1", message.clone(), false), + state.supervisor_sessions.deliver_config( + sandbox_id, + "session-1", + message.clone(), + false + ), DeliveryDisposition::Queued ); let Some(gateway_message::Payload::ConfigUpdate(update)) = diff --git a/crates/openshell-supervisor/src/lib.rs b/crates/openshell-supervisor/src/lib.rs index 97065ef9f..7bde126bf 100644 --- a/crates/openshell-supervisor/src/lib.rs +++ b/crates/openshell-supervisor/src/lib.rs @@ -11614,5 +11614,4 @@ network_policies: assert_eq!(duplicate, result); assert_eq!(ctx.opa_engine.current_generation(), rejected_generation); } - } diff --git a/sdk/typescript/src/client.test.ts b/sdk/typescript/src/client.test.ts index 6aa17fe15..46d5ef500 100644 --- a/sdk/typescript/src/client.test.ts +++ b/sdk/typescript/src/client.test.ts @@ -28,9 +28,9 @@ import { ConfigUpdateConsistency, ConfigUpdateOperationState, OpenShell, + ServiceAuthorizationMode as ProtoServiceAuthorizationMode, SandboxPhase, SandboxRestartPolicy, - ServiceAuthorizationMode as ProtoServiceAuthorizationMode, ServiceStatus, } from './gen/openshell_pb.js'; import { PolicySource, SettingScope } from './gen/sandbox_pb.js';