diff --git a/tests/e2e/README.md b/tests/e2e/README.md index 5766c9471..a54eb8517 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -22,7 +22,7 @@ 所有crateの`test-support`依存はテスト側の非default `fixture-support` featureだけで有効にする。製品binary providerは別Cargo invocationと専用`target/e2e-products`へ`yoi`・`yoi-server`・`yoi-runtime`を追加featureなしでビルドする。support test自身をこのtargetへビルドしてはならない。`YOI_E2E_BIN`等の明示binary overrideを使用する場合も、呼出し側が同じ分離を保証する。既存TUI用`e2e-test` featureは別の`target/e2e-tui-products`へビルドし、restore製品binaryへ混入しない。 -固定資産の正常restore・並行隔離E2E: +固定資産の正常restore・並行隔離・応答喪失回復E2E: ```sh cargo test -p yoi-e2e --features e2e,fixture-support --test worker_restore @@ -33,6 +33,12 @@ YOI_E2E_RESTORE_FIXTURE=target/fixture-candidate-v1 \ 標準資産は`tests/e2e/fixtures/worker-codex-v1`の`manifest.json`・`worker-template.json`・`expected.json`・`workspace-config.dcdl`。manifestの各asset hash/size、schemaと固定ID、生成元commit、modelを検証する。Workspace configのimport参照を黙って最新へ置換せず、Gitで生成元commitに対する`resources`と`crates/builtin-source`の一致を確認する(Git履歴がない場合も失敗)。移植したSessionの正規公開projectionを供給された会話行と照合し、その全snapshotを実executionのWebSocketと比較する。欠落fixture・hash不一致・未実行は成功扱いしない。 +応答喪失ケースは同じ`WorkerE2eEnvironment`内の単発loopback relayで公開restore要求をそのまま転送し、完全なupstream応答を受け取った後に呼出元へ0byteで切断する。呼出元のtransport失敗とrelay側のAccepted観測は別の事実として記録する。回収は元のrequest ID・観測tokenをそのまま使い、fresh intentへの古いtokenはHTTP409を要求する。 + +process終了ケースはそのAccepted receipt確認後に環境が所有するRuntime childだけをkill/reapし、同じroot・Runtime identity・trustから再起動する。seedの再実行・tokenの差替えはしない。再起動前後のcanonical owner全体のhash・operation ID・owner数が一致し、公開APIの保存結果とfull Session snapshotを回収できることを確認する。公開Worker投影の観測tokenは再起動で変わるため、そのfieldだけはcurrent detailとの一致を別に確認する。canonical診断はboundedな読取りのみで、要求・receiptの内容や認証情報はartifactへ保存しない。 + +これは**commit済みかつterminal receipt保存後**のprocess終了であり、未commit候補やcommit直後・receipt前の終了証明ではない。固定Codex資産へSubmit/Notifyを送らないため、通常入力・Paused admissionの製品E2Eの証明でもない。検証jobはこの明示targetを選択する必要があり、通常suiteだけでは代替しない。 + 通常のharness回帰試験(製品processは起動しない): ```sh diff --git a/tests/e2e/src/worker_environment.rs b/tests/e2e/src/worker_environment.rs index 1fbdc4b0b..749ef4c83 100644 --- a/tests/e2e/src/worker_environment.rs +++ b/tests/e2e/src/worker_environment.rs @@ -1,3 +1,6 @@ +#[path = "worker_environment/response_loss.rs"] +mod response_loss; + use std::fs::{self, File, OpenOptions}; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; @@ -571,12 +574,7 @@ impl WorkerE2eEnvironment { if self.runtime_public_identity.is_none() { self.initialize_runtime_identity(self.runtime_id.clone())?; } - let runtime_identity = self - .runtime_public_identity - .clone() - .expect("initialized identity"); let bundle_path = self.root.join("workspace-public-identity.json"); - let server_url = self.required_server_url()?.to_string(); let workspace_id = self.required_workspace_id()?.to_string(); let runtime_binary = self.binaries.binary("yoi-runtime")?; self.run_command( @@ -616,6 +614,20 @@ impl WorkerE2eEnvironment { seed(self)?; self.write_metadata("runtime-seeded")?; + self.start_runtime_process() + } + + fn start_runtime_process(&mut self) -> Result<()> { + if self.runtime.is_some() { + return Err(HarnessError::Protocol("Runtime is already running".into())); + } + let runtime_binary = self.binaries.binary("yoi-runtime")?; + let runtime_identity = self + .runtime_public_identity + .clone() + .ok_or_else(|| HarnessError::Protocol("Runtime identity is not initialized".into()))?; + let server_url = self.required_server_url()?.to_string(); + let workspace_id = self.required_workspace_id()?.to_string(); let mut command = self.isolated_command(&runtime_binary); command.args([ "--bind", @@ -669,6 +681,36 @@ impl WorkerE2eEnvironment { Ok(()) } + /// 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<()> { + let runtime = self + .runtime + .as_mut() + .ok_or_else(|| HarnessError::Protocol("Runtime is not running".into()))?; + let pid = runtime.id(); + let status = runtime.kill_and_reap(EXIT_WAIT)?; + runtime.join_readers_until(Instant::now() + EXIT_WAIT)?; + fs::write( + self.artifacts_dir.join("runtime-crash.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "owned_pid": pid, "reaped": true, "exit_status": status.to_string(), + "same_runtime_id": self.runtime_id, + }))?, + )?; + self.runtime.take(); + for stream in ["stdout", "stderr"] { + fs::rename( + self.artifacts_dir.join(format!("runtime.{stream}.log")), + self.artifacts_dir + .join(format!("runtime-before-crash.{stream}.log")), + )?; + } + self.runtime_url = None; + self.runtime_shutdown = WorkerProcessShutdownOutcome::NotStarted; + self.start_runtime_process() + } + pub fn assert_connected(&self) -> Result<()> { let server_url = self.required_server_url()?; let workspace_id = self.required_workspace_id()?; @@ -752,6 +794,38 @@ impl WorkerE2eEnvironment { ) } + /// 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. + pub fn restore_worker_losing_response( + &self, + worker_id: &str, + request_id: &str, + expected_observation_token: &str, + ) -> Result { + 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) + .header(reqwest::header::CONNECTION, "close") + .timeout(STARTUP_WAIT + EXIT_WAIT) + .json(&serde_json::json!({ + "request_id": request_id, + "expected_observation_token": expected_observation_token, + })) + .build() + .map_err(|_| HarnessError::Protocol("build response-loss request".into()))?; + let evidence = response_loss::drop_response(&self.client, request)?; + self.append_operation( + "restore response consumed by relay, caller transport failed", + 0, + )?; + Ok(evidence) + } + pub fn stop_worker(&self, worker_id: &str) -> Result { self.request_json( self.client diff --git a/tests/e2e/src/worker_environment/response_loss.rs b/tests/e2e/src/worker_environment/response_loss.rs new file mode 100644 index 000000000..ee6654c85 --- /dev/null +++ b/tests/e2e/src/worker_environment/response_loss.rs @@ -0,0 +1,198 @@ +//! A single bounded response-loss injection in the shared product harness. +use std::io::{Read, Write}; +use std::net::{IpAddr, TcpListener, TcpStream}; +use std::time::Instant; + +use reqwest::blocking::{Client, Request}; +use serde_json::Value; + +use super::{EXIT_WAIT, HTTP_WAIT_STEP, STARTUP_WAIT}; +use crate::{HarnessError, Result}; + +const MAX_HEADER: usize = 16 * 1024; +const MAX_MESSAGE: usize = 2 * 1024 * 1024; + +fn failure(message: &str) -> HarnessError { + HarnessError::Protocol(message.into()) +} + +pub(super) fn drop_response(client: &Client, mut request: Request) -> Result { + let host: IpAddr = request + .url() + .host_str() + .ok_or_else(|| failure("missing relay host"))? + .parse() + .map_err(|_| failure("relay target must be a literal loopback address"))?; + if !host.is_loopback() || request.url().scheme() != "http" { + return Err(failure( + "response-loss relay only permits owned loopback HTTP", + )); + } + let port = request + .url() + .port() + .ok_or_else(|| failure("missing relay port"))?; + let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))?; + listener.set_nonblocking(true)?; + let relay_address = listener.local_addr()?; + request + .url_mut() + .set_port(Some(relay_address.port())) + .map_err(|_| failure("set relay port"))?; + request + .url_mut() + .set_host(Some("127.0.0.1")) + .map_err(|_| failure("set relay host"))?; + std::thread::scope(|scope| { + let relay = scope.spawn(move || -> Result { + let deadline = Instant::now() + STARTUP_WAIT; + let mut downstream = loop { + match listener.accept() { + Ok((stream, peer)) if peer.ip().is_loopback() => break stream, + Ok(_) => return Err(failure("non-loopback relay peer")), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + if Instant::now() >= deadline { + return Err(failure("response-loss relay accept deadline exceeded")); + } + std::thread::sleep(HTTP_WAIT_STEP); + } + Err(error) => return Err(error.into()), + } + }; + let request_bytes = read_message(&mut DeadlineStream { + stream: &mut downstream, + deadline, + })?; + let mut upstream = TcpStream::connect_timeout(&(host, port).into(), EXIT_WAIT)?; + upstream.set_write_timeout(Some(EXIT_WAIT))?; + // The body, request identity and authorization bytes are forwarded + // unchanged. Neither request nor response headers enter diagnostics. + upstream.write_all(&request_bytes)?; + let response = read_message(&mut DeadlineStream { + stream: &mut upstream, + deadline, + })?; + let boundary = + header_end(&response).ok_or_else(|| failure("missing upstream header"))?; + let status = std::str::from_utf8(&response[..boundary]) + .ok() + .and_then(|header| header.lines().next()) + .and_then(|line| line.split_whitespace().nth(1)); + if status != Some("200") { + return Err(failure("restore relay upstream did not return HTTP 200")); + } + let value = serde_json::from_slice(&response[boundary..])?; + // Intentionally send zero bytes back; the caller must observe loss. + drop(downstream); + Ok(value) + }); + let caller_failed = client.execute(request).is_err(); + let evidence = relay + .join() + .map_err(|_| failure("response-loss relay panicked"))??; + if !caller_failed { + return Err(failure("caller unexpectedly received a response")); + } + Ok(evidence) + }) +} + +struct DeadlineStream<'a> { + stream: &'a mut TcpStream, + deadline: Instant, +} + +impl Read for DeadlineStream<'_> { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + let remaining = self.deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "relay deadline exceeded", + )); + } + self.stream.set_read_timeout(Some(remaining))?; + self.stream.read(buffer) + } +} + +fn header_end(bytes: &[u8]) -> Option { + bytes + .windows(4) + .position(|window| window == b"\r\n\r\n") + .map(|i| i + 4) +} + +fn read_message(stream: &mut impl Read) -> Result> { + let mut bytes = Vec::new(); + let mut buffer = [0; 4096]; + let mut expected = None; + loop { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(failure("relay peer closed before complete HTTP message")); + } + bytes.extend_from_slice(&buffer[..count]); + if bytes.len() > MAX_MESSAGE { + return Err(failure("relay HTTP byte budget exceeded")); + } + if expected.is_none() { + if let Some(boundary) = header_end(&bytes) { + if boundary > MAX_HEADER { + return Err(failure("relay HTTP header budget exceeded")); + } + let header = std::str::from_utf8(&bytes[..boundary]) + .map_err(|_| failure("invalid relay HTTP header"))?; + let mut lengths = header + .lines() + .filter_map(|line| line.split_once(':')) + .filter(|(name, _)| name.eq_ignore_ascii_case("content-length")) + .map(|(_, value)| value.trim().parse::()); + let length = lengths + .next() + .ok_or_else(|| failure("relay requires Content-Length"))? + .map_err(|_| failure("invalid relay Content-Length"))?; + if lengths.next().is_some() + || length > MAX_MESSAGE - boundary + || header + .lines() + .filter_map(|line| line.split_once(':')) + .any(|(name, _)| name.eq_ignore_ascii_case("transfer-encoding")) + { + return Err(failure("unsupported relay HTTP framing")); + } + expected = Some(boundary + length); + } else if bytes.len() > MAX_HEADER { + return Err(failure("relay HTTP header budget exceeded")); + } + } + if let Some(expected) = expected { + if bytes.len() == expected { + return Ok(bytes); + } + if bytes.len() > expected { + return Err(failure("unexpected relay trailing bytes")); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn framing_rejects_incomplete_ambiguous_or_unbounded_messages() { + for bytes in [ + b"HTTP/1.1 200 OK\r\n\r\n".as_slice(), + b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nshort", + b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nContent-Length: 0\r\n\r\n", + b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nTransfer-Encoding: chunked\r\n\r\n", + b"HTTP/1.1 200 OK\r\nContent-Length: 999999999\r\n\r\n", + ] { + assert!(read_message(&mut &bytes[..]).is_err()); + } + let valid = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}"; + assert_eq!(read_message(&mut &valid[..]).unwrap(), valid); + } +} diff --git a/tests/e2e/tests/worker_restore.rs b/tests/e2e/tests/worker_restore.rs index 3d1619847..75ef29cd2 100644 --- a/tests/e2e/tests/worker_restore.rs +++ b/tests/e2e/tests/worker_restore.rs @@ -143,6 +143,186 @@ fn concurrent_fixed_id_restores_keep_sessions_storage_and_cleanup_isolated() -> result } +#[test] +fn lost_http_response_recovers_exact_request_and_rejects_fresh_stale_intent() -> Result<()> { + response_loss_recovery(false) +} + +#[test] +fn committed_response_loss_survives_owned_runtime_process_kill_and_restart() -> Result<()> { + response_loss_recovery(true) +} + +fn response_loss_recovery(crash_runtime: bool) -> Result<()> { + let fixture = Fixture::load()?; + let result = (|| { + let mut installed = seed::start(&fixture)?; + let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?; + let token = stopped + .restore_observation_token + .as_deref() + .filter(|token| !token.is_empty()) + .ok_or_else(|| failure("missing stopped observation token"))?; + let request_id = "fixed-fixture-response-loss"; + phase( + &installed.environment, + "drop-http-response-after-complete-upstream", + )?; + let observed = installed.environment.restore_worker_losing_response( + &fixture.manifest.worker_id, + request_id, + token, + )?; + let receipt: WorkerRestoreResponse = serde_json::from_value(observed.clone())?; + if receipt.result.state != WorkerRestoreState::Accepted { + return Err(failure( + "upstream did not reach Accepted before response loss", + )); + } + fs::write( + installed + .environment + .artifacts_dir + .join("response-loss.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "caller_received_response": false, + "relay_upstream_state": receipt.result.state, + "request_id": request_id, + "crash_after_committed_terminal_receipt": crash_runtime, + }))?, + )?; + let owner_before = restore_owner_evidence(&installed.environment, &fixture, request_id)?; + if crash_runtime { + phase( + &installed.environment, + "kill-owned-runtime-after-committed-receipt", + )?; + installed.environment.crash_and_restart_runtime()?; + } + phase( + &installed.environment, + "same-request-original-token-recovery", + )?; + let recovered = + installed + .environment + .restore_worker(&fixture.manifest.worker_id, request_id, token)?; + if crash_runtime { + // The public response projects the current observation fence. Runtime + // startup invalidates prior active observations; this is not a change + // to the original request token or its saved outcome. + let current = installed + .environment + .worker_detail(&fixture.manifest.worker_id)?; + assert_eq!( + recovered.pointer("/result/worker/restore_observation_token"), + current.get("restore_observation_token") + ); + let mut expected = observed.clone(); + expected["result"]["worker"]["restore_observation_token"] = + recovered["result"]["worker"]["restore_observation_token"].clone(); + assert_eq!( + recovered, expected, + "restart may change only the public observation fence, not the saved outcome" + ); + } else { + assert_eq!( + recovered, observed, + "exact request must recover the saved receipt" + ); + } + assert_single_state(&installed.environment, &fixture, "idle")?; + let mut subscription = WorkerSessionSubscription::connect( + &installed.environment, + &fixture.manifest.worker_id, + &installed.snapshot, + )?; + subscription.assert_alive()?; + let stale = installed.environment.restore_worker( + &fixture.manifest.worker_id, + "fixed-fixture-fresh-stale-intent", + token, + ); + match stale { + Err(error) if error.to_string().contains("HTTP 409") => {} + _ => { + return Err(failure( + "fresh intent with old observation was not HTTP 409", + )); + } + } + let owner_after = restore_owner_evidence(&installed.environment, &fixture, request_id)?; + assert_eq!( + owner_after, owner_before, + "saved owner/operation/receipt must not change" + ); + fs::write( + installed + .environment + .artifacts_dir + .join("restore-owner-evidence.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "before": owner_before, "after": owner_after, + "request_token_replaced": false, + }))?, + )?; + installed.environment.assert_connected()?; + subscription.assert_alive()?; + stop_and_cleanup(installed, vec![subscription], &fixture) + })(); + fixture.source.assert_unchanged()?; + result +} + +// Read-only diagnostic of the product-written aggregate; never seeds or edits it. +// Persist only a digest, operation ID and owner count, not prompts or credentials. +fn restore_owner_evidence( + environment: &WorkerE2eEnvironment, + fixture: &Fixture, + request_id: &str, +) -> Result { + use sha2::{Digest, Sha256}; + use std::io::Read; + let path = environment + .runtime_data_dir + .join("runtime/workers") + .join(&fixture.manifest.worker_id) + .join("worker.json"); + let mut bytes = Vec::new(); + fs::File::open(path)? + .take(8 * 1024 * 1024 + 1) + .read_to_end(&mut bytes)?; + if bytes.len() > 8 * 1024 * 1024 { + return Err(failure("Worker aggregate byte budget exceeded")); + } + let aggregate: serde_json::Value = serde_json::from_slice(&bytes)?; + let owners = aggregate + .pointer("/restore_guard/owners") + .and_then(serde_json::Value::as_object) + .ok_or_else(|| failure("canonical Worker omitted restore owners"))?; + let owner = owners + .get(request_id) + .ok_or_else(|| failure("canonical owner missing"))?; + if owner + .pointer("/receipt/state") + .and_then(serde_json::Value::as_str) + != Some("accepted") + { + return Err(failure("canonical owner has no Accepted receipt")); + } + let operation = owner + .get("operation_id") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| failure("canonical owner omitted operation ID"))?; + let digest: String = Sha256::digest(serde_json::to_vec(owner)?) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + Ok( + serde_json::json!({"operation_id": operation, "owner_sha256": digest, "owner_count": owners.len()}), + ) +} + fn restore( installed: &Installed, fixture: &Fixture,