From b3209c2012f07a33cd87f436f78df41c6e88cefa Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 26 Oct 2026 19:38:24 +0900 Subject: [PATCH] test: verify real orphan pin repair across owned restore process death --- .../worker-runtime/src/restore_checkpoints.rs | 1 + crates/worker-runtime/src/runtime.rs | 6 + .../src/runtime/restore_checkpoint.rs | 3 +- .../src/runtime/restore_checkpoint_tests.rs | 12 +- .../worker_backend_restore_candidate_tests.rs | 34 +++ tests/e2e/README.md | 4 +- tests/e2e/tests/worker_restore/checkpoint.rs | 2 + tests/e2e/tests/worker_restore/repair.rs | 246 ++++++++++++++++++ tests/e2e/tests/worker_restore/seed.rs | 18 +- 9 files changed, 321 insertions(+), 5 deletions(-) create mode 100644 tests/e2e/tests/worker_restore/repair.rs diff --git a/crates/worker-runtime/src/restore_checkpoints.rs b/crates/worker-runtime/src/restore_checkpoints.rs index 91321afa6..0232eb976 100644 --- a/crates/worker-runtime/src/restore_checkpoints.rs +++ b/crates/worker-runtime/src/restore_checkpoints.rs @@ -25,6 +25,7 @@ pub const MAX_RESTORE_CHECKPOINT_ACK_BYTES: usize = 4096; pub enum RestoreCheckpointStage { PreparedNotCommitted, CommittedPreterminal, + ActivatedPreterminal, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 755509f7b..8eef2a877 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -2871,6 +2871,12 @@ impl Runtime { sanitize_worker_create_failure_message(&error), )); } + #[cfg(feature = "restore-checkpoints")] + self.test_restore_checkpoint( + worker_ref, + operation_id, + crate::restore_checkpoints::RestoreCheckpointStage::ActivatedPreterminal, + ); if let Err(error) = self.activate_restore_candidate_context(worker_ref, &candidate_context) { diff --git a/crates/worker-runtime/src/runtime/restore_checkpoint.rs b/crates/worker-runtime/src/runtime/restore_checkpoint.rs index 96da2eeb9..d94aa4fa1 100644 --- a/crates/worker-runtime/src/runtime/restore_checkpoint.rs +++ b/crates/worker-runtime/src/runtime/restore_checkpoint.rs @@ -75,7 +75,8 @@ impl Runtime { }; let expected = match config.stage { RestoreCheckpointStage::PreparedNotCommitted => RestoreCommitFact::NotCommitted, - RestoreCheckpointStage::CommittedPreterminal => RestoreCommitFact::Committed, + RestoreCheckpointStage::CommittedPreterminal + | RestoreCheckpointStage::ActivatedPreterminal => RestoreCommitFact::Committed, }; if config.worker_id != worker_ref.worker_id || pending.operation_id != operation_id diff --git a/crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs b/crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs index 7369d6c6b..8103c2fe0 100644 --- a/crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs +++ b/crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs @@ -14,6 +14,7 @@ fn checkpoint_stages_ack_canonical_decision_and_eof_resumes_same_operation_once( (RestoreCheckpointStage::PreparedNotCommitted, "release"), (RestoreCheckpointStage::CommittedPreterminal, "eof"), (RestoreCheckpointStage::CommittedPreterminal, "timeout"), + (RestoreCheckpointStage::ActivatedPreterminal, "release"), ] { let dir = tempfile::tempdir().unwrap(); let backend = Arc::new(TestExecutionBackend::default()); @@ -79,14 +80,19 @@ fn checkpoint_stages_ack_canonical_decision_and_eof_resumes_same_operation_once( match stage { RestoreCheckpointStage::PreparedNotCommitted => RestoreCheckpointCommit::NotCommitted, - RestoreCheckpointStage::CommittedPreterminal => RestoreCheckpointCommit::Committed, + RestoreCheckpointStage::CommittedPreterminal + | RestoreCheckpointStage::ActivatedPreterminal => RestoreCheckpointCommit::Committed, } ); assert_eq!( backend.restore_operation_ids.lock().unwrap().as_slice(), &[ack.operation_id] ); - assert!(backend.activation_operation_ids.lock().unwrap().is_empty()); + if stage == RestoreCheckpointStage::ActivatedPreterminal { + assert_eq!(backend.activation_operation_ids.lock().unwrap().as_slice(), &[ack.operation_id]); + } else { + assert!(backend.activation_operation_ids.lock().unwrap().is_empty()); + } assert!(received.try_recv().is_err()); let canonical = runtime_store(&runtime) .read_durable_worker_record(&worker.worker_ref) @@ -255,6 +261,8 @@ fn checkpoint_verification_requires_canonical_worker_operation_commit_and_absent .restore_checkpoint_ack(&config, &worker.worker_ref, operation) .is_err() ); + config.stage = RestoreCheckpointStage::ActivatedPreterminal; + assert!(runtime.restore_checkpoint_ack(&config, &worker.worker_ref, operation).is_err()); config.stage = RestoreCheckpointStage::PreparedNotCommitted; runtime_store(&runtime).fail_next_worker_reconcile_sync(); assert!( 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 690a4025f..894beee9b 100644 --- a/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs @@ -557,6 +557,40 @@ mod restore_candidate { } } + #[test] + #[serial_test::serial(worker_allocation)] + fn restore_run_reclamation_preserves_live_allocation_owner() { + let fixture = Fixture::no_model(); + let factory = fixture.backend.factory.clone(); + let request = fixture.request.clone(); + let run = factory + .worker_restore_run_dir(&request.worker_ref, request.operation_id) + .unwrap(); + fs::create_dir_all(&run).unwrap(); + fs::write(run.join("retained-owner"), b"live allocation").unwrap(); + let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&request.worker_ref); + let allocation = worker::runtime::worker_allocation::install_top_level( + name.clone(), std::process::id(), run.join("worker.sock"), + Vec::new(), fixture.segment_id, + ).unwrap(); + let failure = fixture.backend.run_on_adapter_runtime(async move { + match factory.prepare_restore(request).await { + Err(failure) => Ok(failure), + Ok(_) => Err("live allocation unexpectedly reclaimed".into()), + } + }).unwrap(); + assert_eq!(failure.stage, RuntimeRestoreStage::Controller); + assert_eq!(failure.cause, "restore run directory still has an allocation owner"); + let table = worker::runtime::worker_allocation::LockFileGuard::open( + &worker::runtime::worker_allocation::default_allocation_path().unwrap(), + ).unwrap(); + assert_eq!(table.data().find(&name).unwrap().pid, std::process::id()); + assert_eq!(fs::read(run.join("retained-owner")).unwrap(), b"live allocation"); + drop(table); + drop(allocation); + fixture.assert_quiet(); + } + #[test] #[serial_test::serial(worker_allocation)] fn session_availability_requires_attachable_current_controller() { diff --git a/tests/e2e/README.md b/tests/e2e/README.md index 4622758ab..857a28a09 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -41,7 +41,7 @@ process終了ケースはそのAccepted receipt確認後に環境が所有する ## Opt-in Runtime checkpoint process試験 -上のfeatureless製品4case（target内の既存seed projection試験1件とは別）に加え、`restore-checkpoints`を明示した実Runtimeで2つのlive windowを試験する。このfeatureはoffline fixture-supportと独立しており、通常RuntimeではCLI入口も環境変数による入口も無い。 +上のfeatureless製品4case（target内の既存seed projection試験1件とは別）に加え、`restore-checkpoints`を明示した実Runtimeでdecision前後とactivation後・terminal前のlive windowを試験する。このfeatureはoffline fixture-supportと独立しており、通常RuntimeではCLI入口も環境変数による入口も無い。 ```sh # 同期を有効にしたRuntimeだけを専用targetへ。通常製品binaryを上書きしない。 @@ -55,6 +55,8 @@ cargo test --locked -p yoi-e2e --features e2e,restore-checkpoints \ 共有harnessはloopback ackのstage/Worker/operation/PID/nonceを確認し、直ちに所有childをkill/reapする。**offline正本を確認してから**同roots/identity/trustを再起動するため、ack後のdeadlineとの競合を成功扱いしない。NotCommittedの実私有候補は同operationのcleanup-only→RolledBack/Stopped、Committed/preterminalは同じdecision→実protocol/full Session snapshot→Acceptedへ回収する。元のrequest/token、owner数1、同operation、stale fresh intent拒否、保存source hash、restore run path数、最終cleanupを検証する。JSONでpending状態を作らず、seedは初回のみ。 +orphan upload pinの代表repairは、隔離copyへ正規Session storeのupload/pin APIで作った未結付けpinを、実Profileのcommit後activationが解放する経路を使う。Prepared窓は未修復のsource保持、Activated窓は実修復bytes/hashの保持を区別し、kill/reap後に正本を読んでから再起動する。materialized headの固定資産を巻き戻さないため、この試験は先行repair済みNotCommittedのprocess証明ではない。Activated後のstale allocationは通常登録と同じallocation authorityで回収し、live ownerが残る場合はrun artifactsを回収しない。 + 明示解除・peer EOF・observer scopeの意図的panicも試験する。gateの8秒deadlineとscoped HTTP callの15秒deadlineが待機を制限し、通常operationの継続後に環境のRAIIがprocess/rootを回収する。これはtest scope失敗の回収証明であり、OSによるharness強制終了や電源断の一般的な保証ではない。原fixtureへのSubmit/Notifyや実モデル呼出しは行わない。checkpoint ack/at-death/recovery、kill/reap、binary/hashとcleanup artifactはfeaturelessログと分けて報告する。 通常のharness回帰試験（製品processは起動しない）: diff --git a/tests/e2e/tests/worker_restore/checkpoint.rs b/tests/e2e/tests/worker_restore/checkpoint.rs index e55e815be..312c5beaa 100644 --- a/tests/e2e/tests/worker_restore/checkpoint.rs +++ b/tests/e2e/tests/worker_restore/checkpoint.rs @@ -1,4 +1,6 @@ //! Opt-in real-process checkpoint cases; never substitute the featureless lane. +#[path = "repair.rs"] +mod repair; use super::*; use std::collections::BTreeMap; use std::io::{Read, Write}; diff --git a/tests/e2e/tests/worker_restore/repair.rs b/tests/e2e/tests/worker_restore/repair.rs new file mode 100644 index 000000000..9e502d642 --- /dev/null +++ b/tests/e2e/tests/worker_restore/repair.rs @@ -0,0 +1,246 @@ +//! Real orphan-pin repair on an installed fixed-asset copy, not a synthetic +//! pending head. Only the postcommit activation path can reconcile these pins. +use super::*; +use session_store::{Store, UploadedFileLimits, WorkerSessionStore}; + +const PIN_OWNER: &str = "interrupted-pin-before-queue-checkpoint"; +const UPLOAD: &[u8] = b"isolated orphan upload; never model input"; + +#[test] +fn orphan_pin_not_committed_process_death_preserves_unrepaired_source() -> Result<()> { + repair_crash(RestoreCheckpointStage::PreparedNotCommitted) +} + +#[test] +fn orphan_pin_activated_process_death_preserves_real_repair_and_committed_decision() -> Result<()> { + repair_crash(RestoreCheckpointStage::ActivatedPreterminal) +} + +fn repair_crash(stage: RestoreCheckpointStage) -> Result<()> { + let fixture = Fixture::load()?; + let result = (|| { + let mut environment = WorkerE2eEnvironment::new()?; + let mut peer = Peer::configure(&mut environment, &fixture, stage)?; + let session_id = fixture::support(fixture.manifest.session_id.parse())?; + let mut upload = None; + let mut installed = seed::start_in_with_source_setup(&fixture, environment, |aggregate| { + let store = fixture::support(WorkerSessionStore::new(aggregate.join("session")))?; + let reference = fixture::support(store.write_uploaded_file( + session_id, + "orphan.txt", + "text/plain", + UPLOAD, + UploadedFileLimits::default(), + ))?; + fixture::support(store.pin_uploaded_file(session_id, &reference, PIN_OWNER))?; + upload = Some(reference); + Ok(()) + })?; + let upload = upload.ok_or_else(|| failure("offline upload setup not called"))?; + let relative = format!("session/artifacts/paste/{}.file.json", upload.artifact_id); + let pin_path = aggregate_path(&installed.environment, &fixture).join(&relative); + let pin_before: serde_json::Value = serde_json::from_slice(&fs::read(&pin_path)?)?; + assert_eq!(pin_before["pending_owner_id"], PIN_OWNER); + let before = saved_source_hashes(&installed.environment, &fixture)?; + let initial = canonical(&installed.environment, &fixture)?; + let token = assert_single_state(&installed.environment, &fixture, "stopped")? + .restore_observation_token + .ok_or_else(|| failure("missing original token"))?; + let request_id = "real-orphan-repair-crash-request"; + let committed = stage == RestoreCheckpointStage::ActivatedPreterminal; + let call = installed.environment.pending_restore_call( + &fixture.manifest.worker_id, + request_id, + &token, + )?; + let (ack, at_death) = std::thread::scope(|scope| -> Result<_> { + let request = scope.spawn(call); + let ack = peer.acknowledge(&installed.environment)?; + installed.environment.crash_runtime()?; + let saved = canonical(&installed.environment, &fixture)?; + let op = &saved["execution_state"]["execution"]["intent"]; + let owner = &saved["restore_guard"]["owners"][request_id]; + assert_eq!(saved["restore_guard"]["active_request_id"], request_id); + assert_eq!(op["operation_id"], serde_json::to_value(ack.operation_id)?); + assert_eq!(owner["operation_id"], op["operation_id"]); + assert_eq!(owner["request"]["expected_observation_token"], token); + assert!(owner["receipt"].is_null()); + assert_eq!( + op["commit"], + if committed { + "committed" + } else { + "not_committed" + } + ); + assert_eq!( + ack.commit, + if committed { + RestoreCheckpointCommit::Committed + } else { + RestoreCheckpointCommit::NotCommitted + } + ); + // The durable Committed decision authorizes activation effects. The + // precommit repair callback is not used for a materialized head. + assert_eq!(op["source_repair_possible"], false); + let at_death = saved_source_hashes(&installed.environment, &fixture)?; + let pin_after: serde_json::Value = serde_json::from_slice(&fs::read(&pin_path)?)?; + let mut expected_pin = pin_before.clone(); + if committed { + expected_pin + .as_object_mut() + .unwrap() + .remove("pending_owner_id"); + } + assert_eq!( + pin_after, expected_pin, + "real repair changes only orphan ownership" + ); + if committed { + assert_ne!( + at_death[&relative], before[&relative], + "actual pin bytes must change" + ); + } else { + assert_eq!( + at_death, before, + "preparation cannot repair pin or metadata" + ); + } + // In both windows the original conversation is immutable. + for (path, hash) in &before { + if path == "session/session.json" || path.starts_with("session/segments/") { + assert_eq!(at_death.get(path), Some(hash), "{path}"); + } + } + let store = fixture::support(WorkerSessionStore::open_read_only( + aggregate_path(&installed.environment, &fixture).join("session"), + ))?; + assert_eq!( + fixture::support(store.read_uploaded_file(session_id, &upload))?, + UPLOAD + ); + assert_eq!( + restore_runs(&installed.environment, &fixture)?, + vec![format!("restore-{}", ack.operation_id)] + ); + fs::write( + installed + .environment + .artifacts_dir + .join("repair-at-death.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "ack":ack, "initial_execution":initial["execution_state"], + "canonical_commit":op["commit"], "source_repair_possible":op["source_repair_possible"], + "operation_id":op["operation_id"], "owner_operation_id":owner["operation_id"], + "receipt_absent":owner["receipt"].is_null(), "original_request_and_token":true, + "pin_pending_owner_before":pin_before["pending_owner_id"], + "pin_pending_owner_after":pin_after["pending_owner_id"], + "hashes_before":before, "hashes_at_death":at_death, + "restore_runs":restore_runs(&installed.environment, &fixture)?, + }))?, + )?; + let outcome = request + .join() + .map_err(|_| failure("restore caller panicked"))?; + if let Ok(value) = outcome { + let receipt: WorkerRestoreResponse = serde_json::from_value(value)?; + assert_ne!(receipt.result.state, WorkerRestoreState::Accepted); + } + installed.environment.restart_runtime_after_crash()?; + assert_ne!(installed.environment.runtime_pid(), Some(ack.pid)); + Ok((ack, at_death)) + })?; + drop(peer); + let recovered = installed.environment.restore_worker( + &fixture.manifest.worker_id, + request_id, + &token, + )?; + let receipt: WorkerRestoreResponse = serde_json::from_value(recovered.clone())?; + fs::write( + installed + .environment + .artifacts_dir + .join("repair-first-recovery.json"), + serde_json::to_vec_pretty(&recovered)?, + )?; + assert_eq!( + receipt.result.state, + if committed { + WorkerRestoreState::Accepted + } else { + WorkerRestoreState::RolledBack + } + ); + assert_eq!( + installed.environment.restore_worker( + &fixture.manifest.worker_id, + request_id, + &token + )?, + recovered + ); + let final_saved = canonical(&installed.environment, &fixture)?; + let owners = final_saved["restore_guard"]["owners"] + .as_object() + .ok_or_else(|| failure("missing owners"))?; + assert_eq!(owners.len(), 1); + let owner = &owners[request_id]; + assert_eq!( + owner["operation_id"], + serde_json::to_value(ack.operation_id)? + ); + assert_eq!(owner["request"]["expected_observation_token"], token); + assert_eq!( + owner["receipt"]["state"], + if committed { "accepted" } else { "rolled_back" } + ); + let after = saved_source_hashes(&installed.environment, &fixture)?; + assert_eq!( + after, at_death, + "restart/replay cannot restore the orphan pin or change saved source" + ); + let stale = installed.environment.restore_worker( + &fixture.manifest.worker_id, + "fresh-stale-repair-intent", + &token, + ); + assert!(matches!(stale, Err(error) if error.to_string().contains("HTTP 409"))); + fs::write( + installed + .environment + .artifacts_dir + .join("repair-recovery.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "ack":ack, "state":receipt.result.state, "diagnostics":receipt.result.diagnostics, + "operation_id":owner["operation_id"], "owner_count":owners.len(), + "hashes_after_recovery":after, "restore_runs":restore_runs(&installed.environment, &fixture)?, + "same_original_request_and_token":true, "fresh_stale_rejected":true, + }))?, + )?; + if committed { + assert_single_state(&installed.environment, &fixture, "idle")?; + assert_eq!( + restore_runs(&installed.environment, &fixture)?, + vec![format!("restore-{}", ack.operation_id)] + ); + let mut subscription = WorkerSessionSubscription::connect( + &installed.environment, + &fixture.manifest.worker_id, + &installed.snapshot, + )?; + subscription.assert_alive()?; + stop_and_cleanup(installed, vec![subscription], &fixture) + } else { + assert_single_state(&installed.environment, &fixture, "stopped")?; + assert!(restore_runs(&installed.environment, &fixture)?.is_empty()); + let report = installed.environment.cleanup()?; + assert!(report.cleanup_success && !report.fixture_root_exists_after); + Ok(()) + } + })(); + fixture.source.assert_unchanged()?; + result +} diff --git a/tests/e2e/tests/worker_restore/seed.rs b/tests/e2e/tests/worker_restore/seed.rs index f95085b06..3640db189 100644 --- a/tests/e2e/tests/worker_restore/seed.rs +++ b/tests/e2e/tests/worker_restore/seed.rs @@ -45,7 +45,17 @@ pub fn start(fixture: &Fixture) -> Result { start_in(fixture, WorkerE2eEnvironment::new()?) } -pub fn start_in(fixture: &Fixture, mut environment: WorkerE2eEnvironment) -> Result { +pub fn start_in(fixture: &Fixture, environment: WorkerE2eEnvironment) -> Result { + start_in_with_source_setup(fixture, environment, |_| Ok(())) +} + +// The callback runs only on the installed isolation copy, while Runtime is offline. +// Restart never calls the seed or this setup again. +pub fn start_in_with_source_setup( + fixture: &Fixture, + mut environment: WorkerE2eEnvironment, + setup: impl FnOnce(&Path) -> Result<()>, +) -> Result { super::phase(&environment, "fixture-copy-verify")?; let copy = environment.root.join("provided-fixture"); fixture.source.copy_to(©)?; @@ -114,6 +124,12 @@ pub fn start_in(fixture: &Fixture, mut environment: WorkerE2eEnvironment) -> Res backend_job: None, }; support(template.install(&options, request, &bootstrap.manifest))?; + setup( + &options + .root + .join("workers") + .join(template.worker_id().to_string()), + )?; let sessions = support(session_store::WorkerSessionStore::open_read_only( options .root