From 0edf3237eb9af290bc3bc5bc2226a0753945706e Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Tue, 6 Oct 2026 14:46:05 -0700 Subject: [PATCH] fix(config): finish untracked operations on every polling path - Treat a missing or untracked operation as commit-only in the UpdateConfig wait wrapper, so waited updates succeed in poll mode and clients fall back to policy status. - Finish pending operations from the reconciler when push delivery is off or the local session polls, while still publishing to snapshot-only sessions. Also check on session acceptance and on the owner replica for peer hints, which covers supervisors never registered with the scheduler. - Finish operations from a build only while its own session is current, so a replacement session that applies configuration completes them. - Return the committed result from the TypeScript SDK when a waited update has no operation, and drop a duplicated delete_if span. Signed-off-by: Piotr Mlocek --- .../src/config_delivery/mod.rs | 21 +++++- .../src/config_update_operation.rs | 65 ++++++++++++++++ crates/openshell-server/src/grpc/mod.rs | 19 +++-- crates/openshell-server/src/grpc/policy.rs | 75 +++++++++++++++++++ .../openshell-server/src/persistence/mod.rs | 17 ----- .../src/supervisor_session.rs | 35 +++++++-- sdk/typescript/src/client.test.ts | 16 ++++ sdk/typescript/src/client.ts | 9 ++- 8 files changed, 221 insertions(+), 36 deletions(-) diff --git a/crates/openshell-server/src/config_delivery/mod.rs b/crates/openshell-server/src/config_delivery/mod.rs index a738edea6..b763ec325 100644 --- a/crates/openshell-server/src/config_delivery/mod.rs +++ b/crates/openshell-server/src/config_delivery/mod.rs @@ -592,11 +592,14 @@ async fn run_build(state: &Arc, ticket: &BuildTicket) -> BuildResul .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. + // pending operations bound to this snapshot. A replaced session + // says nothing about its successor, which may apply. if require_acknowledgement - && !state + && disposition != DeliveryDisposition::NoActiveSession + && state .supervisor_sessions .session_applies_config(&key.sandbox_id, session_id) + == Some(false) && let Err(error) = crate::config_update_operation::finish_pending_untracked(state, &key.sandbox_id) .await @@ -900,6 +903,20 @@ pub fn handle_peer_config_update_hint( scheduler.publish_sandbox(&target.sandbox_id, components, Instant::now()) }); kick(state); + // A session without snapshot support is never registered, so + // no build reaches the operations a peer is reconciling. + let state = Arc::clone(state); + let sandbox_id = target.sandbox_id; + tokio::spawn(async move { + if let Err(error) = crate::config_update_operation::finish_pending_if_polling( + &state, + &sandbox_id, + ) + .await + { + warn!(sandbox_id, error = %error, "failed to finish configuration operations for a polling supervisor"); + } + }); } else { response.stale_owner = true; } diff --git a/crates/openshell-server/src/config_update_operation.rs b/crates/openshell-server/src/config_update_operation.rs index 4fdf3c173..f6d9cac50 100644 --- a/crates/openshell-server/src/config_update_operation.rs +++ b/crates/openshell-server/src/config_update_operation.rs @@ -489,6 +489,28 @@ pub async fn finish_untracked(state: &ServerState, operation_id: &str) -> Result .await } +/// True for an operation finished because completion was not tracked. +pub fn is_untracked(operation: &ConfigUpdateOperation) -> bool { + operation.state == i32::from(ConfigUpdateOperationState::Inactive) + && operation.sanitized_error == UNTRACKED_COMPLETION +} + +/// Finish pending operations when this replica's current session for the +/// sandbox keeps polling. Sessions on other replicas are handled there. +pub async fn finish_pending_if_polling( + state: &ServerState, + sandbox_id: &str, +) -> Result<(), Status> { + if state + .supervisor_sessions + .current_session_applies_config(sandbox_id) + == Some(false) + { + finish_pending_untracked(state, sandbox_id).await?; + } + Ok(()) +} + /// 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 @@ -668,6 +690,34 @@ async fn reconcile_records_for_sandbox( _ => {} } + // Without push delivery, or with a local supervisor that keeps polling, + // no apply result can complete these operations. This also drains records + // left pending by a rollback from push to poll mode. + if !crate::config_delivery::push_enabled(state) + || state + .supervisor_sessions + .current_session_applies_config(&sandbox_id) + == Some(false) + { + let components = if records + .iter() + .any(|record| operation_dimension(record) == OperationDimension::Policy) + { + crate::config_delivery::ConfigComponents::SANDBOX_AND_PROVIDER + } else { + crate::config_delivery::ConfigComponents::SANDBOX_CONFIG + }; + for record in records { + if let Some(operation) = record.operation.as_ref() { + finish_untracked(state, &operation.operation_id).await?; + } + } + // A snapshot-only session still receives the change. Without push + // delivery this publishes nothing. + crate::config_delivery::publish_sandbox_components(state, &sandbox_id, components); + return Ok(()); + } + let now = current_time_ms(); let mut claimed_records = Vec::new(); for record in records { @@ -1033,6 +1083,21 @@ mod tests { ) } + #[tokio::test] + async fn reconcile_drains_pending_operations_without_push_delivery() { + // The fixture state polls, as after a rollback from push mode. + let (state, record) = pending_test_operation().await; + let operation_id = record.operation.as_ref().unwrap().operation_id.clone(); + reconcile_one(&state, &operation_id).await.unwrap(); + let operation = get_record(&state, &operation_id) + .await + .unwrap() + .unwrap() + .operation + .unwrap(); + assert!(is_untracked(&operation)); + } + #[test] fn authoritative_phase_classification_is_explicit() { assert_eq!( diff --git a/crates/openshell-server/src/grpc/mod.rs b/crates/openshell-server/src/grpc/mod.rs index 864904717..c8bb4a8dd 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -643,18 +643,21 @@ impl OpenShell for OpenShellService { } let timeout = policy::config_wait_timeout(request.get_ref().wait_timeout.as_ref())?; let mut response = mutation_replay::run(&self.state, request).await?; - if wait { - let id = response + // A response without an operation means completion is not tracked, + // for example because the gateway keeps configuration polling. Clients + // then confirm application through policy status. + if wait + && let Some(id) = response .get_ref() .operation .as_ref() - .ok_or_else(|| Status::internal("gateway omitted completion operation"))? - .operation_id - .clone(); - response.get_mut().operation = Some( + .map(|operation| operation.operation_id.clone()) + { + let operation = crate::config_update_operation::wait_for_terminal(&self.state, &id, timeout) - .await?, - ); + .await?; + response.get_mut().operation = + (!crate::config_update_operation::is_untracked(&operation)).then_some(operation); } Ok(response) } diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 9986575f8..2c0c041c8 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -13343,6 +13343,81 @@ mod tests { ); } + fn waited_setting_update(sandbox: &str) -> UpdateConfigRequest { + UpdateConfigRequest { + sandbox: sandbox.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() + } + } + + #[tokio::test] + async fn poll_mode_waited_update_succeeds_through_the_service() { + use openshell_core::proto::open_shell_server::OpenShell as _; + + let state = test_server_state().await; + let sandbox = test_sandbox( + "sb-poll-service", + "poll-service", + ProtoSandboxPolicy::default(), + Vec::new(), + ); + state.store.put_message(&sandbox).await.unwrap(); + let response = crate::grpc::OpenShellService::new(Arc::clone(&state)) + .update_config(with_user(Request::new(waited_setting_update( + "poll-service", + )))) + .await + .expect("a waited update succeeds when completion is not tracked") + .into_inner(); + assert!(response.operation.is_none()); + } + + #[tokio::test] + async fn push_mode_waited_update_finishes_for_a_legacy_supervisor() { + use openshell_core::proto::open_shell_server::OpenShell as _; + + let mut state = test_server_state().await; + crate::grpc::test_support::enable_push_delivery(&mut state); + let sandbox = test_sandbox( + "sb-legacy-operation", + "legacy-operation", + ProtoSandboxPolicy::default(), + Vec::new(), + ); + state.store.put_message(&sandbox).await.unwrap(); + // A supervisor without snapshot support is never registered with the + // delivery scheduler. + let (tx, _rx) = tokio::sync::mpsc::channel(1); + state.supervisor_sessions.register( + "sb-legacy-operation".into(), + "session-1".into(), + tx, + tokio::sync::oneshot::channel().0, + ); + + let response = tokio::time::timeout( + std::time::Duration::from_secs(3), + crate::grpc::OpenShellService::new(Arc::clone(&state)).update_config(with_user( + Request::new(waited_setting_update("legacy-operation")), + )), + ) + .await + .expect("an untracked operation must not wait for its timeout") + .unwrap() + .into_inner(); + assert!(response.operation.is_none()); + } + #[tokio::test] async fn push_mode_operation_finishes_for_a_polling_supervisor() { let mut state = test_server_state().await; diff --git a/crates/openshell-server/src/persistence/mod.rs b/crates/openshell-server/src/persistence/mod.rs index 820baeaf3..7882b3727 100644 --- a/crates/openshell-server/src/persistence/mod.rs +++ b/crates/openshell-server/src/persistence/mod.rs @@ -579,23 +579,6 @@ impl Store { store_dispatch!(self.count_pending_config_update_operations()) } - /// Delete an object by id with compare-and-swap support. - /// - /// # Arguments - /// * `object_type` - Type discriminator for the object - /// * `id` - Stable object identifier - /// * `expected_resource_version` - Required resource version for the delete to proceed - /// - /// # Returns - /// * `Ok(true)` - Object was deleted - /// * `Ok(false)` - Object not found - /// * `Err(Conflict)` - Resource version mismatch - #[tracing::instrument( - name = "store", - skip_all, - fields(otel.name = "store.delete_if", otel.status_code = tracing::field::Empty, object_type = %object_type, object.id = %id) - )] - /// Delete an object by id with compare-and-swap support. /// /// # Arguments diff --git a/crates/openshell-server/src/supervisor_session.rs b/crates/openshell-server/src/supervisor_session.rs index f5890219a..b4987dc8b 100644 --- a/crates/openshell-server/src/supervisor_session.rs +++ b/crates/openshell-server/src/supervisor_session.rs @@ -1141,16 +1141,25 @@ 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 { + /// Whether this exact session applies and acknowledges streamed + /// configuration, or `None` when it is no longer the current session. + pub fn session_applies_config(&self, sandbox_id: &str, session_id: &str) -> Option { self.sessions .lock() .unwrap() .get(sandbox_id) - .is_some_and(|session| { - session.session_id == session_id && session.config_sequences.acknowledged - }) + .filter(|session| session.session_id == session_id) + .map(|session| session.config_sequences.acknowledged) + } + + /// Whether the current local session applies streamed configuration, or + /// `None` when this replica has no session for the sandbox. + pub fn current_session_applies_config(&self, sandbox_id: &str) -> Option { + self.sessions + .lock() + .unwrap() + .get(sandbox_id) + .map(|session| session.config_sequences.acknowledged) } pub fn is_current_session(&self, sandbox_id: &str, session_id: &str) -> bool { @@ -2815,6 +2824,20 @@ async fn accept_supervisor_session(setup: SessionSetup) -> Result<(), Status> { .replay_pending_relays(&sandbox_id, &session_id, &session_tx) .await; + // Operations committed before a polling supervisor connected can never + // receive its apply result. + if crate::config_delivery::push_enabled(&state) && mode != SessionMode::ConfigApply { + let state = Arc::clone(&state); + let sandbox_id = sandbox_id.clone(); + tokio::spawn(async move { + if let Err(error) = + crate::config_update_operation::finish_pending_if_polling(&state, &sandbox_id).await + { + warn!(sandbox_id, error = %error, "failed to finish configuration operations for a polling supervisor"); + } + }); + } + if let Some(setup) = config_push { crate::config_delivery::register_session( &state, diff --git a/sdk/typescript/src/client.test.ts b/sdk/typescript/src/client.test.ts index 46d5ef500..0416568dd 100644 --- a/sdk/typescript/src/client.test.ts +++ b/sdk/typescript/src/client.test.ts @@ -1575,6 +1575,22 @@ describe('config / policy', () => { expect(result.operationError).toBe('optional component unavailable'); }); + it('setPolicy returns the committed result when completion is not tracked', async () => { + const sandbox = client({ + getSandbox: () => readySandbox('sb', 'sb-id'), + updateConfig: () => ({ + version: 5, + policyHash: 'target', + settingsRevision: 10n, + deleted: false, + }), + }); + + const result = await sandbox.setPolicy('sb', { version: 1, networkPolicies: {} }, { wait: true }); + expect(result.version).toBe(5); + expect(result.operationId).toBeUndefined(); + }); + it('setPolicy rejects a terminal failed durable operation', async () => { const sandbox = client({ getSandbox: () => readySandbox('sb', 'sb-id'), diff --git a/sdk/typescript/src/client.ts b/sdk/typescript/src/client.ts index 7856c92f1..b01d48ff3 100644 --- a/sdk/typescript/src/client.ts +++ b/sdk/typescript/src/client.ts @@ -452,7 +452,11 @@ export interface SandboxConfig { export interface SetPolicyOptions extends SandboxWorkspaceOptions { /** Pin the sandbox resource version for optimistic concurrency (u64 as string). */ expectedResourceVersion?: string; - /** Ask the gateway to wait for a durable terminal apply result. */ + /** + * Ask the gateway to wait for a durable terminal apply result. Gateways that + * do not track completion, such as those that keep configuration polling, + * return the committed result without an operation. + */ wait?: boolean; /** Bound the server-side wait in seconds. Default 60. */ waitTimeoutSecs?: number; @@ -1767,8 +1771,7 @@ export class SandboxClient { waitTimeout: durationFromMs(Math.max(0, (options?.waitTimeoutSecs ?? 60) * 1000)), }); const result = updateConfigResult(resp); - if (options?.wait) { - if (!resp.operation) throw new SdkError('rpc', 'gateway omitted the requested apply operation'); + if (options?.wait && resp.operation) { if ( resp.operation.state === ConfigUpdateOperationState.FAILED || resp.operation.state === ConfigUpdateOperationState.SUPERSEDED ||