diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 97318f00b2..5fc8a59e80 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -2661,6 +2661,7 @@ async fn persist_existing_policy_projection( annotations: &HashMap, current_annotations: &HashMap, backfill_policy: Option<&ProtoSandboxPolicy>, + replace_startup_baseline: bool, ) -> Result, Status> { let annotations_match = annotations .iter() @@ -2678,7 +2679,7 @@ async fn persist_existing_policy_projection( crate::policy_store::permits_initial_static_policy_repair(sandbox); if let Some(policy) = backfill_policy.as_ref() && let Some(spec) = sandbox.spec.as_mut() - && (spec.policy.is_none() || startup_blocked) + && (spec.policy.is_none() || (startup_blocked && replace_startup_baseline)) { spec.policy = Some(policy.clone()); } @@ -4234,6 +4235,7 @@ async fn handle_update_config_inner( &req.annotations, &response_annotations, None, + false, ) .await? }; @@ -4382,6 +4384,7 @@ async fn handle_update_config_inner( &req.annotations, &response_annotations, backfill_policy.as_ref(), + startup_blocked && !sandbox_caller, ) .await?; if backfill_policy.is_some() { @@ -4411,6 +4414,7 @@ async fn handle_update_config_inner( expected_resource_version: req.expected_resource_version, annotations: req.annotations.clone(), backfill_policy: backfill_policy.clone(), + replace_startup_baseline: startup_blocked && !sandbox_caller, }; match state.store.put_policy_revision_atomic(&write).await { @@ -7333,6 +7337,7 @@ async fn apply_merge_operations_with_retry( expected_resource_version: context.expected_resource_version, annotations: context.annotations.clone(), backfill_policy: None, + replace_startup_baseline: false, }) .await .map(Some) diff --git a/crates/openshell-server/src/persistence/tests.rs b/crates/openshell-server/src/persistence/tests.rs index 7a3b435ce9..309eabaf9e 100644 --- a/crates/openshell-server/src/persistence/tests.rs +++ b/crates/openshell-server/src/persistence/tests.rs @@ -1379,6 +1379,7 @@ async fn policy_atomic_write_commits_revision_provenance_and_sandbox_projection( expected_resource_version: current_version, annotations: provenance.clone(), backfill_policy: Some(policy.clone()), + replace_startup_baseline: false, }) .await .unwrap(); @@ -1448,6 +1449,7 @@ async fn policy_atomic_write_rolls_back_sandbox_when_revision_insert_conflicts() expected_resource_version: before_version, annotations: StdHashMap::from([("signature".to_string(), "new".to_string())]), backfill_policy: Some(policy), + replace_startup_baseline: false, }) .await .unwrap_err(); @@ -1493,6 +1495,7 @@ async fn policy_atomic_write_persists_workspace() { expected_resource_version: current_version, annotations: StdHashMap::new(), backfill_policy: Some(policy), + replace_startup_baseline: false, }) .await .unwrap(); diff --git a/crates/openshell-server/src/policy_store.rs b/crates/openshell-server/src/policy_store.rs index 8ac1092802..4869e98246 100644 --- a/crates/openshell-server/src/policy_store.rs +++ b/crates/openshell-server/src/policy_store.rs @@ -37,6 +37,10 @@ pub struct AtomicPolicyRevisionWrite { /// Populate the create-time baseline, or replace it while startup admission /// is blocked and no workload has consumed the static restrictions. pub backfill_policy: Option, + /// Allow `backfill_policy` to replace an existing baseline while startup + /// admission is blocked. Only a management repair may do this; a + /// sandbox-originated write must conflict instead of overwriting it. + pub replace_startup_baseline: bool, } pub fn policy_record_for_atomic_write( @@ -108,7 +112,7 @@ pub fn project_policy_revision_onto_sandbox( changed = true; } Some(current) if current == backfill_policy => {} - Some(_) if startup_blocked => { + Some(_) if startup_blocked && write.replace_startup_baseline => { spec.policy = Some(backfill_policy.clone()); changed = true; } @@ -710,7 +714,7 @@ mod tests { .unwrap() .read_write .push("/new-static-path".to_string()); - let write = AtomicPolicyRevisionWrite { + let mut write = AtomicPolicyRevisionWrite { id: "revision".to_string(), sandbox_id: "sandbox".to_string(), workspace: "default".to_string(), @@ -721,6 +725,7 @@ mod tests { expected_resource_version: 0, annotations: HashMap::new(), backfill_policy: Some(replacement.clone()), + replace_startup_baseline: true, }; for activated in [None, Some(false), Some(true)] { for state in [Admission::Pending, Admission::Rejected, Admission::Accepted] { @@ -753,5 +758,27 @@ mod tests { } } } + // A sandbox-originated write built from a stale read must not replace an + // operator's repair, even while startup admission is still blocked. + write.replace_startup_baseline = false; + let blocked = Sandbox { + spec: Some(SandboxSpec { + policy: Some(baseline), + ..Default::default() + }), + status: Some(SandboxStatus { + configuration_activated: Some(false), + configuration_admission: Some(SandboxConfigurationAdmission { + state: Admission::Pending.into(), + ..Default::default() + }), + ..Default::default() + }), + ..Default::default() + }; + assert!(matches!( + project_policy_revision_onto_sandbox(&write, &blocked.encode_to_vec(), 1), + Err(PersistenceError::Conflict { .. }) + )); } } diff --git a/crates/openshell-supervisor/src/lib.rs b/crates/openshell-supervisor/src/lib.rs index 71cfebe63a..50be7a1045 100644 --- a/crates/openshell-supervisor/src/lib.rs +++ b/crates/openshell-supervisor/src/lib.rs @@ -2062,6 +2062,36 @@ where )) } +/// Retry transport failures while preserving the snapshot used by the request. +/// `None` means the gateway returned `ABORTED`: the caller must back off and +/// fetch fresh configuration instead of resending this snapshot's request. +async fn grpc_retry_snapshot(op_name: &str, f: F) -> Result> +where + T: Send, + F: Fn() -> Fut + Send + Sync, + Fut: Future> + Send, +{ + grpc_retry(op_name, || async { + match f().await { + Ok(value) => Ok(Some(value)), + Err(error) => { + let mut source: Option<&dyn std::error::Error> = Some(error.as_ref()); + while let Some(cause) = source { + if let Some(status) = cause.downcast_ref::() { + if status.code() == tonic::Code::Aborted { + return Ok(None); + } + break; + } + source = cause.source(); + } + Err(error) + } + } + }) + .await +} + #[tonic::async_trait] trait StartupGateway: Send + Sync { async fn snapshot( @@ -2429,12 +2459,19 @@ async fn load_policy_with_gateway( // Sync and re-fetch over a single connection to avoid extra // TLS handshakes. let ws = snapshot.workspace.clone(); - let synced = grpc_retry("Image policy synchronization", || { + let synced = grpc_retry_snapshot("Image policy synchronization", || { gateway.sync(sandbox, &discovered, &ws) }) .await; snapshot = match synced { - Ok(synced) => synced, + Ok(Some(synced)) => synced, + Ok(None) => { + tokio::time::sleep(Duration::from_secs( + 1u64 << reconciliation_attempts.min(2), + )) + .await; + continue; + } Err(error) => { // The gateway stored nothing, so report the rejection // against the snapshot this upload was built from and @@ -2489,12 +2526,19 @@ async fn load_policy_with_gateway( let sync_policy = proto_sync_payload_for_enriched_policy(&proto_policy, enriched) .filter(|_| snapshot.policy_source == openshell_core::proto::PolicySource::Sandbox); if let Some(sync_policy) = sync_policy { - let synced = grpc_retry("Enriched policy synchronization", || { + let synced = grpc_retry_snapshot("Enriched policy synchronization", || { gateway.sync(sandbox, &sync_policy, &snapshot.workspace) }) .await; let canonical = match synced { - Ok(canonical) => canonical, + Ok(Some(canonical)) => canonical, + Ok(None) => { + tokio::time::sleep(Duration::from_secs( + 1u64 << reconciliation_attempts.min(2), + )) + .await; + continue; + } Err(error) => { // The stored policy is unchanged, so report the // rejection against the snapshot it was read from and @@ -6577,6 +6621,176 @@ network_policies: assert_eq!(bundle.2, Some(repaired)); } + async fn assert_startup_write_aborted_refetches(image_upload: bool) { + use openshell_core::proto::{ConfigurationAdmissionState, PolicySource}; + // Inject ABORTED while the fixture changes the desired policy. This + // checks the supervisor's response to a reported conflict; it does not + // show that the gateway rejects an unfenced stale policy write. + let operator_policy = proto_policy_fixture(); + let (calls, mut observed) = tokio::sync::mpsc::unbounded_channel(); + let gateway = WriteRefusingStartupGateway { + desired: Arc::new(std::sync::Mutex::new(if image_upload { + unset_policy_snapshot() + } else { + settings_poll_result(Some(proto_tcp_policy_fixture()), 1, PolicySource::Sandbox) + })), + refusals: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from([ + SyncRefusal::new( + tonic::Code::Aborted, + "persist policy revision failed due to concurrent modification", + ), + ]))), + repair_after_refusal: Some(settings_poll_result( + Some(operator_policy.clone()), + 2, + PolicySource::Sandbox, + )), + rejection_report_errors: Arc::default(), + calls, + }; + let started = tokio::time::Instant::now(); + let policy = load_startup_test_policy(&gateway) + .await + .expect("startup accepts the operator's policy"); + let mut trace = Vec::new(); + while let Ok(call) = observed.try_recv() { + trace.push(call); + } + let refused = trace + .iter() + .position(|call| *call == StartupCall::Sync) + .expect("startup attempts the policy write"); + assert_eq!( + trace.get(refused + 1), + Some(&StartupCall::Snapshot), + "after ABORTED, startup must fetch a fresh snapshot instead of resending the stale payload; calls: {trace:?}" + ); + assert_eq!( + trace + .iter() + .filter(|call| **call == StartupCall::Sync) + .count(), + 1, + "startup must not resend the payload built from the stale read; calls: {trace:?}" + ); + assert!( + !trace.iter().any(|call| matches!( + call, + StartupCall::Report(report) + if report.state == ConfigurationAdmissionState::Rejected + )), + "a stale write is not a configuration error; calls: {trace:?}" + ); + assert_eq!( + policy, operator_policy, + "startup must install the operator's policy, not the stale write-back" + ); + assert!(started.elapsed() >= Duration::from_secs(1)); + assert_eq!(gateway.desired.lock().unwrap().version, 2); + let accepted = trace.last().expect("startup reports acceptance"); + assert!(matches!(accepted, StartupCall::Report(report) + if report.state == ConfigurationAdmissionState::Accepted + && report.generation == Some((2, "hash-v2".to_string(), 200, 0)))); + } + + #[tokio::test(start_paused = true)] + async fn startup_conflict_writes_refetch_instead_of_resending() { + for image_upload in [true, false] { + Box::pin(assert_startup_write_aborted_refetches(image_upload)).await; + } + } + + /// Load through the production startup loop, retaining a timeout so a + /// mistaken conflict retry cannot leave a test waiting for repair forever. + async fn load_startup_test_policy( + gateway: &impl StartupGateway, + ) -> Result { + let bundle = timeout( + Duration::from_mins(1), + load_policy_with_gateway( + Some("sandbox-id".to_string()), + Some("sandbox".to_string()), + Some("http://unused.invalid".to_string()), + None, + None, + &openshell_extension_core::ExtensionCredentialStore::new(), + LocalPolicyIdentity::Required, + None, + Some(ImagePolicyDiscovery::Policy(Box::new( + proto_tcp_policy_fixture(), + ))), + gateway, + ), + ) + .await + .expect("startup must finish within its retry budget")?; + Ok(bundle + .2 + .expect("gateway startup returns its accepted policy")) + } + + #[tokio::test(start_paused = true)] + async fn startup_conflict_write_transport_errors_are_retried() { + use openshell_core::proto::PolicySource; + for image_upload in [true, false] { + let (calls, mut observed) = tokio::sync::mpsc::unbounded_channel(); + let gateway = WriteRefusingStartupGateway { + desired: Arc::new(std::sync::Mutex::new(if image_upload { + unset_policy_snapshot() + } else { + settings_poll_result(Some(proto_tcp_policy_fixture()), 1, PolicySource::Sandbox) + })), + refusals: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from([ + SyncRefusal::new(tonic::Code::Unavailable, "transport failed"), + ]))), + repair_after_refusal: None, + rejection_report_errors: Arc::default(), + calls, + }; + load_startup_test_policy(&gateway) + .await + .expect("startup retries a transport failure"); + let syncs = std::iter::from_fn(|| observed.try_recv().ok()) + .filter(|call| *call == StartupCall::Sync) + .count(); + assert_eq!(syncs, 2, "the same write is retried once"); + } + } + + #[tokio::test(start_paused = true)] + async fn startup_conflict_writes_keep_reconciliation_budget() { + use openshell_core::proto::PolicySource; + for image_upload in [true, false] { + let (calls, mut observed) = tokio::sync::mpsc::unbounded_channel(); + let gateway = WriteRefusingStartupGateway { + desired: Arc::new(std::sync::Mutex::new(if image_upload { + unset_policy_snapshot() + } else { + settings_poll_result(Some(proto_tcp_policy_fixture()), 1, PolicySource::Sandbox) + })), + refusals: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from( + vec![SyncRefusal::new(tonic::Code::Aborted, "configuration keeps changing"); 5], + ))), + repair_after_refusal: None, + rejection_report_errors: Arc::default(), + calls, + }; + let error = load_startup_test_policy(&gateway) + .await + .expect_err("conflicts must have a bounded retry budget"); + assert!( + error + .to_string() + .contains("did not stabilize after 5 attempts"), + "{error}" + ); + let syncs = std::iter::from_fn(|| observed.try_recv().ok()) + .filter(|call| *call == StartupCall::Sync) + .count(); + assert_eq!(syncs, 5, "each conflict spends one attempt, no resends"); + } + } + #[tokio::test(start_paused = true)] async fn startup_policy_write_refusal_image_repair_before_report_refetches() { Box::pin(assert_startup_repair_before_rejection_refetches( diff --git a/docs/how-it-works/sandboxes/overview.mdx b/docs/how-it-works/sandboxes/overview.mdx index 2e824c730e..a327be6a77 100644 --- a/docs/how-it-works/sandboxes/overview.mdx +++ b/docs/how-it-works/sandboxes/overview.mdx @@ -969,6 +969,8 @@ Management operations remain available while startup is blocked. After repair, the supervisor completes startup without recreating the sandbox. Starting a stopped sandbox repeats configuration admission before launching its workload. +While startup is blocked, an operator's policy repair is never replaced by the supervisor's startup write. If the repair lands while the write is in flight, the gateway returns `ABORTED` and the supervisor reads the latest configuration instead of resending the write. Repeated conflicts can still exhaust the startup retry budget. + Image preparation and initial supervisor startup have an absolute deadline of 1800 seconds by default, independently of the CLI wait timeout. Operators can set `image_preparation_timeout_seconds` in `[openshell.gateway]` to any value from 1 through 86400. Each new create or explicit start stores its deadline. Progress events, configuration edits, and gateway or driver restarts cannot extend it. Changing the gateway setting affects new attempts only. The first authenticated configuration report from the current supervisor ends preparation and starts a separate 300-second admission repair window. A slow image download therefore does not consume time reserved for fixing policy. During admission, an effective policy, settings, provider, profile, or attachment change resets the window from its stored change time. The first failed configuration load for that change grants another full window. Repeated failures, duplicate reports, and reconnects do not extend it; reaching `Ready` clears it.