From 907392cce8d7915913c2ab785056867ebd76246d Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 23 Oct 2026 00:21:02 +0900 Subject: [PATCH] fix: preserve restore effects and fence prepared attachment replay --- crates/worker-runtime/src/runtime.rs | 112 +++- crates/worker-runtime/src/worker_backend.rs | 69 +- .../worker_backend_restore_candidate_tests.rs | 206 ++++++ .../worker_backend_restore_failure_tests.rs | 595 +++++++++++++++++- 4 files changed, 952 insertions(+), 30 deletions(-) diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index eaec092dc..fc53df360 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -2150,6 +2150,16 @@ impl Runtime { { return Err(conflict()); } + if let Err(error) = owner.validate_workdir_preparation_authority(&candidate) { + return Ok(( + Some(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::ReconciliationRequired, + "worker_restore_authority_changed", + sanitize_worker_create_failure_message(&error.to_string()), + )), + None, + )); + } } else { // Startup/older persisted pending operations have a lifecycle // owner but no public request receipt. Bind recovery to that @@ -2215,12 +2225,18 @@ impl Runtime { candidate.pending_restore = Some(pending); candidate.restore_guard.rotate(); candidate.restore_guard.active_request_id = Some(request.request_id.clone()); + let workdir_attachment_preparation_base = request + .preparation + .as_ref() + .and_then(|preparation| preparation.workdir_attachments.as_ref()) + .map(|_| candidate.logical_workdir_attachments.clone()); candidate.restore_guard.owners.insert( request.request_id.clone(), RestoreRequestOwner { fingerprint, operation_id, request, + workdir_attachment_preparation_base, preparation_applied: recovering_pending, execution_started: recovering_pending, receipt: None, @@ -2433,14 +2449,15 @@ impl Runtime { { return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail())); } - let preparation = worker + let owner = worker .restore_guard .active_request_id .as_ref() - .and_then(|id| worker.restore_guard.owners.get(id)) - .filter(|owner| !owner.preparation_applied) - .and_then(|owner| owner.request.preparation.as_ref()); - let prepared_worker = match worker.prepared_restore_record(preparation) { + .and_then(|id| worker.restore_guard.owners.get(id)); + let prepared_worker = match owner.map_or_else( + || Ok(worker.clone()), + |owner| owner.prepared_worker_record(worker), + ) { Ok(prepared) => prepared, Err(error) => { return Ok(RuntimeWorkerRestoreResult::failed( @@ -4398,8 +4415,7 @@ impl Runtime { worker_id: worker_ref.worker_id, }); } - let mut candidate = - previous.prepared_restore_record(owner.request.preparation.as_ref())?; + let mut candidate = owner.prepared_worker_record(previous)?; candidate .restore_guard .owners @@ -6277,12 +6293,48 @@ pub(crate) struct RestoreRequestOwner { fingerprint: String, operation_id: WorkerLifecycleOperationId, request: runtime_api::WorkerRestoreRequest, + // This is the admission comparison input, not a second attachment authority. + // Without it an old preparation could hide a later canonical revocation. + // Older owners lacking this fact cannot safely reapply attachment overrides. + #[serde(default, skip_serializing_if = "Option::is_none")] + workdir_attachment_preparation_base: Option>, preparation_applied: bool, #[serde(default)] execution_started: bool, receipt: Option, } +impl RestoreRequestOwner { + fn validate_workdir_preparation_authority( + &self, + worker: &WorkerRecord, + ) -> Result<(), RuntimeError> { + if !self.preparation_applied + && self + .request + .preparation + .as_ref() + .is_some_and(|preparation| preparation.workdir_attachments.is_some()) + && self.workdir_attachment_preparation_base.as_ref() + != Some(&worker.logical_workdir_attachments) + { + return Err(RuntimeError::InvalidRequest( + "current Worker authority differs from admitted Restore attachment preparation" + .into(), + )); + } + Ok(()) + } + + fn prepared_worker_record(&self, worker: &WorkerRecord) -> Result { + if self.preparation_applied { + return Ok(worker.clone()); + } + self.validate_workdir_preparation_authority(worker)?; + worker.prepared_restore_record(self.request.preparation.as_ref()) + } +} + #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] #[serde(deny_unknown_fields)] pub(crate) struct RestoreGuard { @@ -11995,7 +12047,7 @@ mod tests { let mut request = runtime.test_restore_request(&worker.worker_ref); request.preparation = Some(serde_json::from_value(serde_json::json!({ "workspace_api": { "workspace_id": "workspace-a", "base_url": "https://prepared.example" }, - "workdir_attachments": [], "repository_access": [] + "workdir_attachments": [{ "alias": "checkout", "working_directory_id": "prepared-logical-workdir" }], "repository_access": [] })).unwrap()); let gate = Arc::new(RestoreGate::default()); *backend.restore_gate.lock().unwrap() = Some(gate.clone()); @@ -12014,6 +12066,26 @@ mod tests { let state = runtime.lock().unwrap(); let record = state.worker(&worker.worker_ref).unwrap(); assert!(record.request.as_ref().unwrap().workspace_api.is_none()); + assert!(record.logical_workdir_attachments.is_empty()); + let owner = record.restore_guard.owners.values().next().unwrap(); + let mut saved_owner = serde_json::to_value(owner).unwrap(); + let decoded: RestoreRequestOwner = serde_json::from_value(saved_owner.clone()).unwrap(); + assert!( + decoded + .validate_workdir_preparation_authority(record) + .is_ok() + ); + saved_owner + .as_object_mut() + .unwrap() + .remove("workdir_attachment_preparation_base"); + let legacy: RestoreRequestOwner = serde_json::from_value(saved_owner).unwrap(); + assert!( + legacy + .validate_workdir_preparation_authority(record) + .is_err(), + "a missing historical comparison input must not be invented from current authority" + ); assert!( !record .restore_guard @@ -12033,6 +12105,11 @@ mod tests { .base_url, "https://prepared.example" ); + assert_eq!( + backend.restore_requests.lock().unwrap()[0].logical_workdir_attachments[0] + .working_directory_id, + "prepared-logical-workdir" + ); assert!(backend.activation_operation_ids.lock().unwrap().is_empty()); gate.release(); assert_eq!( @@ -12053,6 +12130,25 @@ mod tests { .base_url, "https://prepared.example" ); + let record = state.worker(&worker.worker_ref).unwrap(); + assert_eq!( + record.logical_workdir_attachments[0].working_directory_id, + "prepared-logical-workdir" + ); + let owner = record.restore_guard.owners.values().next().unwrap(); + assert!(owner.preparation_applied); + let mut revoked = record.clone(); + revoked + .apply_logical_workdir_attachments(Vec::new()) + .unwrap(); + assert!( + owner + .prepared_worker_record(&revoked) + .unwrap() + .logical_workdir_attachments + .is_empty(), + "a committed preparation is not a perpetual grant" + ); } #[cfg(feature = "ws-server")] diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index b5ace780f..695a03379 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -360,20 +360,21 @@ async fn prepare_restore_operation( failure })?; if repaired { - Ok(Box::new(RepairedRestoreCandidate { candidate })) + Ok(Box::new(EffectRetainingRestoreCandidate { candidate })) } else { Ok(candidate) } } -// A completed repair is a canonical effect even when later activation fails -// before its own mutation boundary. Carry that evidence with the resource owner. -struct RepairedRestoreCandidate { +// Completed repair or an earlier activation may have changed canonical state. +// Keep that operation-owned evidence through reconstruction and even an +// activation failure before the replacement candidate's own mutation boundary. +struct EffectRetainingRestoreCandidate { candidate: Box, } #[async_trait] -impl RuntimeRestoreCandidate for RepairedRestoreCandidate { +impl RuntimeRestoreCandidate for EffectRetainingRestoreCandidate { fn initial_state(&self) -> protocol::WorkerStateSnapshot { self.candidate.initial_state() } @@ -2953,6 +2954,9 @@ struct PendingRuntimeRestore { task: tokio::sync::Mutex>>, result: Mutex, RuntimeRestoreFailure>>>, activated: Mutex>>, + // Monotonic for this operation, not a claim that its current candidate has + // started. A reconstructed candidate has a new result and no completion in + // `activated`; prior activation/effect evidence must survive that handoff. activation_started: AtomicBool, discard_started: AtomicBool, discard_result: Mutex>>, @@ -2960,6 +2964,16 @@ struct PendingRuntimeRestore { } impl PendingRuntimeRestore { + fn owns_prepared_candidate(&self) -> bool { + matches!( + self.result + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref(), + Some(Ok(_)) + ) + } + async fn wait(&self, timeout: Duration) -> Result<(), String> { let mut task = self.task.lock().await; if let Some(message) = self @@ -3279,23 +3293,40 @@ where && (failure.durable_effects_possible || owner.activation_started.load(Ordering::Acquire)) { + let prior_effects = failure.durable_effects_possible + || owner.activation_started.load(Ordering::Acquire); let result = match prepare_restore_operation(factory.as_ref(), owner.request.clone()) .await { - Ok(candidate) => Ok(candidate), - Err(failure) => Err(finish_restore_failure( - factory.as_ref(), - &owner.request, - failure, - ) - .await), + Ok(candidate) => { + Ok(Box::new(EffectRetainingRestoreCandidate { candidate }) + as Box) + } + Err(mut failure) => { + failure.durable_effects_possible |= prior_effects; + Err(finish_restore_failure( + factory.as_ref(), + &owner.request, + failure, + ) + .await) + } }; - owner.activation_started.store(false, Ordering::Release); + // A completed reprepare replaces attempt resources, never + // the operation's effect history. Only a ready candidate + // may clear the old activation completion for a new attempt. + let activated = match &result { + Ok(_) => None, + Err(failure) if owner.activation_started.load(Ordering::Acquire) => { + Some(Err(failure.clone())) + } + Err(_) => None, + }; *owner .activated .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = None; + .unwrap_or_else(std::sync::PoisonError::into_inner) = activated; *owner .result .lock() @@ -4019,7 +4050,9 @@ where failure.to_string(), ); } - if pending.activation_started.load(Ordering::Acquire) { + if pending.activation_started.load(Ordering::Acquire) + && !pending.owns_prepared_candidate() + { let mut activated = match pending.activated.lock() { Ok(result) => result, Err(_) => { @@ -5205,7 +5238,8 @@ where }; } } - if pending.activation_started.load(Ordering::Acquire) { + if pending.activation_started.load(Ordering::Acquire) && !pending.owns_prepared_candidate() + { let activated = pending .activated .lock() @@ -5330,7 +5364,8 @@ where return Err("restored Worker operation identity does not match".into()); } self.wait_pending_restore(&pending, self.spawn_restore_timeout)?; - if !pending.activation_started.swap(true, Ordering::AcqRel) { + if pending.owns_prepared_candidate() { + pending.activation_started.store(true, Ordering::Release); let owner = pending.clone(); let factory = self.factory.clone(); let task = self.spawn_on_adapter_runtime(async move { diff --git a/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs b/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs index a5b0312f9..443fbac6f 100644 --- a/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs @@ -895,4 +895,210 @@ mod restore_candidate { assert_eq!(model.call_count.load(Ordering::SeqCst), 0); } } + + #[test] + #[serial_test::serial(worker_allocation)] + fn attachment_preparation_replay_cannot_hide_logical_authority_revocation() { + use crate::fs_store::{ + FsRuntimeStoreOptions, PersistedWorkerExecution, PersistedWorkerExecutionState, + PersistedWorkerRecord, + }; + let mut fixture = Fixture::new(); + Arc::get_mut(&mut fixture.backend) + .unwrap() + .spawn_restore_timeout = Duration::from_secs(2); + let worker_ref = fixture.request.worker_ref.clone(); + let scope = fixture.request.workspace_scope.clone().unwrap(); + let options = FsRuntimeStoreOptions { + root: fixture._root.path().join("catalog"), + runtime_id: "candidate-runtime".into(), + display_name: None, + }; + let attachment = crate::catalog::LogicalWorkdirAttachment { + alias: WorkdirAttachmentAlias::new("checkout").unwrap(), + working_directory_id: "workspace-only-workdir".into(), + capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY, + }; + // Only the initial Runtime catalog is seeded. The saved Profile, queue, + // restore admission, revocation, replay and Stop use their real owners. + let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( + options.clone(), + fixture.backend.clone(), + ) + .unwrap(); + let create = fixture.request.request.clone(); + runtime + .test_runtime_store() + .write_worker_record(&PersistedWorkerRecord { + restore_guard: crate::runtime::RestoreGuard::default(), + worker_ref: worker_ref.clone(), + worker_id: worker_ref.worker_id, + profile: create.profile.clone(), + display_name: None, + profile_source: create.profile_source.reference(), + config_bundle: create.config_bundle.clone(), + created_at_ms: None, + status: crate::catalog::WorkerStatus::Stopped, + last_finished_submission_request_id: None, + execution_state: PersistedWorkerExecutionState::Available( + PersistedWorkerExecution { + request: create, + binding: None, + restore_intent: crate::catalog::WorkerRestoreIntent::Explicit, + }, + ), + workspace_id: Some(scope.workspace_id.clone()), + workdir_attachments: Vec::new(), + logical_workdir_attachments: vec![attachment.clone()], + }) + .unwrap(); + drop(runtime); + let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( + options.clone(), + fixture.backend.clone(), + ) + .unwrap(); + assert!( + runtime + .worker_detail(&worker_ref) + .unwrap() + .workdir_attachments + .is_empty() + ); + let before = fixture.canonical_files(); + let authority = || -> serde_json::Value { + serde_json::from_slice( + &fs::read( + options + .root + .join("workers") + .join(worker_ref.worker_id.to_string()) + .join("worker.json"), + ) + .unwrap(), + ) + .unwrap() + }; + let release = Arc::new(tokio::sync::Notify::new()); + let (entered_tx, entered) = mpsc::channel(); + *fixture + .backend + .factory + .restore_faults + .factory_gate + .lock() + .unwrap() = Some(release.clone()); + *fixture + .backend + .factory + .restore_faults + .factory_entered + .lock() + .unwrap() = Some(entered_tx); + let mut request = runtime.test_restore_request(&worker_ref); + request.preparation.as_mut().unwrap().workdir_attachments = + Some(serde_json::from_value(serde_json::to_value(vec![attachment]).unwrap()).unwrap()); + let first = runtime + .restore_worker_operation_scoped(&scope, &worker_ref, request.clone()) + .unwrap(); + entered.recv_timeout(WAIT).unwrap(); + assert_eq!( + first.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "{first:?}" + ); + let pending = fixture + .backend + .pending_restore(&worker_ref) + .unwrap() + .unwrap(); + assert!(!pending.request.logical_workdir_attachments.is_empty()); + assert!(pending.request.previous_workdir_attachments.is_empty()); + assert!(pending.request.deferred_workdir_sessions.is_empty()); + runtime + .replace_worker_workdir_attachments_scoped(&scope, &worker_ref, Vec::new()) + .unwrap(); + assert!(authority().get("logical_workdir_attachments").is_none()); + release.notify_one(); + let replay = runtime + .restore_worker_operation_scoped(&scope, &worker_ref, request.clone()) + .unwrap(); + assert_eq!( + replay.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "revocation hidden by saved preparation: {replay:?}" + ); + assert!( + replay + .message + .as_deref() + .unwrap_or_default() + .contains("current Worker authority differs"), + "{replay:?}" + ); + let record = authority(); + assert!(record.get("logical_workdir_attachments").is_none()); + let owner = &record["restore_guard"]["owners"][&request.request_id]; + assert_eq!( + owner["operation_id"], + serde_json::to_value(pending.request.operation_id).unwrap() + ); + assert!(owner["receipt"].is_null()); + assert_eq!( + owner["workdir_attachment_preparation_base"] + .as_array() + .unwrap() + .len(), + 1 + ); + let loaded = runtime.test_runtime_store().load_runtime_state().unwrap(); + let durable_guard = + serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap(); + assert_eq!(durable_guard["owners"][&request.request_id], *owner); + let (coordinated, preparation) = runtime + .coordinate_worker_restore_operation( + &worker_ref, + runtime_api::WorkerRestoreCoordinationRequest { + expected_observation_token: request.expected_observation_token.clone(), + request_id: request.request_id.clone(), + preparation: None, + }, + Some(&scope), + ) + .unwrap(); + assert_eq!( + coordinated.unwrap().state, + server_api::WorkerRestoreState::ReconciliationRequired + ); + assert!( + preparation.is_none(), + "coordinator must not export the revoked preparation again" + ); + assert!(fixture.backend.workers.lock().unwrap().is_empty()); + assert!(Arc::ptr_eq( + &pending, + &fixture + .backend + .pending_restore(&worker_ref) + .unwrap() + .unwrap() + )); + assert!(!pending.activation_started.load(Ordering::SeqCst)); + fixture.assert_quiet(); + assert_eq!(fixture.canonical_files(), before); + runtime + .stop_worker_scoped(&scope, &worker_ref, None) + .unwrap(); + fixture.assert_released(); + fixture.assert_quiet(); + assert_eq!(fixture.canonical_files(), before); + assert!(authority().get("logical_workdir_attachments").is_none()); + assert_ne!( + runtime + .restore_worker_operation_scoped(&scope, &worker_ref, request) + .unwrap() + .state, + server_api::WorkerRestoreState::Accepted + ); + } } diff --git a/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs b/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs index 40de8168c..91530ab28 100644 --- a/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs @@ -100,6 +100,148 @@ impl WorkingDirectoryMaterializer for RestoreOpenCountingMaterializer { } } +// Substitute only initial creation: a new aggregate/Session is left pending +// before its first head, without ever materializing or rewinding a saved head. +// All Restore preflight, repair, retry, activation and Stop use the real backend. +struct PendingProfileSeedBackend { + inner: Arc, +} + +impl WorkerExecutionBackend for PendingProfileSeedBackend { + fn backend_id(&self) -> &str { + self.inner.backend_id() + } + + fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { + let aggregate = self + .inner + .factory + .worker_aggregate_dir(&request.worker_ref) + .unwrap(); + let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&request.worker_ref); + self.inner + .run_on_adapter_runtime(async move { + assert!( + !aggregate.join("session").exists(), + "pending fixture requires a new Session aggregate" + ); + let manifest = manifest::WorkerManifest::from_toml(&format!( + r#" +[worker] +name = "{name}" +pwd = "./" +[scope] +allow = [] +[model] +scheme = "anthropic" +model_id = "test-model" +auth = {{ kind = "none" }} +[engine] +max_tokens = 100 +"# + )) + .unwrap(); + let sessions = WorkerSessionStore::new(aggregate.join("session")).unwrap(); + let metadata = WorkerAggregateStore::new(&aggregate, &name).unwrap(); + assert!(metadata.read_by_name(&name).unwrap().is_none()); + let engine = + Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated( + MockClient::new(Vec::new()), + ); + let mut head = Worker::new( + manifest.clone(), + engine, + CombinedStore::new(sessions.clone(), metadata.clone()), + WorkerWorkspaceContext::unavailable( + None, + "initial creation interrupted before first head", + ), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + head.enable_worker_metadata_write_through().unwrap(); + let mut saved = metadata.read_by_name(&name).unwrap().unwrap(); + let active = saved.active.as_ref().unwrap(); + assert!(active.segment_id.is_none()); + assert!(sessions.list_segments(active.session_id).unwrap().is_empty()); + saved.resolved_manifest_snapshot = + Some(manifest::write_persisted_worker_manifest_snapshot(&manifest).unwrap()); + metadata.write(&saved).unwrap(); + drop(head); + Ok(()) + }) + .unwrap(); + WorkerExecutionSpawnResult::Connected { + worker_state: protocol::WorkerStateSnapshot::initial(), + workdir_attachments: Vec::new(), + } + } + + fn fetch_workspace_config( + &self, + request: WorkspaceConfigFetchRequest, + ) -> Result { + self.inner.fetch_workspace_config(request) + } + + fn observe_workspace_prompt_projection( + &self, + projection: worker::WorkspacePromptProjection, + ) -> Result<(), String> { + self.inner.observe_workspace_prompt_projection(projection) + } + + fn preflight_restore( + &self, + request: &WorkerExecutionRestoreRequest, + ) -> Result<(), WorkerExecutionResult> { + self.inner.preflight_restore(request) + } + + fn completed_restore_failure( + &self, + operation_id: crate::execution::WorkerLifecycleOperationId, + worker_ref: &WorkerRef, + ) -> Option { + self.inner + .completed_restore_failure(operation_id, worker_ref) + } + + fn restore_worker(&self, request: WorkerExecutionRestoreRequest) -> WorkerExecutionSpawnResult { + self.inner.restore_worker(request) + } + + fn reconcile_restore( + &self, + request: WorkerExecutionRestoreRequest, + ) -> WorkerExecutionSpawnResult { + self.inner.reconcile_restore(request) + } + + fn activate_restored_worker( + &self, + operation_id: crate::execution::WorkerLifecycleOperationId, + worker_ref: &WorkerRef, + ) -> Result<(), String> { + self.inner + .activate_restored_worker(operation_id, worker_ref) + } + + fn dispatch_input(&self, worker_ref: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult { + self.inner.dispatch_input(worker_ref, input) + } + + fn dispatch_method(&self, worker_ref: &WorkerRef, method: Method) -> WorkerExecutionResult { + self.inner.dispatch_method(worker_ref, method) + } + + fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult { + self.inner.stop_worker_operation(request) + } +} + struct ProfileFailureFixture { runtime: EmbeddedRuntime, backend: Arc, @@ -115,6 +257,14 @@ impl ProfileFailureFixture { } fn with_saved_inputs(timeout: Duration, model: Option) -> Self { + Self::with_seed(timeout, model, false) + } + + fn with_pending_head(timeout: Duration) -> Self { + Self::with_seed(timeout, None, true) + } + + fn with_seed(timeout: Duration, model: Option, pending_head: bool) -> Self { let root = restore_failure_tempdir(); let repo = root.path().join("g"); fs::create_dir(&repo).unwrap(); @@ -141,22 +291,31 @@ impl ProfileFailureFixture { opens: provider_opens.clone(), }), ); + let execution: Arc = if pending_head { + Arc::new(PendingProfileSeedBackend { + inner: backend.clone(), + }) + } else { + backend.clone() + }; let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: store, runtime_id: "r".into(), display_name: None, }, - backend.clone(), + execution, ) .unwrap(); runtime.store_config_bundle(test_bundle()).unwrap(); let mut request = create_request("restore failure recovery"); request.profile = ProfileSelector::Builtin("default".into()); - request.workdir_attachment_requests = vec![WorkingDirectoryAttachmentRequest { - alias: WorkdirAttachmentAlias::new("checkout").unwrap(), - working_directory: working_directory_request(&repo), - }]; + if !pending_head { + request.workdir_attachment_requests = vec![WorkingDirectoryAttachmentRequest { + alias: WorkdirAttachmentAlias::new("checkout").unwrap(), + working_directory: working_directory_request(&repo), + }]; + } let worker = runtime.create_worker(request).unwrap(); let seed_handle = queue_inputs.then(|| { backend.workers.lock().unwrap()[&worker.worker_ref] @@ -203,6 +362,127 @@ impl ProfileFailureFixture { *self.backend.factory.restore_faults.fail_at.lock().unwrap() = Some(stage); } + fn pin_unclaimed_restore_upload(&self) -> (PathBuf, BTreeMap>) { + let aggregate = self + .backend + .factory + .worker_aggregate_dir(&self.worker.worker_ref) + .unwrap(); + let name = + ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&self.worker.worker_ref); + let metadata = WorkerAggregateStore::open_read_only(&aggregate, name.clone()).unwrap(); + let session_id = metadata + .read_by_name(&name) + .unwrap() + .unwrap() + .active + .unwrap() + .session_id; + let session = WorkerSessionStore::new(aggregate.join("session")).unwrap(); + let orphan = session + .write_uploaded_file( + session_id, + "orphan.txt", + "text/plain", + b"restore effect evidence", + session_store::UploadedFileLimits::default(), + ) + .unwrap(); + session + .pin_uploaded_file(session_id, &orphan, "uncommitted-restore-input") + .unwrap(); + let artifacts = aggregate.join("session/artifacts/paste"); + let pinned = persisted_files(&artifacts); + assert!(!pinned.is_empty()); + (artifacts, pinned) + } + + fn assert_retained_restore_effects( + &self, + request: &runtime_api::WorkerRestoreRequest, + pending: &Arc, + stage: RuntimeRestoreStage, + ) { + let retained = self + .backend + .pending_restore(&self.worker.worker_ref) + .unwrap() + .expect("resource cleanup must not remove the operation's effect owner"); + assert!(Arc::ptr_eq(pending, &retained)); + { + let result = retained.result.lock().unwrap(); + let failure = result.as_ref().unwrap().as_ref().err().unwrap(); + assert_eq!(failure.stage, stage); + assert_eq!(failure.cleanup, RuntimeRestoreCleanup::Confirmed); + assert!( + failure.durable_effects_possible, + "resource cleanup cannot erase prior canonical effects: {failure:?}" + ); + } + let record: serde_json::Value = serde_json::from_slice( + &fs::read( + self.root + .path() + .join("s/workers") + .join(self.worker.worker_ref.worker_id.to_string()) + .join("worker.json"), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!( + record["restore_guard"]["active_request_id"], + request.request_id + ); + let operation = serde_json::to_value(pending.request.operation_id).unwrap(); + let owner = &record["restore_guard"]["owners"][&request.request_id]; + assert_eq!(owner["operation_id"], operation); + assert!( + owner["receipt"].is_null(), + "false terminal receipt: {owner:?}" + ); + assert_eq!( + record["execution_state"]["state"], + "reconciliation_required" + ); + assert_eq!( + record["execution_state"]["execution"]["operation"], + "restore" + ); + assert_eq!( + record["execution_state"]["execution"]["intent"]["operation_id"], + operation + ); + assert!( + self.backend + .completed_restore_failures + .lock() + .unwrap() + .is_empty() + ); + assert!(self.backend.workers.lock().unwrap().is_empty()); + assert!( + self.backend + .factory + .failed_restore_resources + .lock() + .unwrap() + .is_empty(), + "cleanup confirmed resources, not rollback of canonical effects" + ); + assert!( + self.backend + .factory + .observation_hub + .get(&self.worker.worker_ref) + .is_none() + ); + if !pending.request.deferred_workdir_sessions.is_empty() { + self.assert_closed_deferred_sessions(pending); + } + assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0); + } + fn assert_released(&self) { assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0); assert!(self.backend.pending_restores.lock().unwrap().is_empty()); @@ -422,6 +702,311 @@ impl ProfileFailureFixture { } } +#[test] +#[serial_test::serial(worker_allocation)] +fn profile_restore_prior_effects_survive_read_only_reprepare_failure() { + for (activation_stage, read_stage) in [ + (RuntimeRestoreStage::Feature, RuntimeRestoreStage::Metadata), + ( + RuntimeRestoreStage::Controller, + RuntimeRestoreStage::Reconstruction, + ), + ] { + let fixture = ProfileFailureFixture::new(Duration::from_secs(5)); + let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload(); + fixture.fail(activation_stage); + let request = fixture + .runtime + .test_restore_request(&fixture.worker.worker_ref); + let first = fixture + .runtime + .restore_worker_operation(&fixture.worker.worker_ref, request.clone()) + .unwrap(); + assert_eq!( + first.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "{activation_stage:?}: {first:?}" + ); + let pending = fixture + .backend + .pending_restore(&fixture.worker.worker_ref) + .unwrap() + .unwrap(); + assert!(pending.activation_started.load(Ordering::SeqCst)); + fixture.assert_retained_restore_effects(&request, &pending, activation_stage); + let effected = persisted_files(&artifacts); + assert_ne!( + effected, pinned, + "activation must actually reconcile the orphan pin" + ); + + // The next attempt needs no repair, and fails before any new mutation. + // Resource cleanup is confirmed, but the first attempt's pin change + // was not rolled back. Repeated sends must retain the same owner. + for _ in 0..2 { + fixture.fail(read_stage); + let replay = fixture + .runtime + .restore_worker_operation(&fixture.worker.worker_ref, request.clone()) + .unwrap(); + assert_eq!( + replay.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "{activation_stage:?} then {read_stage:?}: {replay:?}" + ); + fixture.assert_retained_restore_effects(&request, &pending, read_stage); + assert!(pending.activation_started.load(Ordering::SeqCst)); + assert_eq!(persisted_files(&artifacts), effected); + } + fixture.resume_same_operation(request, pending.request.operation_id); + } +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn profile_restore_prior_effects_from_pending_head_repair_survive_read_only_reprepare_failure() { + let fixture = ProfileFailureFixture::with_pending_head(Duration::from_secs(5)); + let aggregate = fixture + .backend + .factory + .worker_aggregate_dir(&fixture.worker.worker_ref) + .unwrap(); + let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&fixture.worker.worker_ref); + let metadata = WorkerAggregateStore::new(&aggregate, &name).unwrap(); + let initial = metadata.read_by_name(&name).unwrap().unwrap(); + let pending_session = initial.active.as_ref().unwrap().session_id; + assert!(initial.active.as_ref().unwrap().segment_id.is_none()); + { + let before = WorkerSessionStore::new(aggregate.join("session")).unwrap(); + assert!(before.list_sessions().unwrap().is_empty()); + assert!(before.list_segments(pending_session).unwrap().is_empty()); + } + // The independent new Worker has never published a head. Its first segment + // and active pointer must be created by production repair, not fixture + // rewinding/deletion of a materialized Session. + fixture.fail(RuntimeRestoreStage::Metadata); + let request = fixture + .runtime + .test_restore_request(&fixture.worker.worker_ref); + let first = fixture + .runtime + .restore_worker_operation(&fixture.worker.worker_ref, request.clone()) + .unwrap(); + assert_eq!( + first.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "{first:?}" + ); + let pending = fixture + .backend + .pending_restore(&fixture.worker.worker_ref) + .unwrap() + .unwrap(); + assert!(!pending.activation_started.load(Ordering::SeqCst)); + assert!(pending.request.deferred_workdir_sessions.is_empty()); + fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::Metadata); + let failure = pending + .result + .lock() + .unwrap() + .as_ref() + .unwrap() + .as_ref() + .err() + .unwrap() + .clone(); + let repaired = WorkerAggregateStore::open_read_only(&aggregate, &name) + .unwrap() + .read_by_name(&name) + .unwrap() + .unwrap(); + eprintln!( + "pending-head repair first={first:?} failure={failure:?} active={:?}", + repaired.active + ); + let active = repaired + .active + .as_ref() + .expect("real repair must retain the pending Session's active pointer"); + assert_eq!(active.session_id, pending_session); + // Store handles opened before the first Manifest retain session_id=None; + // they cannot observe that another owner has since published this Session. + // Reopen read-only to require the canonical Manifest as well as its logs. + let session = WorkerSessionStore::open_read_only(aggregate.join("session")) + .unwrap_or_else(|error| { + panic!( + "repair Manifest reread failed: {error}; first={first:?}; failure={failure:?}; active={active:?}" + ) + }); + assert_eq!(session.list_sessions().unwrap(), vec![pending_session]); + let segments = session.list_segments(pending_session).unwrap(); + assert_eq!( + segments.len(), + 1, + "real repair must commit exactly one seed; failure={failure:?}; active={active:?}" + ); + assert_eq!(active.segment_id, Some(segments[0])); + let entries = session + .read_all_read_only(pending_session, segments[0]) + .unwrap(); + assert!(matches!( + entries.first(), + Some(LogEntry::AnnotatedSegmentStart { .. }) + )); + assert!( + session_store::collect_state(&entries) + .system_prompt + .is_some() + ); + assert_ne!( + repaired, initial, + "repair must publish the first active head" + ); + let repaired_session = persisted_files(&aggregate.join("session")); + + for _ in 0..2 { + fixture.fail(RuntimeRestoreStage::Reconstruction); + let replay = fixture + .runtime + .restore_worker_operation(&fixture.worker.worker_ref, request.clone()) + .unwrap(); + assert_eq!( + replay.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "repair effect cannot become a rollback on the next read failure: {replay:?}" + ); + fixture.assert_retained_restore_effects( + &request, + &pending, + RuntimeRestoreStage::Reconstruction, + ); + let current_metadata = WorkerAggregateStore::open_read_only(&aggregate, &name).unwrap(); + assert_eq!( + current_metadata.read_by_name(&name).unwrap().unwrap(), + repaired + ); + let current_session = + WorkerSessionStore::open_read_only(aggregate.join("session")).unwrap(); + assert_eq!( + current_session.list_sessions().unwrap(), + vec![pending_session] + ); + assert_eq!( + current_session.list_segments(pending_session).unwrap(), + segments + ); + assert_eq!( + persisted_files(&aggregate.join("session")), + repaired_session + ); + } + fixture.resume_same_operation(request, pending.request.operation_id); +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn profile_restore_prior_effects_survive_rebuilt_candidate_pre_effect_activation_failure() { + let fixture = ProfileFailureFixture::new(Duration::from_secs(5)); + let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload(); + fixture.fail(RuntimeRestoreStage::Feature); + let request = fixture + .runtime + .test_restore_request(&fixture.worker.worker_ref); + let first = fixture + .runtime + .restore_worker_operation(&fixture.worker.worker_ref, request.clone()) + .unwrap(); + assert_eq!( + first.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "{first:?}" + ); + let pending = fixture + .backend + .pending_restore(&fixture.worker.worker_ref) + .unwrap() + .unwrap(); + fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::Feature); + let effected = persisted_files(&artifacts); + assert_ne!(effected, pinned); + + let release = Arc::new(tokio::sync::Notify::new()); + let (entered_tx, entered) = mpsc::channel(); + *fixture + .backend + .factory + .restore_faults + .factory_gate + .lock() + .unwrap() = Some(release.clone()); + *fixture + .backend + .factory + .restore_faults + .factory_entered + .lock() + .unwrap() = Some(entered_tx); + let runtime = fixture.runtime.clone(); + let worker_ref = fixture.worker.worker_ref.clone(); + let resend = request.clone(); + let retry = std::thread::spawn(move || runtime.restore_worker_operation(&worker_ref, resend)); + let reached = entered.recv_timeout(Duration::from_secs(5)); + // This gate is after real read-only reconstruction. Change only saved + // metadata, so the rebuilt candidate's first activation revalidation fails + // before allocation/pin/metadata/delegation mutation. No factory or repair + // implementation is replaced by a fake. + let aggregate = fixture + .backend + .factory + .worker_aggregate_dir(&fixture.worker.worker_ref) + .unwrap(); + let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&fixture.worker.worker_ref); + let metadata = WorkerAggregateStore::new(&aggregate, &name).unwrap(); + let mut changed = metadata.read_by_name(&name).unwrap().unwrap(); + changed.workspace_root = Some(fixture.root.path().join("changed-after-reconstruction")); + metadata.write(&changed).unwrap(); + release.notify_one(); + let rebuilt = retry.join().unwrap().unwrap(); + reached.expect("same-operation retry must reach the real reconstructed candidate"); + assert_eq!( + rebuilt.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "{rebuilt:?}" + ); + assert_eq!( + fixture + .backend + .factory + .restore_faults + .executions + .load(Ordering::SeqCst), + 2 + ); + assert!(pending.activation_started.load(Ordering::SeqCst)); + fixture.assert_retained_restore_effects( + &request, + &pending, + RuntimeRestoreStage::Reconstruction, + ); + assert_eq!(metadata.read_by_name(&name).unwrap().unwrap(), changed); + assert_eq!(persisted_files(&artifacts), effected); + + fixture.fail(RuntimeRestoreStage::Metadata); + let replay = fixture + .runtime + .restore_worker_operation(&fixture.worker.worker_ref, request.clone()) + .unwrap(); + assert_eq!( + replay.state, + server_api::WorkerRestoreState::ReconciliationRequired, + "successful reconstruction must not erase the older effect: {replay:?}" + ); + fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::Metadata); + assert_eq!(metadata.read_by_name(&name).unwrap().unwrap(), changed); + assert_eq!(persisted_files(&artifacts), effected); + fixture.resume_same_operation(request, pending.request.operation_id); +} + #[test] #[serial_test::serial(worker_allocation)] fn profile_restore_timeout_attachment_revocation_cannot_reactivate_old_candidate() {