From 8fed7353cc7086363bbf93807b606cf2d88365cf Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 18:01:14 +0000 Subject: [PATCH 1/3] D-WAL-6: version clock tests for LanceCycleWriter (red, stub returns None) Clock equals storage at every version on a store the writer created; a reopened writer seeds at the head and answers None before it, as the VersionClock contract (#1562) says; deleted rows keep their positions; the reconciled append keeps the clock equal to storage. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- crates/lance-graph/src/graph/cycle_sink.rs | 155 +++++++++++++++++++++ 1 file changed, 155 insertions(+) diff --git a/crates/lance-graph/src/graph/cycle_sink.rs b/crates/lance-graph/src/graph/cycle_sink.rs index d9f80ed54..1c7a6005b 100644 --- a/crates/lance-graph/src/graph/cycle_sink.rs +++ b/crates/lance-graph/src/graph/cycle_sink.rs @@ -93,6 +93,7 @@ use lance::session::Session; use lance_graph_contract::collapse_gate::MailboxId; use lance_graph_contract::kanban::{ExecTarget, KanbanColumn, KanbanMove}; use lance_graph_contract::scheduler::DatasetVersion; +use lance_graph_contract::temporal_pov::VersionClock; use lance_graph_planner::persist_sink::{ CommitError, CommitOutcome, CycleId, DetachedCycleBatch, FrameMeta, LandedSlot, SweepSlot, WalSink, WriteFailed, @@ -715,6 +716,15 @@ impl LanceCycleWriter { Ok(()) } + /// The store's version clock: rows durable through each version, so a + /// temporal read resolves a version range to a row range without touching + /// storage (D-WAL-6). `None` before the store exists and for a store with + /// deleted rows, where row positions no longer follow append order. + #[must_use] + pub fn version_clock(&self) -> Option<&VersionClock> { + None + } + /// How many reconciliation scans ([`find_frame`](Self::find_frame)) this /// writer has EVER run. Zero across a run of fresh monotonic commits — /// the "zero scans on the normal path" falsifier reads this. @@ -2291,4 +2301,149 @@ mod tests { "no frame added" ); } + + /// Rows a reader of version `v` sees, read from storage: the oracle the + /// in-memory clock must match. + async fn rows_in_storage(path: &str, v: u64) -> u64 { + let ds = Dataset::open(path).await.unwrap(); + let ds = ds.checkout_version(v).await.unwrap(); + ds.count_rows(None).await.unwrap() as u64 + } + + /// Commits of different sizes on a bootstrapped store (version 1 holds no + /// rows): one batch per cycle, `c` landings on `c` distinct rows. + async fn varied_store(path: &str) -> LanceCycleWriter { + let mut w = LanceCycleWriter::open(path).await.unwrap(); + w.bootstrap().await.unwrap(); + for c in 1..=6u64 { + let casts = (0..c).map(|i| artifact(c, i, 42, i)).collect(); + let head = w.head(); + persist_cycle(&mut w, CycleFrame::new(CycleId(c), head), casts) + .await + .unwrap(); + } + w + } + + /// D-WAL-6 wiring: the live clock answers, for every version, exactly the + /// rows a reader of that version finds in storage. + #[tokio::test] + async fn the_version_clock_matches_storage_at_every_version() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let path = path.to_str().unwrap(); + let w = varied_store(path).await; + let head = w.head().0; + assert_eq!(head, 7, "bootstrap plus six commits"); + let clock = w.version_clock().expect("a written store has a clock"); + assert_eq!(clock.head(), head); + for v in 1..=head { + assert_eq!( + clock.rows_through(v), + Some(rows_in_storage(path, v).await), + "version {v}" + ); + } + assert_eq!(clock.rows_through(1), Some(0), "bootstrap adds no rows"); + } + + /// Physical row positions durable at version `v`: rows plus deleted + /// rows, which keep their position (`VersionClock` contract). + async fn positions_in_storage(path: &str, v: u64) -> u64 { + let ds = Dataset::open(path).await.unwrap(); + let ds = ds.checkout_version(v).await.unwrap(); + ds.fragments() + .iter() + .map(|f| f.physical_rows.unwrap() as u64) + .sum() + } + + /// A reopened writer seeds its clock at the store's head, as the + /// `VersionClock` contract says: versions before the head read `None`, + /// never a guess; commits after it are recorded. + #[tokio::test] + async fn a_reopened_writer_seeds_its_clock_at_the_head() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let path = path.to_str().unwrap(); + drop(varied_store(path).await); + let mut w = LanceCycleWriter::open(path).await.unwrap(); + let head = w.head().0; + let clock = w.version_clock().expect("an existing store has a clock"); + assert_eq!(clock.base(), head); + assert_eq!(clock.rows_through(head), Some(rows_in_storage(path, head).await)); + assert_eq!(clock.rows_through(head - 1), None, "before the seed: no guess"); + let at = w.head(); + persist_cycle( + &mut w, + CycleFrame::new(CycleId(7), at), + vec![artifact(7, 0, 42, 0), artifact(7, 1, 42, 1)], + ) + .await + .unwrap(); + let next = w.head().0; + assert_eq!(next, head + 1); + let clock = w.version_clock().unwrap(); + assert_eq!(clock.rows_through(next), Some(rows_in_storage(path, next).await)); + } + + /// Deleted rows stay inside their range (the caller masks them): the + /// clock counts positions, not live rows, so rows appended after a delete + /// get the positions storage gives them. + #[tokio::test] + async fn deleted_rows_keep_their_positions() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let path = path.to_str().unwrap(); + drop(varied_store(path).await); + let mut ds = Dataset::open(path).await.unwrap(); + ds.delete(&format!("kind = {KIND_IMAGE} AND cycle = 3")) + .await + .unwrap(); + let mut w = LanceCycleWriter::open(path).await.unwrap(); + let head = w.head().0; + let live = rows_in_storage(path, head).await; + let positions = positions_in_storage(path, head).await; + assert!(live < positions, "the fixture deleted rows"); + assert_eq!(w.version_clock().unwrap().rows_through(head), Some(positions)); + let at = w.head(); + persist_cycle( + &mut w, + CycleFrame::new(CycleId(7), at), + vec![artifact(7, 0, 42, 0)], + ) + .await + .unwrap(); + let next = w.head().0; + assert_eq!( + w.version_clock().unwrap().rows_through(next), + Some(positions_in_storage(path, next).await) + ); + } + + /// The reconciled path (a published append whose acknowledgement was + /// lost) keeps the clock equal to storage. + #[tokio::test] + async fn the_clock_survives_a_reconciled_append() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let path = path.to_str().unwrap(); + let mut w = varied_store(path).await; + w.fault + .fail_append_published + .store(true, std::sync::atomic::Ordering::Relaxed); + let at = w.head(); + let out = persist_cycle( + &mut w, + CycleFrame::new(CycleId(7), at), + vec![artifact(7, 0, 42, 0), artifact(7, 1, 42, 1)], + ) + .await + .unwrap(); + assert!(matches!(out, CommitOutcome::Reconciled { .. }), "{out:?}"); + let head = w.head().0; + let clock = w.version_clock().expect("clock after reconcile"); + assert_eq!(clock.head(), head); + assert_eq!(clock.rows_through(head), Some(rows_in_storage(path, head).await)); + } } From a02f9630fafb1c5b29fba1c83ca2a4266ff416a8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 18:01:48 +0000 Subject: [PATCH 2/3] D-WAL-6: wire the version clock into LanceCycleWriter The clock is seeded at the store's head on open (and on bootstrap) from the physical row counts in the held manifest, then recorded after every commit outcome. Deleted rows keep their positions; a version whose rows went back starts a new clock. No reads beyond the manifest the handle holds. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- crates/lance-graph/src/graph/cycle_sink.rs | 85 +++++++++++++++++++--- 1 file changed, 75 insertions(+), 10 deletions(-) diff --git a/crates/lance-graph/src/graph/cycle_sink.rs b/crates/lance-graph/src/graph/cycle_sink.rs index 1c7a6005b..3f596fcb0 100644 --- a/crates/lance-graph/src/graph/cycle_sink.rs +++ b/crates/lance-graph/src/graph/cycle_sink.rs @@ -202,6 +202,9 @@ pub struct LanceCycleWriter { /// [`WalSink::commit_cycle`] refuses to run a SECOND `Create` and /// `DatasetNotFound` is never read as proof of absence. create_unknown: bool, + /// Rows durable per version since this writer first saw the store + /// (D-WAL-6). Seeded at the head on open, recorded after every commit. + clock: Option, /// The RAII registry claim. Held from BEFORE `open`'s first `.await`, so /// a cancelled or failed `open` releases its slot through `Drop` — no /// manual removal on any path, no leaked reservation. @@ -361,6 +364,7 @@ impl LanceCycleWriter { reconcile_scans: AtomicU64::new(0), committed_through: None, create_unknown: false, + clock: None, claim, #[cfg(test)] fault: TestFaults::default(), @@ -374,6 +378,7 @@ impl LanceCycleWriter { // mechanism, never a second head ledger. w.committed_through = w.max_cycle().await?; } + w.sync_clock(); Ok(w) } @@ -713,6 +718,7 @@ impl LanceCycleWriter { } // Storage has now shown the store to exist, by either route. self.create_unknown = false; + self.sync_clock(); Ok(()) } @@ -722,7 +728,37 @@ impl LanceCycleWriter { /// deleted rows, where row positions no longer follow append order. #[must_use] pub fn version_clock(&self) -> Option<&VersionClock> { - None + self.clock.as_ref() + } + + /// Bring the clock to the handle's version: record it when it is newer, + /// start a new clock at it when the rows went back (a rewrite) or the + /// writer has none yet. Row positions are physical rows: a deleted row + /// keeps its position and the reader masks it (`VersionClock` contract). + /// Reads only the manifest the handle already holds; no I/O. + fn sync_clock(&mut self) { + let Some(ds) = self.ds.as_ref() else { + return; + }; + let version = ds.version().version; + let Some(rows) = ds + .fragments() + .iter() + .map(|f| f.physical_rows.map(|n| n as u64)) + .sum::>() + else { + // A fragment without a physical row count: positions are unknown. + self.clock = None; + return; + }; + let keep = match self.clock.as_mut() { + Some(c) if version > c.head() => c.record(version, rows).is_ok(), + Some(c) => version == c.head() && c.head_rows() == rows, + None => false, + }; + if !keep { + self.clock = Some(VersionClock::new(version, rows)); + } } /// How many reconciliation scans ([`find_frame`](Self::find_frame)) this @@ -772,10 +808,10 @@ impl LanceCycleWriter { } } -impl WalSink for LanceCycleWriter { - /// THE single durable commit for a whole cycle — reconciliation-first, - /// fence second, append third; the outcome is fully honored. - async fn commit_cycle( +impl LanceCycleWriter { + /// The body of [`WalSink::commit_cycle`]; the trait method syncs the + /// version clock after it, whatever the outcome. + async fn commit_cycle_inner( &mut self, batch: DetachedCycleBatch, ) -> Result { @@ -1006,6 +1042,19 @@ impl WalSink for LanceCycleWriter { } } } +} + +impl WalSink for LanceCycleWriter { + /// THE single durable commit for a whole cycle — reconciliation-first, + /// fence second, append third; the outcome is fully honored. + async fn commit_cycle( + &mut self, + batch: DetachedCycleBatch, + ) -> Result { + let out = self.commit_cycle_inner(batch).await; + self.sync_clock(); + out + } /// Committed landing METADATA in stored canonical order, bounded by /// `after_cycle` (pushed into the Lance scan). Payloads are NOT read here @@ -2371,8 +2420,15 @@ mod tests { let head = w.head().0; let clock = w.version_clock().expect("an existing store has a clock"); assert_eq!(clock.base(), head); - assert_eq!(clock.rows_through(head), Some(rows_in_storage(path, head).await)); - assert_eq!(clock.rows_through(head - 1), None, "before the seed: no guess"); + assert_eq!( + clock.rows_through(head), + Some(rows_in_storage(path, head).await) + ); + assert_eq!( + clock.rows_through(head - 1), + None, + "before the seed: no guess" + ); let at = w.head(); persist_cycle( &mut w, @@ -2384,7 +2440,10 @@ mod tests { let next = w.head().0; assert_eq!(next, head + 1); let clock = w.version_clock().unwrap(); - assert_eq!(clock.rows_through(next), Some(rows_in_storage(path, next).await)); + assert_eq!( + clock.rows_through(next), + Some(rows_in_storage(path, next).await) + ); } /// Deleted rows stay inside their range (the caller masks them): the @@ -2405,7 +2464,10 @@ mod tests { let live = rows_in_storage(path, head).await; let positions = positions_in_storage(path, head).await; assert!(live < positions, "the fixture deleted rows"); - assert_eq!(w.version_clock().unwrap().rows_through(head), Some(positions)); + assert_eq!( + w.version_clock().unwrap().rows_through(head), + Some(positions) + ); let at = w.head(); persist_cycle( &mut w, @@ -2444,6 +2506,9 @@ mod tests { let head = w.head().0; let clock = w.version_clock().expect("clock after reconcile"); assert_eq!(clock.head(), head); - assert_eq!(clock.rows_through(head), Some(rows_in_storage(path, head).await)); + assert_eq!( + clock.rows_through(head), + Some(rows_in_storage(path, head).await) + ); } } From 6ced78872e0be10b7296ae692388ddf15a72d8f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 18:05:51 +0000 Subject: [PATCH 3/3] D-WAL-6: board and plan record the writer wiring Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- .claude/board/STATUS_BOARD.md | 2 +- .claude/plans/wal-shootout-v1.md | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 900808b03..8d7c76ccd 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -9,7 +9,7 @@ Plan: `.claude/plans/wal-shootout-v1.md`. Feeds D-LNC-7; decides nothing. | **D-WAL-3** | 60-minute soak per arm, CSV every 30 s | Measured, truncated at 27.5 min / 1,649 cycles by a container restart (§4a): M commit p50 flat 2.0–2.8 ms, L 5.8 → ~10.5 ms; M disk 1.57× L (WAL kept after each flush); both scans linear in rows; L RSS 95 → 762 MiB unexplained | drift, growth and RSS reported | | **D-WAL-4** | where L's resident memory goes | Measured: every Lance commit caches a full manifest clone per version (lance 13 `record_successful_commit`); append-only adds one fragment per version, so the cache grows as V²/2 to its 1 GiB default. 1,600 versions: default session 628 MiB cache / 570 MiB RSS, 8 MiB-capped session 7.8 MiB / 52 MiB | `examples/wal_rss_probe.rs`, default vs capped arm in separate processes | | **D-WAL-5** | cap the writer's Lance session | Done: open and both Create writes use a session with an 8 MiB metadata cache. 400 commits: 42.6 MB uncapped, under 8 MiB capped; reopen path covered (127 MB uncapped at 800) | `session_cache_stays_under_the_cap_over_many_commits`; disables on the Create write and on open each turn it red | -| **D-WAL-6** | version clock: a point of view as one row range | Done: `contract::temporal_pov::VersionClock`; v4 4.2 ns per `rows_at` vs 127 µs per-row scan at 400k rows | row-version oracle over 200 versions with gaps; disables (gap fill, start offset, base guard) each turn it red. Not yet wired into a writer or reader; seeding from storage at open is open | +| **D-WAL-6** | version clock: a point of view as one row range | Done: `contract::temporal_pov::VersionClock`; v4 4.2 ns per `rows_at` vs 127 µs per-row scan at 400k rows | row-version oracle over 200 versions with gaps; disables (gap fill, start offset, base guard) each turn it red. Wired into `LanceCycleWriter` 2026-10-15: seeded at the head on open from the manifest's physical row counts, recorded after every commit; deleted rows keep their positions; four tests against storage, three disables red. Reader wiring open | | **D-WAL-7** | `temporal.rs` without row copies | Done: `deinterlace_indices`, `local_trajectories_indices`, `local_trajectory_indices_of`; copying forms keep their signatures as wrappers | clone counter at 0 for the index form; removing the sort turns the two ordering tests red | ## D-MTP — Hyperon/MeTTa over the masking-algebra polyfill (2026-10-14) diff --git a/.claude/plans/wal-shootout-v1.md b/.claude/plans/wal-shootout-v1.md index 4d129d36d..6feef49a9 100644 --- a/.claude/plans/wal-shootout-v1.md +++ b/.claude/plans/wal-shootout-v1.md @@ -243,8 +243,10 @@ compactor) does to both curves. reconciliation asks for them. - **D-WAL-6.** `contract::temporal_pov::VersionClock`: rows of an append-only lane per version, eight bytes per version; a `TemporalPov` resolves to one row range in 4.2 ns on - v4 (a per-row scan of 400k rows: 127 µs). Not yet wired: a writer records into it after - each commit; seeding it from storage at open is open. + v4 (a per-row scan of 400k rows: 127 µs). Wired into `LanceCycleWriter` (2026-10-15): + seeded at the store's head on open from the physical row counts of the held manifest (no + reads), recorded after every commit outcome; versions before the seed read `None`; deleted + rows keep their positions; rows going back start a new clock. A reader is not wired yet. - **D-WAL-7.** `temporal.rs` index forms (`deinterlace_indices`, `local_trajectories_indices`, `local_trajectory_indices_of`); the copying forms wrap them. - **D-LNC-7, decided:** `LanceCycleWriter` stays the default with the capped session;