From 9dc394bff792b6cca68e511f109d004c19f17738 Mon Sep 17 00:00:00 2001 From: Nikhil Unni Date: Wed, 7 Oct 2026 16:37:54 -0700 Subject: [PATCH] feat(uffd-handler): post-copy demand faults fetch a window of chunks (ADR 0045 C2 addendum) A live move of an 8 GiB guest reached Active only after a 22 s harness handshake: every page agentd touched was one remote fault, 12,727 of them at 8 ms each. A demand fault now sends one `NeedWindow` for the faulting chunk plus up to seven sealed, uninstalled neighbours in its region (`--fault-around-chunks`, env `ENGRAM_UFFD_FAULT_AROUND`, default 8, range 1..=256; 1 is the old behaviour). - The fault loop installs and wakes only the faulting chunk from the first streamed response, then returns. A prefetch worker on its own fault connection installs the suffix through the existing install claims, outside the fault-active guard, so a slow suffix never stalls another vCPU or the drain. An overlapping demand goes out as a one-chunk window on the demand connection and the claims deduplicate. - Every response shape is validated against the request's `req_id` and expected `chunk_offset`; a reordered or duplicate reply marks the peer lost before any install. Retries request only the unfinished suffix. - Worker connects are bounded and cancellation-aware; `stop` covers lock acquisition and the join under one deadline and detaches with a warn; blocking fetches run outside the install lock with a cancellation recheck before any copy. - `fault_around_chunks_installed` rides the handler stats, the control report and the host's fault-path totals. Page-channel `PROTO_VERSION` 5 -> 6; coordinator `WIRE_VERSION` 34 -> 35 so hosts roll in lockstep. Tests: streamed installation, suffix-only retry, corrupt and reordered replies, one-chunk equivalence with `NeedAt`, window selection over random seals (proptest), concurrent drain dedup, faulting-page-only wake, overlapping and unrelated faults and the drain bypassing a stalled suffix, cancellation during a pending connect, a worker that misses the stop deadline, a stalled AltSource fetch under stop; the two-host KVM test checks the counter is reported. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01UHxL6gj4o8EvxYpgtwEWaM --- Cargo.lock | 1 + crates/engram-host-agent/src/migrate_peer.rs | 80 ++ .../engram-host-agent/src/pooled_backend.rs | 2 + .../tests/two_host_live_teleport.rs | 23 + crates/engram-migrate-proto/src/lib.rs | 20 +- .../tests/golden/from_source_hello_ack.bin | Bin 24 -> 24 bytes .../golden/handler_control_drain_done.bin | Bin 60 -> 68 bytes .../tests/golden/to_source_hello.bin | Bin 38 -> 38 bytes .../tests/golden/to_source_need_window.bin | Bin 0 -> 36 bytes .../engram-migrate-proto/tests/support/mod.rs | 20 +- .../engram-migrate-proto/tests/wire_golden.rs | 17 +- crates/engram-protocol/src/wire.rs | 3 +- crates/engram-protocol/tests/wire_golden.rs | 4 +- crates/engram-uffd-handler/Cargo.toml | 1 + crates/engram-uffd-handler/src/main.rs | 18 + crates/engram-uffd-handler/src/peer.rs | 613 ++++++++- crates/engram-uffd-handler/src/runtime.rs | 1195 ++++++++++++++++- docs/adr/0045-live-session-teleport.md | 36 + 18 files changed, 1927 insertions(+), 106 deletions(-) create mode 100644 crates/engram-migrate-proto/tests/golden/to_source_need_window.bin diff --git a/Cargo.lock b/Cargo.lock index ac0b569af..d6e9c34db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3037,6 +3037,7 @@ dependencies = [ "engram-substrate-proto", "engram-telemetry", "libc", + "proptest", "sendfd", "serde", "serde_json", diff --git a/crates/engram-host-agent/src/migrate_peer.rs b/crates/engram-host-agent/src/migrate_peer.rs index a692eb265..dec690e18 100644 --- a/crates/engram-host-agent/src/migrate_peer.rs +++ b/crates/engram-host-agent/src/migrate_peer.rs @@ -528,6 +528,37 @@ impl PeerServer { .write_us .fetch_add(t_write.elapsed().as_micros() as u64, Ordering::Relaxed); } + ToSource::NeedWindow { + req_id, + chunk_offsets, + } => { + if chunk_offsets.is_empty() + || chunk_offsets.len() > engram_migrate_proto::MAX_FAULT_AROUND_CHUNKS + { + write_frame( + &mut stream, + &FromSource::Error { + req_id: Some(req_id), + message: "invalid fault window size".into(), + }, + )?; + continue; + } + for offset in chunk_offsets { + export.touch(); + let resp = self.serve_need_at(export, req_id, offset, purpose); + let failed = matches!(resp, FromSource::Error { .. }); + let t_write = crate::time_source::metrics_now(); + write_frame(&mut stream, &resp)?; + export + .serve + .write_us + .fetch_add(t_write.elapsed().as_micros() as u64, Ordering::Relaxed); + if failed { + break; + } + } + } ToSource::DrainDone { pulled, alt_sourced, @@ -862,6 +893,55 @@ mod tests { ); } + #[cfg(target_os = "linux")] + #[tokio::test] + async fn fault_window_streams_in_request_order_with_per_chunk_integrity() { + let server = PeerServer::new(9102); + let memory: Vec = (0..16384).map(|i| (i / 4096 + 1) as u8).collect(); + let mut export = test_export("t"); + for i in 0..4 { + export.seal.set(i); + } + export.vmas = vec![GuestVma { + start: memory.as_ptr() as u64, + end: memory.as_ptr() as u64 + memory.len() as u64, + file_offset: 0, + }]; + server.register(export); + let offsets = vec![8192, 0, 4096]; + let got = talk( + server, + vec![ + hello("exp-1", "t"), + ToSource::NeedWindow { + req_id: 7, + chunk_offsets: offsets.clone(), + }, + ], + 5, + ) + .await; + assert_eq!(got.len(), 5); + for (response, offset) in got[2..].iter().zip(offsets) { + let FromSource::Page { + req_id, + chunk_offset, + bytes, + hash, + lz4, + } = response + else { + panic!("expected Page, got {response:?}") + }; + assert_eq!((*req_id, *chunk_offset), (7, offset)); + assert_eq!(*hash, engram_migrate_proto::wire_hash(bytes)); + assert_eq!( + engram_migrate_proto::decompress_page(bytes.clone(), *lz4).unwrap(), + memory[offset as usize..offset as usize + 4096] + ); + } + } + #[tokio::test] async fn good_hello_acks_and_pushes_seal_then_validates_need_at() { let server = PeerServer::new(9102); diff --git a/crates/engram-host-agent/src/pooled_backend.rs b/crates/engram-host-agent/src/pooled_backend.rs index 55693b59f..95c5d68e2 100644 --- a/crates/engram-host-agent/src/pooled_backend.rs +++ b/crates/engram-host-agent/src/pooled_backend.rs @@ -8333,6 +8333,7 @@ impl SandboxBackend for PooledBackend { faults, fault_us, fault_max_us, + fault_around_chunks_installed, } => { // Restore-tail attribution: the fault-path // totals are the serial P2P cost inside the FC @@ -8342,6 +8343,7 @@ impl SandboxBackend for PooledBackend { faults, fault_ms = fault_us / 1000, fault_max_us, + fault_around_chunks_installed, "post-copy memory drain done (handler fault-path totals)", ); return Ok(DrainOutcome::Done { diff --git a/crates/engram-host-agent/tests/two_host_live_teleport.rs b/crates/engram-host-agent/tests/two_host_live_teleport.rs index 218865f63..2b5eaed07 100644 --- a/crates/engram-host-agent/tests/two_host_live_teleport.rs +++ b/crates/engram-host-agent/tests/two_host_live_teleport.rs @@ -287,6 +287,25 @@ async fn two_host_live_teleport_preserves_post_checkpoint_state() { let (moved, cap) = common::postcopy::move_guest(&client_a, &client_b, vm, ckpt.clone(), host_a.addr).await; common::postcopy::drain(&host_b.pooled, moved).await; + let stats_path = host_b + .pooled + .prefault_stats_path(moved) + .expect("handler stats path"); + assert!( + common::poll_until_async( + std::time::Duration::from_secs(5), + std::time::Duration::from_millis(20), + || async { + std::fs::read(&stats_path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .and_then(|stats| stats["fault_around_chunks_installed"].as_u64()) + .is_some() + }, + ) + .await, + "handler must report the demand-window install counter" + ); // Commit destroys A's frozen source; the moved VM lives on B. client_a @@ -299,6 +318,10 @@ async fn two_host_live_teleport_preserves_post_checkpoint_state() { .expect("commit on A"); assert!(!host_a.pooled.list().await.unwrap().contains(&vm)); assert!(host_b.pooled.list().await.unwrap().contains(&moved)); + assert!( + stats_path.is_file(), + "successful move retains handler stats" + ); // G2: the post-checkpoint sentinel survives on B. let check = exec( diff --git a/crates/engram-migrate-proto/src/lib.rs b/crates/engram-migrate-proto/src/lib.rs index c62de698a..e421d06ab 100644 --- a/crates/engram-migrate-proto/src/lib.rs +++ b/crates/engram-migrate-proto/src/lib.rs @@ -26,7 +26,7 @@ //! ``` //! //! A handler may open MULTIPLE connections per export (one for the fault -//! path, one or more for the background drain) — the token is per-export, +//! path, one for fault-around prefetch, and one or more for the background drain) — the token is per-export, //! not per-connection, and every authenticated connection receives the //! `Seal` push. `GET_STATE` from the original ADR sketch is deliberately //! ABSENT: `state.bin` rides the gRPC `MigrationFetch` (`StateBin` @@ -73,7 +73,11 @@ use serde::{de::DeserializeOwned, Deserialize, Serialize}; /// that is the chunk-store's content address, not wire integrity. /// /// v5: remove the unused by-hash request and reply. -pub const PROTO_VERSION: u32 = 5; +/// v6: ordered fault windows and fault-around installation accounting. +pub const PROTO_VERSION: u32 = 6; + +/// Bound a demand window to limit request and response work. +pub const MAX_FAULT_AROUND_CHUNKS: usize = 256; /// Wire-integrity hash for `Page` payloads (v4: blake3 over the wire /// bytes — the SAME bytes shipped, compressed or raw). One helper so @@ -86,8 +90,8 @@ pub fn wire_hash(bytes: &[u8]) -> [u8; 32] { /// it (see the v3 note on [`PROTO_VERSION`]). #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] pub enum ConnPurpose { - /// The handler's fault-loop connection: latency-critical, single - /// in-flight request, a stalled vCPU behind every frame. + /// Demand or fault-around connection: latency-critical first response, + /// one in-flight request. A fault window may have a speculative suffix. Fault, /// Background drain: throughput-oriented, pipelined, yields to /// faults on the dest side. @@ -197,6 +201,12 @@ pub enum ToSource { alt_sourced: u64, zero_chunks: u64, }, + /// One response per offset, in request order, with the same req_id. + /// Stop the window at the first Error response. + NeedWindow { + req_id: u64, + chunk_offsets: Vec, + }, } /// source host-agent → dest uffd-handler. @@ -299,6 +309,7 @@ pub enum HandlerControl { faults: u64, fault_us: u64, fault_max_us: u64, + fault_around_chunks_installed: u64, }, /// The peer is gone (dial/reconnect exhausted, frame error, or sha /// mismatch) with sealed chunks still uninstalled. FATAL by design: @@ -458,6 +469,7 @@ mod tests { faults: 42, fault_us: 55_000, fault_max_us: 9_000, + fault_around_chunks_installed: 0, }); round_trip(&HandlerControl::PeerLost { remaining: 3, diff --git a/crates/engram-migrate-proto/tests/golden/from_source_hello_ack.bin b/crates/engram-migrate-proto/tests/golden/from_source_hello_ack.bin index 2f94ce23c4338d689f1f79259c59ed4ffacbc7e0..179098f6062ff26debdaa64092447563a11f667b 100644 GIT binary patch literal 24 UcmZQzU|?VaVg?2dFbQJ-00Aff7XSbN literal 24 UcmZQzU|?VcVg?2dFbQJ-009&L761SM diff --git a/crates/engram-migrate-proto/tests/golden/handler_control_drain_done.bin b/crates/engram-migrate-proto/tests/golden/handler_control_drain_done.bin index 15555d59424d603fd45748a2afceaa485a6eba43..098a25a32aaf3036a768f4fdd6a4ea5aaa396032 100644 GIT binary patch delta 13 OcmcCey#N3J literal 38 hcmZQzU|?VcViq6 impl Strategy { purpose, } ), + (any::(), prop::collection::vec(any::(), 1..=256)).prop_map( + |(req_id, chunk_offsets)| ToSource::NeedWindow { + req_id, + chunk_offsets + } + ), (any::(), any::()).prop_map(|(req_id, chunk_offset)| ToSource::NeedAt { req_id, chunk_offset @@ -125,9 +131,19 @@ pub fn handler_control() -> impl Strategy { any::(), any::(), any::(), + any::(), ) .prop_map( - |(pulled, alt_sourced, zero_chunks, ms, faults, fault_us, fault_max_us)| { + |( + pulled, + alt_sourced, + zero_chunks, + ms, + faults, + fault_us, + fault_max_us, + fault_around_chunks_installed, + )| { HandlerControl::DrainDone { pulled, alt_sourced, @@ -136,6 +152,7 @@ pub fn handler_control() -> impl Strategy { faults, fault_us, fault_max_us, + fault_around_chunks_installed, } } ), @@ -161,6 +178,7 @@ fn _exhaustiveness_to_source(m: &ToSource) { match m { ToSource::Hello { .. } => {} ToSource::NeedAt { .. } => {} + ToSource::NeedWindow { .. } => {} ToSource::DrainDone { .. } => {} } } diff --git a/crates/engram-migrate-proto/tests/wire_golden.rs b/crates/engram-migrate-proto/tests/wire_golden.rs index 19710c2a6..08ce3ba86 100644 --- a/crates/engram-migrate-proto/tests/wire_golden.rs +++ b/crates/engram-migrate-proto/tests/wire_golden.rs @@ -21,7 +21,7 @@ //! ## Regenerating the corpus (only when you INTENTIONALLY evolve a type) //! //! A wire break requires a protocol-version bump and coordinated endpoint -//! updates. Version 5 removes the unused by-hash request and reply. Regenerate: +//! updates. Version 6 adds demand windows and install accounting. Regenerate: //! //! ```text //! cargo test -p engram-migrate-proto --test wire_golden -- --ignored regen_golden @@ -123,6 +123,12 @@ fn to_source_golden_and_variant_indices() { zero_chunks: 42, }; + let window = ToSource::NeedWindow { + req_id: 8, + chunk_offsets: vec![0, 1048576], + }; + assert_golden("to_source_need_window", &window); + assert_variant_index(&window, 3, "ToSource::NeedWindow"); assert_golden("to_source_hello", &hello); assert_golden("to_source_need_at", &need_at); assert_golden("to_source_drain_done", &drain_done); @@ -205,6 +211,7 @@ fn handler_control_golden_and_variant_indices() { faults: 42, fault_us: 55_000, fault_max_us: 9_000, + fault_around_chunks_installed: 0, }; let peer_lost = HandlerControl::PeerLost { remaining: 3, @@ -243,6 +250,13 @@ fn regen_golden() { eprintln!("wrote {} ({} bytes)", path.display(), bytes.len()); } + write( + "to_source_need_window", + &ToSource::NeedWindow { + req_id: 8, + chunk_offsets: vec![0, 1048576], + }, + ); write( "to_source_hello", &ToSource::Hello { @@ -343,6 +357,7 @@ fn regen_golden() { faults: 42, fault_us: 55_000, fault_max_us: 9_000, + fault_around_chunks_installed: 0, }, ); write( diff --git a/crates/engram-protocol/src/wire.rs b/crates/engram-protocol/src/wire.rs index 86a415c44..ce72523d5 100644 --- a/crates/engram-protocol/src/wire.rs +++ b/crates/engram-protocol/src/wire.rs @@ -169,7 +169,8 @@ use serde::{Deserialize, Serialize}; // payloads do not change. The bump fences mixed fleets: an older host refuses // the same snapshot with an untyped error, which the resume verb counts toward // the Dead budget instead of recovering from disk. -pub const WIRE_VERSION: u32 = 35; +// v36 gates page-channel v6 fault windows for a lockstep host roll. +pub const WIRE_VERSION: u32 = 36; /// gRPC metadata (header) key carrying the caller's [`WIRE_VERSION`] on /// every coord→host request (issue #229). ASCII, lowercase — tonic diff --git a/crates/engram-protocol/tests/wire_golden.rs b/crates/engram-protocol/tests/wire_golden.rs index 675f76bfb..6272f248f 100644 --- a/crates/engram-protocol/tests/wire_golden.rs +++ b/crates/engram-protocol/tests/wire_golden.rs @@ -699,8 +699,10 @@ fn wire_version_pinned() { // Existing bincode goldens do not embed presetup; their bytes do not change. // 34 -> 35: the memory image refusal is typed. No payload changes; the // bump fences old hosts that refuse untyped. + // 35 -> 36: page-channel v6 fault windows require a lockstep host roll. + // Coordinator bincode payload bytes are unchanged. assert_eq!( - WIRE_VERSION, 35, + WIRE_VERSION, 36, "WIRE_VERSION changed — confirm payload goldens were regenerated too" ); } diff --git a/crates/engram-uffd-handler/Cargo.toml b/crates/engram-uffd-handler/Cargo.toml index b64d5d99f..6de0fe81c 100644 --- a/crates/engram-uffd-handler/Cargo.toml +++ b/crates/engram-uffd-handler/Cargo.toml @@ -71,4 +71,5 @@ workspace-hack = { version = "0.1", path = "../../workspace-hack" } userfaultfd = { version = "0.9", features = ["linux5_13"] } [dev-dependencies] +proptest.workspace = true tempfile = "3" diff --git a/crates/engram-uffd-handler/src/main.rs b/crates/engram-uffd-handler/src/main.rs index 3eca2df7c..6393f058f 100644 --- a/crates/engram-uffd-handler/src/main.rs +++ b/crates/engram-uffd-handler/src/main.rs @@ -17,6 +17,7 @@ //! --listen /tmp/uffd.sock \ //! --canonical-manifest @v \ //! --session-manifest @v \ +//! [--fault-around-chunks N] (env ENGRAM_UFFD_FAULT_AROUND, default 8) //! [--prefault-trace file:|] \ //! [--publish-trace-host ] \ //! [--trace-key ] \ @@ -191,6 +192,7 @@ mod linux { /// the background. The token rides `ENGRAM_PEER_TOKEN` (argv /// leaks via /proc/*/cmdline); `--peer-token` is a dev override. pub peer_addr: Option, + pub fault_around_chunks: usize, pub peer_export_id: Option, pub peer_token: Option, /// ADR 0045 C2: bind a UnixListener here and stream one-way @@ -220,6 +222,8 @@ mod linux { let mut cache_root: Option = None; let mut recorder_window_ms: u64 = DEFAULT_RECORDER_WINDOW_MS; let mut base_shm: Option = None; + let mut fault_around_chunks = std::env::var("ENGRAM_UFFD_FAULT_AROUND") + .unwrap_or_else(|_| engram_uffd_handler::peer::DEFAULT_FAULT_AROUND_CHUNKS.to_string()); let mut peer_addr: Option = None; let mut peer_export_id: Option = None; let mut peer_token: Option = None; @@ -229,6 +233,11 @@ mod linux { let mut argv = std::env::args().skip(1); while let Some(arg) = argv.next() { match arg.as_str() { + "--fault-around-chunks" => { + fault_around_chunks = argv + .next() + .ok_or("--fault-around-chunks requires a value")?; + } "--listen" => { listen = Some(PathBuf::from( argv.next() @@ -371,7 +380,14 @@ mod linux { } } + let fault_around_chunks: usize = fault_around_chunks + .parse() + .map_err(|_| "invalid fault-around chunk count")?; + if !(1..=engram_migrate_proto::MAX_FAULT_AROUND_CHUNKS).contains(&fault_around_chunks) { + return Err("fault-around chunk count must be in 1..=256".into()); + } Ok(Args { + fault_around_chunks, listen, canonical_manifest, session_manifest, @@ -665,6 +681,7 @@ mod linux { handle, engram_uffd_handler::runtime::RunListenerOpts { prefault_trace: prefault, + fault_around_chunks: args.fault_around_chunks, recorder_window: window, trace_output: trace_out, base_shm, @@ -784,6 +801,7 @@ mod linux { --listen \\ --canonical-manifest @v \\ --session-manifest @v \\ + [--fault-around-chunks N] (env ENGRAM_UFFD_FAULT_AROUND, default 8) [--prefault-trace file:|] \\ [--publish-trace-host ] \\ [--cache-root ] \\ diff --git a/crates/engram-uffd-handler/src/peer.rs b/crates/engram-uffd-handler/src/peer.rs index 9e6f96fd0..2f7701900 100644 --- a/crates/engram-uffd-handler/src/peer.rs +++ b/crates/engram-uffd-handler/src/peer.rs @@ -6,9 +6,8 @@ //! Deliberately minimal and fully synchronous: the fault loop is a //! blocking thread, and this client must never grow tonic/tokio. One //! connection serves the fault path (single in-flight request — the -//! fault loop serves one fault at a time, so pipelining buys nothing -//! there); the background drain opens its own connection(s) and -//! pipelines. +//! fault loop requests one chunk at a time). Fault-around uses its own +//! worker connection; the background drain opens its own connections and pipelines. //! //! ## Lifecycle / soundness //! @@ -26,7 +25,7 @@ //! session lost — the caller reports `PeerLost` and the fault loop //! exits loud rather than papering over missing state. -use std::net::TcpStream; +use std::net::{TcpStream, ToSocketAddrs}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Mutex; @@ -53,8 +52,19 @@ const RECONNECT_BACKOFF: Duration = Duration::from_millis(500); /// exceeds the source's worst-case `process_vm_readv` service: it serves /// from the PAUSED guest's resident RAM (no disk, no network), and the /// drain's hot-first ordering keeps any single chunk's serve sub-ms. -const PEER_READ_TIMEOUT: Duration = Duration::from_secs(5); +pub(crate) const PEER_READ_TIMEOUT: Duration = Duration::from_secs(5); const PEER_WRITE_TIMEOUT: Duration = Duration::from_secs(5); + +/// First response from a new worker: each attempt can connect, send Hello, +/// read HelloAck and Seal, send NeedWindow, then read the first page. +/// Include the initial attempt and every redial, plus reconnect backoff. +#[cfg(target_os = "linux")] +pub(crate) const WINDOW_FIRST_RESPONSE_TIMEOUT: Duration = PEER_READ_TIMEOUT + .saturating_mul(4) + .saturating_add(PEER_WRITE_TIMEOUT.saturating_mul(2)) + .saturating_mul(RECONNECT_ATTEMPTS + 1) + .saturating_add(RECONNECT_BACKOFF.saturating_mul(RECONNECT_ATTEMPTS)); + /// TCP keepalive so a dropped path with no in-flight data still surfaces /// (the read/write timeouts only fire while a request is outstanding; /// keepalive catches an idle fault conn between faults). Idle 10 s, then @@ -129,6 +139,29 @@ pub enum PeerPage { AltSource([u8; 32]), } +/// Default number of chunks in a demand window. +pub const DEFAULT_FAULT_AROUND_CHUNKS: usize = 8; + +/// Select sealed, uninstalled chunks from the fault through this region. +/// The faulting chunk can straddle a region boundary; later chunks cannot. +pub fn fault_window( + seal: &SealBitmap, + first: u64, + region_end: u64, + limit: usize, + installed: impl Fn(usize) -> bool, +) -> Vec { + (first / seal.chunk_size..seal.chunk_count) + .map(|idx| (idx, idx * seal.chunk_size)) + .take_while(|&(_, offset)| { + offset == first || offset.saturating_add(seal.chunk_size) <= region_end + }) + .filter(|&(idx, _)| seal.get(idx) && !installed(idx as usize)) + .take(limit) + .map(|(_, offset)| offset) + .collect() +} + /// Drain accounting (rides `ToSource::DrainDone` + the control sock). #[derive(Clone, Copy, Debug, Default)] pub struct DrainStats { @@ -196,9 +229,67 @@ fn dial( expect_total_bytes: u64, purpose: engram_migrate_proto::ConnPurpose, ) -> Result<(TcpStream, SealBitmap), PeerError> { - let mut stream = TcpStream::connect(addr)?; + dial_with_socket( + addr, + export_id, + token, + expect_chunk_size, + expect_total_bytes, + purpose, + |_| Ok(true), + ) +} + +/// `None` checks cancellation before and after each connect attempt. `Some` +/// registers the connected socket before the handshake can block. +fn connect_with_socket( + addr: &str, + connected: &impl Fn(Option<&TcpStream>) -> Result, +) -> Result { + if !connected(None)? { + return Err(PeerError::Lost("connection canceled".into())); + } + let deadline = std::time::Instant::now() + PEER_READ_TIMEOUT; + let mut last_error = std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "peer address resolved to no addresses", + ); + for address in addr.to_socket_addrs()? { + if !connected(None)? { + return Err(PeerError::Lost("connection canceled".into())); + } + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + if remaining.is_zero() { + return Err(std::io::Error::from(std::io::ErrorKind::TimedOut).into()); + } + let result = TcpStream::connect_timeout(&address, remaining); + if !connected(None)? { + return Err(PeerError::Lost("connection canceled".into())); + } + match result { + Ok(stream) => return Ok(stream), + Err(error) => last_error = error, + } + } + Err(last_error.into()) +} + +fn dial_with_socket( + addr: &str, + export_id: &str, + token: &str, + expect_chunk_size: u64, + expect_total_bytes: u64, + purpose: engram_migrate_proto::ConnPurpose, + connected: impl Fn(Option<&TcpStream>) -> Result, +) -> Result<(TcpStream, SealBitmap), PeerError> { + let mut stream = connect_with_socket(addr, &connected)?; stream.set_nodelay(true)?; apply_peer_sockopts(&stream)?; + // Register worker sockets before any blocking handshake read or write. + if !connected(Some(&stream))? { + return Err(PeerError::Lost("connection canceled".into())); + } write_frame( &mut stream, &ToSource::Hello { @@ -356,26 +447,86 @@ impl PeerSession { Ok(stream) } - /// Fault-path request: one round-trip on the shared connection, - /// sha-verified. Reconnects through `RECONNECT_ATTEMPTS` before - /// declaring the peer lost. - pub fn need_at(&self, chunk_offset: u64) -> Result { + /// Fault windows use their own connection; the source permits multiple + /// Fault connections per export. This does not hold the drain priority gate. + #[cfg(target_os = "linux")] + pub(crate) fn open_prefetch_conn( + &self, + connected: impl Fn(Option<&TcpStream>) -> Result, + ) -> Result { + dial_with_socket( + &self.addr, + &self.export_id, + &self.token, + self.chunk_size, + self.expected_total, + engram_migrate_proto::ConnPurpose::Fault, + connected, + ) + .map(|(stream, _)| stream) + } + + /// Run a one-chunk demand window on the fault connection. + pub fn need_at(&self, offset: u64) -> Result { + let mut page = None; + self.need_window(&[offset], |_, next| { + page = Some(next); + Ok::<_, std::convert::Infallible>(()) + })? + .unwrap(); + Ok(page.expect("one response")) + } + + /// Stream a window; retry only the suffix not yet installed. + /// The outer error is a peer failure. The inner error is from the + /// installer; it stops the stream and the caller must exit the handler. + pub fn need_window( + &self, + offsets: &[u64], + install: impl FnMut(u64, PeerPage) -> Result<(), E>, + ) -> Result, PeerError> { if self.is_lost() { return Err(PeerError::Lost("peer already marked lost".into())); } let _fault = self.begin_fault(); let mut conn = self.fault_conn.lock().expect("fault conn poisoned"); + let result = self.stream_window(&mut conn, offsets, install, || false, |_| Ok(true)); + if result.is_err() { + self.mark_lost(); + } + result + } + + /// Shared streaming and suffix-only retry logic. The prefetch caller owns + /// its connection and handles errors under its cancellation lock, so a + /// canceled window cannot latch the session lost after DrainDone. + pub(crate) fn stream_window( + &self, + conn: &mut TcpStream, + offsets: &[u64], + mut install: impl FnMut(u64, PeerPage) -> Result<(), E>, + cancelled: impl Fn() -> bool, + connected: impl Fn(Option<&TcpStream>) -> Result, + ) -> Result, PeerError> { + let mut completed = 0; let mut last_err: Option = None; for attempt in 0..=RECONNECT_ATTEMPTS { + if cancelled() { + return Ok(Ok(())); + } if attempt > 0 { std::thread::sleep(RECONNECT_BACKOFF); - match dial( + if cancelled() { + return Ok(Ok(())); + } + match dial_with_socket( &self.addr, &self.export_id, &self.token, self.chunk_size, self.expected_total, engram_migrate_proto::ConnPurpose::Fault, + &connected, ) { Ok((fresh, _seal)) => *conn = fresh, Err(e) => { @@ -385,15 +536,43 @@ impl PeerSession { } } } - match request_chunk(&mut conn, &self.next_req, chunk_offset) { - Ok(page) => return Ok(page), + if !connected(Some(conn))? { + return Ok(Ok(())); + } + let result = (|| { + let req_id = self.next_req.fetch_add(1, Ordering::Relaxed); + write_frame( + &mut *conn, + &ToSource::NeedWindow { + req_id, + chunk_offsets: offsets[completed..].to_vec(), + }, + )?; + while completed < offsets.len() { + let offset = offsets[completed]; + let response = read_frame(&mut *conn)?; + if cancelled() { + return Ok(Ok(())); + } + let page = decode_page(response, req_id, offset)?; + if let Err(e) = install(offset, page) { + return Ok(Err(e)); + } + completed += 1; + } + Ok(Ok(())) + })(); + if cancelled() { + return Ok(Ok(())); + } + match result { + Ok(result) => return Ok(result), // Terminal classifications never retry: corrupt content, a // connection-fatal server error (`Error{req_id: None}`), // or geometry skew are all unrecoverable on any conn. Err(e @ PeerError::ShaMismatch { .. }) | Err(e @ PeerError::Server(_)) | Err(e @ PeerError::GeometryMismatch(_)) => { - self.mark_lost(); return Err(e); } // Per-request failures (`Error{req_id: Some}`) and @@ -410,12 +589,10 @@ impl PeerSession { Err(e @ PeerError::Lost(_)) => { // Should not surface from request_chunk, but treat as // terminal if it ever does. - self.mark_lost(); return Err(e); } } } - self.mark_lost(); Err(PeerError::Lost(format!( "reconnects exhausted: {}", last_err.map(|e| e.to_string()).unwrap_or_default() @@ -428,7 +605,7 @@ impl PeerSession { /// drain's park condition), and on drop — every return path — /// accumulates the elapsed wall, tracks the max, and stamps the /// completion time for the drain's recent-fault window. - fn begin_fault(&self) -> impl Drop + '_ { + pub(crate) fn begin_fault(&self) -> impl Drop + '_ { self.fault_count.fetch_add(1, Ordering::Relaxed); self.faults_inflight.fetch_add(1, Ordering::Relaxed); struct G<'a>(&'a PeerSession, std::time::Instant); @@ -447,9 +624,9 @@ impl PeerSession { } } -/// One `NeedAt` round-trip on `conn` (used by the fault path and, with -/// its own connection, the drain). -pub fn request_chunk( +/// Legacy single-chunk request, used to check one-chunk equivalence. +#[cfg(test)] +fn request_chunk( conn: &mut TcpStream, next_req: &AtomicU64, chunk_offset: u64, @@ -498,19 +675,24 @@ pub fn decode_page( .map_err(PeerError::Server)?; Ok(PeerPage::Bytes(raw)) } - FromSource::ZeroChunk { req_id, .. } if req_id == want_req => Ok(PeerPage::Zero), + FromSource::ZeroChunk { + req_id, + chunk_offset, + } if req_id == want_req && chunk_offset == want_offset => Ok(PeerPage::Zero), FromSource::AltSource { req_id, durable_sha256, - .. - } if req_id == want_req => Ok(PeerPage::AltSource(durable_sha256)), + chunk_offset, + } if req_id == want_req && chunk_offset == want_offset => { + Ok(PeerPage::AltSource(durable_sha256)) + } // Honor the wire contract: `req_id: Some` ⇒ THAT request failed // (the conn keeps serving — retryable); `req_id: None` ⇒ // connection-fatal (terminal). FromSource::Error { - req_id: Some(_), + req_id: Some(req_id), message, - } => Err(PeerError::RequestFailed(message)), + } if req_id == want_req => Err(PeerError::RequestFailed(message)), FromSource::Error { req_id: None, message, @@ -602,7 +784,14 @@ impl ControlInner { /// invariant on the fields). Never blocks beyond the per-write timeout; /// dead subscribers are pruned. fn report(&self, msg: HandlerControl) { + self.report_if(msg, || true); + } + + fn report_if(&self, msg: HandlerControl, publish: impl FnOnce() -> bool) { let mut backlog = self.backlog.lock().expect("backlog poisoned"); + if !publish() { + return; + } backlog.push(msg.clone()); let mut subs = self.subscribers.lock().expect("subscribers poisoned"); subs.retain_mut(|s| write_frame(s, &msg).is_ok()); @@ -644,6 +833,13 @@ impl ControlTx { self.inner.report(msg); } + /// Check publication after waiting for the backlog lock. The caller can + /// serialize cancellation here without holding its lock during socket I/O. + #[cfg(target_os = "linux")] + pub(crate) fn report_if(&self, msg: HandlerControl, publish: impl FnOnce() -> bool) { + self.inner.report_if(msg, publish); + } + /// Test-only: install a hook fired between a subscriber's backlog /// snapshot and its registration (while the backlog lock is held). #[cfg(test)] @@ -653,7 +849,7 @@ impl ControlTx { } #[cfg(test)] -mod tests { +pub(crate) mod tests { use std::net::TcpListener; use super::*; @@ -661,10 +857,100 @@ mod tests { const CHUNK: u64 = 4096; const TOTAL: u64 = 4 * 4096; + /// A full Linux accept queue makes the next TCP handshake stay pending. + /// No routing assumptions or external black-hole address are needed. + #[cfg(target_os = "linux")] + #[test] + fn cancellation_during_pending_connect_is_bounded() { + let listener = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::STREAM, + Some(socket2::Protocol::TCP), + ) + .unwrap(); + listener + .bind( + &"127.0.0.1:0" + .parse::() + .unwrap() + .into(), + ) + .unwrap(); + listener.listen(0).unwrap(); + let addr = listener.local_addr().unwrap().as_socket().unwrap(); + // Linux permits backlog + 1 established connections. Never accept it. + let _queued = TcpStream::connect_timeout(&addr, Duration::from_secs(1)).unwrap(); + let stopped = std::sync::Arc::new(AtomicBool::new(false)); + let worker_stopped = stopped.clone(); + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + let result = dial_with_socket( + &addr.to_string(), + "e", + "tok", + CHUNK, + TOTAL, + engram_migrate_proto::ConnPurpose::Fault, + |socket| { + assert!( + socket.is_none(), + "full accept queue must prevent connection" + ); + let _ = entered_tx.send(()); + Ok(!worker_stopped.load(Ordering::SeqCst)) + }, + ); + done_tx.send(result).unwrap(); + }); + // The two checks occur before resolution and immediately before connect. + entered_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + entered_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(matches!( + done_rx.recv_timeout(Duration::from_millis(100)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) + )); + let started = std::time::Instant::now(); + stopped.store(true, Ordering::SeqCst); + let result = done_rx + .recv_timeout(PEER_READ_TIMEOUT + Duration::from_secs(1)) + .expect("canceled connect exceeded its deadline"); + assert!( + matches!(result, Err(PeerError::Lost(ref message)) if message == "connection canceled") + ); + assert!(started.elapsed() < PEER_READ_TIMEOUT + Duration::from_secs(1)); + worker.join().unwrap(); + } + + #[test] + fn canceled_connect_does_not_resolve_or_open_a_socket() { + let result = dial_with_socket( + "invalid address", + "e", + "tok", + CHUNK, + TOTAL, + engram_migrate_proto::ConnPurpose::Fault, + |_| Ok(false), + ); + assert!( + matches!(result, Err(PeerError::Lost(ref message)) if message == "connection canceled") + ); + } + /// A minimal fake source: accepts conns, answers Hello/Ack/Seal, /// then serves canned NeedAt responses. - fn fake_source( + pub(crate) fn fake_source( + seal_bits: Vec, + respond: impl Fn(u64, u64) -> FromSource + Send + Clone + 'static, + ) -> std::net::SocketAddr { + fake_source_geometry(seal_bits, CHUNK, TOTAL, respond) + } + + pub(crate) fn fake_source_geometry( seal_bits: Vec, + chunk_size: u64, + total_bytes: u64, respond: impl Fn(u64, u64) -> FromSource + Send + Clone + 'static, ) -> std::net::SocketAddr { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -693,12 +979,12 @@ mod tests { &mut s, &FromSource::HelloAck { version: PROTO_VERSION, - chunk_size: CHUNK, - total_bytes: TOTAL, + chunk_size, + total_bytes, }, ) .unwrap(); - let mut bitmap = SealBitmap::new(CHUNK, TOTAL / CHUNK); + let mut bitmap = SealBitmap::new(chunk_size, total_bytes / chunk_size); for b in &seal_bits { bitmap.set(*b); } @@ -714,6 +1000,21 @@ mod tests { return; } } + ToSource::NeedWindow { + req_id, + chunk_offsets, + } => { + for offset in chunk_offsets { + let resp = respond(req_id, offset); + let failed = matches!(resp, FromSource::Error { .. }); + if write_frame(&mut s, &resp).is_err() { + return; + } + if failed { + break; + } + } + } ToSource::DrainDone { .. } => return, _ => return, } @@ -724,7 +1025,7 @@ mod tests { addr } - fn page_resp(req_id: u64, chunk_offset: u64) -> FromSource { + pub(crate) fn page_resp(req_id: u64, chunk_offset: u64) -> FromSource { // Through the REAL compression path — the canned server // serves exactly what the prod source serves. let (bytes, lz4) = engram_migrate_proto::compress_page(vec![0xAB; CHUNK as usize]); @@ -738,6 +1039,244 @@ mod tests { } } + #[test] + fn every_page_shape_checks_request_and_offset_before_install() { + for response in [ + page_resp(7, CHUNK), + FromSource::ZeroChunk { + req_id: 7, + chunk_offset: CHUNK, + }, + FromSource::AltSource { + req_id: 7, + chunk_offset: CHUNK, + durable_sha256: [0; 32], + }, + ] { + assert!(matches!( + decode_page(response.clone(), 7, 0), + Err(PeerError::Server(_)) + )); + assert!(matches!( + decode_page(response, 8, CHUNK), + Err(PeerError::Server(_)) + )); + } + assert!(matches!( + decode_page( + FromSource::Error { + req_id: Some(8), + message: "wrong request".into(), + }, + 7, + 0 + ), + Err(PeerError::Server(_)) + )); + } + + #[test] + fn reordered_mixed_window_fails_before_any_install() { + let addr = fake_source(vec![0, 1], |req_id, offset| { + if offset == 0 { + FromSource::ZeroChunk { + req_id, + chunk_offset: CHUNK, + } + } else { + FromSource::AltSource { + req_id, + chunk_offset: 0, + durable_sha256: [0; 32], + } + } + }); + let peer = + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), CHUNK, TOTAL).unwrap(); + let mut installs = 0; + let result = peer.need_window(&[0, CHUNK], |_, _| { + installs += 1; + Ok::<_, ()>(()) + }); + assert!(matches!(result, Err(PeerError::Server(_)))); + assert_eq!(installs, 0); + assert!(peer.is_lost()); + } + + #[test] + fn duplicate_from_previous_window_fails_before_any_new_install() { + let saved = std::sync::Arc::new(Mutex::new(None)); + let addr = fake_source(vec![0, 1], move |req_id, offset| { + saved + .lock() + .unwrap() + .get_or_insert_with(|| FromSource::ZeroChunk { + req_id, + chunk_offset: offset, + }) + .clone() + }); + let peer = + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), CHUNK, TOTAL).unwrap(); + peer.need_at(0).unwrap(); + let mut installs = 0; + let result = peer.need_window(&[CHUNK], |_, _| { + installs += 1; + Ok::<_, ()>(()) + }); + assert!(matches!(result, Err(PeerError::Server(_)))); + assert_eq!(installs, 0); + assert!(peer.is_lost()); + } + + #[test] + fn duplicate_reply_fails_before_installing_duplicate() { + let addr = fake_source(vec![0, 1], |req_id, _| FromSource::ZeroChunk { + req_id, + chunk_offset: 0, + }); + let peer = + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), CHUNK, TOTAL).unwrap(); + let mut installed = Vec::new(); + let result = peer.need_window(&[0, CHUNK], |offset, _| { + installed.push(offset); + Ok::<_, ()>(()) + }); + assert!(matches!(result, Err(PeerError::Server(_)))); + // The valid prefix stays installed; the duplicate never reaches the installer. + assert_eq!(installed, [0]); + assert!(peer.is_lost()); + } + + #[test] + fn window_streams_verified_pages_and_keeps_prefix_on_corruption() { + let addr = fake_source(vec![0, 1, 2], |req_id, offset| { + if offset == 2 * CHUNK { + FromSource::Page { + req_id, + chunk_offset: offset, + bytes: vec![7], + hash: [0; 32], + lz4: false, + } + } else { + page_resp(req_id, offset) + } + }); + let sess = + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), CHUNK, TOTAL).unwrap(); + let mut installed = Vec::new(); + let err = sess + .need_window(&[0, CHUNK, 2 * CHUNK], |offset, page| { + let PeerPage::Bytes(bytes) = page else { + panic!("expected bytes") + }; + assert_eq!(bytes, vec![0xAB; CHUNK as usize]); + installed.push(offset); + Ok::<_, ()>(()) + }) + .unwrap_err(); + assert!(matches!(err, PeerError::ShaMismatch { .. })); + assert_eq!(installed, vec![0, CHUNK]); + assert!(sess.is_lost()); + assert_eq!(sess.fault_stats().0, 1); + } + + #[test] + fn window_installs_before_next_response_and_retries_only_suffix() { + let (tx, rx) = std::sync::mpsc::channel(); + let rx = std::sync::Arc::new(Mutex::new(rx)); + let seen = std::sync::Arc::new(Mutex::new(Vec::new())); + let seen_source = seen.clone(); + let failed = std::sync::Arc::new(AtomicBool::new(false)); + let addr = fake_source(vec![0, 1, 2], move |req, offset| { + seen_source.lock().unwrap().push(offset); + if offset == CHUNK && !failed.swap(true, Ordering::SeqCst) { + // The server will not send chunk 1 until chunk 0 was installed. + rx.lock() + .unwrap() + .recv_timeout(Duration::from_secs(2)) + .unwrap(); + return FromSource::Error { + req_id: Some(req), + message: "transient read failure".into(), + }; + } + page_resp(req, offset) + }); + let sess = + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), CHUNK, TOTAL).unwrap(); + let mut installed = Vec::new(); + sess.need_window(&[0, CHUNK, 2 * CHUNK], |offset, _| { + installed.push(offset); + if offset == 0 { + tx.send(()).unwrap(); + } + Ok::<_, ()>(()) + }) + .unwrap() + .unwrap(); + assert_eq!(installed, vec![0, CHUNK, 2 * CHUNK]); + assert_eq!(*seen.lock().unwrap(), vec![0, CHUNK, CHUNK, 2 * CHUNK]); + assert_eq!(sess.fault_stats().0, 1); + assert!(!sess.is_lost()); + } + + #[test] + fn one_chunk_window_matches_need_at() { + let addr = fake_source(vec![0], page_resp); + let sess = + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), CHUNK, TOTAL).unwrap(); + let PeerPage::Bytes(single) = + request_chunk(&mut sess.fault_conn.lock().unwrap(), &sess.next_req, 0).unwrap() + else { + panic!("bytes") + }; + sess.need_window(&[0], |offset, page| { + assert_eq!(offset, 0); + let PeerPage::Bytes(bytes) = page else { + panic!("bytes") + }; + assert_eq!(bytes, single); + Ok::<_, ()>(()) + }) + .unwrap() + .unwrap(); + } + + #[test] + fn window_skips_installed_and_unsealed_and_region_tail() { + let mut seal = SealBitmap::new(4096, 8); + for i in [0, 2, 3, 4, 5, 6, 7] { + seal.set(i); + } + assert_eq!( + fault_window(&seal, 0, 5 * 4096 + 100, 8, |i| i == 2), + vec![0, 3 * 4096, 4 * 4096] + ); + assert_eq!(fault_window(&seal, 0, 8 * 4096, 1, |_| false), vec![0]); + } + + proptest::proptest! { + #![proptest_config(proptest::test_runner::Config { + failure_persistence: Some(Box::new(proptest::test_runner::FileFailurePersistence::Direct("proptest-regressions/fault_window.txt"))), + ..proptest::test_runner::Config::default() + })] + #[test] + fn selected_window_is_sealed_uninstalled_and_bounded( + bits in proptest::collection::vec((proptest::bool::ANY, proptest::bool::ANY), 1..128), + first in 0usize..128, end in 1usize..129, limit in 1usize..257, + ) { + let end = end.min(bits.len()); + let first = first.min(end - 1); + let mut seal = SealBitmap::new(4096, bits.len() as u64); + for (i, &(sealed, _)) in bits.iter().enumerate() { if sealed { seal.set(i as u64); } } + let got = fault_window(&seal, first as u64 * 4096, end as u64 * 4096, limit, |i| bits[i].1); + let expected: Vec<_> = (first..end).filter(|&i| bits[i].0 && !bits[i].1).take(limit).map(|i| i as u64 * 4096).collect(); + proptest::prop_assert_eq!(got, expected); + } + } + #[test] fn connect_validates_geometry_and_holds_seal() { let addr = fake_source(vec![0, 2], page_resp); @@ -1024,13 +1563,15 @@ mod tests { if drop_after_seal { return; // connection dies before serving } - while let Ok(ToSource::NeedAt { + while let Ok(ToSource::NeedWindow { req_id, - chunk_offset, + chunk_offsets, }) = read_frame::<_, ToSource>(&mut s) { - if write_frame(&mut s, &page_resp(req_id, chunk_offset)).is_err() { - return; + for offset in chunk_offsets { + if write_frame(&mut s, &page_resp(req_id, offset)).is_err() { + return; + } } } }); @@ -1124,6 +1665,7 @@ mod tests { faults: 0, fault_us: 0, fault_max_us: 0, + fault_around_chunks_installed: 0, }); }); std::thread::sleep(Duration::from_millis(200)); @@ -1181,6 +1723,7 @@ mod tests { faults: 0, fault_us: 0, fault_max_us: 0, + fault_around_chunks_installed: 0, }); }) }; diff --git a/crates/engram-uffd-handler/src/runtime.rs b/crates/engram-uffd-handler/src/runtime.rs index 2e1f57a08..2a280b0e7 100644 --- a/crates/engram-uffd-handler/src/runtime.rs +++ b/crates/engram-uffd-handler/src/runtime.rs @@ -37,6 +37,7 @@ use std::os::fd::{AsRawFd, FromRawFd}; use std::os::unix::net::UnixStream; use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::sync::{Condvar, Mutex}; use std::time::Duration; @@ -126,6 +127,9 @@ pub struct PrefaultStats { /// reaches that chunk. #[serde(default)] pub peer_live_faults: u64, + /// Chunks installed by demand windows, including the faulting chunk. + #[serde(default)] + pub fault_around_chunks_installed: u64, } /// Filename for [`PrefaultStats`], colocated with @@ -387,6 +391,66 @@ enum Claim { Done, } +/// One bounded prefetch slot. Requests are never queued behind a window. +struct FaultWindow { + progress: Arc, + thread: std::thread::JoinHandle<()>, +} + +struct WindowProgress { + stopped: AtomicBool, + // Serializes the final cancellation check with installs and report publication. + socket: Mutex>, +} + +impl WindowProgress { + fn new() -> Self { + Self { + stopped: AtomicBool::new(false), + socket: Mutex::new(None), + } + } + + fn stopped(&self) -> bool { + self.stopped.load(Ordering::SeqCst) + } +} + +impl FaultWindow { + fn stop(self) { + let deadline = std::time::Instant::now() + crate::peer::PEER_READ_TIMEOUT; + loop { + match self.progress.socket.try_lock() { + Ok(socket) => { + self.progress.stopped.store(true, Ordering::SeqCst); + if let Some(socket) = &*socket { + let _ = socket.shutdown(std::net::Shutdown::Both); + } + break; + } + Err(std::sync::TryLockError::Poisoned(_)) => panic!("window poisoned"), + Err(std::sync::TryLockError::WouldBlock) => {} + } + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + if remaining.is_zero() { + self.progress.stopped.store(true, Ordering::SeqCst); + tracing::warn!("fault-around install lock missed stop deadline; detaching"); + return; + } + std::thread::sleep(remaining.min(Duration::from_millis(5))); + } + while !self.thread.is_finished() { + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + if remaining.is_zero() { + tracing::warn!("fault-around worker did not stop before deadline; detaching"); + return; + } + std::thread::sleep(remaining.min(Duration::from_millis(5))); + } + let _ = self.thread.join(); + } +} + /// All the state the event loop needs after the handshake. pub struct Runtime { mappings: Vec, @@ -439,6 +503,9 @@ pub struct Runtime { /// task pulls the rest in the background. `None` ⇒ everything /// above is independent of the source. peer: Option>, + fault_window: Mutex>, + fault_around_chunks: usize, + fault_around_chunks_installed: std::sync::atomic::AtomicU64, /// One-way progress/failure reports to the host-agent (peer mode). control: Option>, /// ADR 0045 C2: latched once the background drain has pulled every @@ -526,6 +593,9 @@ impl Runtime { base_shm, zeropage_ok: std::sync::atomic::AtomicBool::new(true), peer: None, + fault_window: Mutex::new(None), + fault_around_chunks: crate::peer::DEFAULT_FAULT_AROUND_CHUNKS, + fault_around_chunks_installed: std::sync::atomic::AtomicU64::new(0), control: None, drain_done: std::sync::atomic::AtomicBool::new(false), window_dump_done: std::sync::atomic::AtomicBool::new(false), @@ -1025,6 +1095,17 @@ impl Runtime { Claim::Claimed => {} _ => return Ok(false), } + self.install_claimed_chunk_at(byte_offset, bytes, wake) + } + + fn install_claimed_chunk_at( + &self, + byte_offset: u64, + bytes: Bytes, + wake: bool, + ) -> Result { + let chunk_size = self.backend.chunk_size(); + let chunk_idx = (byte_offset / chunk_size) as usize; // The chunk's bytes occupy file offsets // [byte_offset, byte_offset + total_len). Walk every region the @@ -1064,6 +1145,12 @@ impl Runtime { Claim::Claimed => {} _ => return Ok(false), } + self.install_claimed_zero_at(byte_offset, wake) + } + + fn install_claimed_zero_at(&self, byte_offset: u64, wake: bool) -> Result { + let chunk_size = self.backend.chunk_size(); + let chunk_idx = (byte_offset / chunk_size) as usize; // Zero the chunk's whole [byte_offset, +total_len) range, // walking every region it spans (clamped to the manifest's @@ -1180,13 +1267,22 @@ impl Runtime { /// mappings, in which case fall back to a private COPY of a zeroed /// buffer (correct, costs one private page per touched page). fn install_zero_substrate(&self, byte_offset: u64, wake: bool) -> Result { - use std::sync::atomic::Ordering; let chunk_size = self.backend.chunk_size(); let chunk_idx = (byte_offset / chunk_size) as usize; match self.claim_for_install(chunk_idx) { Claim::Claimed => {} _ => return Ok(false), } + self.install_claimed_zero_substrate(byte_offset, wake) + } + + fn install_claimed_zero_substrate( + &self, + byte_offset: u64, + wake: bool, + ) -> Result { + let chunk_size = self.backend.chunk_size(); + let chunk_idx = (byte_offset / chunk_size) as usize; // Walk every region the chunk spans (clamped to the manifest's // total). Per segment try ZEROPAGE; on a kernel that rejects it @@ -1235,7 +1331,7 @@ impl Runtime { /// Run forever, draining events from the UFFD and serving each /// page fault. Returns `Ok(())` cleanly when the UFFD is closed /// (Firecracker exited / sandbox destroyed). - pub fn run(&self) -> Result<(), HandlerError> { + pub fn run(self: &Arc) -> Result<(), HandlerError> { let mut faults_served: u64 = 0; loop { match self.uffd.read_event() { @@ -1285,7 +1381,71 @@ impl Runtime { } } - fn serve_pagefault(&self, fault_addr: u64) -> Result<(), HandlerError> { + /// Install one verified peer response without waking adjacent pages. + fn install_peer_page( + &self, + offset: u64, + page: crate::peer::PeerPage, + ) -> Result { + let bytes = Self::prepare_peer_page(&self.backend, &self.handle, offset, page)?; + self.install_prepared_peer_page(offset, bytes) + } + + /// Resolve durable bytes without a runtime reference or the install lock. + fn prepare_peer_page( + backend: &ChunkedMemoryBackend, + handle: &TokioHandle, + offset: u64, + page: crate::peer::PeerPage, + ) -> Result, HandlerError> { + use crate::peer::PeerPage; + Ok(match page { + PeerPage::Bytes(bytes) => Some(Bytes::from(bytes)), + PeerPage::Zero => None, + PeerPage::AltSource(_) => { + let hash = match backend.resolve(offset) { + Some(ResolvedPage::Chunk { hash }) => Some(hash), + Some(ResolvedPage::Canonical { canonical_offset }) => { + backend.canonical_chunk_hash(canonical_offset) + } + Some(ResolvedPage::Zero { .. }) => None, + None => { + return Err(HandlerError::PeerLost(format!( + "AltSource for unresolvable offset {offset:#x}" + ))) + } + }; + hash.map(|hash| handle.block_on(backend.fetch_chunk(hash))) + .transpose()? + } + }) + } + + fn install_prepared_peer_page( + &self, + offset: u64, + bytes: Option, + ) -> Result { + let chunk_idx = (offset / self.backend.chunk_size()) as usize; + if self.claim_for_install(chunk_idx) != Claim::Claimed { + return Ok(false); + } + self.install_claimed_peer_page(offset, bytes) + } + + fn install_claimed_peer_page( + &self, + offset: u64, + bytes: Option, + ) -> Result { + match bytes { + Some(bytes) => self.install_claimed_chunk_at(offset, bytes, false), + None if self.base_shm.is_some() => self.install_claimed_zero_substrate(offset, false), + None => self.install_claimed_zero_at(offset, false), + } + } + + fn serve_pagefault(self: &Arc, fault_addr: u64) -> Result<(), HandlerError> { // Locate the region this fault came from to translate // host_va → byte_offset (the inverse of `locate_offset`). let region = self @@ -1330,60 +1490,29 @@ impl Runtime { not dialing torn-down peer, falling through to resolve()" ); } else { - match peer.need_at(chunk_byte_offset) { - Ok(crate::peer::PeerPage::Bytes(bytes)) => { - let installed = - self.install_chunk_at(chunk_byte_offset, bytes.into(), true)?; - if !installed { - self.wake_page(page_aligned, page_size)?; - } - return Ok(()); - } - Ok(crate::peer::PeerPage::Zero) => { - // Private zero install — NEVER the shared base - // (sealed content is divergence by definition). - let installed = if self.base_shm.is_some() { - self.install_zero_substrate(chunk_byte_offset, true)? - } else { - self.install_zero_at(chunk_byte_offset, true)? - }; - if !installed { - self.wake_page(page_aligned, page_size)?; - } + let offsets = crate::peer::fault_window( + peer.seal(), + chunk_byte_offset, + region.offset + region.size as u64, + self.fault_around_chunks, + |idx| self.chunk_installed(idx), + ); + if offsets.is_empty() { + self.wake_page(page_aligned, page_size)?; + return Ok(()); + } + match self.serve_fault_window(peer, offsets, chunk_byte_offset) { + Ok(()) => { + self.wake_page(page_aligned, page_size)?; return Ok(()); } - Ok(crate::peer::PeerPage::AltSource(_durable)) => { - // Over-approximation demote: the source proved this - // chunk equals the durable manifest entry, so the - // normal resolve() arms below serve it (class 2). - } Err(e) => { - // Issue #227 (a): before fatally escalating, re-check - // whether the chunk landed meanwhile — the drain or a - // sibling fault may have installed it while THIS fault - // sat parked mid-redial (a multi-second window), or the - // drain may have just finished and torn down the export - // out from under us. If it's installed now, the peer - // was never actually needed: wake and return Ok rather - // than rewinding a healthy destination. if self.chunk_installed(chunk_idx) { - tracing::info!( - chunk_idx, - error = %e, - "sealed fault errored but chunk landed via drain/sibling; \ - waking instead of escalating PeerLost" - ); self.wake_page(page_aligned, page_size)?; return Ok(()); } - peer.mark_lost(); - if let Some(control) = self.control.as_ref() { - control.report(engram_migrate_proto::HandlerControl::PeerLost { - remaining: self.sealed_uninstalled_count(peer), - detail: e.to_string(), - }); - } - return Err(HandlerError::PeerLost(e.to_string())); + self.report_window_failure(peer, &e.to_string()); + return Err(e); } } } @@ -1495,13 +1624,199 @@ impl Runtime { .unwrap_or(false) } - /// Latch the drain-complete flag (DrainDone). Once set, the fault - /// path stops dialing the peer (see `drain_done` field doc). - fn mark_drain_done(&self) { + fn report_window_failure(&self, peer: &crate::peer::PeerSession, detail: &str) { + peer.mark_lost(); + if let Some(control) = &self.control { + control.report(engram_migrate_proto::HandlerControl::PeerLost { + remaining: self.sealed_uninstalled_count(peer), + detail: detail.to_owned(), + }); + } + } + + fn install_window_page( + &self, + offset: u64, + page: crate::peer::PeerPage, + ) -> Result<(), HandlerError> { + if self.install_peer_page(offset, page)? { + self.fault_around_chunks_installed + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + Ok(()) + } + + /// Only the requested chunk holds fault priority. A busy prefetch slot + /// drops new speculative work; every demand uses the fault connection. + fn serve_fault_window( + self: &Arc, + peer: &Arc, + offsets: Vec, + fault_offset: u64, + ) -> Result<(), HandlerError> { + use crate::peer::PEER_READ_TIMEOUT; + let mut slot = self.fault_window.lock().expect("window slot poisoned"); + if slot.as_ref().is_some_and(|job| job.thread.is_finished()) { + slot.take().unwrap().stop(); + } + if self.is_drain_done() + || self.chunk_installed((fault_offset / self.backend.chunk_size()) as usize) + { + return Ok(()); + } + if slot.is_some() { + drop(slot); + // Overlapping demands also bypass the worker. Install claims + // deduplicate its later response without waiting for the suffix. + let page = peer + .need_at(fault_offset) + .map_err(|e| HandlerError::PeerLost(e.to_string()))?; + return self.install_window_page(fault_offset, page); + } + if peer.is_lost() { + return Err(HandlerError::PeerLost("peer already marked lost".into())); + } + let _fault = peer.begin_fault(); + let progress = Arc::new(WindowProgress::new()); + let worker_progress = progress.clone(); + let runtime = Arc::downgrade(self); + let backend = self.backend.clone(); + let handle = self.handle.clone(); + let worker_peer = peer.clone(); + let (first_tx, first_rx) = std::sync::mpsc::sync_channel(1); + let (installed_tx, installed_rx) = std::sync::mpsc::sync_channel(1); + let thread = std::thread::Builder::new() + .name("uffd-fault-around".into()) + .spawn(move || { + let connected = |conn: Option<&std::net::TcpStream>| { + let mut socket = worker_progress.socket.lock().expect("window poisoned"); + if worker_progress.stopped() { + return Ok(false); + } + if let Some(conn) = conn { + *socket = Some(conn.try_clone()?); + } + Ok(true) + }; + let result = (|| -> Result<(), String> { + let mut conn = worker_peer + .open_prefetch_conn(connected) + .map_err(|e| e.to_string())?; + worker_peer + .stream_window( + &mut conn, + &offsets, + |offset, page| { + if worker_progress.stopped() { + return Ok(()); + } + if offset == fault_offset { + first_tx.send(page).map_err(|e| e.to_string())?; + installed_rx + .recv_timeout(PEER_READ_TIMEOUT) + .map_err(|e| e.to_string())?; + } else { + // Fetches may outlive stop(). They own only the backend, + // never the runtime or its mappings. + let bytes = + Self::prepare_peer_page(&backend, &handle, offset, page) + .map_err(|e| e.to_string())?; + if worker_progress.stopped() { + return Ok(()); + } + let Some(rt) = runtime.upgrade() else { + return Ok(()); + }; + let chunk_idx = (offset / backend.chunk_size()) as usize; + // A competing installer may be fetching durable bytes too. + // Wait for its claim outside the cancellation lock. + if rt.claim_for_install(chunk_idx) != Claim::Claimed { + return Ok(()); + } + let _guard = + worker_progress.socket.lock().expect("window poisoned"); + if worker_progress.stopped() { + rt.release_chunk(chunk_idx); + return Ok(()); + } + if rt + .install_claimed_peer_page(offset, bytes) + .map_err(|e| e.to_string())? + { + rt.fault_around_chunks_installed + .fetch_add(1, Ordering::Relaxed); + } + } + Ok::<_, String>(()) + }, + || worker_progress.stopped(), + connected, + ) + .map_err(|e| e.to_string())??; + Ok(()) + })(); + if let Err(error) = result { + if let Some(rt) = runtime.upgrade() { + let message = engram_migrate_proto::HandlerControl::PeerLost { + remaining: rt.sealed_uninstalled_count(&worker_peer), + detail: error, + }; + let publish = || { + let _guard = worker_progress.socket.lock().expect("window poisoned"); + if worker_progress.stopped() { + return false; + } + if rt.sealed_uninstalled_count(&worker_peer) == 0 { + tracing::debug!( + "discarding fault-window failure: all sealed chunks installed" + ); + return false; + } + worker_peer.mark_lost(); + true + }; + if let Some(control) = &rt.control { + control.report_if(message, publish); + } else { + publish(); + } + } + } + })?; + *slot = Some(FaultWindow { progress, thread }); + drop(slot); + let page = first_rx + .recv_timeout(crate::peer::WINDOW_FIRST_RESPONSE_TIMEOUT) + .map_err(|e| HandlerError::PeerLost(format!("fault window first response: {e}")))?; + self.install_window_page(fault_offset, page)?; + // The fault loop installs the first response. The worker owns all later + // responses and never wakes a page or holds the fault-active guard. + let _ = installed_tx.send(()); + Ok(()) + } + + fn stop_fault_window(&self) { + let mut slot = self.fault_window.lock().expect("window slot poisoned"); + if let Some(job) = slot.take() { + job.stop(); + } + // Keep the slot locked until new windows are prohibited. self.drain_done .store(true, std::sync::atomic::Ordering::SeqCst); } + /// Cancel the prefetch worker and bound the join before publishing DrainDone. + fn mark_drain_done(&self) { + self.stop_fault_window(); + } + + #[cfg(test)] + fn finish_fault_window(&self) { + if let Some(job) = self.fault_window.lock().unwrap().take() { + job.thread.join().unwrap(); + } + } + /// True once the background drain has reported DrainDone. fn is_drain_done(&self) -> bool { self.drain_done.load(std::sync::atomic::Ordering::SeqCst) @@ -1775,6 +2090,7 @@ impl Runtime { /// open and treats our close as a protocol error). pub struct RunListenerOpts { pub prefault_trace: Option, + pub fault_around_chunks: usize, pub recorder_window: Duration, pub trace_output: Option, pub base_shm: Option, @@ -1794,6 +2110,7 @@ pub fn run_listener( ) -> Result { let RunListenerOpts { prefault_trace, + fault_around_chunks, recorder_window, trace_output, base_shm, @@ -1817,6 +2134,7 @@ pub fn run_listener( "handshake complete", ); let mut rt = Runtime::new(mappings, uffd, backend, handle, recorder_window, base_shm)?; + rt.fault_around_chunks = fault_around_chunks; if let Some(path) = trace_output { rt.set_trace_output(path); } @@ -1866,6 +2184,9 @@ pub fn run_listener( // path must already know not to dial the peer by // the time that frame is observed. rt.mark_drain_done(); + let fault_around_chunks_installed = rt + .fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed); let (faults, fault_us, fault_max_us) = peer.fault_stats(); // Review finding 6: stash the drain + live-fault // snapshot so it can be patched onto @@ -1893,6 +2214,7 @@ pub fn run_listener( faults, fault_us, fault_max_us, + fault_around_chunks_installed, }); } } @@ -1944,9 +2266,12 @@ pub fn run_listener( if let Some(peer_drain_stats) = peer_drain_stats { if let Some(path) = rt.prefault_stats_path() { let existing = std::fs::read(&path).ok(); - if let Some(patched) = + if let Some(mut patched) = patch_peer_fields(existing.as_deref(), peer_drain_stats) { + patched.fault_around_chunks_installed = rt + .fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed); write_prefault_stats(&path, &patched); } } @@ -1957,6 +2282,7 @@ pub fn run_listener( tracing::info!(pid, "starting fault loop"); let result = rt.run(); tracing::info!(pid, ?result, "fault loop returned"); + rt.stop_fault_window(); // Join the prefault producer before freezing the recorder so it's // quiesced. Once the uffd closes (FC exited) the prefault's next install // errors out, so this returns promptly rather than blocking teardown. @@ -2001,6 +2327,7 @@ mod tests { peer_alt_sourced: 0, peer_zero_chunks: 0, peer_live_faults: 0, + fault_around_chunks_installed: 0, }; write_prefault_stats(&stats_path, &stats); @@ -2068,6 +2395,7 @@ mod tests { peer_alt_sourced: 1, peer_zero_chunks: 4, peer_live_faults: 7, + fault_around_chunks_installed: 0, }, ); } @@ -3113,17 +3441,16 @@ mod tests { // (no privileged userfaultfd required); the end-to-end retry through // a real UFFD is covered by `fetch_failure_rolls_back_then_retries`. - /// Build a minimal single-page-per-chunk Runtime with NO real UFFD - /// region wired to the guest — only the install bookkeeping is - /// exercised. The UFFD is created over a throwaway anonymous page so - /// `Runtime::new` has a valid fd; tests here never call the install - /// ioctls, only `claim_chunk`/`mark_installed`/`release_chunk`/ - /// `wait_until_settled`/`chunk_installed`. Returns `None` if the - /// kernel forbids unprivileged uffd (so the test SKIPs, like the - /// real-region tests above). + /// Build a small runtime with a real anonymous UFFD mapping, for + /// install-state tests and demand-window tests. Returns None when + /// the kernel forbids userfaultfd. Tests that use the mapping must + /// keep it live until their reader and installer threads have joined. fn bookkeeping_runtime(n_chunks: usize) -> Option> { + test_runtime_with_chunk_size(n_chunks, 4096) + } + + fn test_runtime_with_chunk_size(n_chunks: usize, chunk_size: u64) -> Option> { let page_size = 4096u64; - let chunk_size = page_size; let total = chunk_size * n_chunks as u64; let dir = tempfile::tempdir().unwrap(); @@ -3200,6 +3527,748 @@ mod tests { )) } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stalled_suffix_does_not_block_another_fault_or_drain_and_late_reply_is_dropped() { + use crate::peer::{ + tests::{fake_source, page_resp}, + PeerSession, + }; + let Some(mut rt) = bookkeeping_runtime(4) else { + return; + }; + let (stalled_tx, stalled_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let release_rx = Arc::new(Mutex::new(release_rx)); + let stalled = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let addr = fake_source(vec![0, 1, 2, 3], move |req, offset| { + if offset == 4096 && !stalled.swap(true, std::sync::atomic::Ordering::SeqCst) { + stalled_tx.send(()).unwrap(); + release_rx.lock().unwrap().recv().unwrap(); + // This invalid late frame must not mark a drained peer lost. + return engram_migrate_proto::FromSource::ZeroChunk { + req_id: req, + chunk_offset: 0, + }; + } + page_resp(req, offset) + }); + let peer = tokio::task::spawn_blocking(move || { + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 16384) + }) + .await + .unwrap() + .unwrap(); + let peer = Arc::new(peer); + let mutable = Arc::get_mut(&mut rt).unwrap(); + mutable.fault_around_chunks = 2; + mutable.set_peer(peer.clone(), None); + let base = rt.mappings[0].base_host_virt_addr; + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let fault_rt = rt.clone(); + let first = + std::thread::spawn(move || done_tx.send(fault_rt.serve_pagefault(base)).unwrap()); + stalled_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + done_rx + .recv_timeout(Duration::from_secs(2)) + .expect("first fault waited for suffix") + .unwrap(); + first.join().unwrap(); + assert!(!peer.fault_active_within(Duration::ZERO)); + assert_eq!( + rt.fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed), + 1 + ); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let fault_rt = rt.clone(); + let second = std::thread::spawn(move || { + done_tx.send(fault_rt.serve_pagefault(base + 8192)).unwrap() + }); + done_rx + .recv_timeout(Duration::from_secs(2)) + .expect("second fault waited for suffix") + .unwrap(); + second.join().unwrap(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let drain_rt = rt.clone(); + let drain_peer = peer.clone(); + let drain = std::thread::spawn(move || { + let result = drain_rt.drain_from_peer(&drain_peer, None); + drain_rt.mark_drain_done(); + done_tx.send(result).unwrap(); + }); + done_rx + .recv_timeout(Duration::from_secs(2)) + .expect("drain waited for suffix") + .unwrap(); + drain.join().unwrap(); + let count = rt + .fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed); + release_tx.send(()).unwrap(); + assert!(!peer.is_lost()); + assert_eq!( + rt.fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed), + count + ); + assert!((0..4).all(|i| rt.chunk_installed(i))); + // SAFETY: DrainDone fenced all later installs; no installer remains. + unsafe { + libc::munmap(base as *mut _, 16384); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn overlapping_and_unrelated_faults_and_drain_bypass_stalled_suffix() { + use crate::peer::{ + tests::{fake_source, page_resp}, + PeerSession, + }; + let Some(mut rt) = bookkeeping_runtime(4) else { + return; + }; + let (stalled_tx, stalled_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let release_rx = Arc::new(Mutex::new(release_rx)); + let stalled = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let seen = Arc::new(Mutex::new(Vec::new())); + let source_seen = seen.clone(); + let addr = fake_source(vec![0, 1, 2, 3], move |req, offset| { + source_seen.lock().unwrap().push(offset); + if offset == 4096 && !stalled.swap(true, std::sync::atomic::Ordering::SeqCst) { + stalled_tx.send(()).unwrap(); + release_rx.lock().unwrap().recv().unwrap(); + } + page_resp(req, offset) + }); + let peer = tokio::task::spawn_blocking(move || { + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 16384) + }) + .await + .unwrap() + .unwrap(); + let peer = Arc::new(peer); + let mutable = Arc::get_mut(&mut rt).unwrap(); + mutable.fault_around_chunks = 3; + mutable.set_peer(peer.clone(), None); + let base = rt.mappings[0].base_host_virt_addr; + // Exercise the same serial fault loop: A, overlapping C, unrelated D. + // B stays blocked throughout all three faults and the drain. + for offset in [0, 8192, 12288] { + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let fault_rt = rt.clone(); + let fault = std::thread::spawn(move || { + done_tx + .send(fault_rt.serve_pagefault(base + offset)) + .unwrap(); + }); + done_rx + .recv_timeout(Duration::from_secs(2)) + .expect("demand waited for speculative B") + .unwrap(); + fault.join().unwrap(); + if offset == 0 { + stalled_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + } + assert!(!peer.fault_active_within(Duration::ZERO)); + } + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let drain_rt = rt.clone(); + let drain_peer = peer.clone(); + let drain = std::thread::spawn(move || { + done_tx + .send(drain_rt.drain_from_peer(&drain_peer, None)) + .unwrap(); + }); + let stats = done_rx + .recv_timeout(Duration::from_secs(2)) + .expect("drain waited for speculative B") + .unwrap(); + drain.join().unwrap(); + assert_eq!(stats.pulled, 1); + assert!((0..4).all(|i| rt.chunk_installed(i))); + assert_eq!( + rt.fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed), + 3 + ); + + // Let the worker attempt both duplicate installs, without cancellation. + release_tx.send(()).unwrap(); + rt.finish_fault_window(); + assert_eq!( + rt.fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed), + 3 + ); + let mut requests = seen.lock().unwrap().clone(); + requests.sort_unstable(); + assert_eq!(requests, [0, 4096, 4096, 8192, 8192, 12288]); + assert!(!peer.is_lost()); + // SAFETY: all faults, the drain, and the prefetch worker have joined. + unsafe { + let bytes = std::slice::from_raw_parts(base as *const u8, 16384); + assert!(bytes.iter().all(|byte| *byte == 0xAB)); + libc::munmap(base as *mut _, 16384); + } + } + + #[test] + fn fault_window_stop_detaches_a_worker_that_misses_the_deadline() { + let progress = Arc::new(WindowProgress::new()); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let (exited_tx, exited_rx) = std::sync::mpsc::channel(); + let thread = std::thread::spawn(move || { + release_rx.recv().unwrap(); + exited_tx.send(()).unwrap(); + }); + let job = FaultWindow { + progress: progress.clone(), + thread, + }; + let started = std::time::Instant::now(); + job.stop(); + assert!(started.elapsed() < crate::peer::PEER_READ_TIMEOUT + Duration::from_secs(1)); + assert!(progress.stopped()); + release_tx.send(()).unwrap(); + exited_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + } + + #[test] + fn fault_window_stop_deadline_includes_install_lock() { + let progress = Arc::new(WindowProgress::new()); + let worker_progress = progress.clone(); + let (locked_tx, locked_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let (exited_tx, exited_rx) = std::sync::mpsc::channel(); + let thread = std::thread::spawn(move || { + let _guard = worker_progress.socket.lock().unwrap(); + locked_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + assert!(worker_progress.stopped()); + exited_tx.send(()).unwrap(); + }); + locked_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + let job = FaultWindow { + progress: progress.clone(), + thread, + }; + let started = std::time::Instant::now(); + job.stop(); + let elapsed = started.elapsed(); + // Release even when the timing assertion fails. + release_tx.send(()).unwrap(); + exited_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(elapsed < crate::peer::PEER_READ_TIMEOUT + Duration::from_secs(1)); + assert!(progress.stopped()); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stalled_alt_source_does_not_block_stop_or_install_after_cancellation() { + use crate::peer::{ + tests::{fake_source, page_resp}, + PeerSession, + }; + use engram_substrate_proto::{read_frame, write_frame, FromWriter, ToWriter}; + use std::os::fd::AsFd; + use std::os::unix::net::UnixListener; + + // Both late success and late failure must be discarded. + for fail in [false, true] { + let Some(mut rt) = bookkeeping_runtime(2) else { + return; + }; + let dir = tempfile::tempdir().unwrap(); + let blob: Arc = + Arc::new(LocalBlobStorage::new(dir.path().join("blob"))); + let store = ChunkStore::new(blob); + let bytes = vec![0xCD; 4096]; + let hash = store.put_chunk(&bytes).await.unwrap(); + let manifest = Manifest { + schema_version: MANIFEST_SCHEMA_VERSION, + kind: ManifestKind::Memory, + chunk_size: ChunkSize::bytes(4096), + total_bytes: 8192, + chunks: vec![ChunkRef { offset: 4096, hash }], + parent: None, + working_set_trace: None, + annotations: serde_json::Value::Null, + }; + let path = dir.path().join("populate.sock"); + let listener = UnixListener::bind(&path).unwrap(); + let file_path = dir.path().join("chunk"); + std::fs::write(&file_path, bytes).unwrap(); + let (fetch_tx, fetch_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let durable = std::thread::spawn(move || { + let (mut socket, _) = listener.accept().unwrap(); + let request: ToWriter = read_frame(&mut socket).unwrap(); + assert!(matches!(request, ToWriter::Populate { .. })); + fetch_tx.send(()).unwrap(); + // No answer until stop has returned. This is below the client's + // 15-second timeout, so the worker cannot finish on its own. + release_rx.recv().unwrap(); + if fail { + write_frame( + &mut socket, + &FromWriter::PopulateErr { + msg: "durable source failed".into(), + }, + ) + .unwrap(); + } else { + write_frame(&mut socket, &FromWriter::Populated { len: 4096 }).unwrap(); + let file = std::fs::File::open(file_path).unwrap(); + engram_substrate_proto::send_fd(&socket, file.as_fd()).unwrap(); + } + }); + Arc::get_mut(&mut rt).unwrap().backend = Arc::new( + ChunkedMemoryBackend::new( + &manifest, + &manifest, + engram_chunk_store::reader::ChunkCacheReader::new(dir.path().join("cache")), + Some(Arc::new(crate::populate_client::PopulateClient::new(path))), + store, + ) + .unwrap(), + ); + let addr = fake_source(vec![0, 1], move |req, offset| { + if offset == 4096 { + engram_migrate_proto::FromSource::AltSource { + req_id: req, + chunk_offset: offset, + durable_sha256: *hash.as_bytes(), + } + } else { + page_resp(req, offset) + } + }); + let peer = Arc::new( + tokio::task::spawn_blocking(move || { + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 8192) + }) + .await + .unwrap() + .unwrap(), + ); + Arc::get_mut(&mut rt).unwrap().set_peer(peer.clone(), None); + let base = rt.mappings[0].base_host_virt_addr; + let fault_rt = rt.clone(); + std::thread::spawn(move || fault_rt.serve_pagefault(base)) + .join() + .unwrap() + .unwrap(); + fetch_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + let progress = rt + .fault_window + .lock() + .unwrap() + .as_ref() + .unwrap() + .progress + .clone(); + assert!( + progress.socket.try_lock().is_ok(), + "durable fetch held install lock" + ); + assert_eq!(Arc::strong_count(&rt), 1, "fetch retained runtime mappings"); + let started = std::time::Instant::now(); + rt.mark_drain_done(); + let elapsed = started.elapsed(); + assert!(progress.stopped()); + assert!(rt.is_drain_done()); + release_tx.send(()).unwrap(); + durable.join().unwrap(); + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while Arc::strong_count(&progress) != 1 { + assert!( + std::time::Instant::now() < deadline, + "detached worker did not exit" + ); + std::thread::sleep(Duration::from_millis(5)); + } + assert!(elapsed < crate::peer::PEER_READ_TIMEOUT + Duration::from_secs(1)); + assert!(!rt.chunk_installed(1)); + assert_eq!(rt.fault_around_chunks_installed.load(Ordering::Relaxed), 1); + assert!(!peer.is_lost()); + // SAFETY: the fault and detached prefetch worker have exited. + unsafe { + libc::munmap(base as *mut _, 8192); + } + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn demand_window_installs_all_shapes_and_wakes_only_faulting_page() { + use crate::peer::{tests::fake_source_geometry, PeerSession}; + let Some(mut rt) = test_runtime_with_chunk_size(4, 8192) else { + return; + }; + let addr = + fake_source_geometry(vec![0, 1, 2, 3], 8192, 32768, |req, offset| match offset { + 8192 => engram_migrate_proto::FromSource::ZeroChunk { + req_id: req, + chunk_offset: offset, + }, + 16384 => engram_migrate_proto::FromSource::AltSource { + req_id: req, + chunk_offset: offset, + durable_sha256: [0; 32], + }, + _ => { + let (bytes, lz4) = engram_migrate_proto::compress_page(vec![0xAB; 8192]); + let hash = engram_migrate_proto::wire_hash(&bytes); + engram_migrate_proto::FromSource::Page { + req_id: req, + chunk_offset: offset, + bytes, + hash, + lz4, + } + } + }); + let peer = tokio::task::spawn_blocking(move || { + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 8192, 32768) + }) + .await + .unwrap() + .unwrap(); + Arc::get_mut(&mut rt) + .unwrap() + .set_peer(Arc::new(peer), None); + let base = rt.mappings[0].base_host_virt_addr; + // Queue faults in the first chunk and the next chunk before the install. + let (tx, rx) = std::sync::mpsc::channel(); + let readers: Vec<_> = [0, 4096, 8192] + .into_iter() + .map(|offset| { + let tx = tx.clone(); + std::thread::spawn(move || { + // SAFETY: the registered mapping remains live until all readers join. + let byte = unsafe { std::ptr::read_volatile((base + offset) as *const u8) }; + tx.send((offset, byte)).unwrap(); + }) + }) + .collect(); + for _ in 0..3 { + assert!(matches!( + rt.uffd.read_event().unwrap(), + Some(Event::Pagefault { .. }) + )); + } + let serve = rt.clone(); + std::thread::spawn(move || serve.serve_pagefault(base)) + .join() + .unwrap() + .unwrap(); + assert_eq!(rx.recv_timeout(Duration::from_secs(2)).unwrap(), (0, 0xAB)); + assert!( + rx.recv_timeout(Duration::from_millis(20)).is_err(), + "adjacent page was woken" + ); + rt.finish_fault_window(); + assert!((0..4).all(|i| rt.chunk_installed(i))); + assert_eq!( + rt.fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed), + 4 + ); + rt.serve_pagefault(base + 4096).unwrap(); + assert_eq!( + rx.recv_timeout(Duration::from_secs(2)).unwrap(), + (4096, 0xAB) + ); + rt.serve_pagefault(base + 8192).unwrap(); + assert_eq!(rx.recv_timeout(Duration::from_secs(2)).unwrap(), (8192, 0)); + for reader in readers { + reader.join().unwrap(); + } + // SAFETY: no reader or installer still uses the mapping. + unsafe { + libc::munmap(base as *mut _, 32768); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn demand_window_deduplicates_concurrent_drain_install() { + use crate::peer::{ + tests::{fake_source, page_resp}, + PeerSession, + }; + let Some(mut rt) = bookkeeping_runtime(4) else { + return; + }; + let barrier = Arc::new(std::sync::Barrier::new(2)); + let source_barrier = barrier.clone(); + let addr = fake_source(vec![0, 1, 2, 3], move |req, offset| { + if offset == 4096 { + source_barrier.wait(); + source_barrier.wait(); + } + page_resp(req, offset) + }); + let peer = tokio::task::spawn_blocking(move || { + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 16384) + }) + .await + .unwrap() + .unwrap(); + Arc::get_mut(&mut rt) + .unwrap() + .set_peer(Arc::new(peer), None); + let base = rt.mappings[0].base_host_virt_addr; + let fault_rt = rt.clone(); + let fault = std::thread::spawn(move || fault_rt.serve_pagefault(base)); + barrier.wait(); + assert!(rt + .install_chunk_at(4096, Bytes::from(vec![0xCD; 4096]), false) + .unwrap()); + barrier.wait(); + fault.join().unwrap().unwrap(); + rt.finish_fault_window(); + assert_eq!( + rt.fault_around_chunks_installed + .load(std::sync::atomic::Ordering::Relaxed), + 3 + ); + // SAFETY: all chunks are installed, and the fault worker has joined. + assert_eq!(unsafe { std::ptr::read((base + 4096) as *const u8) }, 0xCD); + unsafe { + libc::munmap(base as *mut _, 16384); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn demand_window_skips_unsealed_and_installed_chunks() { + use crate::peer::{ + tests::{fake_source, page_resp}, + PeerSession, + }; + let Some(mut rt) = bookkeeping_runtime(4) else { + return; + }; + let seen = Arc::new(Mutex::new(Vec::new())); + let source_seen = seen.clone(); + let addr = fake_source(vec![0, 1, 3], move |req, offset| { + source_seen.lock().unwrap().push(offset); + page_resp(req, offset) + }); + let peer = tokio::task::spawn_blocking(move || { + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 16384) + }) + .await + .unwrap() + .unwrap(); + Arc::get_mut(&mut rt) + .unwrap() + .set_peer(Arc::new(peer), None); + assert!(rt + .install_chunk_at(4096, Bytes::from(vec![0xCD; 4096]), false) + .unwrap()); + let base = rt.mappings[0].base_host_virt_addr; + let fault_rt = rt.clone(); + std::thread::spawn(move || fault_rt.serve_pagefault(base)) + .join() + .unwrap() + .unwrap(); + rt.finish_fault_window(); + assert_eq!(*seen.lock().unwrap(), vec![0, 12288]); + assert!(!rt.chunk_installed(2)); + // SAFETY: the fault worker has joined and no thread uses the mapping. + unsafe { + libc::munmap(base as *mut _, 16384); + } + } + + /// Read the control backlog after the worker has exited. Any earlier + /// PeerLost must appear before DrainDone and fail this assertion. + fn assert_window_drain_done(rt: &Runtime, control_path: &std::path::Path) { + use engram_migrate_proto::{read_frame, HandlerControl}; + rt.mark_drain_done(); + rt.control + .as_ref() + .unwrap() + .report(HandlerControl::DrainDone { + pulled: 0, + alt_sourced: 0, + zero_chunks: 0, + ms: 0, + faults: 0, + fault_us: 0, + fault_max_us: 0, + fault_around_chunks_installed: 0, + }); + let mut subscriber = std::os::unix::net::UnixStream::connect(control_path).unwrap(); + subscriber + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + assert!(matches!( + read_frame::<_, HandlerControl>(&mut subscriber).unwrap(), + HandlerControl::DrainDone { .. } + )); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn demand_window_first_read_timeout_reconnects_before_first_response_deadline() { + use crate::peer::{ + tests::{fake_source, page_resp}, + ControlTx, PeerSession, + }; + // The single-chunk setting must retain the same retry policy. + for window in [1, 2] { + let Some(mut rt) = bookkeeping_runtime(4) else { + return; + }; + let requests = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let source_requests = requests.clone(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let release_rx = Arc::new(Mutex::new(release_rx)); + let addr = fake_source(vec![0], move |req, offset| { + if source_requests.fetch_add(1, Ordering::SeqCst) == 0 { + // Only a read timeout can start the retry: keep the first + // connection open without sending any response bytes. + let _ = release_rx.lock().unwrap().recv(); + } + page_resp(req, offset) + }); + let peer = Arc::new( + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 16384) + .unwrap(), + ); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("control.sock"); + let control = Arc::new(ControlTx::bind(&path).unwrap()); + let mutable = Arc::get_mut(&mut rt).unwrap(); + mutable.fault_around_chunks = window; + mutable.set_peer(peer.clone(), Some(control)); + let base = rt.mappings[0].base_host_virt_addr; + let fault_rt = rt.clone(); + let result = std::thread::spawn(move || fault_rt.serve_pagefault(base)) + .join() + .unwrap(); + release_tx.send(()).unwrap(); + result.unwrap(); + rt.finish_fault_window(); + assert_eq!(requests.load(Ordering::SeqCst), 2); + assert!(rt.chunk_installed(0)); + assert!(!peer.is_lost()); + assert_window_drain_done(&rt, &path); + // SAFETY: the fault and prefetch worker have exited. + unsafe { + libc::munmap(base as *mut _, 16384); + } + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn obsolete_suffix_failure_before_cancellation_does_not_report_peer_lost() { + use crate::peer::{ + tests::{fake_source, page_resp}, + ControlTx, PeerSession, + }; + let Some(mut rt) = bookkeeping_runtime(4) else { + return; + }; + let (pending_tx, pending_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let release_rx = Arc::new(Mutex::new(release_rx)); + let addr = fake_source(vec![0, 1], move |req, offset| { + if offset == 4096 { + pending_tx.send(()).unwrap(); + release_rx.lock().unwrap().recv().unwrap(); + engram_migrate_proto::FromSource::Error { + req_id: None, + message: "obsolete suffix failed".into(), + } + } else { + page_resp(req, offset) + } + }); + let peer = Arc::new( + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 16384).unwrap(), + ); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("control.sock"); + let control = Arc::new(ControlTx::bind(&path).unwrap()); + Arc::get_mut(&mut rt) + .unwrap() + .set_peer(peer.clone(), Some(control)); + let base = rt.mappings[0].base_host_virt_addr; + let fault_rt = rt.clone(); + std::thread::spawn(move || fault_rt.serve_pagefault(base)) + .join() + .unwrap() + .unwrap(); + pending_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + assert!(rt + .install_chunk_at(4096, Bytes::from(vec![0xCD; 4096]), false) + .unwrap()); + assert_eq!(rt.sealed_uninstalled_count(&peer), 0); + assert!(!rt.is_drain_done()); + assert!(!rt + .fault_window + .lock() + .unwrap() + .as_ref() + .unwrap() + .progress + .stopped()); + release_tx.send(()).unwrap(); + // Let the error reach publication BEFORE mark_drain_done cancels work. + rt.finish_fault_window(); + assert!(!peer.is_lost()); + assert_window_drain_done(&rt, &path); + // SAFETY: the fault and prefetch worker have exited. + unsafe { + libc::munmap(base as *mut _, 16384); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn demand_window_error_preserves_installed_prefix_and_reports_peer_lost() { + use crate::peer::{ + tests::{fake_source, page_resp}, + PeerSession, + }; + let Some(mut rt) = bookkeeping_runtime(4) else { + return; + }; + let addr = fake_source(vec![0, 1, 2, 3], |req, offset| { + if offset == 4096 { + engram_migrate_proto::FromSource::Error { + req_id: None, + message: "source lost".into(), + } + } else { + page_resp(req, offset) + } + }); + let peer = tokio::task::spawn_blocking(move || { + PeerSession::connect(addr.to_string(), "e".into(), "tok".into(), 4096, 16384) + }) + .await + .unwrap() + .unwrap(); + Arc::get_mut(&mut rt) + .unwrap() + .set_peer(Arc::new(peer), None); + let base = rt.mappings[0].base_host_virt_addr; + let fault_rt = rt.clone(); + std::thread::spawn(move || fault_rt.serve_pagefault(base)) + .join() + .unwrap() + .unwrap(); + rt.finish_fault_window(); + assert!(rt.chunk_installed(0)); + assert!(!rt.chunk_installed(1)); + assert!(rt.peer.as_ref().unwrap().is_lost()); + // SAFETY: the fault worker has joined and no thread uses the mapping. + unsafe { + libc::munmap(base as *mut _, 16384); + } + } + /// Acceptance #2 + #1 (state level): a claim that ends in error rolls /// the chunk back to `Empty` so the SAME chunk can be re-claimed and /// retried — the bit is never left set across a failure, which is the diff --git a/docs/adr/0045-live-session-teleport.md b/docs/adr/0045-live-session-teleport.md index 2cf3d8c22..23b4d0c5a 100644 --- a/docs/adr/0045-live-session-teleport.md +++ b/docs/adr/0045-live-session-teleport.md @@ -339,6 +339,42 @@ NBD independence after source destruction, and failed-restore cleanup checks. The destination starts restore before capture, drains memory and disk before commit, and uses a full checkpoint for durable memory after the move. +## Addendum — 2026-10-07: demand fault windows + +C2 demand faults request up to eight sealed, uninstalled chunks in one +`NeedWindow` frame. The faulting chunk is first. Following chunks stay +inside its memory region. Each response retains its own framing and wire +hash. Every response must match both the request ID and the next requested +offset, including ZeroChunk and AltSource. A dedicated prefetch worker sends +the window on its own Fault connection; the source accepts multiple such +connections per export. The fault loop installs the first response, wakes +the faulting page, and returns. The worker installs the suffix without holding +the fault-active guard. All copies use DONTWAKE; only the faulting page gets +an explicit wake. The existing install claims deduplicate concurrent drain +work. Retries request only the unfinished suffix. + +There is one prefetch slot and no waiting queue. While it is busy, a new fault +uses the separate demand connection, including faults inside the pending window. +Install claims deduplicate the worker's later response. New speculative windows +are dropped. DrainDone and teardown cancel the socket and wait at most five +seconds for the worker to exit, then detach it with a warning if necessary. +Cancellation prevents late installs and PeerLost reports; a detached worker +holds only a weak runtime reference between installs. Initial and retry TCP +connects have a five-second bound and check cancellation before and after +connecting. The first response also uses the existing five-second timeout. + +`--fault-around-chunks` or `ENGRAM_UFFD_FAULT_AROUND` selects 1 through 256 +chunks (default 8). A value of 1 preserves single-chunk demand behavior. +`fault_around_chunks_installed` counts successful demand installs, including +the faulting chunk, in handler stats and the drain completion report. +The drain order, hot-first policy, and seal rules are unchanged. + +The page-channel `PROTO_VERSION` advances from 5 to 6. Existing request +variant numbers stay fixed. The separate coordinator/host `WIRE_VERSION` +advances from 35 to 36. Its exact-match gate keeps hosts on the same version +during a roll, as for the page-channel v5 bump. The handler and host must use +page-channel version 6 together. Coordinator bincode payload bytes do not change. + ## Sources Carries forward **ADR 0042** (`docs/adr/0042-substrate-architecture-survey.md` + `0072-substrate-survey-evidence.md`): QEMU post-copy docs + Hines'09; Firecracker snapshot/UFFD docs + discussions #3119/#2938 (FC has no native live migration; UFFD restore is page-source-pluggable); `loopholelabs/drafter` + `silo` v0.2.21 (the unified-device post-copy-with-durable-backstop reference, AGPL — design reference only); REAP/FaaSnap/Catalyzer (working-set prefetch); CodeSandbox engineering blogs (`MAP_SHARED` continuous flush, local-NVMe memory). Kernel references for the substrate: `userfaultfd(2)` + `Documentation/admin-guide/mm/userfaultfd.rst` (MINOR/CONTINUE since 5.13 shmem/hugetlbfs; WP on shmem since 5.19; MISSING scope), `mmap(2)` MAP_FIXED semantics, KVM memslot/GUP behavior under address-space changes. Codebase seams verified against the current tree (the UFFD fill seam, `effective_restore_mode`, the disk flush scheduler, the eviction pipeline + lease, the operator `roll_node` + `desired_hosts`, the node-assets / detect-changes pipeline, the web admin UX). Direct read of the `loopholelabs/firecracker` compare diff happened **only** for the superseded v1 surface (recorded in Phase B history); the substrate (v2) design deliberately derives from kernel primitives alone.