From 78be7aca7748e7e47b84b56700bd794b51e2182c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 29 May 2026 23:59:49 +0000 Subject: [PATCH 1/4] feat(kvs-lance): read-only Timeline view over Lance version history (Rubicon step 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the "SurrealDB-as-view-over-Lance" surface: a time-series view into the SoA backed entirely by Lance 6.0.0's native dataset versioning. No new storage — the Lance dataset already IS the source of truth (the Rubicon ruling: Lance leads, SurrealDB is one read-only view, never a store). New: - kvs/lance/timeline.rs — `Timeline` enumerates the dataset version history (`Dataset::versions()` -> Vec) and hands out `TimelineView`s, each an immutable snapshot pinned via `checkout_version(u64)`. `VersionInfo` carries {version:u64, timestamp_us}. Reads are tombstone-aware (a delete at/before the pinned version reads back absent). Read-only BY CONSTRUCTION: the view owns a checked-out snapshot and exposes no set/del/commit, so SurrealDB structurally cannot mutate the leading store. - kvs/lance/mod.rs — `Datastore::timeline()` shares the live dataset handle (no second open). - kvs/mvcc_source.rs — `MvccSource` trait + `LocalGeneratedMvcc`, recovered verbatim from the reverted PR #24 (additive, dead_code-gated until its consumer lands). - kvs/lance/tests.rs — 2 timeline tests (versions grow + monotone with commits; a historical view reads the SoA as it stood: present at the write version, absent before it). Every Lance call is confirmed against fetched lance-6.0.0 source (Version struct dataset.rs:202; versions() dataset.rs:2000; Scanner project/filter/limit) and against in-org usage in lance-graph graph/versioned.rs. Additive only — no existing signature changes. Verified: `cargo check -p surrealdb-core --features kv-lance` → Finished, 0 errors. Remaining warnings are never-used on the not-yet-wired consumer side (resolve when the ractor/kanban consumer calls `timeline()` in the next step). https://claude.ai/code/session_01R9AWgFa65uPnLyS2my2d2R --- .claude/board/AGENT_LOG.md | 23 ++ .claude/board/EPIPHANIES.md | 19 ++ surrealdb/core/src/kvs/lance/mod.rs | 19 ++ surrealdb/core/src/kvs/lance/tests.rs | 91 ++++++++ surrealdb/core/src/kvs/lance/timeline.rs | 264 +++++++++++++++++++++++ surrealdb/core/src/kvs/mod.rs | 1 + surrealdb/core/src/kvs/mvcc_source.rs | 170 +++++++++++++++ 7 files changed, 587 insertions(+) create mode 100644 surrealdb/core/src/kvs/lance/timeline.rs create mode 100644 surrealdb/core/src/kvs/mvcc_source.rs diff --git a/.claude/board/AGENT_LOG.md b/.claude/board/AGENT_LOG.md index 7be8d5743650..60a10ff0a1a7 100644 --- a/.claude/board/AGENT_LOG.md +++ b/.claude/board/AGENT_LOG.md @@ -72,3 +72,26 @@ one work path). **Commit(s):** _filled in by the committing session_ + +## 2026-05-29 — kvs-lance time-series view (SoA/Rubicon step 1) +**Branch:** claude/sleepy-cori-aRK2x +**Added:** +- `kvs/lance/timeline.rs` (264 LOC) — `Timeline` (read-only view over Lance + version history) + `TimelineView` (immutable snapshot at one version) + + `VersionInfo{version:u64, timestamp_us:Option}`. Uses only confirmed + Lance 6.0.0 surface: versions(), checkout_version(), version().version, + scan().project()/filter(). Tombstone-aware reads. +- `kvs/lance/mod.rs` — `Datastore::timeline()` accessor (shares the live + dataset handle, no second open). +- `kvs/mvcc_source.rs` (170 LOC) — `MvccSource` trait + `LocalGeneratedMvcc`, + borrowed verbatim from reverted PR #24 (2a54a32); additive, dead_code-gated + until its consumer (kv-tikv native MVCC / lance version source) lands. +- `kvs/lance/tests.rs` — 2 tests: versions grow+monotone with commits; a + historical TimelineView reads the SoA as it stood (present at write version, + absent before). +**Verify:** `cargo check -p surrealdb-core --features kv-lance` → Finished, 0 +errors (6m43s cold). Timeline tests: see commit (run pending at log time). +**Deferred (per user):** thinking-style i4-32 `I4x32::pack/unpack` are todo!() +in lance-graph-contract (carrier glitch) — NOT touched; wiring first. +**Next:** ractor mailbox owns SoA → publishes link onto this timeline (kanban); +EpisodicWitness64; replace BindSpace; wire deprecated→cognitive-shader-driver. diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index f2b78028ea0d..e444c3609b63 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -57,3 +57,22 @@ guard invariant 1; invariant 2 needs new range-scan tests once **Cross-ref:** `lance/mod.rs:362-417` (get path), `lance/mod.rs:607-642` (scan_impl), `lance-backend/README.md` § "Transaction Model". + +## 2026-05-29 — kvs-lance Timeline: Lance-native versioning IS the time-series view +**Status:** FINDING +**Scope:** surrealdb/core/src/kvs/lance/{timeline.rs,mod.rs} + +The "SurrealDB-as-view-over-Lance" (Rubicon) surface needs no new storage: +Lance 6.0.0 already exposes the full timeline. `Dataset::versions() -> +Vec, metadata}>` enumerates the +history; `checkout_version(u64)` pins an immutable snapshot. Confirmed against +fetched lance-6.0.0 source (dataset.rs:202 Version struct; dataset.rs:2000 +versions()) AND against in-org usage in lance-graph +crates/lance-graph/src/graph/versioned.rs:432. The new `Timeline` / +`TimelineView` types are read-only BY CONSTRUCTION (they own a checked-out +snapshot, expose no set/del/commit), so "SurrealDB never mutates the leading +store" is a type-system guarantee, not a convention. Per-key time-travel +(`checkout_version` + tombstone-as-data) was already wired in get()/scan_impl(); +this only adds the timeline *enumeration* + a read-only view handle. Compiles +clean under `cargo check -p surrealdb-core --features kv-lance` (Finished, 0 +errors; the only warnings are never-used on the not-yet-wired consumer side). diff --git a/surrealdb/core/src/kvs/lance/mod.rs b/surrealdb/core/src/kvs/lance/mod.rs index 8a55f07dfb7b..62a6d56b4b16 100644 --- a/surrealdb/core/src/kvs/lance/mod.rs +++ b/surrealdb/core/src/kvs/lance/mod.rs @@ -66,9 +66,17 @@ mod commit_gate; mod flusher; mod memtable; mod schema; +mod timeline; mod tx_buffer; mod wal; +// `Timeline` is consumed now (the `Datastore::timeline()` return type); +// `TimelineView` + `VersionInfo` are the read-side surface a kanban/replay +// consumer reaches for next. Re-exported crate-wide so that wiring lands +// without churn; `allow(unused_imports)` until the first in-tree consumer. +#[allow(unused_imports)] +pub(crate) use timeline::{Timeline, TimelineView, VersionInfo}; + use std::ops::Range; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -400,6 +408,17 @@ impl Datastore { self.dataset.read().await.inner.version().version } + /// Open a read-only [`Timeline`] over this datastore's version history. + /// + /// This is the "SurrealDB-as-view-over-Lance" surface (the Rubicon + /// ruling): the timeline enumerates Lance's native dataset versions and + /// hands out immutable [`TimelineView`]s. It shares the same dataset + /// handle as live transactions — no second open — and exposes reads + /// only, so it cannot mutate the leading store. + pub(crate) fn timeline(&self) -> Timeline { + Timeline::new(Arc::clone(&self.dataset)) + } + /// Test-only accessor for the underlying dataset Arc. /// /// Lets `lance::tests` exercise alternative write paths (notably diff --git a/surrealdb/core/src/kvs/lance/tests.rs b/surrealdb/core/src/kvs/lance/tests.rs index 278f46a911ca..ff280d6952cf 100644 --- a/surrealdb/core/src/kvs/lance/tests.rs +++ b/surrealdb/core/src/kvs/lance/tests.rs @@ -1900,3 +1900,94 @@ async fn writepath_legacy_commit_gate_provides_snapshot_iso() { ds.shutdown().await.expect("shutdown"); } + +// ────────────────────────────────────────────────────────────────────────── +// Timeline — read-only time-series view (the Rubicon "SurrealDB-as-view") +// ────────────────────────────────────────────────────────────────────────── + +/// The timeline enumerates Lance's native version history and that history +/// grows by (at least) one entry per committed transaction. +#[tokio::test] +async fn test_timeline_versions_grow_with_commits() { + let path = unique_tmp_path(); + let path_str = path.to_str().expect("path is valid UTF-8"); + let ds = Datastore::new(path_str, LanceConfig::default()).await.expect("create"); + + let timeline = ds.timeline(); + let v_start = timeline.versions().await.expect("versions @ start").len(); + + // Two committed write transactions → at least two new Lance versions. + for (k, v) in [(b"a".as_ref(), b"1".as_ref()), (b"b".as_ref(), b"2".as_ref())] { + let tx = ds.transaction(true, false).await.expect("tx"); + tx.set(k.to_vec(), v.to_vec()).await.expect("set"); + tx.commit().await.expect("commit"); + } + + let versions = timeline.versions().await.expect("versions @ end"); + assert!( + versions.len() >= v_start + 2, + "expected ≥{} versions after 2 commits, got {}", + v_start + 2, + versions.len() + ); + // Version numbers are monotone non-decreasing along the timeline. + for w in versions.windows(2) { + assert!(w[0].version <= w[1].version, "timeline not monotone: {:?}", versions); + } + // The latest entry matches the datastore's current version. + let latest = timeline.latest_version().await; + assert_eq!( + versions.last().map(|vi| vi.version), + Some(latest), + "timeline tail must equal current_version" + ); + + ds.shutdown().await.expect("shutdown"); +} + +/// A historical [`TimelineView`] reads the SoA as it stood at that version: +/// a key written at version N is absent from a view pinned before N and +/// present from the view at/after N. +#[tokio::test] +async fn test_timeline_view_reads_historical_soa() { + let path = unique_tmp_path(); + let path_str = path.to_str().expect("path is valid UTF-8"); + let ds = Datastore::new(path_str, LanceConfig::default()).await.expect("create"); + + let timeline = ds.timeline(); + let v_before = timeline.latest_version().await; + + // Commit a single key. + { + let tx = ds.transaction(true, false).await.expect("tx"); + tx.set(b"hist".to_vec(), b"present".to_vec()).await.expect("set"); + tx.commit().await.expect("commit"); + } + let v_after = timeline.latest_version().await; + assert!(v_after > v_before, "commit did not advance the dataset version"); + + // View at the latest version sees the value. + let view_after = timeline.view_at(v_after).await.expect("view @ after"); + assert_eq!(view_after.version(), v_after); + assert_eq!( + view_after.get(&b"hist".to_vec()).await.expect("get @ after").as_deref(), + Some(b"present".as_ref()), + "view at the write version must see the key" + ); + + // View at the pre-write version must NOT see the value. + let view_before = timeline.view_at(v_before).await.expect("view @ before"); + assert!( + view_before.get(&b"hist".to_vec()).await.expect("get @ before").is_none(), + "view before the write must not see the key (time-travel violated)" + ); + + // scan() at the latest version surfaces the live row. + let rows = view_after.scan().await.expect("scan @ after"); + assert!( + rows.iter().any(|(k, v)| k == b"hist" && v == b"present"), + "timeline scan must surface the committed row; got {rows:?}" + ); + + ds.shutdown().await.expect("shutdown"); +} diff --git a/surrealdb/core/src/kvs/lance/timeline.rs b/surrealdb/core/src/kvs/lance/timeline.rs new file mode 100644 index 000000000000..24e63d1c8aab --- /dev/null +++ b/surrealdb/core/src/kvs/lance/timeline.rs @@ -0,0 +1,264 @@ +#![cfg(feature = "kv-lance")] + +//! # Timeline — read-only time-series view into the Lance-backed SoA +//! +//! ## Role (the Rubicon ruling) +//! +//! lance-graph is the leading store / source of truth; SurrealDB is **one +//! VIEW over it** (the Rubicon kanban), **never a store**. This module is +//! the concrete surface for that ruling: a [`Timeline`] enumerates the +//! Lance dataset's native version history and hands out [`TimelineView`]s — +//! each an *immutable* snapshot pinned to one version. The view exposes +//! reads only; it owns no write path, so "SurrealDB never mutates the SoA" +//! is enforced by the type system, not by convention. +//! +//! ## What it builds on (all confirmed Lance 6.0.0 surface) +//! +//! - `Dataset::versions() -> Vec` — the version +//! timeline (same call the lance-graph `VersionedGraph` uses). +//! - `Dataset::checkout_version(u64)` — pin a read-only snapshot. +//! - `Dataset::version().version -> u64` — the latest version number. +//! - `Scanner::project(&["key","val","version"])` + `filter` — the same +//! scan idiom as [`super::Transaction::scan_impl`]. +//! +//! ## Wall-clock time (owned, not inferred) +//! +//! The `version` column written by [`super::schema::KvSchema`] carries the +//! MVCC version *at write time* (a `u64`). Wall-clock timestamps for the +//! kanban come from the Lance [`Version`] records' own ordering; we surface +//! the raw `version: u64` (monotone, gap-tolerant) as the canonical +//! timeline axis and expose the Lance-reported timestamp opportunistically. +//! Consumers that need a guaranteed wall-clock column should follow the +//! lance-graph `VersionedGraph` pattern of writing an owned +//! `timestamp_micros` column; the SurrealDB KV schema deliberately keeps +//! only `version` to stay byte-minimal. +//! +//! ## Additive shape +//! +//! New module; no existing item is modified. `Datastore::timeline()` is the +//! single new accessor (see `mod.rs`). + +use std::sync::Arc; + +use futures::TryStreamExt; +use tokio::sync::RwLock; + +use super::DatasetHandle; +use super::schema::KvSchema; +use crate::kvs::err::{Error, Result}; +use crate::kvs::{Key, Val}; + +/// One entry in the version history of the underlying Lance dataset. +/// +/// `version` is the canonical, monotone (gap-tolerant) timeline axis. It is +/// the same `u64` that [`super::Transaction::get`]'s `version: Option` +/// parameter accepts, so any `VersionInfo` returned here can be fed straight +/// back into a point-in-time read. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct VersionInfo { + /// Lance native dataset version number. Monotone non-decreasing across + /// commits; gaps are permitted. + pub version: u64, + /// Lance-reported creation time of this version, in microseconds since + /// the Unix epoch, when available. `None` if Lance does not surface a + /// timestamp for the version. + pub timestamp_us: Option, +} + +/// A read-only time-series view over the Lance-backed SoA. +/// +/// Constructed from the same `Arc>` the live +/// [`super::Datastore`] holds, so it observes committed history without a +/// second open. It exposes **no** write methods — the Rubicon "view, never +/// a store" invariant is structural. +#[derive(Clone)] +pub struct Timeline { + dataset: Arc>, +} + +impl Timeline { + /// Build a timeline over the given dataset handle. + pub(super) fn new(dataset: Arc>) -> Self { + Self { + dataset, + } + } + + /// The latest (current) version of the dataset. + pub async fn latest_version(&self) -> u64 { + self.dataset.read().await.inner.version().version + } + + /// Enumerate the full version history, oldest → newest. + /// + /// Mirrors `lance-graph::VersionedGraph::versions`: delegates to + /// `Dataset::versions()` and projects each Lance `Version` onto the + /// minimal [`VersionInfo`] the kanban needs. + pub async fn versions(&self) -> Result> { + let ds = self.dataset.read().await; + let raw = ds + .inner + .versions() + .await + .map_err(|e| Error::Datastore(format!("lance timeline versions: {e}")))?; + Ok(raw + .into_iter() + .map(|v| VersionInfo { + version: v.version, + // Lance `Version::timestamp` is a `chrono::DateTime`; + // fold it to epoch-micros for a dependency-light surface. + // If a future Lance bump changes the shape, only this line + // moves. + timestamp_us: Some(v.timestamp.timestamp_micros()), + }) + .collect()) + } + + /// Open an immutable view pinned to `version`. + /// + /// Returns `Err` if the version does not exist in the dataset history + /// (e.g. it was pruned by the background optimizer's + /// `cleanup_old_versions`). + pub async fn view_at(&self, version: u64) -> Result { + let ds = self.dataset.read().await; + let snapshot = ds + .inner + .checkout_version(version) + .await + .map_err(|e| Error::Datastore(format!("lance timeline checkout {version}: {e}")))?; + Ok(TimelineView { + version, + snapshot, + }) + } + + /// Open an immutable view pinned to the latest version. + pub async fn view_latest(&self) -> Result { + let version = self.latest_version().await; + self.view_at(version).await + } +} + +/// An immutable snapshot of the SoA at one version. +/// +/// Holds a checked-out Lance dataset (`checkout_version` yields an owned, +/// read-pinned `Dataset`). All methods are read-only; there is deliberately +/// no `set`/`del`/`commit` here. +pub struct TimelineView { + version: u64, + snapshot: lance::Dataset, +} + +impl TimelineView { + /// The version this view is pinned to. + pub fn version(&self) -> u64 { + self.version + } + + /// Point read of a single key as it existed at this version. + /// + /// Skips tombstoned rows (a delete recorded at-or-before this version + /// reads back as absent), matching [`super::Transaction::get`]'s + /// historical-read semantics. + pub async fn get(&self, key: &Key) -> Result> { + let filter = KvSchema::build_get_predicate(key); + let mut scanner = self.snapshot.scan(); + scanner + .filter(&filter) + .map_err(|e| Error::Datastore(format!("lance timeline get filter: {e}")))? + .project(&["val", "tombstone"]) + .map_err(|e| Error::Datastore(format!("lance timeline get project: {e}")))? + .limit(Some(1), None) + .map_err(|e| Error::Datastore(format!("lance timeline get limit: {e}")))?; + + let mut stream = scanner + .try_into_stream() + .await + .map_err(|e| Error::Datastore(format!("lance timeline get stream: {e}")))?; + + while let Some(batch) = stream + .try_next() + .await + .map_err(|e| Error::Datastore(format!("lance timeline get next: {e}")))? + { + if batch.num_rows() == 0 { + continue; + } + let tombstone_col = batch + .column_by_name("tombstone") + .ok_or_else(|| Error::Datastore("lance timeline get: missing tombstone".into()))? + .as_any() + .downcast_ref::() + .ok_or_else(|| { + Error::Datastore("lance timeline get: tombstone type mismatch".into()) + })?; + if tombstone_col.value(0) { + return Ok(None); + } + let val_col = batch + .column_by_name("val") + .ok_or_else(|| Error::Datastore("lance timeline get: missing val".into()))? + .as_any() + .downcast_ref::() + .ok_or_else(|| Error::Datastore("lance timeline get: val type mismatch".into()))?; + return Ok(Some(val_col.value(0).to_vec())); + } + Ok(None) + } + + /// Scan all live `(key, val)` pairs visible at this version, ascending + /// by key. Tombstoned keys are omitted. + /// + /// This is the "transparent view into the SoA" surface: a downstream + /// kanban / replay consumer walks `timeline.versions()` and calls + /// `view_at(v).scan()` to materialise the SoA as it stood at each + /// Rubicon commit — without any write path back into Lance. + pub async fn scan(&self) -> Result> { + let mut scanner = self.snapshot.scan(); + scanner + .project(&["key", "val", "tombstone"]) + .map_err(|e| Error::Datastore(format!("lance timeline scan project: {e}")))?; + + let mut stream = scanner + .try_into_stream() + .await + .map_err(|e| Error::Datastore(format!("lance timeline scan stream: {e}")))?; + + let mut out: Vec<(Key, Val)> = Vec::new(); + while let Some(batch) = stream + .try_next() + .await + .map_err(|e| Error::Datastore(format!("lance timeline scan next: {e}")))? + { + let key_col = batch + .column_by_name("key") + .ok_or_else(|| Error::Datastore("lance timeline scan: missing key".into()))? + .as_any() + .downcast_ref::() + .ok_or_else(|| Error::Datastore("lance timeline scan: key type mismatch".into()))?; + let val_col = batch + .column_by_name("val") + .ok_or_else(|| Error::Datastore("lance timeline scan: missing val".into()))? + .as_any() + .downcast_ref::() + .ok_or_else(|| Error::Datastore("lance timeline scan: val type mismatch".into()))?; + let tombstone_col = batch + .column_by_name("tombstone") + .ok_or_else(|| Error::Datastore("lance timeline scan: missing tombstone".into()))? + .as_any() + .downcast_ref::() + .ok_or_else(|| { + Error::Datastore("lance timeline scan: tombstone type mismatch".into()) + })?; + + for i in 0..batch.num_rows() { + if tombstone_col.value(i) { + continue; + } + out.push((key_col.value(i).to_vec(), val_col.value(i).to_vec())); + } + } + out.sort_by(|a, b| a.0.cmp(&b.0)); + Ok(out) + } +} diff --git a/surrealdb/core/src/kvs/mod.rs b/surrealdb/core/src/kvs/mod.rs index c136276db2f2..c3029300b6dc 100644 --- a/surrealdb/core/src/kvs/mod.rs +++ b/surrealdb/core/src/kvs/mod.rs @@ -36,6 +36,7 @@ mod indxdb; #[cfg(feature = "kv-lance")] mod lance; mod mem; +mod mvcc_source; mod rocksdb; mod surrealkv; mod tikv; diff --git a/surrealdb/core/src/kvs/mvcc_source.rs b/surrealdb/core/src/kvs/mvcc_source.rs new file mode 100644 index 000000000000..c6fafab52b6a --- /dev/null +++ b/surrealdb/core/src/kvs/mvcc_source.rs @@ -0,0 +1,170 @@ +//! Pluggable MVCC version-number source for SurrealDB storage backends. +//! +//! # Why this exists +//! +//! Different storage backends require different strategies for generating +//! monotonically-increasing version numbers used to implement MVCC +//! (multi-version concurrency control): +//! +//! - **Local backends** (mem, RocksDB, SurrealKV, Lance): a process-local +//! [`AtomicU64`] counter is sufficient because there is only one writer +//! process. [`LocalGeneratedMvcc`] provides this default. +//! +//! - **Distributed backends** (kv-tikv with native-MVCC): the PD (Placement +//! Driver) cluster issues HLC (Hybrid Logical Clock) timestamps that are +//! globally ordered. Sprint 2 will add `impl MvccSource for +//! TikvNativeMvccTxn` that delegates to the TiKV client — that +//! implementation is **out of scope here** and will be added +//! in `kvs/tikv/native_mvcc.rs` without touching this file. +//! +//! Making the source pluggable through a trait keeps the transaction layer +//! free from backend-specific timestamp logic and makes it straightforward +//! to compose alternative clock implementations in tests. +//! +//! # Additive shape +//! +//! This module introduces a **new** public trait and a single concrete +//! impl. No existing trait or struct is modified. The orchestrator +//! adds the `pub mod mvcc_source;` declaration to `kvs/mod.rs` during +//! consolidation; individual backends opt in by implementing the trait. +//! +//! # References +//! +//! See `.claude/plans/integration-plan.md §1` (Contracts table — +//! `MvccSource` trait sketch) and `§5b` (kv-tikv-native-mvcc consumer). + +// Sprint 0/1 scaffold (borrowed from the reverted PR #24): the trait + local +// impl land first; the consumers (kv-tikv native MVCC, and the lance timeline +// version source) wire in a later sprint. Allow dead_code until then so the +// additive module does not warn the workspace. +#![allow(dead_code)] + +use std::future::Future; +use std::sync::atomic::{AtomicU64, Ordering}; + +/// A source of monotonically non-decreasing MVCC version numbers. +/// +/// Implementors are expected to be cheaply cloneable (e.g. `Arc`-wrapped) +/// and safe to share across async tasks. +/// +/// # Contract +/// +/// - Each call to [`next_version`] MUST return a value `≥` any value +/// previously returned by the same instance. +/// - Values need not be consecutive; gaps are permitted (for example, an +/// HLC-backed source will return wall-clock milliseconds which can jump). +/// - Implementations must be `Send + Sync` so they can be stored inside +/// `Arc` and shared across Tokio tasks. +/// +/// [`next_version`]: MvccSource::next_version +pub trait MvccSource: Send + Sync { + /// Return the next MVCC version number. + /// + /// The returned `Future` must be `Send` so that callers can `.await` it + /// inside Tokio multi-threaded runtimes. + fn next_version(&self) -> impl Future> + Send; +} + +// --------------------------------------------------------------------------- +// Default implementation — process-local atomic counter +// --------------------------------------------------------------------------- + +/// A process-local MVCC version source backed by an [`AtomicU64`] counter. +/// +/// This matches SurrealDB's current behaviour for single-node backends: a +/// simple fetch-and-increment that returns a strictly increasing `u64` on +/// every call. It is **not** suitable for distributed deployments where +/// multiple writer processes must agree on a global ordering. +/// +/// # Example +/// +/// ```rust,ignore +/// use crate::kvs::mvcc_source::{LocalGeneratedMvcc, MvccSource}; +/// +/// let src = LocalGeneratedMvcc::new(); +/// let v1 = src.next_version().await.unwrap(); +/// let v2 = src.next_version().await.unwrap(); +/// assert!(v2 > v1); +/// ``` +pub struct LocalGeneratedMvcc { + /// The underlying monotonic counter. Starts at 1 so that version 0 can + /// be used as a sentinel "no version" value by callers if needed. + counter: AtomicU64, +} + +impl LocalGeneratedMvcc { + /// Create a new counter starting at version `1`. + pub fn new() -> Self { + Self { + counter: AtomicU64::new(1), + } + } +} + +impl Default for LocalGeneratedMvcc { + fn default() -> Self { + Self::new() + } +} + +impl MvccSource for LocalGeneratedMvcc { + /// Atomically increment the counter and return the previous value. + /// + /// This is an infallible, synchronous operation wrapped in an immediately- + /// ready `Future` so it conforms to the async trait contract. The + /// [`Ordering::Relaxed`] store is sufficient because the counter is + /// always accessed through a single `AtomicU64` with sequentially- + /// consistent fetch-add semantics on the same process. + fn next_version(&self) -> impl Future> + Send { + let v = self.counter.fetch_add(1, Ordering::SeqCst); + std::future::ready(Ok(v)) + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::{LocalGeneratedMvcc, MvccSource}; + + /// Verify that successive calls to `next_version` are strictly increasing. + #[tokio::test] + async fn local_mvcc_is_monotonically_increasing() { + let src = LocalGeneratedMvcc::new(); + + let mut prev = src.next_version().await.expect("next_version must not fail"); + for _ in 0..1_000 { + let next = src.next_version().await.expect("next_version must not fail"); + assert!( + next > prev, + "expected version {next} > {prev} but it was not strictly greater" + ); + prev = next; + } + } + + /// Verify that independent instances do not share state. + #[tokio::test] + async fn local_mvcc_instances_are_independent() { + let a = LocalGeneratedMvcc::new(); + let b = LocalGeneratedMvcc::new(); + + let va = a.next_version().await.unwrap(); + let vb = b.next_version().await.unwrap(); + + // Both start at 1; they must be equal, not one ahead of the other. + assert_eq!(va, vb, "independent instances must start from the same seed"); + } + + /// Smoke-test the `Default` impl. + #[test] + fn default_starts_at_one() { + let src = LocalGeneratedMvcc::default(); + // fetch_add returns the *old* value, so the first call returns 1. + let rt = tokio::runtime::Builder::new_current_thread().build().unwrap(); + let v = rt.block_on(src.next_version()).unwrap(); + assert_eq!(v, 1, "first version from a fresh counter must be 1"); + } +} From 9205453720aeb93a575430aaff89a0a12ec9d66f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 30 May 2026 00:08:39 +0000 Subject: [PATCH 2/4] fix(kvs-lance): timeline tests use LegacyCommitGate for per-commit versions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 2026-05-29 timeline tests assumed 1 commit = 1 Lance dataset version, but that only holds on WritePath::LegacyCommitGate. On the default LsmWithWal path, commits batch through WAL+memtable and the background flusher migrates them into Lance asynchronously, so the version timeline reflects flush boundaries, not individual commits (observed: 2 commits -> 1 version; single commit left latest_version unchanged). Both timeline tests now construct LegacyCommitGate configs, where commit() returns only after its own Lance commit lands. Result: 2 passed; 0 failed. The timeline.rs code was correct — only the test harness used the wrong write-path. Design note recorded in EPIPHANIES: the ractor/kanban consumer that publishes onto the timeline must run on the gate path (or force a flush) to get one timeline entry per Rubicon commit. https://claude.ai/code/session_01R9AWgFa65uPnLyS2my2d2R --- .claude/board/EPIPHANIES.md | 18 +++++++++++++++ surrealdb/core/src/kvs/lance/tests.rs | 32 ++++++++++++++++++++++++--- 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index e444c3609b63..95845bbb83dc 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -76,3 +76,21 @@ store" is a type-system guarantee, not a convention. Per-key time-travel this only adds the timeline *enumeration* + a read-only view handle. Compiles clean under `cargo check -p surrealdb-core --features kv-lance` (Finished, 0 errors; the only warnings are never-used on the not-yet-wired consumer side). + +## 2026-05-30 — kvs-lance timeline granularity = write-path-dependent (corrects 2026-05-29) +**Status:** FINDING +**Scope:** surrealdb/core/src/kvs/lance/{timeline.rs,tests.rs} + +The 2026-05-29 timeline tests wrongly assumed "1 commit = 1 Lance version". +On the DEFAULT `WritePath::LsmWithWal`, commits land in WAL+memtable and the +background flusher batches them into Lance asynchronously — so the timeline +reflects FLUSH BOUNDARIES, not individual commits (observed: 2 commits → 1 +version; a single commit left latest_version unchanged). For per-commit +timeline granularity (which the Rubicon kanban needs — each commit/plan/prune +a distinct entry) the datastore must use `WritePath::LegacyCommitGate`, where +`Transaction::commit` returns only after its own Lance commit lands. Tests +fixed to construct LegacyCommitGate configs; both pass (2/2). The timeline CODE +was correct; the test HARNESS used the wrong write-path. Design consequence: +the ractor/kanban consumer that publishes onto the timeline must run on the +gate path (or call an explicit flush) to get one timeline entry per Rubicon +commit. Cross-ref: config.rs WritePath docs; writepath_legacy_commit_gate_smoke. diff --git a/surrealdb/core/src/kvs/lance/tests.rs b/surrealdb/core/src/kvs/lance/tests.rs index ff280d6952cf..39cc5fcde027 100644 --- a/surrealdb/core/src/kvs/lance/tests.rs +++ b/surrealdb/core/src/kvs/lance/tests.rs @@ -1906,12 +1906,27 @@ async fn writepath_legacy_commit_gate_provides_snapshot_iso() { // ────────────────────────────────────────────────────────────────────────── /// The timeline enumerates Lance's native version history and that history -/// grows by (at least) one entry per committed transaction. +/// grows by one entry per committed transaction. +/// +/// Uses `WritePath::LegacyCommitGate`: only the gate path makes every commit +/// land synchronously as its own Lance dataset version. On the default +/// `LsmWithWal` path, commits batch through the WAL+memtable and the +/// background flusher migrates them into Lance asynchronously, so the +/// timeline reflects *flush boundaries*, not individual commits — the +/// correct granularity for a per-commit Rubicon kanban is the gate path. #[tokio::test] async fn test_timeline_versions_grow_with_commits() { let path = unique_tmp_path(); let path_str = path.to_str().expect("path is valid UTF-8"); - let ds = Datastore::new(path_str, LanceConfig::default()).await.expect("create"); + let ds = Datastore::new( + path_str, + LanceConfig { + write_path: WritePath::LegacyCommitGate, + ..LanceConfig::default() + }, + ) + .await + .expect("create"); let timeline = ds.timeline(); let v_start = timeline.versions().await.expect("versions @ start").len(); @@ -1952,7 +1967,18 @@ async fn test_timeline_versions_grow_with_commits() { async fn test_timeline_view_reads_historical_soa() { let path = unique_tmp_path(); let path_str = path.to_str().expect("path is valid UTF-8"); - let ds = Datastore::new(path_str, LanceConfig::default()).await.expect("create"); + // LegacyCommitGate: each commit lands synchronously as its own Lance + // version, so `v_before < v_after` holds per-commit (see the companion + // test's note on the LSM flush-boundary semantics). + let ds = Datastore::new( + path_str, + LanceConfig { + write_path: WritePath::LegacyCommitGate, + ..LanceConfig::default() + }, + ) + .await + .expect("create"); let timeline = ds.timeline(); let v_before = timeline.latest_version().await; From 5997eea9dd8b34207eca9a03dcf36899f2c77c9b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 30 May 2026 07:01:29 +0000 Subject: [PATCH 3/4] fix(kvs-lance): collapse write+delete commit into one Lance version A batch carrying both writes and deletes applied writes via MergeInsertBuilder::execute_reader and then deletes via a separate Dataset::delete, producing TWO native Lance versions. The intermediate (writes-applied, delete-pending) version was hidden from live readers by the datastore write lock, but Timeline::versions() surfaced it, so a replayer's view_at() could materialize a torn state that was never an atomic SurrealDB commit (codex P1 on PR #29). Fold deletes into tombstone rows (tombstone=true) streamed into the SAME merge_insert, so a commit/flush is exactly one Lance version. The read path already filters `tombstone = false`, so get/scan/keys are unchanged; the physical Dataset::delete path is removed. Add Transaction::build_tombstone_batch_lance (mirrors build_write_batch_lance) and retire the never-read delete_via_tombstone_row config flag. Regression: test_timeline_write_delete_commit_is_single_atomic_version asserts a mixed write+delete commit adds exactly one version and the resulting snapshot reflects all ops atomically. https://claude.ai/code/session_01R9AWgFa65uPnLyS2my2d2R --- .claude/board/AGENT_LOG.md | 37 ++++++++++ .claude/board/EPIPHANIES.md | 20 ++++++ surrealdb/core/src/kvs/config.rs | 5 -- surrealdb/core/src/kvs/lance/commit_gate.rs | 79 +++++++++++++-------- surrealdb/core/src/kvs/lance/flusher.rs | 76 +++++++++++--------- surrealdb/core/src/kvs/lance/mod.rs | 46 ++++++++++++ surrealdb/core/src/kvs/lance/tests.rs | 79 +++++++++++++++++++++ 7 files changed, 274 insertions(+), 68 deletions(-) diff --git a/.claude/board/AGENT_LOG.md b/.claude/board/AGENT_LOG.md index 60a10ff0a1a7..e47f735b1264 100644 --- a/.claude/board/AGENT_LOG.md +++ b/.claude/board/AGENT_LOG.md @@ -95,3 +95,40 @@ errors (6m43s cold). Timeline tests: see commit (run pending at log time). in lance-graph-contract (carrier glitch) — NOT touched; wiring first. **Next:** ractor mailbox owns SoA → publishes link onto this timeline (kanban); EpisodicWitness64; replace BindSpace; wire deprecated→cognitive-shader-driver. + +## 2026-05-30 — codex P1 fix: write+delete commit = ONE Lance version (PR #29) +**Branch:** claude/kvs-lance-timeline +**Scope:** +- `kvs/lance/commit_gate.rs`, `kvs/lance/flusher.rs` — `single_lance_commit` +- `kvs/lance/mod.rs` — `build_tombstone_batch_lance` helper +- `kvs/config.rs` — retire dead `delete_via_tombstone_row` flag +- `kvs/lance/tests.rs` — regression test +**Verdict:** PASS + +**What was done (max 5 lines):** +- Codex P1 on PR #29 was VALID: a batch with BOTH writes and deletes ran + `merge_insert` (writes) THEN `Dataset::delete` (deletes) = TWO Lance + versions; the intermediate write-before-delete version leaked through + `Timeline::versions()` as a snapshot that was never an atomic commit. +- Fix: fold deletes into tombstone rows (`tombstone=true`) in the SAME + `merge_insert` → exactly ONE version per commit/flush. New + `Transaction::build_tombstone_batch_lance` mirrors `build_write_batch_lance`. +- Read path already filters `tombstone = false` (schema.rs:145,152), so + get/scan/keys stay correct; physical `Dataset::delete` fully removed. +- Retired the never-read `delete_via_tombstone_row` config flag — the fix is + unconditional; a toggle would only re-open the torn-timeline hole. + +**Tests run:** +- `cargo check -p surrealdb-core --features kv-lance` → Finished, 0 errors +- `cargo test … kvs::lance::tests::test_timeline` → 3 passed; 0 failed (incl. + new `test_timeline_write_delete_commit_is_single_atomic_version`) + +**Open questions / follow-ups:** +- Tombstone rows now accumulate (one dead row per created-then-deleted key) + until compaction; the background optimizer should GC tombstones past the + retention horizon — queued for the compaction pass. +- NOT run through `cargo +nightly fmt`: the crate is not fmt-clean under + `.rustfmt.toml`'s unstable opts (whole-crate churn across 22+ untouched + files), so hand-formatted to match surrounding `lance/` style. + +**Commit(s):** (this commit) diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index 95845bbb83dc..17f8bc8dc59c 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -94,3 +94,23 @@ was correct; the test HARNESS used the wrong write-path. Design consequence: the ractor/kanban consumer that publishes onto the timeline must run on the gate path (or call an explicit flush) to get one timeline entry per Rubicon commit. Cross-ref: config.rs WritePath docs; writepath_legacy_commit_gate_smoke. + +## 2026-05-30 — A SurrealDB commit with writes+deletes was TWO Lance versions, not one +**Status:** FINDING +**Scope:** `kvs/lance/commit_gate.rs`, `kvs/lance/flusher.rs`, `kvs/lance/mod.rs` + +`single_lance_commit` applied writes via `MergeInsertBuilder::execute_reader` +and deletes via a SEPARATE `Dataset::delete` — each its own native Lance +commit. So any batch carrying both produced two versions: an intermediate +(writes applied, deletes pending) and the final. The datastore write lock hid +the intermediate from live readers, but `Timeline::versions()` enumerates raw +`Dataset::versions()` and surfaced it, letting a replayer `view_at()` a torn +state that never atomically existed. The schema was already built for the fix +(a `tombstone` Boolean column + read predicates filtering `tombstone = false`): +folding deletes as tombstone rows into the same `merge_insert` makes +1 commit = 1 version *structurally*, not by convention. Trade-off accepted: +tombstone rows accumulate until a compaction/GC pass (physical `Dataset::delete` +previously reclaimed that space immediately). + +**Cross-ref:** codex P1 on PR #29 (discussion_r3328296248); fix in this +commit; regression `test_timeline_write_delete_commit_is_single_atomic_version`. diff --git a/surrealdb/core/src/kvs/config.rs b/surrealdb/core/src/kvs/config.rs index d346ecc55b6b..6f1b6b073bac 100644 --- a/surrealdb/core/src/kvs/config.rs +++ b/surrealdb/core/src/kvs/config.rs @@ -336,10 +336,6 @@ pub struct LanceConfig { /// `Dataset::checkout(version)`). pub versioned: bool, - /// Whether to write deletions as explicit tombstone rows - /// (in addition to using Lance's native deletion vectors). - pub delete_via_tombstone_row: bool, - /// Which write-path to use. See [`WritePath`] for the two /// options and their semantics. Defaults to /// [`WritePath::LsmWithWal`] — the Sprint AA hot path. @@ -367,7 +363,6 @@ impl Default for LanceConfig { fn default() -> Self { Self { versioned: true, - delete_via_tombstone_row: false, write_path: WritePath::default(), disable_background_flusher: false, } diff --git a/surrealdb/core/src/kvs/lance/commit_gate.rs b/surrealdb/core/src/kvs/lance/commit_gate.rs index 3c3b6d837c54..35ba7e95eb4a 100644 --- a/surrealdb/core/src/kvs/lance/commit_gate.rs +++ b/surrealdb/core/src/kvs/lance/commit_gate.rs @@ -57,7 +57,6 @@ use tokio::sync::{RwLock, mpsc, oneshot}; use tracing::{trace, warn}; use super::DatasetHandle; -use super::schema::KvSchema; use crate::kvs::err::{Error, Result}; use crate::kvs::{Key, Val}; @@ -342,9 +341,23 @@ async fn execute_batch(dataset: &Arc>, batch: Vec