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; diff --git a/crates/lance-graph/src/graph/cycle_sink.rs b/crates/lance-graph/src/graph/cycle_sink.rs index d9f80ed54..3f596fcb0 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, @@ -201,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. @@ -360,6 +364,7 @@ impl LanceCycleWriter { reconcile_scans: AtomicU64::new(0), committed_through: None, create_unknown: false, + clock: None, claim, #[cfg(test)] fault: TestFaults::default(), @@ -373,6 +378,7 @@ impl LanceCycleWriter { // mechanism, never a second head ledger. w.committed_through = w.max_cycle().await?; } + w.sync_clock(); Ok(w) } @@ -712,9 +718,49 @@ impl LanceCycleWriter { } // Storage has now shown the store to exist, by either route. self.create_unknown = false; + self.sync_clock(); 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> { + 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 /// writer has EVER run. Zero across a run of fresh monotonic commits — /// the "zero scans on the normal path" falsifier reads this. @@ -762,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 { @@ -996,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 @@ -2291,4 +2350,165 @@ 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) + ); + } }