From 1b6e38f7bf887061101caf3a2bfa94e631d2aa85 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 25 Oct 2026 00:42:38 +0000 Subject: [PATCH 1/2] lab(drf-w4): hand flushed row images to the real CycleWriter (W4-W7) Feature handoff: RowImage::into_slot / into_slots turn flush output into owned SweepSlots (row bytes moved, one cast per distinct row, stream positions continued across cycles). RowImage::row is now Vec so the move into the payload is not a second copy. Tests drive lance-graph-supervisor's cycle_writer over a reconciliation- first fake WAL: W4 the source changes after flush and the landed image is the snapshot, durability waits for the WAL; W2 through the writer (raw per-patch casts lose a facet to the freeze's whole-payload last-wins); W6 lost ack -> ResubmitFrozen with the same casts -> Reconciled, while regenerating from the changed source -> HashConflict/Escalate; backpressure returns casts; W7 an empty cycle never calls the WAL. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G22yT6htkcdyXsihxxXdrg --- crates/drf-focus-probe/Cargo.toml | 14 +- crates/drf-focus-probe/src/handoff.rs | 52 ++ crates/drf-focus-probe/src/lib.rs | 12 +- .../tests/integration/focus.rs | 2 +- .../tests/integration/handoff.rs | 453 ++++++++++++++++++ .../drf-focus-probe/tests/integration/main.rs | 2 + 6 files changed, 530 insertions(+), 5 deletions(-) create mode 100644 crates/drf-focus-probe/src/handoff.rs create mode 100644 crates/drf-focus-probe/tests/integration/handoff.rs diff --git a/crates/drf-focus-probe/Cargo.toml b/crates/drf-focus-probe/Cargo.toml index 882432648..7d2489abb 100644 --- a/crates/drf-focus-probe/Cargo.toml +++ b/crates/drf-focus-probe/Cargo.toml @@ -2,7 +2,8 @@ # stale/foreign-key refusal, intra-focus prefetch lookahead. Timings only in # examples/p2_breakeven.rs, reported, not gated. # Own [workspace] root, excluded from the lance-graph workspace. Run from here: -# cargo test +# cargo test (W1-W3, focus, prefetch) +# cargo test --features handoff (W4-W7 against the real CycleWriter) [package] name = "drf-focus-probe" version = "0.0.0" @@ -14,6 +15,17 @@ description = "Lab probe D-DRF-P1: sealed 64x64 u16 map, full-witness FocusKey, [dependencies] ndarray = { git = "https://github.com/AdaWorldAPI/ndarray", branch = "master", default-features = false, features = ["std"] } +# W4-W7 handoff to the real CycleWriter (`cargo test --features handoff`). +lance-graph-planner = { path = "../lance-graph-planner", optional = true } + +[features] +handoff = ["dep:lance-graph-planner"] + +[dev-dependencies] +lance-graph-supervisor = { path = "../lance-graph-supervisor", features = ["cycle-driver"] } +lance-graph-planner = { path = "../lance-graph-planner" } +lance-graph-contract = { path = "../lance-graph-contract" } +tokio = { version = "1", features = ["rt", "macros", "sync"] } [patch."https://github.com/AdaWorldAPI/ndarray"] ndarray = { path = "../../../ndarray" } diff --git a/crates/drf-focus-probe/src/handoff.rs b/crates/drf-focus-probe/src/handoff.rs new file mode 100644 index 000000000..ffe0f4a66 --- /dev/null +++ b/crates/drf-focus-probe/src/handoff.rs @@ -0,0 +1,52 @@ +//! D-DRF W4–W7: hand flushed row images to the real `CycleWriter` as owned +//! casts (feature `handoff`). +//! +//! A [`RowImage`] owns its bytes, so once [`Writeback::flush`] returns, the +//! source rows and the focus can be dropped or changed while the writer is +//! still sealing. The compiler checks that; the tests check that what lands is +//! the flushed snapshot. +//! +//! One cast per distinct row. The writer's freeze folds same-row casts by +//! replacing the whole payload (later stream position wins), so two casts for +//! one row would drop one of them. [`Writeback::flush`] already coalesces per +//! row, which is why the handoff takes its output and nothing else. +//! +//! [`Writeback::flush`]: crate::Writeback::flush + +use lance_graph_planner::persist_sink::{CycleId, MailboxId, SweepSlot}; + +use crate::RowImage; + +impl RowImage { + /// This image as one cast for `owner`. The row bytes are moved, not copied. + pub fn into_slot(self, cycle: CycleId, stream_position: u64, owner: MailboxId) -> SweepSlot { + SweepSlot { + cycle, + stream_position, + owner, + row: u64::from(self.ord), + paired_move: None, + payload: self.row, + } + } +} + +/// One cast per image, in the images' ordinal order, at stream positions +/// `next_position..next_position + images.len()`. Returns the casts and the +/// next free position. Positions must keep increasing across cycles (the +/// writer's recovery watermark depends on it), so pass the returned value to +/// the next cycle. +pub fn into_slots( + images: Vec, + cycle: CycleId, + owner: MailboxId, + next_position: u64, +) -> (Vec, u64) { + let n = images.len() as u64; + let slots = images + .into_iter() + .zip(next_position..) + .map(|(img, pos)| img.into_slot(cycle, pos, owner)) + .collect(); + (slots, next_position + n) +} diff --git a/crates/drf-focus-probe/src/lib.rs b/crates/drf-focus-probe/src/lib.rs index d704240bf..cffeb3918 100644 --- a/crates/drf-focus-probe/src/lib.rs +++ b/crates/drf-focus-probe/src/lib.rs @@ -8,6 +8,9 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; +#[cfg(feature = "handoff")] +pub mod handoff; + /// Canonical SoA row ABI. pub const ROW: usize = 512; /// One SoA row. @@ -301,9 +304,11 @@ struct Patch { } /// The final image of one dirty canonical row, owned for handoff to a writer. +/// `row` is exactly [`ROW`] bytes. It is a `Vec` so the handoff can move it +/// into the writer's payload without a second copy. pub struct RowImage { pub ord: u16, - pub row: Row, + pub row: Vec, } /// Writes staged against one focus, coalesced per canonical row at flush @@ -410,13 +415,14 @@ impl Writeback<'_> { if out.last().is_none_or(|img| img.ord != p.ord) { // The focus was built from this same source, so every ordinal // was range-checked then; refuse rather than index if not. - let row = *soa + let row = soa .rows .get(p.ord as usize) .ok_or(Refusal::OrdinalOutOfRange { slot: usize::MAX, ord: p.ord, - })?; + })? + .to_vec(); out.push(RowImage { ord: p.ord, row }); } let img = out.last_mut().expect("pushed above"); diff --git a/crates/drf-focus-probe/tests/integration/focus.rs b/crates/drf-focus-probe/tests/integration/focus.rs index 40294daba..be160c875 100644 --- a/crates/drf-focus-probe/tests/integration/focus.rs +++ b/crates/drf-focus-probe/tests/integration/focus.rs @@ -408,7 +408,7 @@ fn w2_two_aliasing_slots_writing_different_facets_both_survive_in_any_order() { } let imgs = wb.flush(&soa).unwrap(); assert_eq!(imgs.len(), 1); - imgs[0].row + imgs.into_iter().next().unwrap().row }; let (a, b) = (image(true), image(false)); assert_eq!(a, b, "order must not matter"); diff --git a/crates/drf-focus-probe/tests/integration/handoff.rs b/crates/drf-focus-probe/tests/integration/handoff.rs new file mode 100644 index 000000000..957970a7d --- /dev/null +++ b/crates/drf-focus-probe/tests/integration/handoff.rs @@ -0,0 +1,453 @@ +//! D-DRF W4–W7: flushed row images handed to the real `CycleWriter` +//! (lance-graph-supervisor) over a reconciliation-first fake WAL. +use drf_focus_probe::handoff::into_slots; +use drf_focus_probe::*; +use lance_graph_contract::scheduler::DatasetVersion; +use lance_graph_planner::persist_sink::{ + CommitError, CommitOutcome, CycleId, DetachedCycleBatch, FrameMeta, LandedSlot, SweepSlot, + WalSink, WriteFailed, +}; +use lance_graph_supervisor::cycle_driver::SealRecovery; +use lance_graph_supervisor::cycle_writer::{SealError, SubmitError, cycle_writer}; +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use tokio::sync::Semaphore; + +const NROWS: usize = 1000; +const OFF: usize = 16; +const OFF2: usize = 64; +const OWNER: u32 = 7; + +fn rows() -> Vec { + (0..NROWS) + .map(|i| { + let mut r = [0u8; ROW]; + r[OFF..OFF + 4].copy_from_slice(&(i as i32).to_le_bytes()); + r + }) + .collect() +} +fn sid() -> SourceId { + SourceId { + dataset: 1, + branch: 1, + version: 1, + index_seal: 9, + } +} +fn map_from(id: u64, mut f: impl FnMut(usize) -> u16) -> Arc { + let mut o = Box::new([0u16; SLOTS]); + for (s, v) in o.iter_mut().enumerate() { + *v = f(s); + } + Arc::new(SealedMap::new(MapId(id), 9, o)) +} +fn facet(off: usize) -> Facet { + Facet::new(off).unwrap() +} + +/// What landed: one entry per durable commit. +#[derive(Clone)] +struct Landed { + cycle: CycleId, + batch_hash: u64, + image: BTreeMap>, +} + +/// One-writer fake WAL. Reconciles by `(cycle, batch_hash)` before appending, +/// fences on `base_version`, can hold commits behind a gate, and can lose the +/// acknowledgement of one cycle (commit lands, caller sees `Ambiguous`). +struct FakeWal { + head: u64, + landed: Arc>>, + gate: Option>, + entered: Arc, + lose_ack_of: Option, +} + +impl FakeWal { + fn new() -> Self { + Self { + head: 0, + landed: Arc::default(), + gate: None, + entered: Arc::default(), + lose_ack_of: None, + } + } +} + +impl WalSink for FakeWal { + async fn commit_cycle( + &mut self, + batch: DetachedCycleBatch, + ) -> Result { + self.entered.fetch_add(1, Ordering::SeqCst); + if let Some(gate) = &self.gate { + gate.acquire().await.expect("gate open").forget(); + } + let (cycle, hash) = (batch.frame.cycle, batch.batch_hash); + let existing = self + .landed + .lock() + .unwrap() + .iter() + .find(|l| l.cycle == cycle) + .map(|l| l.batch_hash); + match existing { + Some(h) if h == hash => { + return Ok(CommitOutcome::Reconciled { + current_head: DatasetVersion(self.head), + cycle, + batch_hash: hash, + }); + } + Some(h) => { + return Err(CommitError::HashConflict { + cycle, + stored_hash: h, + offered_hash: hash, + }); + } + None => {} + } + if batch.frame.base_version.0 != self.head { + return Err(CommitError::Fenced { + current_head: DatasetVersion(self.head), + }); + } + for (row, p) in &batch.image { + if p.len() != ROW { + return Err(CommitError::InvalidArtifact { + row: *row, + len: p.len(), + }); + } + } + self.head += 1; + self.landed.lock().unwrap().push(Landed { + cycle, + batch_hash: hash, + image: batch.image, + }); + if self.lose_ack_of.take() == Some(cycle) { + return Err(CommitError::Ambiguous { + cycle, + batch_hash: hash, + cause: "ack lost".into(), + }); + } + Ok(CommitOutcome::Committed { + version: DatasetVersion(self.head), + cycle, + batch_hash: hash, + }) + } + + async fn scan_sealed(&self, _after: Option) -> Result, WriteFailed> { + Ok(Vec::new()) + } + + async fn timeline(&self) -> Result, WriteFailed> { + Ok(Vec::new()) + } +} + +/// Stage `n` writes over 37 distinct rows (slot s -> row (s % 37) * 11) and +/// flush them into casts for `cycle`. +fn dirty_cycle(src: &[Row], cycle: u64, next_pos: u64) -> (Vec, u64) { + let soa = Soa { + rows: src, + id: sid(), + }; + let m = map_from(400 + cycle, |s| ((s % 37) * 11) as u16); + let mgr = FocusManager::new(); + let f = mgr.load(&soa, &m, Scope(1), facet(OFF)).unwrap(); + let mut wb = f.writeback(); + for s in 0..64 { + wb.stage(s, facet(OFF), 1000 + ((s % 37) as i32)).unwrap(); + } + into_slots(wb.flush(&soa).unwrap(), CycleId(cycle), OWNER, next_pos) +} + +async fn until(n: &AtomicUsize, v: usize) { + while n.load(Ordering::SeqCst) < v { + tokio::task::yield_now().await; + } +} + +#[tokio::test(flavor = "current_thread")] +async fn w4_owned_handoff_outlives_the_source_and_durability_waits_for_the_wal() { + let mut src = rows(); + let (slots, next) = dirty_cycle(&src, 1, 0); + assert_eq!(slots.len(), 37, "one cast per distinct row"); + assert_eq!(next, 37); + let expect: BTreeMap> = slots.iter().map(|s| (s.row, s.payload.clone())).collect(); + // The focus and the borrow of `src` ended inside `dirty_cycle`; the thought + // goes on and changes the source while the writer still holds the casts. + for r in src.iter_mut() { + r[OFF] ^= 0xFF; + } + + let mut wal = FakeWal::new(); + let gate = Arc::new(Semaphore::new(0)); + wal.gate = Some(gate.clone()); + let (entered, landed) = (wal.entered.clone(), wal.landed.clone()); + let (writer, mut h) = cycle_writer(wal, DatasetVersion(0), 4); + let run = tokio::spawn(writer.run()); + + let mut ticket = h.try_submit(CycleId(1), slots).expect("queue has room"); + until(&entered, 1).await; + // The WAL is holding the commit: nothing is durable yet. + assert!(ticket.try_sealed().is_none()); + assert_eq!(h.durable().through, None); + assert!(landed.lock().unwrap().is_empty()); + + gate.add_permits(1); + let sealed = ticket.sealed().await.expect("sealed"); + assert!(matches!(sealed.outcome, CommitOutcome::Committed { .. })); + let mark = h.wait_durable(CycleId(1)).await.expect("durable"); + assert_eq!(mark.through, Some(CycleId(1))); + assert_eq!(mark.head, DatasetVersion(1)); + // What landed is the flushed snapshot, not the source as it is now. + assert_eq!(landed.lock().unwrap()[0].image, expect); + + // A second cycle continues the stream positions (recovery watermark). + let (slots2, next2) = dirty_cycle(&src, 2, next); + assert_eq!(slots2[0].stream_position, 37); + assert_eq!(next2, 74); + gate.add_permits(1); + h.try_submit(CycleId(2), slots2) + .unwrap() + .sealed() + .await + .unwrap(); + drop(h); + let exit = run.await.unwrap(); + assert_eq!(exit.mark.through, Some(CycleId(2))); + assert_eq!(landed.lock().unwrap().len(), 2); +} + +#[tokio::test(flavor = "current_thread")] +async fn w2_through_the_writer_two_facets_of_one_row_both_land() { + let src = rows(); + let soa = Soa { + rows: &src, + id: sid(), + }; + let m = map_from(500, |s| if s < 2 { 500 } else { (s % NROWS) as u16 }); + let mgr = FocusManager::new(); + let f = mgr.load(&soa, &m, Scope(1), facet(OFF)).unwrap(); + let mut wb = f.writeback(); + wb.stage(0, facet(OFF), 7).unwrap(); + wb.stage(1, facet(OFF2), 9).unwrap(); + let (slots, _) = into_slots(wb.flush(&soa).unwrap(), CycleId(1), OWNER, 0); + + // Contrast arm, built by hand: one cast per staged write, each the sealed + // row with only its own facet. The writer's freeze keeps the later + // payload whole, so the first write is lost. This is why the handoff takes + // `flush` output and never raw patches. + let raw: Vec = [(OFF, 7i32), (OFF2, 9)] + .into_iter() + .zip(10u64..) + .map(|((off, v), pos)| { + let mut p = src[500].to_vec(); + p[off..off + 4].copy_from_slice(&v.to_le_bytes()); + SweepSlot { + cycle: CycleId(2), + stream_position: pos, + owner: OWNER, + row: 500, + paired_move: None, + payload: p, + } + }) + .collect(); + + let wal = FakeWal::new(); + let landed = wal.landed.clone(); + let (writer, h) = cycle_writer(wal, DatasetVersion(0), 4); + let run = tokio::spawn(writer.run()); + h.try_submit(CycleId(1), slots) + .unwrap() + .sealed() + .await + .unwrap(); + h.try_submit(CycleId(2), raw) + .unwrap() + .sealed() + .await + .unwrap(); + drop(h); + run.await.unwrap(); + + let l = landed.lock().unwrap(); + let get = |c: usize, off: usize| { + i32::from_le_bytes(l[c].image[&500][off..off + 4].try_into().unwrap()) + }; + assert_eq!( + (get(0, OFF), get(0, OFF2)), + (7, 9), + "coalesced: both facets" + ); + assert_eq!(get(1, OFF), 500, "raw: the first facet write was replaced"); + assert_eq!(get(1, OFF2), 9); +} + +#[tokio::test(flavor = "current_thread")] +async fn w6_lost_ack_resubmits_the_same_frozen_casts_and_resampling_fails_closed() { + let mut src = rows(); + let (slots, _) = dirty_cycle(&src, 1, 0); + let original: Vec<(u64, Vec)> = slots.iter().map(|s| (s.row, s.payload.clone())).collect(); + + let mut wal = FakeWal::new(); + wal.lose_ack_of = Some(CycleId(1)); + let landed = wal.landed.clone(); + let (writer, h) = cycle_writer(wal, DatasetVersion(0), 4); + let run = tokio::spawn(writer.run()); + let err = h + .try_submit(CycleId(1), slots) + .unwrap() + .sealed() + .await + .unwrap_err(); + let failure = match err { + SealError::Failed(f) => f, + other => panic!("expected Failed, got {other:?}"), + }; + assert_eq!(failure.recovery(), SealRecovery::ResubmitFrozen); + // The handed-back casts are the frozen batch, byte for byte. + let back: Vec<(u64, Vec)> = failure + .casts + .iter() + .map(|s| (s.row, s.payload.clone())) + .collect(); + assert_eq!(back, original); + // The writer is poisoned: the next submission comes back with its casts. + // The writer closes its queue before it answers the failed ticket. + assert!(matches!( + h.try_submit(CycleId(2), Vec::new()), + Err(SubmitError::Closed { .. }) + )); + drop(h); + let exit = run.await.unwrap(); + assert_eq!(landed.lock().unwrap().len(), 1, "it did land once"); + + // Meanwhile the source moved on. + for r in src.iter_mut() { + r[OFF] ^= 0x55; + } + + // Right: re-submit the SAME frozen casts on a fresh writer. Reconciled, no + // second append. + let (writer, h) = cycle_writer(exit.sink, exit.mark.head, 4); + let run = tokio::spawn(writer.run()); + let ok = h + .try_submit(CycleId(1), failure.casts.clone()) + .unwrap() + .sealed() + .await + .unwrap(); + assert!(matches!(ok.outcome, CommitOutcome::Reconciled { .. })); + assert_eq!(landed.lock().unwrap().len(), 1); + + // Wrong: regenerate cycle 1 from the live (changed) source. Different + // bytes, different hash: the WAL fails closed instead of overwriting. + let (resampled, _) = dirty_cycle(&src, 1, 0); + drop(h); + let exit = run.await.unwrap(); + let (writer, h) = cycle_writer(exit.sink, exit.mark.head, 4); + let run = tokio::spawn(writer.run()); + let err = h + .try_submit(CycleId(1), resampled) + .unwrap() + .sealed() + .await + .unwrap_err(); + match err { + SealError::Failed(f) => assert_eq!(f.recovery(), SealRecovery::Escalate), + other => panic!("expected Failed, got {other:?}"), + } + assert_eq!(landed.lock().unwrap().len(), 1); + drop(h); + run.await.unwrap(); +} + +#[tokio::test(flavor = "current_thread")] +async fn w6_backpressure_hands_the_casts_back_untouched() { + let src = rows(); + let mut wal = FakeWal::new(); + let gate = Arc::new(Semaphore::new(0)); + wal.gate = Some(gate.clone()); + let entered = wal.entered.clone(); + let (writer, h) = cycle_writer(wal, DatasetVersion(0), 1); + let run = tokio::spawn(writer.run()); + let (a, next) = dirty_cycle(&src, 1, 0); + let t1 = h.try_submit(CycleId(1), a).unwrap(); + until(&entered, 1).await; // cycle 1 is in the WAL, the queue is empty + let (b, next) = dirty_cycle(&src, 2, next); + let t2 = h.try_submit(CycleId(2), b).unwrap(); // fills the queue (depth 1) + let (c, _) = dirty_cycle(&src, 3, next); + let want: Vec> = c.iter().map(|s| s.payload.clone()).collect(); + match h.try_submit(CycleId(3), c) { + Err(SubmitError::Full { casts, .. }) => { + assert_eq!( + casts.iter().map(|s| s.payload.clone()).collect::>(), + want + ) + } + other => panic!("expected Full, got {:?}", other.map(|t| t.cycle)), + } + gate.add_permits(2); + t1.sealed().await.unwrap(); + t2.sealed().await.unwrap(); + drop(h); + run.await.unwrap(); +} + +#[tokio::test(flavor = "current_thread")] +async fn w7_a_cycle_with_no_dirty_row_publishes_nothing() { + let src = rows(); + let soa = Soa { + rows: &src, + id: sid(), + }; + let m = map_from(700, |s| (s % NROWS) as u16); + let mgr = FocusManager::new(); + let f = mgr.load(&soa, &m, Scope(1), facet(OFF)).unwrap(); + let (none, next) = into_slots(f.writeback().flush(&soa).unwrap(), CycleId(1), OWNER, 0); + assert!(none.is_empty()); + assert_eq!(next, 0); + + let wal = FakeWal::new(); + let (entered, landed) = (wal.entered.clone(), wal.landed.clone()); + let (writer, mut h) = cycle_writer(wal, DatasetVersion(0), 4); + let run = tokio::spawn(writer.run()); + let s = h + .try_submit(CycleId(1), none) + .unwrap() + .sealed() + .await + .unwrap(); + assert!(matches!(s.outcome, CommitOutcome::NoChange { .. })); + assert_eq!(entered.load(Ordering::SeqCst), 0, "the WAL is never called"); + assert_eq!( + h.wait_durable(CycleId(1)).await.unwrap().head, + DatasetVersion(0) + ); + // A dirty cycle after it publishes exactly one version. + let (slots, _) = dirty_cycle(&src, 2, next); + h.try_submit(CycleId(2), slots) + .unwrap() + .sealed() + .await + .unwrap(); + assert_eq!( + h.wait_durable(CycleId(2)).await.unwrap().head, + DatasetVersion(1) + ); + assert_eq!(landed.lock().unwrap().len(), 1); + drop(h); + run.await.unwrap(); +} diff --git a/crates/drf-focus-probe/tests/integration/main.rs b/crates/drf-focus-probe/tests/integration/main.rs index f25a0a451..2f3a53dbc 100644 --- a/crates/drf-focus-probe/tests/integration/main.rs +++ b/crates/drf-focus-probe/tests/integration/main.rs @@ -1 +1,3 @@ pub mod focus; +#[cfg(feature = "handoff")] +pub mod handoff; From 44794a4567390c250cf1fcc9ec29c9ddedcc3876 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 25 Oct 2026 00:43:50 +0000 Subject: [PATCH 2/2] board: D-DRF-W4 handoff entry (W4-W7) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G22yT6htkcdyXsihxxXdrg --- .../entries/2026-10-25-drf-w4-handoff.md | 28 +++++++++++++++++++ .claude/board/entries/README.md | 3 +- 2 files changed, 30 insertions(+), 1 deletion(-) create mode 100644 .claude/board/entries/2026-10-25-drf-w4-handoff.md diff --git a/.claude/board/entries/2026-10-25-drf-w4-handoff.md b/.claude/board/entries/2026-10-25-drf-w4-handoff.md new file mode 100644 index 000000000..b1a485447 --- /dev/null +++ b/.claude/board/entries/2026-10-25-drf-w4-handoff.md @@ -0,0 +1,28 @@ +# D-DRF-W4 — W4–W7: flushed row images handed to the real CycleWriter (lab) + +**Status:** TEST-PINNED (lab, `cargo test --features handoff`; no CI job runs this crate). +**Code:** `crates/drf-focus-probe/src/handoff.rs` (`RowImage::into_slot`, `into_slots`). +**Tests:** `tests/integration/handoff.rs`, driving `lance-graph-supervisor::cycle_writer` +over a reconciliation-first fake WAL. Plan: `.claude/plans/d-dr64-focus-cache-falsifiers-v1.md` §4. + +- **W4 PASS:** 64 writes over 37 rows → 37 owned casts. The focus and the borrow end at + flush; the source is then changed while the WAL holds the commit. Nothing is durable until + the WAL releases (`durable().through == None`, ticket unresolved); what lands is the flushed + snapshot. Stream positions continue across cycles (37 → 74). +- **W2 through the writer PASS:** two facets of one row land together. Contrast arm: one cast + per raw patch loses the first facet, because the writer's freeze keeps the later payload + whole. The per-row coalesce in `flush` is what makes the handoff safe. +- **W6 PASS:** lost acknowledgement → `ResubmitFrozen`, handed-back casts equal the frozen + batch byte for byte, re-submit on a fresh writer → `Reconciled`, one append. Regenerating + cycle 1 from the changed source → `HashConflict` → `Escalate`. A full queue hands the casts + back unchanged; a poisoned writer refuses new submissions. +- **W7 PASS:** a cycle with no dirty row gives zero casts → `NoChange`, the WAL is never + called, the head stays; the next dirty cycle publishes exactly one version. +- **Disables, each red:** one image per patch (W1–W3 tests and the through-writer W2 test); + `into_slots` not advancing the stream position (W4). +- **Change:** `RowImage::row` is now `Vec` (was `[u8; 512]`) so the move into the + cast's payload is not a second copy. + +OPEN: W5 (descriptor-only cast pointing into live SoA, lease held until capture) needs a +descriptor payload the writer does not have; W8 (eviction under memory pressure with live +readers) is not run. The fake WAL stands in for Lance; a real `WalSink` run is not done. diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md index b12cec492..46350fbda 100644 --- a/.claude/board/entries/README.md +++ b/.claude/board/entries/README.md @@ -25,10 +25,11 @@ index row, (3) no duplicate entry id. Checks 1 and 2 are deliberately opposite directions; the stranding this convention prevents shows up in exactly one of them, never both. -468 entries, 2026-08-06 .. 2026-10-25. +469 entries, 2026-08-06 .. 2026-10-25. | date | entry id | finding | file | |---|---|---|---| +| 2026-10-25 | `D-DRF-W4` | | [2026-10-25-drf-w4-handoff.md](2026-10-25-drf-w4-handoff.md) | | 2026-10-25 | `D-DRF-W3` | | [2026-10-25-drf-w3-ordered-apply.md](2026-10-25-drf-w3-ordered-apply.md) | | 2026-10-25 | `D-DRF-W1` | | [2026-10-25-drf-w1-writeback.md](2026-10-25-drf-w1-writeback.md) | | 2026-10-24 | `tanh-precision-symmetric-i8` | | [2026-10-24-tanh-precision-symmetric-i8.md](2026-10-24-tanh-precision-symmetric-i8.md) |