feat(uffd-handler): post-copy demand faults fetch a window of chunks (ADR 0045 C2 addendum)

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<u8> = (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);

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 {

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::<serde_json::Value>(&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(

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<u64>,
+    },
 }
 
 /// 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,

crates/engram-migrate-proto/tests/support/mod.rs

@@ -46,6 +46,12 @@ pub fn to_source() -> impl Strategy<Value = ToSource> {
                 purpose,
             }
         ),
+        (any::<u64>(), prop::collection::vec(any::<u64>(), 1..=256)).prop_map(
+            |(req_id, chunk_offsets)| ToSource::NeedWindow {
+                req_id,
+                chunk_offsets
+            }
+        ),
         (any::<u64>(), any::<u64>()).prop_map(|(req_id, chunk_offset)| ToSource::NeedAt {
             req_id,
             chunk_offset
@@ -125,9 +131,19 @@ pub fn handler_control() -> impl Strategy<Value = HandlerControl> {
             any::<u64>(),
             any::<u64>(),
             any::<u64>(),
+            any::<u64>(),
         )
             .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<Value = HandlerControl> {
                         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 { .. } => {}
     }
 }

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(

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

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"
     );
 }

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"