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 <pmlocek@nvidia.com>
This commit is contained in:
Piotr Mlocek
2026-10-06 14:46:05 -07:00
parent 866e39c2ef
commit 0edf3237eb
8 changed files with 221 additions and 36 deletions
@@ -592,11 +592,14 @@ async fn run_build(state: &Arc<ServerState>, 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;
}
@@ -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!(
+11 -8
View File
@@ -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)
}
@@ -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;
@@ -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
@@ -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<bool> {
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<bool> {
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,
+16
View File
@@ -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'),
+6 -3
View File
@@ -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 ||