diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index e61b4fe4f..c57cfa6a8 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -21,6 +21,8 @@ default = ["ws-server", "fs-store"] fs-store = ["dep:fs4"] # Offline fixture tooling only; never enabled by the product defaults. test-support = ["fs-store", "ws-server"] +# Explicit process-test synchronization; independent of offline fixture tooling. +restore-checkpoints = ["fs-store"] http-server = ["dep:axum", "dep:tower", "dep:reqwest"] ws-server = ["http-server", "axum/ws", "tokio/sync"] diff --git a/crates/worker-runtime/README.md b/crates/worker-runtime/README.md index e4f041745..37c4db1f0 100644 --- a/crates/worker-runtime/README.md +++ b/crates/worker-runtime/README.md @@ -39,3 +39,30 @@ not stripped or rewritten. Portability/secret-marker checks do not replace review of the supplied artificial conversation or its independently verified hashes. The installer refuses any destination content other than initialized `auth.toml`; interrupted publication is fail-closed, not an overwrite/recovery facility. + +## Owned-process Restore checkpoints + +The independent, nondefault `restore-checkpoints` feature exists only for explicit +process tests in the shared `tests/e2e` harness. Offline `test-support` does not enable +it. Normal binaries reject `--test-restore-checkpoint`; no environment variable or +HTTP endpoint activates this facility. Do not deploy a checkpoint-enabled binary +as a replacement for the default product. + +A test-enabled Runtime accepts `--test-restore-checkpoint ` with a +bounded regular JSON file selecting a loopback socket, Worker ID, one stage and a +diagnostic nonce. The nonce is only a handshake label, never execution identity, +request authority or routing state. The one-shot stages are real prepared candidate +with `NotCommitted`, and durable `Committed` before activation/terminal receipt. +Before acknowledging, the existing bounded canonical reader validates the exact +Worker/operation/decision, syncs its file/directories and requires no saved receipt. +No pending JSON is fabricated and no lifecycle work is skipped. + +The Runtime sends bounded newline-delimited JSON evidence, then waits at most eight +seconds for byte `R`. EOF, invalid reply, unavailable peer or deadline releases the +wait into the original operation. The Runtime-state mutex is not held across that +wait; existing per-Worker ownership still covers the actual candidate and cleanup. +The harness verifies the ack, kills/reaps only its owned child and inspects the +canonical aggregate offline before restarting the same roots/identity without the +checkpoint flag or another fixture install. These tests are a separate cfg/binary +lane from featureless product regression; they do not certify arbitrary OS failure, +external power loss or a real model call. diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index bb72c71e3..c00c5fe77 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -16,6 +16,8 @@ pub mod management; pub mod observation; pub mod profile_archive; pub mod resource; +#[cfg(feature = "restore-checkpoints")] +pub mod restore_checkpoints; #[cfg(test)] #[path = "../../../tests/support/restore_environment.rs"] mod restore_test_environment; diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index d8b985f43..e40577695 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -117,6 +117,12 @@ fn run() -> Result<(), ProcessError> { let listener = tokio::net::TcpListener::bind(config.http.bind_addr).await?; let local_addr = listener.local_addr()?; let worker_runtime = build_runtime(&config)?; + #[cfg(feature = "restore-checkpoints")] + if let Some(checkpoint) = &config.test_restore_checkpoint { + worker_runtime + .install_test_restore_checkpoint(checkpoint.clone()) + .map_err(ProcessError::Runtime)?; + } eprintln!( "yoi-runtime listening on {local_addr}; intended client is a trusted backend/proxy, not a browser" ); @@ -391,6 +397,21 @@ where let (flag, inline_value) = split_flag_value(arg)?; match flag.as_str() { "--help" | "-h" => return Ok(None), + #[cfg(feature = "restore-checkpoints")] + "--test-restore-checkpoint" => { + let path = take_value(&flag, inline_value, &mut args)?; + if config.test_restore_checkpoint.is_some() { + return Err(ProcessError::usage( + "--test-restore-checkpoint may only be specified once".into(), + )); + } + config.test_restore_checkpoint = Some( + worker_runtime::restore_checkpoints::RestoreCheckpointConfig::from_path( + Path::new(&path), + ) + .map_err(ProcessError::Runtime)?, + ); + } "--bind" => { let value = take_value(&flag, inline_value, &mut args)?; config.http.bind_addr = value.parse::().map_err(|error| { @@ -465,6 +486,12 @@ where } apply_store_selection(&mut config); + #[cfg(feature = "restore-checkpoints")] + if config.test_restore_checkpoint.is_some() && config.no_store { + return Err(ProcessError::usage( + "--test-restore-checkpoint requires an fs Runtime store".into(), + )); + } Ok(Some(config)) } @@ -521,6 +548,8 @@ struct ProcessConfig { no_store: bool, backend_resource_endpoint: Option, backend_resource_token: Option, + #[cfg(feature = "restore-checkpoints")] + test_restore_checkpoint: Option, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -542,6 +571,8 @@ impl ProcessConfig { no_store: false, backend_resource_endpoint: None, backend_resource_token: None, + #[cfg(feature = "restore-checkpoints")] + test_restore_checkpoint: None, }) } @@ -1227,6 +1258,47 @@ Auth commands: mod tests { use super::*; + #[test] + fn checkpoint_cli_is_absent_by_default() { + let config = parse_args([] as [&str; 0]).unwrap().unwrap(); + #[cfg(feature = "restore-checkpoints")] + assert!(config.test_restore_checkpoint.is_none()); + #[cfg(not(feature = "restore-checkpoints"))] + { + let _ = config; + let error = parse_args(["--test-restore-checkpoint", "never-opened.json"]).unwrap_err(); + assert!(error.to_string().contains("unknown argument")); + } + } + + #[cfg(feature = "restore-checkpoints")] + #[test] + fn checkpoint_cli_requires_valid_explicit_file_and_fs_store() { + use worker_runtime::restore_checkpoints::{ + RestoreCheckpointConfig, RestoreCheckpointStage, + }; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("checkpoint.json"); + let checkpoint = RestoreCheckpointConfig { + endpoint: "127.0.0.1:1234".parse().unwrap(), + worker_id: protocol::WorkerId::now_v7(), + stage: RestoreCheckpointStage::CommittedPreterminal, + nonce: "cli-test".into(), + }; + std::fs::write(&path, serde_json::to_vec(&checkpoint).unwrap()).unwrap(); + let args = ["--test-restore-checkpoint", path.to_str().unwrap()]; + let config = parse_args(args).unwrap().unwrap(); + assert_eq!(config.test_restore_checkpoint, Some(checkpoint)); + assert!(parse_args([args[0], args[1], "--no-store"]).is_err()); + assert!(parse_args([args[0], args[1], args[0], args[1]]).is_err()); + assert!(parse_args([args[0]]).is_err()); + std::fs::write(&path, b"private-invalid-body").unwrap(); + let error = parse_args(args).unwrap_err().to_string(); + assert!(!error.contains("private-invalid-body")); + std::fs::remove_file(&path).unwrap(); + assert!(parse_args(args).is_err()); + } + #[test] fn help_does_not_start_server() { assert!(parse_args(["--help"]).unwrap().is_none()); diff --git a/crates/worker-runtime/src/restore_checkpoints.rs b/crates/worker-runtime/src/restore_checkpoints.rs new file mode 100644 index 000000000..91321afa6 --- /dev/null +++ b/crates/worker-runtime/src/restore_checkpoints.rs @@ -0,0 +1,143 @@ +//! Explicit, ephemeral synchronization for owned Runtime process tests only. +//! +//! The nonce authenticates no Runtime action and is never a routing identity. +//! Socket failure, EOF and the fixed deadline only release the gate; the lifecycle +//! owner continues its original operation and remains responsible for cleanup. + +use std::fs::OpenOptions; +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpStream}; +use std::path::Path; +use std::time::{Duration, Instant}; + +use protocol::WorkerId; +use serde::{Deserialize, Serialize}; + +use crate::error::RuntimeError; +use crate::execution::WorkerLifecycleOperationId; + +pub const RESTORE_CHECKPOINT_TIMEOUT: Duration = Duration::from_secs(8); +pub const MAX_RESTORE_CHECKPOINT_CONFIG_BYTES: usize = 4096; +pub const MAX_RESTORE_CHECKPOINT_ACK_BYTES: usize = 4096; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RestoreCheckpointStage { + PreparedNotCommitted, + CommittedPreterminal, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RestoreCheckpointCommit { + NotCommitted, + Committed, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RestoreCheckpointConfig { + pub endpoint: SocketAddr, + pub worker_id: WorkerId, + pub stage: RestoreCheckpointStage, + pub nonce: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RestoreCheckpointAck { + pub nonce: String, + pub stage: RestoreCheckpointStage, + pub runtime_id: String, + pub worker_id: WorkerId, + pub operation_id: WorkerLifecycleOperationId, + pub commit: RestoreCheckpointCommit, + pub receipt_absent: bool, + pub pid: u32, +} + +fn invalid_config() -> RuntimeError { + // Do not include serde errors, paths or input contents in process diagnostics. + RuntimeError::InvalidRequest("invalid test restore checkpoint configuration".into()) +} + +impl RestoreCheckpointConfig { + pub fn validate(&self) -> Result<(), RuntimeError> { + if !self.endpoint.ip().is_loopback() + || self.endpoint.port() == 0 + || self.nonce.is_empty() + || self.nonce.len() > 256 + || self.nonce.chars().any(char::is_control) + { + return Err(invalid_config()); + } + Ok(()) + } + + pub fn from_path(path: &Path) -> Result { + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + // A FIFO must not stall startup before its regular-file check. + options.custom_flags(libc::O_NONBLOCK | libc::O_NOFOLLOW); + } + let file = options.open(path).map_err(|_| invalid_config())?; + let metadata = file.metadata().map_err(|_| invalid_config())?; + if !metadata.is_file() || metadata.len() > MAX_RESTORE_CHECKPOINT_CONFIG_BYTES as u64 { + return Err(invalid_config()); + } + let mut bytes = Vec::new(); + file.take(MAX_RESTORE_CHECKPOINT_CONFIG_BYTES as u64 + 1) + .read_to_end(&mut bytes) + .map_err(|_| invalid_config())?; + if bytes.len() > MAX_RESTORE_CHECKPOINT_CONFIG_BYTES { + return Err(invalid_config()); + } + let config: Self = serde_json::from_slice(&bytes).map_err(|_| invalid_config())?; + config.validate()?; + Ok(config) + } +} + +pub(crate) fn wait_for_release(config: &RestoreCheckpointConfig, ack: &RestoreCheckpointAck) { + let _ = exchange(config, ack); +} + +fn remaining(deadline: Instant) -> std::io::Result { + deadline + .checked_duration_since(Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .ok_or_else(|| std::io::Error::from(std::io::ErrorKind::TimedOut)) +} + +fn exchange(config: &RestoreCheckpointConfig, ack: &RestoreCheckpointAck) -> std::io::Result<()> { + let deadline = Instant::now() + RESTORE_CHECKPOINT_TIMEOUT; + let mut bytes = serde_json::to_vec(ack)?; + bytes.push(b'\n'); + if bytes.len() > MAX_RESTORE_CHECKPOINT_ACK_BYTES { + return Err(std::io::Error::from(std::io::ErrorKind::InvalidData)); + } + let mut stream = TcpStream::connect_timeout(&config.endpoint, remaining(deadline)?)?; + // Each syscall shares one budget; partial writes cannot reset the deadline. + let mut pending = bytes.as_slice(); + while !pending.is_empty() { + stream.set_write_timeout(Some(remaining(deadline)?))?; + let written = stream.write(pending)?; + if written == 0 { + return Err(std::io::Error::from(std::io::ErrorKind::WriteZero)); + } + pending = &pending[written..]; + } + stream.set_read_timeout(Some(remaining(deadline)?))?; + let mut release = [0]; + stream.read_exact(&mut release)?; + if release != [b'R'] { + return Err(std::io::Error::from(std::io::ErrorKind::InvalidData)); + } + Ok(()) +} + +#[cfg(test)] +mod tests; diff --git a/crates/worker-runtime/src/restore_checkpoints/tests.rs b/crates/worker-runtime/src/restore_checkpoints/tests.rs new file mode 100644 index 000000000..16735224d --- /dev/null +++ b/crates/worker-runtime/src/restore_checkpoints/tests.rs @@ -0,0 +1,139 @@ +use super::*; +use std::io::{BufRead, BufReader}; +use std::net::TcpListener; +use std::thread; + +fn config(endpoint: SocketAddr) -> RestoreCheckpointConfig { + RestoreCheckpointConfig { + endpoint, + worker_id: WorkerId::now_v7(), + stage: RestoreCheckpointStage::PreparedNotCommitted, + nonce: "checkpoint-test-handshake".into(), + } +} + +fn ack(config: &RestoreCheckpointConfig) -> RestoreCheckpointAck { + RestoreCheckpointAck { + nonce: config.nonce.clone(), + stage: config.stage, + runtime_id: "checkpoint-runtime".into(), + worker_id: config.worker_id, + operation_id: WorkerLifecycleOperationId::new(), + commit: RestoreCheckpointCommit::NotCommitted, + receipt_absent: true, + pid: std::process::id(), + } +} + +#[test] +fn config_rejects_nonliteral_nonloopback_bad_stage_unknown_fields_and_invalid_worker() { + let valid = serde_json::to_value(config("127.0.0.1:1234".parse().unwrap())).unwrap(); + for (field, value) in [ + ("endpoint", "localhost:1234"), + ("endpoint", "192.0.2.1:1234"), + ("endpoint", "0.0.0.0:1234"), + ("endpoint", "127.0.0.1:0"), + ("endpoint", "[::ffff:127.0.0.1]:1234"), + ("worker_id", "not-a-worker-id"), + ("stage", "after_terminal"), + ("nonce", ""), + ("nonce", "secret\nbody"), + ("timeout", "unbounded"), + ] { + let mut invalid = valid.clone(); + invalid[field] = value.into(); + let parsed = serde_json::from_value::(invalid); + assert!( + parsed.is_err() || parsed.unwrap().validate().is_err(), + "{field}" + ); + } + for endpoint in ["127.0.0.1:1234", "[::1]:1234"] { + config(endpoint.parse().unwrap()).validate().unwrap(); + } +} + +#[test] +fn config_file_bounds_and_errors_do_not_disclose_contents() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("config.json"); + assert!(RestoreCheckpointConfig::from_path(&path).is_err()); + std::fs::write(&path, vec![b'x'; MAX_RESTORE_CHECKPOINT_CONFIG_BYTES + 1]).unwrap(); + assert!(RestoreCheckpointConfig::from_path(&path).is_err()); + std::fs::write( + &path, + br#"{"nonce":"private-config-body","stage":"secret"}"#, + ) + .unwrap(); + let error = RestoreCheckpointConfig::from_path(&path) + .unwrap_err() + .to_string(); + assert!(!error.contains("private-config-body")); + assert!(!error.contains("secret")); + let mut valid = config("127.0.0.1:1234".parse().unwrap()); + valid.nonce = "x".repeat(257); + assert!(valid.validate().is_err()); + valid.nonce = "valid".into(); + std::fs::write(&path, serde_json::to_vec(&valid).unwrap()).unwrap(); + assert_eq!(RestoreCheckpointConfig::from_path(&path).unwrap(), valid); + assert!(RestoreCheckpointConfig::from_path(dir.path()).is_err()); +} + +#[test] +fn explicit_release_eof_and_invalid_release_close_the_owned_stream() { + for release in [Some(b'R'), None, Some(b'X')] { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let config = config(listener.local_addr().unwrap()); + let expected = ack(&config); + let sent = expected.clone(); + let task = thread::spawn(move || exchange(&config, &sent)); + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut line = String::new(); + BufReader::new(&mut stream).read_line(&mut line).unwrap(); + assert!(line.len() <= MAX_RESTORE_CHECKPOINT_ACK_BYTES); + assert_eq!( + serde_json::from_str::(&line).unwrap(), + expected + ); + if let Some(byte) = release { + stream.write_all(&[byte]).unwrap(); + } else { + stream.shutdown(std::net::Shutdown::Write).unwrap(); + } + assert_eq!(task.join().unwrap().is_ok(), release == Some(b'R')); + let mut byte = [0]; + assert_eq!(stream.read(&mut byte).unwrap(), 0); + } +} + +#[test] +fn absent_listener_resumes_without_waiting_or_retrying() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let config = config(listener.local_addr().unwrap()); + drop(listener); + let start = Instant::now(); + wait_for_release(&config, &ack(&config)); + assert!(start.elapsed() < Duration::from_secs(2)); +} + +#[test] +fn silent_peer_hits_fixed_deadline_and_stream_is_dropped() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let config = config(listener.local_addr().unwrap()); + let start = Instant::now(); + let task = thread::spawn(move || exchange(&config, &ack(&config))); + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(12))) + .unwrap(); + let mut reader = BufReader::new(&mut stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert!(task.join().unwrap().is_err()); + assert!(start.elapsed() >= RESTORE_CHECKPOINT_TIMEOUT - Duration::from_millis(100)); + assert!(start.elapsed() < RESTORE_CHECKPOINT_TIMEOUT + Duration::from_secs(3)); + assert_eq!(reader.read(&mut [0]).unwrap(), 0); +} diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index f55d359f8..4f9141cc1 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1,3 +1,5 @@ +#[cfg(feature = "restore-checkpoints")] +mod restore_checkpoint; mod restore_continuation; use crate::catalog::{ ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, ProfileSelector, @@ -2776,6 +2778,12 @@ impl Runtime { worker_state, workdir_attachments, } => { + #[cfg(feature = "restore-checkpoints")] + self.test_restore_checkpoint( + worker_ref, + operation_id, + crate::restore_checkpoints::RestoreCheckpointStage::PreparedNotCommitted, + ); let commit = self.commit_restored_worker_execution( worker_ref, worker_state.clone(), @@ -2784,6 +2792,12 @@ impl Runtime { ); match commit { Ok(_) => { + #[cfg(feature = "restore-checkpoints")] + self.test_restore_checkpoint( + worker_ref, + operation_id, + crate::restore_checkpoints::RestoreCheckpointStage::CommittedPreterminal, + ); if let Err(error) = backend.activate_restored_worker(operation_id, worker_ref) { @@ -5419,6 +5433,8 @@ impl std::fmt::Debug for BackendResourceClientRef { #[derive(Debug)] struct RuntimeState { + #[cfg(feature = "restore-checkpoints")] + restore_checkpoint: Option, display_name: Option, backend: RuntimeBackendKind, /// Backend-bound stable identity used for cross-boundary retention evidence. @@ -5466,6 +5482,8 @@ impl RuntimeState { display_name, backend: RuntimeBackendKind::Memory, runtime_identity: None, + #[cfg(feature = "restore-checkpoints")] + restore_checkpoint: None, persistence: RuntimePersistence::Memory, status: RuntimeStatus::Running, execution_backend: None, @@ -5500,6 +5518,8 @@ impl RuntimeState { display_name, backend: RuntimeBackendKind::FsStore, runtime_identity: None, + #[cfg(feature = "restore-checkpoints")] + restore_checkpoint: None, persistence: RuntimePersistence::Fs(store), status: RuntimeStatus::Running, execution_backend: None, @@ -5550,6 +5570,8 @@ impl RuntimeState { display_name: persisted.display_name, backend: RuntimeBackendKind::FsStore, runtime_identity: None, + #[cfg(feature = "restore-checkpoints")] + restore_checkpoint: None, persistence: RuntimePersistence::Fs(store), status: persisted.status, execution_backend: None, @@ -7553,6 +7575,10 @@ fn subscription_worker_state(status: WorkerStatus) -> SubscriptionWorkerState { #[cfg(test)] mod tests { + #[cfg(feature = "restore-checkpoints")] + mod restore_checkpoint_tests { + include!("runtime/restore_checkpoint_tests.rs"); + } use super::*; use crate::catalog::{ ConfigBundleRef, MaterializerKind, ProfileSelector, RepositoryMaterializationContext, diff --git a/crates/worker-runtime/src/runtime/restore_checkpoint.rs b/crates/worker-runtime/src/runtime/restore_checkpoint.rs new file mode 100644 index 000000000..96da2eeb9 --- /dev/null +++ b/crates/worker-runtime/src/runtime/restore_checkpoint.rs @@ -0,0 +1,111 @@ +use super::*; +use crate::restore_checkpoints::{ + RestoreCheckpointAck, RestoreCheckpointCommit, RestoreCheckpointConfig, RestoreCheckpointStage, + wait_for_release, +}; + +impl Runtime { + /// Install an explicitly selected, one-shot process-test gate after normal startup. + /// It is not persisted or consulted by admission, routing or recovery authority. + pub fn install_test_restore_checkpoint( + &self, + config: RestoreCheckpointConfig, + ) -> Result<(), RuntimeError> { + config.validate()?; + let mut state = self.lock()?; + if !matches!(state.persistence, RuntimePersistence::Fs(_)) { + return Err(RuntimeError::InvalidRequest( + "test restore checkpoint requires an fs Runtime store".into(), + )); + } + state.restore_checkpoint = Some(config); + Ok(()) + } + + // The lifecycle caller holds the per-Worker operation lock. No Runtime-state + // mutex is retained while waiting; the real backend candidate remains owned by + // the same operation throughout this synchronous call. + pub(super) fn test_restore_checkpoint( + &self, + worker_ref: &WorkerRef, + operation_id: WorkerLifecycleOperationId, + stage: RestoreCheckpointStage, + ) { + let config = { + let Ok(mut state) = self.lock() else { + return; + }; + let slot = &mut state.restore_checkpoint; + if !slot.as_ref().is_some_and(|config| { + config.worker_id == worker_ref.worker_id && config.stage == stage + }) { + return; + } + // Consume even if canonical verification or the handshake fails. + slot.take().expect("matched checkpoint") + }; + let Ok(ack) = self.restore_checkpoint_ack(&config, worker_ref, operation_id) else { + return; + }; + wait_for_release(&config, &ack); + } + + pub(super) fn restore_checkpoint_ack( + &self, + config: &RestoreCheckpointConfig, + worker_ref: &WorkerRef, + operation_id: WorkerLifecycleOperationId, + ) -> Result { + let invalid = || { + RuntimeError::InvalidRequest("test restore checkpoint authority did not match".into()) + }; + let state = self.lock()?; + let RuntimePersistence::Fs(store) = &state.persistence else { + return Err(invalid()); + }; + // The existing helper validates the exact bounded canonical aggregate and + // syncs that same file plus its owning directories. Never manufacture a + // journal marker or infer commit from an in-memory projection. + let record = store.read_durable_worker_record(worker_ref)?; + let PersistedWorkerExecutionState::ReconciliationRequired( + PersistedWorkerLifecycleOperation::Restore(pending), + ) = &record.execution_state + else { + return Err(invalid()); + }; + let expected = match config.stage { + RestoreCheckpointStage::PreparedNotCommitted => RestoreCommitFact::NotCommitted, + RestoreCheckpointStage::CommittedPreterminal => RestoreCommitFact::Committed, + }; + if config.worker_id != worker_ref.worker_id + || pending.operation_id != operation_id + || pending.commit != expected + { + return Err(invalid()); + } + let guard = &record.restore_guard; + let id = guard.active_request_id.as_ref().ok_or_else(invalid)?; + let owner = guard.owners.get(id).ok_or_else(invalid)?; + if owner.operation_id != operation_id + || owner.request.request_id != *id + || owner.receipt.is_some() + { + return Err(invalid()); + } + let runtime_id = state.runtime_identity.clone().ok_or_else(invalid)?; + Ok(RestoreCheckpointAck { + nonce: config.nonce.clone(), + stage: config.stage, + runtime_id, + worker_id: record.worker_id, + operation_id, + commit: match expected { + RestoreCommitFact::NotCommitted => RestoreCheckpointCommit::NotCommitted, + RestoreCommitFact::Committed => RestoreCheckpointCommit::Committed, + RestoreCommitFact::LegacyUnknown => unreachable!(), + }, + receipt_absent: true, + pid: std::process::id(), + }) + } +} diff --git a/crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs b/crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs new file mode 100644 index 000000000..7369d6c6b --- /dev/null +++ b/crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs @@ -0,0 +1,296 @@ +use super::*; +use crate::restore_checkpoints::{ + RestoreCheckpointAck, RestoreCheckpointCommit, RestoreCheckpointConfig, RestoreCheckpointStage, +}; +use std::io::{BufRead, BufReader, Write}; +use std::net::TcpListener; +use std::time::Duration; + +#[test] +fn checkpoint_stages_ack_canonical_decision_and_eof_resumes_same_operation_once() { + // Component scope: durable Runtime with a substituted execution backend. + // Real prepared Controllers and process death are owned by the E2E suite. + for (stage, release) in [ + (RestoreCheckpointStage::PreparedNotCommitted, "release"), + (RestoreCheckpointStage::CommittedPreterminal, "eof"), + (RestoreCheckpointStage::CommittedPreterminal, "timeout"), + ] { + let dir = tempfile::tempdir().unwrap(); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = Runtime::with_fs_store_and_execution_backend( + FsRuntimeStoreOptions::new(dir.path().join("runtime")) + .with_runtime_id("checkpoint-runtime"), + backend.clone(), + ) + .unwrap(); + runtime.bind_runtime_identity("checkpoint-runtime").unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("checkpoint original operation")) + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let request = runtime.test_restore_request(&worker.worker_ref); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + runtime + .install_test_restore_checkpoint(RestoreCheckpointConfig { + endpoint: listener.local_addr().unwrap(), + worker_id: worker.worker_id, + stage, + nonce: "runtime-stage-test".into(), + }) + .unwrap(); + let cloned = runtime.clone(); + let worker_ref = worker.worker_ref.clone(); + let sent_request = request.clone(); + let (sent, received) = std::sync::mpsc::channel(); + let task = std::thread::spawn(move || { + sent.send(cloned.restore_worker_operation(&worker_ref, sent_request)) + .unwrap(); + }); + listener.set_nonblocking(true).unwrap(); + let stream = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let listener = tokio::net::TcpListener::from_std(listener).unwrap(); + let (stream, _) = tokio::time::timeout(Duration::from_secs(5), listener.accept()) + .await + .expect("checkpoint did not connect") + .unwrap(); + stream.into_std().unwrap() + }); + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let mut reader = BufReader::new(stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + let ack: RestoreCheckpointAck = serde_json::from_str(&line).unwrap(); + assert_eq!(ack.stage, stage); + assert_eq!(ack.nonce, "runtime-stage-test"); + assert_eq!(ack.runtime_id, "checkpoint-runtime"); + assert_eq!(ack.worker_id, worker.worker_id); + assert_eq!(ack.pid, std::process::id()); + assert!(ack.receipt_absent); + assert_eq!( + ack.commit, + match stage { + RestoreCheckpointStage::PreparedNotCommitted => + RestoreCheckpointCommit::NotCommitted, + RestoreCheckpointStage::CommittedPreterminal => RestoreCheckpointCommit::Committed, + } + ); + assert_eq!( + backend.restore_operation_ids.lock().unwrap().as_slice(), + &[ack.operation_id] + ); + 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) + .unwrap(); + let id = canonical.restore_guard.active_request_id.as_ref().unwrap(); + assert_eq!(id, &request.request_id); + assert!(canonical.restore_guard.owners[id].receipt.is_none()); + // Release, EOF and timeout all resume the original lifecycle owner. + let start = std::time::Instant::now(); + if release == "release" { + reader.get_mut().write_all(b"R").unwrap(); + } else if release == "eof" { + reader + .get_mut() + .shutdown(std::net::Shutdown::Write) + .unwrap(); + } + let result = received + .recv_timeout(Duration::from_secs(12)) + .unwrap() + .unwrap(); + assert!(start.elapsed() < Duration::from_secs(11)); + task.join().unwrap(); + assert_eq!(result.state, WorkerRestoreState::Accepted); + assert_eq!( + backend.activation_operation_ids.lock().unwrap().as_slice(), + &[ack.operation_id] + ); + assert_eq!( + runtime + .restore_worker_operation(&worker.worker_ref, request) + .unwrap(), + result + ); + assert_eq!(*backend.restore_count.lock().unwrap(), 1); + assert!(runtime.lock().unwrap().restore_checkpoint.is_none()); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let next = runtime.test_restore_request(&worker.worker_ref); + assert_eq!( + runtime + .restore_worker_operation(&worker.worker_ref, next) + .unwrap() + .state, + WorkerRestoreState::Accepted + ); + assert_eq!(*backend.restore_count.lock().unwrap(), 2); + } +} + +#[test] +fn checkpoint_disabled_mismatched_or_unreachable_does_not_change_restore_result() { + for mode in ["disabled", "mismatch", "unreachable"] { + let dir = tempfile::tempdir().unwrap(); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = Runtime::with_fs_store_and_execution_backend( + FsRuntimeStoreOptions::new(dir.path().join("runtime")) + .with_runtime_id("checkpoint-runtime"), + backend.clone(), + ) + .unwrap(); + runtime.bind_runtime_identity("checkpoint-runtime").unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("checkpoint absent")) + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let request = runtime.test_restore_request(&worker.worker_ref); + if mode != "disabled" { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = listener.local_addr().unwrap(); + drop(listener); + runtime + .install_test_restore_checkpoint(RestoreCheckpointConfig { + endpoint, + worker_id: if mode == "mismatch" { + WorkerId::now_v7() + } else { + worker.worker_id + }, + stage: RestoreCheckpointStage::PreparedNotCommitted, + nonce: "absent-test".into(), + }) + .unwrap(); + } + let result = runtime + .restore_worker_operation(&worker.worker_ref, request.clone()) + .unwrap(); + assert_eq!(result.state, WorkerRestoreState::Accepted, "{mode}"); + assert_eq!( + runtime + .restore_worker_operation(&worker.worker_ref, request) + .unwrap(), + result + ); + assert_eq!(*backend.restore_count.lock().unwrap(), 1); + assert_eq!( + runtime.lock().unwrap().restore_checkpoint.is_some(), + mode == "mismatch" + ); + } +} + +#[test] +fn checkpoint_verification_requires_canonical_worker_operation_commit_and_absent_receipt() { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().join("runtime"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = Runtime::with_fs_store_and_execution_backend( + FsRuntimeStoreOptions::new(&root).with_runtime_id("checkpoint-runtime"), + backend.clone(), + ) + .unwrap(); + runtime.bind_runtime_identity("checkpoint-runtime").unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("checkpoint validation")) + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let request = runtime.test_restore_request(&worker.worker_ref); + *backend.restore_result.lock().unwrap() = Some(WorkerExecutionSpawnResult::Errored( + WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, "not yet connected"), + )); + assert_eq!( + runtime + .restore_worker_operation(&worker.worker_ref, request.clone()) + .unwrap() + .state, + WorkerRestoreState::ReconciliationRequired + ); + let operation = backend.restore_operation_ids.lock().unwrap()[0]; + let mut config = RestoreCheckpointConfig { + endpoint: "127.0.0.1:1234".parse().unwrap(), + worker_id: worker.worker_id, + stage: RestoreCheckpointStage::PreparedNotCommitted, + nonce: "verification-test".into(), + }; + let path = root + .join("workers") + .join(worker.worker_id.to_string()) + .join("worker.json"); + let before = std::fs::read(&path).unwrap(); + assert!( + runtime + .restore_checkpoint_ack(&config, &worker.worker_ref, operation) + .is_ok() + ); + assert!( + runtime + .restore_checkpoint_ack( + &config, + &worker.worker_ref, + WorkerLifecycleOperationId::new() + ) + .is_err() + ); + config.worker_id = WorkerId::now_v7(); + assert!( + runtime + .restore_checkpoint_ack(&config, &worker.worker_ref, operation) + .is_err() + ); + config.worker_id = worker.worker_id; + config.stage = RestoreCheckpointStage::CommittedPreterminal; + 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!( + runtime + .restore_checkpoint_ack(&config, &worker.worker_ref, operation) + .is_err() + ); + assert_eq!( + std::fs::read(&path).unwrap(), + before, + "verification must not write authority" + ); + // In-memory equality must not hide an unreadable canonical aggregate. + std::fs::write(&path, b"unreadable canonical aggregate").unwrap(); + assert!( + runtime + .restore_checkpoint_ack(&config, &worker.worker_ref, operation) + .is_err() + ); + assert_eq!( + std::fs::read(&path).unwrap(), + b"unreadable canonical aggregate" + ); + std::fs::write(&path, before).unwrap(); + backend.restore_result.lock().unwrap().take(); + assert_eq!( + runtime + .restore_worker_operation(&worker.worker_ref, request) + .unwrap() + .state, + WorkerRestoreState::Accepted + ); + assert!( + runtime + .restore_checkpoint_ack(&config, &worker.worker_ref, operation) + .is_err(), + "terminal operation cannot ack a preterminal checkpoint" + ); +} diff --git a/tests/e2e/Cargo.toml b/tests/e2e/Cargo.toml index 53b60d5e3..000155fe2 100644 --- a/tests/e2e/Cargo.toml +++ b/tests/e2e/Cargo.toml @@ -10,6 +10,7 @@ autotests = false default = [] e2e = [] fixture-support = ["yoi-workspace-server/test-support", "worker-runtime/test-support"] +restore-checkpoints = ["fixture-support", "worker-runtime/restore-checkpoints"] [dependencies] libc.workspace = true diff --git a/tests/e2e/README.md b/tests/e2e/README.md index a54eb8517..4622758ab 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -39,6 +39,24 @@ process終了ケースはそのAccepted receipt確認後に環境が所有する これは**commit済みかつterminal receipt保存後**のprocess終了であり、未commit候補やcommit直後・receipt前の終了証明ではない。固定Codex資産へSubmit/Notifyを送らないため、通常入力・Paused admissionの製品E2Eの証明でもない。検証jobはこの明示targetを選択する必要があり、通常suiteだけでは代替しない。 +## Opt-in Runtime checkpoint process試験 + +上のfeatureless製品4case(target内の既存seed projection試験1件とは別)に加え、`restore-checkpoints`を明示した実Runtimeで2つのlive windowを試験する。このfeatureはoffline fixture-supportと独立しており、通常RuntimeではCLI入口も環境変数による入口も無い。 + +```sh +# 同期を有効にしたRuntimeだけを専用targetへ。通常製品binaryを上書きしない。 +cargo build --locked -p worker-runtime --bin yoi-runtime \ + --features restore-checkpoints --target-dir target/e2e-checkpoint-products +cargo test --locked -p yoi-e2e --features e2e,restore-checkpoints \ + --test worker_restore checkpoint:: -- --test-threads=1 +``` + +`WorkerE2eEnvironment::use_checkpoint_runtime`はその環境のRuntime binary/hashと、隔離root内の明示configだけを選択する。global provider・環境変数・通常起動への暗黙設定は変更しない。configは一度だけCLIへ渡し、再起動時は渡さない。test側のlink featureをfeatureless製品binaryへ混入させない。 + +共有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は初回のみ。 + +明示解除・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は起動しない): ```sh diff --git a/tests/e2e/src/worker_environment.rs b/tests/e2e/src/worker_environment.rs index 749ef4c83..6ff2c40cb 100644 --- a/tests/e2e/src/worker_environment.rs +++ b/tests/e2e/src/worker_environment.rs @@ -266,6 +266,8 @@ pub struct WorkerE2eEnvironment { access_token: Option, client: Client, binaries: WorkerBinaryProviderInfo, + #[cfg(feature = "restore-checkpoints")] + restore_checkpoint_config: Option, } impl WorkerE2eEnvironment { @@ -331,11 +333,49 @@ impl WorkerE2eEnvironment { access_token: None, client, binaries, + #[cfg(feature = "restore-checkpoints")] + restore_checkpoint_config: None, }; environment.write_metadata("created")?; Ok(environment) } + /// Select the separately built test-enabled Runtime for this environment + /// only. The checkpoint file must already belong to its isolated root. + #[cfg(feature = "restore-checkpoints")] + pub fn use_checkpoint_runtime(&mut self, binary: PathBuf, config: &Path) -> Result<()> { + if self.runtime.is_some() || self.server.is_some() { + return Err(HarnessError::Protocol( + "checkpoint selection requires an unstarted environment".into(), + )); + } + let config = config.canonicalize()?; + if !config.starts_with(self.root.canonicalize()?) || !config.is_file() { + return Err(HarnessError::Protocol( + "checkpoint config is outside the owned environment".into(), + )); + } + let record = binary_record( + "yoi-runtime", + binary.canonicalize()?, + "restore-checkpoints-opt-in".into(), + None, + )?; + let runtime = self + .binaries + .binaries + .iter_mut() + .find(|record| record.name == "yoi-runtime") + .ok_or_else(|| HarnessError::Protocol("missing Runtime binary record".into()))?; + *runtime = record; + self.restore_checkpoint_config = Some(config); + fs::write( + self.artifacts_dir.join("binary-provider.json"), + serde_json::to_vec_pretty(&self.binaries)?, + )?; + Ok(()) + } + pub fn start(&mut self) -> Result<()> { self.start_backend()?; self.start_runtime() @@ -639,6 +679,10 @@ impl WorkerE2eEnvironment { "--fs-root", self.runtime_data_dir.to_string_lossy().as_ref(), ]); + #[cfg(feature = "restore-checkpoints")] + if let Some(config) = self.restore_checkpoint_config.take() { + command.arg("--test-restore-checkpoint").arg(config); + } let runtime = ManagedProcess::spawn( "runtime", command, @@ -684,6 +728,13 @@ impl WorkerE2eEnvironment { /// Kill/reap only the Runtime child owned by this environment, retaining roots /// and identities. Restart neither seeds storage nor rotates restore requests. pub fn crash_and_restart_runtime(&mut self) -> Result<()> { + self.crash_runtime()?; + self.start_runtime_process() + } + + /// Leave the owned Runtime reaped and offline for read-only canonical crash + /// inspection before restart. No root, identity or request is recreated. + pub fn crash_runtime(&mut self) -> Result<()> { let runtime = self .runtime .as_mut() @@ -708,6 +759,17 @@ impl WorkerE2eEnvironment { } self.runtime_url = None; self.runtime_shutdown = WorkerProcessShutdownOutcome::NotStarted; + Ok(()) + } + + /// Restart the previously initialized identity without seed or checkpoint flags. + #[cfg(feature = "restore-checkpoints")] + pub fn restart_runtime_after_crash(&mut self) -> Result<()> { + if self.runtime.is_some() || self.runtime_public_identity.is_none() { + return Err(HarnessError::Protocol( + "Runtime is not an offline initialized child".into(), + )); + } self.start_runtime_process() } @@ -794,6 +856,37 @@ impl WorkerE2eEnvironment { ) } + /// Build one authenticated call that may run in a scoped test thread while + /// the owner observes a checkpoint. The HTTP request deadline bounds joining + /// that thread even on test failure; credentials never enter artifacts. + #[cfg(feature = "restore-checkpoints")] + pub fn pending_restore_call( + &self, + worker_id: &str, + request_id: &str, + expected_observation_token: &str, + ) -> Result Result + Send + 'static + use<>> { + let token = self.access_token.as_deref().ok_or_else(|| { + HarnessError::Protocol("Backend access token is not initialized".into()) + })?; + let request = self + .client + .post(self.worker_url(worker_id, "restore")?) + .bearer_auth(token) + .timeout(STARTUP_WAIT) + .json(&serde_json::json!({"request_id": request_id, + "expected_observation_token": expected_observation_token})) + .build() + .map_err(|_| HarnessError::Protocol("build pending restore request".into()))?; + let client = self.client.clone(); + Ok(move || { + let response = client + .execute(request) + .map_err(|_| HarnessError::Protocol("pending restore transport failed".into()))?; + decode_api_response(response, "pending restore") + }) + } + /// Forward one unchanged public restore request through an owned loopback /// relay, consume its complete upstream response, and close without delivering /// any response bytes. The returned JSON is relay evidence, not caller success. diff --git a/tests/e2e/tests/worker_restore.rs b/tests/e2e/tests/worker_restore.rs index 75ef29cd2..accb62ed8 100644 --- a/tests/e2e/tests/worker_restore.rs +++ b/tests/e2e/tests/worker_restore.rs @@ -1,4 +1,7 @@ //! Fixed, externally supplied Worker restore. Never generates a fixture or sends model input. +#[cfg(feature = "restore-checkpoints")] +#[path = "worker_restore/checkpoint.rs"] +mod checkpoint; #[path = "worker_restore/fixture.rs"] mod fixture; #[path = "worker_restore/seed.rs"] diff --git a/tests/e2e/tests/worker_restore/checkpoint.rs b/tests/e2e/tests/worker_restore/checkpoint.rs new file mode 100644 index 000000000..e55e815be --- /dev/null +++ b/tests/e2e/tests/worker_restore/checkpoint.rs @@ -0,0 +1,496 @@ +//! Opt-in real-process checkpoint cases; never substitute the featureless lane. +use super::*; +use std::collections::BTreeMap; +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::path::{Path, PathBuf}; +use worker_runtime::restore_checkpoints::{ + MAX_RESTORE_CHECKPOINT_ACK_BYTES, RestoreCheckpointAck, RestoreCheckpointCommit, + RestoreCheckpointConfig, RestoreCheckpointStage, +}; + +struct Peer { + listener: TcpListener, + config: RestoreCheckpointConfig, + connection: Option, +} + +impl Peer { + fn configure( + environment: &mut WorkerE2eEnvironment, + fixture: &Fixture, + stage: RestoreCheckpointStage, + ) -> Result { + let listener = TcpListener::bind("127.0.0.1:0")?; + listener.set_nonblocking(true)?; + let config = RestoreCheckpointConfig { + endpoint: listener.local_addr()?, + worker_id: fixture::support(fixture.manifest.worker_id.parse())?, + stage, + nonce: format!( + "owned-checkpoint-{}-{}", + std::process::id(), + listener.local_addr()?.port() + ), + }; + let path = environment.root.join("restore-checkpoint.json"); + fs::write(&path, serde_json::to_vec(&config)?)?; + let provider = yoi_e2e::worker_binary_provider_info()?; + let binary = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../target/e2e-checkpoint-products") + .join(provider.profile) + .join("yoi-runtime"); + environment.use_checkpoint_runtime(binary, &path)?; + Ok(Self { + listener, + config, + connection: None, + }) + } + + fn acknowledge(&mut self, environment: &WorkerE2eEnvironment) -> Result { + let deadline = Instant::now() + Duration::from_secs(15); + let (mut stream, address) = loop { + match self.listener.accept() { + Ok(value) => break value, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + if Instant::now() >= deadline { + return Err(failure("checkpoint ack deadline exceeded")); + } + std::thread::sleep(Duration::from_millis(10)); + } + Err(error) => return Err(error.into()), + } + }; + if !address.ip().is_loopback() { + return Err(failure("non-loopback checkpoint peer")); + } + let mut bytes = Vec::new(); + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(failure("checkpoint ack read deadline exceeded")); + } + stream.set_read_timeout(Some(remaining))?; + let mut byte = [0]; + stream.read_exact(&mut byte)?; + if byte == [b'\n'] { + break; + } + bytes.push(byte[0]); + if bytes.len() > MAX_RESTORE_CHECKPOINT_ACK_BYTES { + return Err(failure("checkpoint ack oversized")); + } + } + let ack: RestoreCheckpointAck = serde_json::from_slice(&bytes)?; + if ack.nonce != self.config.nonce + || ack.stage != self.config.stage + || ack.worker_id != self.config.worker_id + || ack.runtime_id != environment.runtime_id + || Some(ack.pid) != environment.runtime_pid() + || !ack.receipt_absent + { + return Err(failure( + "checkpoint ack does not match owned process and stage", + )); + } + fs::write( + environment.artifacts_dir.join("checkpoint-ack.json"), + serde_json::to_vec_pretty(&ack)?, + )?; + self.connection = Some(stream); + Ok(ack) + } + + fn release(&mut self) -> Result<()> { + let mut stream = self + .connection + .take() + .ok_or_else(|| failure("checkpoint not acknowledged"))?; + stream.set_write_timeout(Some(Duration::from_secs(1)))?; + stream.write_all(b"R")?; + Ok(()) + } +} + +fn aggregate_path(environment: &WorkerE2eEnvironment, fixture: &Fixture) -> PathBuf { + environment + .runtime_data_dir + .join("runtime/workers") + .join(&fixture.manifest.worker_id) +} + +fn canonical(environment: &WorkerE2eEnvironment, fixture: &Fixture) -> Result { + let mut bytes = Vec::new(); + fs::File::open(aggregate_path(environment, fixture).join("worker.json"))? + .take(8 * 1024 * 1024 + 1) + .read_to_end(&mut bytes)?; + if bytes.len() > 8 * 1024 * 1024 { + return Err(failure("canonical aggregate oversized")); + } + Ok(serde_json::from_slice(&bytes)?) +} + +fn saved_source_hashes( + environment: &WorkerE2eEnvironment, + fixture: &Fixture, +) -> Result> { + fn visit(root: &Path, path: &Path, files: &mut BTreeMap) -> Result<()> { + if files.len() > 128 { + return Err(failure("saved source file budget exceeded")); + } + let metadata = fs::symlink_metadata(path)?; + if metadata.file_type().is_symlink() { + return Err(failure("unexpected saved source symlink")); + } + if metadata.is_dir() { + for entry in fs::read_dir(path)? { + visit(root, &entry?.path(), files)?; + } + } else if metadata.is_file() { + if metadata.len() > 8 * 1024 * 1024 { + return Err(failure("saved source byte budget exceeded")); + } + use sha2::{Digest, Sha256}; + let digest: String = Sha256::digest(fs::read(path)?) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + files.insert( + path.strip_prefix(root) + .unwrap() + .to_string_lossy() + .into_owned(), + digest, + ); + } else { + return Err(failure("unexpected saved source entry type")); + } + Ok(()) + } + let root = aggregate_path(environment, fixture); + let mut files = BTreeMap::new(); + for name in ["metadata.json", "session"] { + visit(&root, &root.join(name), &mut files)?; + } + if files.len() < 2 { + return Err(failure("saved source inventory is incomplete")); + } + Ok(files) +} + +fn restore_runs(environment: &WorkerE2eEnvironment, fixture: &Fixture) -> Result> { + let runs = aggregate_path(environment, fixture).join("runs"); + if !runs.exists() { + return Ok(Vec::new()); + } + let mut names = Vec::new(); + for entry in fs::read_dir(runs)? { + let entry = entry?; + if entry.file_name().to_string_lossy().starts_with("restore-") { + if !entry.file_type()?.is_dir() { + return Err(failure("restore run is not a directory")); + } + names.push(entry.file_name().to_string_lossy().into_owned()); + } + } + names.sort(); + Ok(names) +} + +#[test] +fn acknowledged_not_committed_candidate_process_death_recovers_same_request_as_rollback() +-> Result<()> { + crash_window(RestoreCheckpointStage::PreparedNotCommitted) +} + +#[test] +fn acknowledged_committed_preterminal_process_death_reconstructs_same_decision() -> Result<()> { + crash_window(RestoreCheckpointStage::CommittedPreterminal) +} + +fn crash_window(stage: RestoreCheckpointStage) -> Result<()> { + let fixture = Fixture::load()?; + let result = (|| { + let mut environment = WorkerE2eEnvironment::new()?; + let mut peer = Peer::configure(&mut environment, &fixture, stage)?; + let mut installed = seed::start_in(&fixture, environment)?; + let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?; + let token = stopped + .restore_observation_token + .ok_or_else(|| failure("missing original observation"))?; + let original_sources = saved_source_hashes(&installed.environment, &fixture)?; + let request_id = "owned-process-checkpoint-request"; + let call = installed.environment.pending_restore_call( + &fixture.manifest.worker_id, + request_id, + &token, + )?; + let ack = std::thread::scope(|scope| -> Result { + let request = scope.spawn(call); + let ack = peer.acknowledge(&installed.environment)?; + installed.environment.crash_runtime()?; + // With the actual child reaped there can be no deadline race while + // inspecting the exact decision and nonterminal owner at process death. + let saved = canonical(&installed.environment, &fixture)?; + let op = saved + .pointer("/execution_state/execution/intent") + .ok_or_else(|| failure("checkpoint lacks canonical pending operation"))?; + let owner = &saved["restore_guard"]["owners"][request_id]; + assert_eq!(op["operation_id"], serde_json::to_value(ack.operation_id)?); + assert_eq!(owner["operation_id"], op["operation_id"]); + assert!(owner["receipt"].is_null()); + assert_eq!(owner["request"]["expected_observation_token"], token); + assert_eq!(op["source_repair_possible"], false); + let committed = stage == RestoreCheckpointStage::CommittedPreterminal; + assert_eq!( + op["commit"], + if committed { + "committed" + } else { + "not_committed" + } + ); + assert_eq!( + ack.commit, + if committed { + RestoreCheckpointCommit::Committed + } else { + RestoreCheckpointCommit::NotCommitted + } + ); + assert_eq!( + saved_source_hashes(&installed.environment, &fixture)?, + original_sources + ); + assert_eq!( + restore_runs(&installed.environment, &fixture)?, + vec![format!("restore-{}", ack.operation_id)] + ); + fs::write( + installed + .environment + .artifacts_dir + .join("checkpoint-at-death.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "runtime_reaped_before_inspection": installed.environment.runtime_pid().is_none(), + "operation_id": op["operation_id"], "commit": op["commit"], + "receipt_absent": owner["receipt"].is_null(), + "source_repair_possible": op["source_repair_possible"], + "same_original_request_and_token": true, + "restore_runs": restore_runs(&installed.environment, &fixture)?, + "saved_source_hashes": saved_source_hashes(&installed.environment, &fixture)?, + }))?, + )?; + let outcome = request + .join() + .map_err(|_| failure("public restore caller panicked"))?; + if let Ok(value) = outcome { + let receipt: WorkerRestoreResponse = serde_json::from_value(value)?; + assert_ne!( + receipt.result.state, + WorkerRestoreState::Accepted, + "no caller Accepted may precede either checkpoint kill" + ); + } + installed.environment.restart_runtime_after_crash()?; + assert_ne!(installed.environment.runtime_pid(), Some(ack.pid)); + Ok(ack) + })?; + drop(peer); + let recovered = installed.environment.restore_worker( + &fixture.manifest.worker_id, + request_id, + &token, + )?; + let receipt: WorkerRestoreResponse = serde_json::from_value(recovered.clone())?; + let committed = stage == RestoreCheckpointStage::CommittedPreterminal; + 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 canonical owners"))?; + assert_eq!(owners.len(), 1); + assert_eq!( + owners[request_id]["operation_id"], + serde_json::to_value(ack.operation_id)? + ); + assert_eq!( + owners[request_id]["request"]["expected_observation_token"], + token + ); + assert_eq!( + owners[request_id]["receipt"]["state"], + if committed { "accepted" } else { "rolled_back" } + ); + let stale = installed.environment.restore_worker( + &fixture.manifest.worker_id, + "new-stale-checkpoint-intent", + &token, + ); + if !matches!(stale, Err(error) if error.to_string().contains("HTTP 409")) { + return Err(failure("fresh stale intent was not rejected")); + } + fs::write( + installed + .environment + .artifacts_dir + .join("checkpoint-recovery.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "ack": ack, "recovered_state": receipt.result.state, + "canonical_operation_id": owners[request_id]["operation_id"], + "owner_count": owners.len(), "same_original_request_and_token": true, + "saved_source_hashes_before": original_sources, + "saved_source_hashes_after": saved_source_hashes(&installed.environment, &fixture)?, + "restore_runs_after": restore_runs(&installed.environment, &fixture)?, + }))?, + )?; + 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()); + assert_eq!( + saved_source_hashes(&installed.environment, &fixture)?, + original_sources + ); + let report = installed.environment.cleanup()?; + assert!(report.cleanup_success && !report.fixture_root_exists_after); + Ok(()) + } + })(); + fixture.source.assert_unchanged()?; + result +} + +#[test] +fn acknowledged_checkpoint_peer_disconnect_resumes_original_operation_and_tears_down() -> Result<()> +{ + observer_completion(false) +} + +#[test] +fn acknowledged_checkpoint_explicit_release_resumes_original_operation_and_tears_down() -> Result<()> +{ + observer_completion(true) +} + +fn observer_completion(release: bool) -> Result<()> { + let fixture = Fixture::load()?; + let result = (|| { + let mut environment = WorkerE2eEnvironment::new()?; + let mut peer = Peer::configure( + &mut environment, + &fixture, + RestoreCheckpointStage::PreparedNotCommitted, + )?; + let installed = seed::start_in(&fixture, environment)?; + let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?; + let token = stopped + .restore_observation_token + .ok_or_else(|| failure("missing observation"))?; + let call = installed.environment.pending_restore_call( + &fixture.manifest.worker_id, + "peer-disconnect-request", + &token, + )?; + std::thread::scope(|scope| -> Result<()> { + let request = scope.spawn(call); + peer.acknowledge(&installed.environment)?; + // EOF models an abandoned observer, not a successful restore result. + if release { + peer.release()?; + } else { + drop(peer); + } + let value = request + .join() + .map_err(|_| failure("public caller panicked"))??; + let response: WorkerRestoreResponse = serde_json::from_value(value)?; + assert_eq!(response.result.state, WorkerRestoreState::Accepted); + Ok(()) + })?; + let mut subscription = WorkerSessionSubscription::connect( + &installed.environment, + &fixture.manifest.worker_id, + &installed.snapshot, + )?; + subscription.assert_alive()?; + stop_and_cleanup(installed, vec![subscription], &fixture) + })(); + fixture.source.assert_unchanged()?; + result +} + +#[test] +fn acknowledged_checkpoint_failed_observer_scope_has_bounded_owned_cleanup() -> Result<()> { + let fixture = Fixture::load()?; + let environment = WorkerE2eEnvironment::new()?; + let root = environment.root.clone(); + let artifacts = environment.artifacts_dir.clone(); + let started = Instant::now(); + let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| -> Result<()> { + let mut environment = environment; + let mut peer = Peer::configure( + &mut environment, + &fixture, + RestoreCheckpointStage::PreparedNotCommitted, + )?; + let installed = seed::start_in(&fixture, environment)?; + let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?; + let token = stopped + .restore_observation_token + .ok_or_else(|| failure("missing observation"))?; + let call = installed.environment.pending_restore_call( + &fixture.manifest.worker_id, + "failed-observer-request", + &token, + )?; + std::thread::scope(|scope| -> Result<()> { + scope.spawn(call); + peer.acknowledge(&installed.environment)?; + // Leave the peer connected while unwinding the scoped HTTP caller. + // The Runtime's own deadline must unblock it, then RAII must reap + // both owned processes and remove the environment, not wait forever. + panic!("intentional checkpoint observer failure"); + }) + })); + assert!(outcome.is_err(), "the observer failure was not exercised"); + assert!( + started.elapsed() < Duration::from_secs(45), + "owned cleanup exceeded scenario budget" + ); + assert!(!root.exists(), "failed observer left its isolated root"); + let cleanup: serde_json::Value = + serde_json::from_slice(&fs::read(artifacts.join("cleanup.json"))?)?; + assert_eq!(cleanup["cleanup_success"], true); + assert_eq!(cleanup["fixture_root_exists_after"], false); + fixture.source.assert_unchanged()?; + Ok(()) +} diff --git a/tests/e2e/tests/worker_restore/seed.rs b/tests/e2e/tests/worker_restore/seed.rs index afc7804e4..f95085b06 100644 --- a/tests/e2e/tests/worker_restore/seed.rs +++ b/tests/e2e/tests/worker_restore/seed.rs @@ -42,7 +42,10 @@ pub struct Installed { } pub fn start(fixture: &Fixture) -> Result { - let mut environment = WorkerE2eEnvironment::new()?; + start_in(fixture, WorkerE2eEnvironment::new()?) +} + +pub fn start_in(fixture: &Fixture, mut environment: WorkerE2eEnvironment) -> Result { super::phase(&environment, "fixture-copy-verify")?; let copy = environment.root.join("provided-fixture"); fixture.source.copy_to(©)?;