From 77b51c8d1225ace8ab868e8a0e4414307f8aab43 Mon Sep 17 00:00:00 2001 From: Joao Henrique Machado Silva Date: Sun, 10 May 2026 12:45:07 +0200 Subject: [PATCH] feat(engine): Phase 11.2 logical clock + active-tx registry (SQLR-22) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second slice of Phase 11 (concurrent writes via MVCC + BEGIN CONCURRENT). Adds the MVCC primitives that 11.3 will plug an in-memory version index into and 11.4 will hand begin/commit timestamps to. New `sqlrite::mvcc` module: - MvccClock — process-wide monotonic u64 over AtomicU64. tick(), now(), observe(value). The observe-CAS-loop guarantees the clock never moves backwards under contention; used at WAL replay to bring the in-memory clock up to the persisted high-water mark. - ActiveTxRegistry — Mutex over in-flight TxId → begin_ts mappings. register(&clock) returns a RAII TxHandle that unregisters on drop. min_active_begin_ts() answers Phase 11.6 GC's "what's still possibly visible" question in O(log N). - TxId newtype + TxTimestampOrId tagged union, defined now so 11.4 can plug in without re-litigating the type shape. WAL format bumps v1 → v2: - Bytes 24..32 of the WAL header (previously reserved-zero) now carry the persisted clock_high_water u64. - v1 WALs open cleanly — those zero bytes read as "clock never advanced" — and the next checkpoint rewrites the header at v2. No offline upgrade step. - New WAL_FORMAT_VERSION_MIN_SUPPORTED = 1 keeps the version-range check explicit. Forward versions (e.g. v3) reject with a clean error rather than misinterpreting bytes. - Wal::clock_high_water() / Wal::set_clock_high_water(value) accessors. The setter rejects regressions with a typed error (same value is a no-op). Not yet wired into the executor — that's 11.3. The clock is standalone today; only the mvcc module's own tests tick it. The plumbing is in place so 11.3 can read Wal::clock_high_water() at open time and feed the in-memory clock into a Database field. 21 new tests: - 8 in mvcc::clock::tests — seed, default, monotonicity, observe, contention (8 threads × 250 ticks unique). - 7 in mvcc::registry::tests — empty-min, register-advances-clock, unique-ids, unregister-arbitrary-order, Send+Sync compile check, concurrent registrations. - 6 in sql::pager::wal::tests — fresh-zero, round-trip-through- truncate, monotonic-persistence-across-truncates, regression-rejected, v1-opens-with-zero-clock, unknown-future-version-rejected. 599/599 workspace tests pass. fmt + clippy clean. cargo doc clean on changed files. Docs: docs/file-format.md WAL header diagram updated for v2 + the v1→v2 compatibility note. docs/roadmap.md marks 11.1 ✅ and 11.2 🚧. docs/design-decisions.md gets a 12b entry on the clock + WAL persistence design. docs/_index.md project-state line picks up the 11.2 summary. Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/_index.md | 2 +- docs/design-decisions.md | 16 ++ docs/file-format.md | 17 ++- docs/roadmap.md | 14 +- src/lib.rs | 6 + src/mvcc/clock.rs | 182 ++++++++++++++++++++++ src/mvcc/mod.rs | 30 ++++ src/mvcc/registry.rs | 322 +++++++++++++++++++++++++++++++++++++++ src/sql/pager/wal.rs | 211 ++++++++++++++++++++++++- 9 files changed, 787 insertions(+), 13 deletions(-) create mode 100644 src/mvcc/clock.rs create mode 100644 src/mvcc/mod.rs create mode 100644 src/mvcc/registry.rs diff --git a/docs/_index.md b/docs/_index.md index 9e5fadb..d30fc30 100644 --- a/docs/_index.md +++ b/docs/_index.md @@ -54,7 +54,7 @@ As of May 2026, SQLRite has: - Full-text search + hybrid retrieval (Phase 8 complete): FTS5-style inverted index with BM25 ranking + `fts_match` / `bm25_score` scalar functions + `try_fts_probe` optimizer hook + on-disk persistence with on-demand v4 → v5 file-format bump (8a-8c), a worked hybrid-retrieval example combining BM25 with vector cosine via raw arithmetic (8d), and a `bm25_search` MCP tool symmetric with `vector_search` (8e). See [`docs/fts.md`](fts.md). - SQL surface + DX follow-ups (Phase 9 complete, v0.2.0 → v0.9.1): DDL completeness — `DEFAULT`, `DROP TABLE` / `DROP INDEX`, `ALTER TABLE` (9a); free-list + manual `VACUUM` (9b) + auto-VACUUM (9c); `IS NULL` / `IS NOT NULL` (9d); `GROUP BY` + aggregates + `DISTINCT` + `LIKE` + `IN` (9e); four flavors of `JOIN` — INNER, LEFT, RIGHT, FULL OUTER (9f); prepared statements + `?` parameter binding with a per-connection LRU plan cache (9g); HNSW probe widened to cosine + dot via `WITH (metric = …)` (9h); `PRAGMA` dispatcher with the `auto_vacuum` knob (9i) - Benchmarks against SQLite + DuckDB (Phase 10 complete, SQLR-4 / SQLR-16): twelve-workload bench harness with a pluggable `Driver` trait, criterion-driven, pinned-host runs published. See [`docs/benchmarks.md`](benchmarks.md). -- Phase 11 (concurrent writes via MVCC + `BEGIN CONCURRENT`, SQLR-22) is in flight. **11.1 multi-connection foundation: shipped on this branch.** `Connection` is now `Send + Sync` and `Connection::connect()` mints a sibling handle that shares the same backing `Database`. Snapshot-isolation reads + `BEGIN CONCURRENT` writes follow in 11.3 / 11.4. Plan: [`docs/concurrent-writes-plan.md`](concurrent-writes-plan.md). +- Phase 11 (concurrent writes via MVCC + `BEGIN CONCURRENT`, SQLR-22) is in flight. **11.1 multi-connection foundation: shipped.** `Connection` is now `Send + Sync` and `Connection::connect()` mints a sibling handle that shares the same backing `Database`. **11.2 logical clock + active-tx registry: shipped on this branch.** New `sqlrite::mvcc` module exposes `MvccClock` (AtomicU64-backed) and `ActiveTxRegistry` (`min_active_begin_ts` for GC). The WAL header bumps v1 → v2 to persist the clock high-water mark; v1 WALs upgrade transparently. Snapshot-isolation reads + `BEGIN CONCURRENT` writes follow in 11.3 / 11.4. Plan: [`docs/concurrent-writes-plan.md`](concurrent-writes-plan.md). - A fully-automated release pipeline that ships every product to its registry on every release with one human action — Rust engine + `sqlrite-ask` + `sqlrite-mcp` to crates.io, Python wheels to PyPI (`sqlrite`), Node.js + WASM to npm (`@joaoh82/sqlrite` + `@joaoh82/sqlrite-wasm`), Go module via `sdk/go/v*` git tag, plus C FFI tarballs, MCP binary tarballs, and unsigned desktop installers as GitHub Release assets (Phase 6 complete) See the [Roadmap](roadmap.md) for the full phase plan. diff --git a/docs/design-decisions.md b/docs/design-decisions.md index 6dbe652..c76fdc9 100644 --- a/docs/design-decisions.md +++ b/docs/design-decisions.md @@ -156,6 +156,22 @@ Decisions are grouped by the engine layer they concern: parser, storage, concurr --- +### 12b. MVCC logical clock persisted in the WAL header (Phase 11.2) + +**Decision.** A new `sqlrite::mvcc` module ships [`MvccClock`](../src/mvcc/clock.rs) (an `AtomicU64`-backed process-wide counter) and [`ActiveTxRegistry`](../src/mvcc/registry.rs) (a `Mutex`-backed set of in-flight `TxId → begin_ts` mappings). The clock's high-water mark is persisted in the WAL header — bytes 24..32, previously reserved-zero — and the WAL format version bumps from 1 to 2. + +**Why.** MVCC visibility (`begin_ts <= ts < end_ts`) requires that timestamps never repeat — including across reopens. Restarting the engine and resuming the clock from zero would silently break that invariant the moment two transactions on either side of a restart picked the same value. Persisting the high-water mark on each checkpoint means reopen seeds the in-memory clock past the last value the previous run handed out. Putting it in the WAL header (rather than the main file) is right because the WAL is the durability boundary already; commits already fsync the WAL, and the existing checkpoint code rewrites the WAL header on truncate. + +**Cost.** A WAL format bump is real but is mitigated by the v1 layout: bytes 24..32 were reserved-zero, so a v1 WAL parses cleanly under v2 rules with `clock_high_water = 0` — exactly what a never-ticked clock would carry. The next checkpoint rewrites the header at v2; no offline upgrade step. The reader still rejects forward versions (e.g. v3) with a typed error rather than silently misinterpreting bytes. + +**Why an `AtomicU64`, not a plain `u64` behind a `Mutex`.** Each transaction calls the clock twice (begin + commit). Under contention, a `Mutex` would funnel every BEGIN through one critical section. Atomic CAS is wait-free and cheaper. The `observe()` helper (used at WAL replay to bring the clock up to a persisted value) is a CAS loop rather than a `store` so two racing observers can't move the clock backwards. + +**Why `Mutex` for the active-tx registry rather than a sharded skip list.** The registry is touched twice per transaction (begin + commit/rollback). Phase 11.6's GC reads `min_active_begin_ts` once per sweep. That's roughly 2N + 1 mutex calls per N transactions — well below contention thresholds for any realistic workload. When the GC profile actually shows the registry on the hot path, a sharded structure is the obvious upgrade; until then, simple wins. + +**Plan-doc reference.** [`concurrent-writes-plan.md`](concurrent-writes-plan.md) §4.1 (clock) and §4.2 (version index — the registry is an extracted slice of MvStore that's standalone in 11.2). Phase 11.3 is the first in-tree consumer. + +--- + ## Query execution ### 13. `NULL`-as-false in `WHERE` clauses diff --git a/docs/file-format.md b/docs/file-format.md index 837c043..2300c27 100644 --- a/docs/file-format.md +++ b/docs/file-format.md @@ -326,14 +326,27 @@ A second file alongside the `.sqlrite`, named `.sqlrite-wal`, records page │ offset │ length │ content │ ├────────┼────────┼─────────────────────────────────────────────────┤ │ 0 │ 8 │ magic: "SQLRWAL\0" │ -│ 8 │ 4 │ format version (u32 LE) = 1 │ +│ 8 │ 4 │ format version (u32 LE) │ +│ │ │ 1 = pre-Phase-11 │ +│ │ │ 2 = Phase 11.2 — adds clock_high_water │ │ 12 │ 4 │ page size (u32 LE) = 4096 │ │ 16 │ 4 │ salt (u32 LE) — rolled each checkpoint │ │ 20 │ 4 │ checkpoint seq (u32 LE) — increments per ckpt │ -│ 24 │ 8 │ reserved / zero │ +│ 24 │ 8 │ clock_high_water (u64 LE) — Phase 11.2 MVCC │ +│ │ │ logical-clock high-water mark; rewritten on │ +│ │ │ each checkpoint. v1 left these bytes as zero │ +│ │ │ (read identically as "no timestamps issued"). │ └────────┴────────┴─────────────────────────────────────────────────┘ ``` +**v1 → v2 compatibility.** The bytes that now hold `clock_high_water` +were reserved-zero in v1, so a pre-Phase-11 WAL opens cleanly: the +parser interprets the zeros as `clock_high_water = 0`, which is +indistinguishable from "fresh checkpoint, clock has never advanced." +The next checkpoint rewrites the header at v2 — there's no offline +upgrade step. Forward versions we don't recognize (e.g. v3) error +out with a clean diagnostic rather than misinterpreting the bytes. + ### Frames Each frame is `FRAME_HEADER_SIZE + PAGE_SIZE` = **4112 bytes**: diff --git a/docs/roadmap.md b/docs/roadmap.md index 35b51a2..7b45d50 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -585,13 +585,19 @@ End-to-end SQLR-4 / SQLR-16 bench harness with twelve workloads across three gro Lift SQLRite past SQLite's single-writer ceiling with multi-version concurrency control and a `BEGIN CONCURRENT` transaction mode, modelled on Turso's experimental MVCC. The plan doc internally numbers sub-phases as "Phase 10.x" (its working title before the roadmap renumbering); they're listed under Phase 11 here because Phase 10 already shipped. -### 🚧 Phase 11.1 — Multi-connection foundation *(in progress, plan-doc "Phase 10.1")* +### ✅ Phase 11.1 — Multi-connection foundation *(plan-doc "Phase 10.1")* -`Connection` is now a thin handle backed by `Arc>`. Call [`Connection::connect`] to mint a sibling that shares the same engine state — typically one per worker thread. The headline contract: `Connection` is `Send + Sync`, and the engine no longer requires the caller to wrap the public API in their own `Mutex`. Today every operation still serializes through the per-database mutex (and the pager's existing process-level flock), so the behaviour change is *capability*, not throughput; concurrent throughput arrives with `BEGIN CONCURRENT` in 11.4. +`Connection` is a thin handle backed by `Arc>`. Call [`Connection::connect`] to mint a sibling that shares the same engine state — typically one per worker thread. The headline contract: `Connection` is `Send + Sync`, and the engine no longer requires the caller to wrap the public API in their own `Mutex`. Today every operation still serializes through the per-database mutex (and the pager's existing process-level flock), so the behaviour change is *capability*, not throughput; concurrent throughput arrives with `BEGIN CONCURRENT` in 11.4. -### Phase 11.2 — Logical clock + active-tx registry *(planned)* +### 🚧 Phase 11.2 — Logical clock + active-tx registry *(in progress, plan-doc "Phase 10.2")* -`MvccClock` (`AtomicU64`) hands out begin / commit timestamps; `ActiveTxRegistry` exposes `min_active_begin_ts()` for GC. WAL header bumps from v1 → v2 to persist the high-water mark. +New [`sqlrite::mvcc`](../src/mvcc/) module: + +- `MvccClock` — process-wide monotonic `u64` over `AtomicU64`. `tick()` hands out begin- / commit-timestamps; `now()` reads the high-water without advancing it; `observe(value)` advances the clock to `value` if greater (used at WAL replay). +- `ActiveTxRegistry` — `Mutex` over in-flight transactions. `register(&clock)` allocates a `TxId`, snapshots `begin_ts`, and returns a RAII `TxHandle`; `min_active_begin_ts()` answers Phase 11.6 GC's "what's still possibly visible" question. +- `TxId` newtype + `TxTimestampOrId` tagged union — defined now so 11.4 can plug in without re-litigating the type shape. + +WAL format bumps **v1 → v2**: bytes 24..32 of the WAL header (previously reserved-zero) now carry the persisted `clock_high_water` `u64`. v1 WALs open cleanly — those zero bytes read as "clock never advanced" — and the next checkpoint rewrites the header at v2. No offline upgrade step. `Wal::set_clock_high_water` / `Wal::clock_high_water` accessors expose the field; the setter rejects regressions with a typed error. The clock isn't wired into the executor yet (that's 11.3); the persistence + restore plumbing is in place so 11.3 just reads the high-water at open and seeds the in-memory clock. ### Phase 11.3 — `MvStore` skeleton + snapshot-isolation reads *(planned)* diff --git a/src/lib.rs b/src/lib.rs index 9edcd52..0d26700 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -61,6 +61,12 @@ extern crate prettytable; pub mod ask; pub mod connection; pub mod error; +// Phase 11.2 — multi-version concurrency control primitives. The +// module is public because Phase 11.3 / 11.4 grow externally-visible +// types (timestamps, busy errors) on top of these foundations; today +// it just exposes the standalone clock + active-tx registry. See +// `docs/concurrent-writes-plan.md`. +pub mod mvcc; pub mod sql; // Phase 5a public API. diff --git a/src/mvcc/clock.rs b/src/mvcc/clock.rs new file mode 100644 index 0000000..7e1b034 --- /dev/null +++ b/src/mvcc/clock.rs @@ -0,0 +1,182 @@ +//! [`MvccClock`] — the logical-clock primitive that hands out +//! begin- and commit-timestamps for MVCC transactions (Phase 11.2). +//! +//! Per [`docs/concurrent-writes-plan.md`](../../../docs/concurrent-writes-plan.md): +//! +//! > A monotonic `u64` counter, per-`Database`. Hands out `begin_ts` +//! > at `BEGIN CONCURRENT` and `commit_ts` at the start of validation. +//! > Wrapped in `AtomicU64`; no contention because each transaction +//! > calls it twice. +//! +//! The clock is persisted to the WAL header on each checkpoint so +//! reopens resume past the highest committed timestamp — see +//! [`crate::sql::pager::wal::WalHeader::clock_high_water`]. Without +//! persistence, two transactions on either side of a reopen could +//! receive the same timestamp and the snapshot-isolation visibility +//! rule (`begin <= ts < end`) would mis-classify one of them. + +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Process-wide logical clock. Cheap to clone — internally an `Arc` +/// over an [`AtomicU64`] in the [`Database`](crate::Database) wiring +/// (added in Phase 11.3). Standalone today. +#[derive(Debug, Default)] +pub struct MvccClock { + counter: AtomicU64, +} + +impl MvccClock { + /// Builds a clock seeded at `initial`. The next [`MvccClock::tick`] + /// returns `initial + 1`. + /// + /// Use this with the value persisted in the WAL header at open + /// time so the clock resumes past the last-checkpointed + /// high-water mark. + pub fn new(initial: u64) -> Self { + Self { + counter: AtomicU64::new(initial), + } + } + + /// Returns the current high-water timestamp without advancing it. + /// Phase 11.6's GC reads this alongside + /// [`super::ActiveTxRegistry::min_active_begin_ts`] to decide + /// which row-version chains are reclaimable. + pub fn now(&self) -> u64 { + self.counter.load(Ordering::Acquire) + } + + /// Bumps the clock by one and returns the new value. Strictly + /// monotonic: every call observes a distinct `u64`. + pub fn tick(&self) -> u64 { + // `fetch_add` returns the *previous* value — adjust to "after" + // semantics so callers see "the timestamp this call hands out". + // Wrap-around is impossible in practice (a billion ticks/s for + // 600 years still fits in `u64`), so saturating-add isn't + // needed. + self.counter.fetch_add(1, Ordering::AcqRel) + 1 + } + + /// Promotes the clock to at least `value`. No-op if `value` is at + /// or below the current high-water mark. Used at WAL replay to + /// bring the in-memory clock up to the persisted high-water + /// without an extra `tick()`. + pub fn observe(&self, value: u64) { + let mut current = self.counter.load(Ordering::Acquire); + while value > current { + // CAS rather than `store` — racing observers shouldn't + // step on each other and shouldn't move the clock + // backwards if a faster `tick` already overtook them. + match self.counter.compare_exchange_weak( + current, + value, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => return, + Err(actual) => current = actual, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::thread; + + #[test] + fn new_seeds_the_counter() { + let c = MvccClock::new(42); + assert_eq!(c.now(), 42); + assert_eq!(c.tick(), 43); + assert_eq!(c.now(), 43); + } + + #[test] + fn default_starts_at_zero() { + let c = MvccClock::default(); + assert_eq!(c.now(), 0); + assert_eq!(c.tick(), 1); + } + + #[test] + fn tick_is_strictly_monotonic_within_a_thread() { + let c = MvccClock::new(0); + let mut last = 0; + for _ in 0..1_000 { + let t = c.tick(); + assert!(t > last, "tick went backwards: {t} after {last}"); + last = t; + } + } + + #[test] + fn observe_only_moves_forward() { + let c = MvccClock::new(100); + c.observe(50); // ignored — below current + assert_eq!(c.now(), 100); + c.observe(200); + assert_eq!(c.now(), 200); + c.observe(150); // ignored — below current + assert_eq!(c.now(), 200); + } + + /// Concurrent ticks across N threads must hand out N × M distinct + /// values (no duplicates, no skipped values). This is the property + /// MVCC visibility relies on. + #[test] + fn ticks_are_unique_under_contention() { + const THREADS: usize = 8; + const PER_THREAD: usize = 250; + let clock = Arc::new(MvccClock::new(0)); + + let handles: Vec<_> = (0..THREADS) + .map(|_| { + let c = Arc::clone(&clock); + thread::spawn(move || { + let mut out = Vec::with_capacity(PER_THREAD); + for _ in 0..PER_THREAD { + out.push(c.tick()); + } + out + }) + }) + .collect(); + + let mut all = Vec::with_capacity(THREADS * PER_THREAD); + for h in handles { + all.extend(h.join().unwrap()); + } + all.sort_unstable(); + // No duplicates. + for w in all.windows(2) { + assert_ne!(w[0], w[1], "duplicate timestamp {}", w[0]); + } + // Range is contiguous 1..=THREADS*PER_THREAD (clock seeded at 0). + assert_eq!(all.first().copied(), Some(1)); + assert_eq!(all.last().copied(), Some((THREADS * PER_THREAD) as u64)); + } + + /// Concurrent `observe`s must not move the clock backwards. + #[test] + fn observe_under_contention_never_regresses() { + const THREADS: usize = 8; + let clock = Arc::new(MvccClock::new(0)); + let handles: Vec<_> = (0..THREADS) + .map(|tid| { + let c = Arc::clone(&clock); + thread::spawn(move || { + // Each thread observes a deterministic distinct + // value; the clock should end at the max. + c.observe((tid as u64 + 1) * 1_000); + }) + }) + .collect(); + for h in handles { + h.join().unwrap(); + } + assert_eq!(clock.now(), THREADS as u64 * 1_000); + } +} diff --git a/src/mvcc/mod.rs b/src/mvcc/mod.rs new file mode 100644 index 0000000..45c1da3 --- /dev/null +++ b/src/mvcc/mod.rs @@ -0,0 +1,30 @@ +//! Multi-version concurrency control primitives (Phase 11). +//! +//! This module is the foundation for SQLRite's `BEGIN CONCURRENT` story +//! — see [`docs/concurrent-writes-plan.md`](../../docs/concurrent-writes-plan.md) +//! for the full sequenced design. As of **Phase 11.2** it carries the +//! standalone primitives that the rest of the work hangs off: +//! +//! - [`MvccClock`] — a process-wide monotonic `u64` counter that hands +//! out begin- and commit-timestamps. Persisted to the WAL header so +//! timestamps don't reuse the same value across reopens. +//! - [`ActiveTxRegistry`] — tracks the begin-timestamps of in-flight +//! transactions. Garbage collection (Phase 11.6) needs +//! [`ActiveTxRegistry::min_active_begin_ts`] to know which versions +//! are still possibly visible to a live reader. +//! - [`TxId`] — opaque newtype around a `u64`, allocated by the clock +//! while a transaction is in flight. After commit the same value is +//! reused as the row version's `begin` timestamp; the discriminator +//! between "in-flight transaction id" and "committed timestamp" +//! lives in [`TxTimestampOrId`]. +//! +//! Nothing in the executor reads from these yet — Phase 11.3 wires +//! them into a new `MvStore` in front of the pager. Keeping the +//! plumbing standalone in 11.2 means the Phase 11.4 `BEGIN CONCURRENT` +//! work can pull them in without re-litigating the foundation. + +pub mod clock; +pub mod registry; + +pub use clock::MvccClock; +pub use registry::{ActiveTxRegistry, TxHandle, TxId, TxTimestampOrId}; diff --git a/src/mvcc/registry.rs b/src/mvcc/registry.rs new file mode 100644 index 0000000..3258930 --- /dev/null +++ b/src/mvcc/registry.rs @@ -0,0 +1,322 @@ +//! [`ActiveTxRegistry`] — the live-transaction table that +//! garbage collection consults to know which row versions are still +//! possibly visible (Phase 11.2). +//! +//! Per [`docs/concurrent-writes-plan.md`](../../../docs/concurrent-writes-plan.md): +//! +//! > Versions whose `end` timestamp is older than the oldest active +//! > reader's begin-timestamp are dead and may be reclaimed. +//! +//! The registry is the source of "oldest active reader's +//! begin-timestamp" — [`ActiveTxRegistry::min_active_begin_ts`]. +//! Phase 11.6 (GC) reads it on every sweep; Phase 11.4 (commit +//! validation) registers each `BEGIN CONCURRENT` transaction here +//! and unregisters at COMMIT/ROLLBACK. +//! +//! The current shape uses a `Mutex` for simplicity. Two +//! reasons not to over-engineer this for v0: +//! +//! 1. The map is only touched twice per transaction (begin + +//! commit/rollback). Even a thousand concurrent writers hit a +//! couple-thousand `lock` calls per second — well below mutex +//! contention thresholds. +//! 2. `min_active_begin_ts` is `O(log N)` on a `BTreeMap` (the +//! smallest key is at `iter().next()`), which is fine for the +//! "GC asks once per sweep" use case. +//! +//! When the GC profile shows the registry on the hot path, swap to +//! a sharded skip list or an `RwLock`-protected sorted set. Until +//! then this is sufficient. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use super::clock::MvccClock; + +/// Opaque transaction identifier. Newtype around a `u64` so a stray +/// timestamp doesn't accidentally pass as a `TxId` and vice versa. +/// Allocated by [`ActiveTxRegistry::register`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct TxId(pub u64); + +impl std::fmt::Display for TxId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "tx{}", self.0) + } +} + +/// Tagged-union "this version is in-flight under transaction `id`" +/// vs. "this version was committed at timestamp `ts`". Per the plan, +/// row versions carry a `begin: TxTimestampOrId` so reads can ignore +/// versions belonging to still-open transactions while still seeing +/// the latest committed version. +/// +/// Phase 11.4 will be the first consumer; defining it here keeps the +/// type stable across the in-flight sub-phases so callers don't have +/// to chase a moving target. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TxTimestampOrId { + /// Committed timestamp — visible to any transaction whose + /// `begin_ts >= this`. + Timestamp(u64), + /// In-flight transaction id — invisible to every other reader + /// until the producing transaction commits and stamps its + /// versions with a timestamp. + Id(TxId), +} + +/// Live-transaction table. Cheap to clone (internally `Arc`-wrapped +/// state); pass clones into worker threads. +#[derive(Clone, Debug, Default)] +pub struct ActiveTxRegistry { + inner: Arc>, +} + +#[derive(Debug, Default)] +struct RegistryInner { + /// `TxId` → `begin_ts`. Deterministic ordering by `TxId` (which + /// matches allocation order) — useful for diagnostic output. + by_id: BTreeMap, + /// Multiset of `begin_ts` values, with a count for each. Lets + /// `min_active_begin_ts` answer in `O(log N)` regardless of + /// `by_id`'s size, and `unregister` just decrements the relevant + /// counter rather than scanning. The shape matters once we have + /// many concurrent transactions all sharing the same begin_ts + /// (rare under MvccClock, but possible if a snapshot is taken + /// without ticking the clock). + by_ts: BTreeMap, +} + +impl ActiveTxRegistry { + /// Creates an empty registry. Equivalent to `Default::default()`. + pub fn new() -> Self { + Self::default() + } + + /// Registers a new transaction. Allocates a fresh [`TxId`] from + /// `clock` and snapshots the current clock value as the + /// transaction's `begin_ts`. + /// + /// Returns a [`TxHandle`] — when the handle drops, the + /// transaction is automatically unregistered. RAII keeps the + /// "did the transaction's caller forget to clean up?" failure + /// mode out of the cold-path code. + pub fn register(&self, clock: &MvccClock) -> TxHandle { + let begin_ts = clock.tick(); + let id = TxId(begin_ts); + let mut g = self.lock(); + g.by_id.insert(id, begin_ts); + *g.by_ts.entry(begin_ts).or_insert(0) += 1; + drop(g); + TxHandle { + id, + begin_ts, + registry: self.clone(), + } + } + + /// Returns the begin-timestamp of the oldest in-flight + /// transaction, or `None` when nothing is in flight. Phase 11.6 + /// uses this to set the GC watermark — versions whose `end` + /// timestamp is strictly less than this value can never be seen + /// again and may be reclaimed. + pub fn min_active_begin_ts(&self) -> Option { + self.lock().by_ts.keys().next().copied() + } + + /// Number of in-flight transactions. Cheap diagnostic accessor; + /// not load-bearing for correctness. + pub fn active_count(&self) -> usize { + self.lock().by_id.len() + } + + /// Internal — dropped through [`TxHandle::drop`]. + fn unregister(&self, id: TxId, begin_ts: u64) { + let mut g = self.lock(); + g.by_id.remove(&id); + if let Some(slot) = g.by_ts.get_mut(&begin_ts) { + *slot = slot.saturating_sub(1); + if *slot == 0 { + g.by_ts.remove(&begin_ts); + } + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, RegistryInner> { + self.inner + .lock() + .unwrap_or_else(|e| panic!("sqlrite: ActiveTxRegistry mutex poisoned: {e}")) + } +} + +/// RAII guard returned by [`ActiveTxRegistry::register`]. Dropping it +/// unregisters the transaction. A typical caller doesn't deal with it +/// explicitly — it lives on the `ConcurrentTx` struct (Phase 11.4) +/// and is dropped when the transaction commits or rolls back. +#[derive(Debug)] +pub struct TxHandle { + id: TxId, + begin_ts: u64, + registry: ActiveTxRegistry, +} + +impl TxHandle { + /// The opaque identifier this transaction was allocated. Stable + /// for the handle's lifetime. + pub fn id(&self) -> TxId { + self.id + } + + /// The timestamp at which this transaction's snapshot was taken. + /// Phase 11.3 reads use this as the visibility cutoff: a row + /// version with `begin <= self.begin_ts() < end` is the visible + /// one. + pub fn begin_ts(&self) -> u64 { + self.begin_ts + } +} + +impl Drop for TxHandle { + fn drop(&mut self) { + self.registry.unregister(self.id, self.begin_ts); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_registry_has_no_minimum() { + let r = ActiveTxRegistry::new(); + assert_eq!(r.min_active_begin_ts(), None); + assert_eq!(r.active_count(), 0); + } + + #[test] + fn register_advances_clock_and_updates_minimum() { + let clock = MvccClock::new(0); + let r = ActiveTxRegistry::new(); + + let h1 = r.register(&clock); + assert_eq!(h1.begin_ts(), 1); + assert_eq!(r.min_active_begin_ts(), Some(1)); + + let h2 = r.register(&clock); + assert_eq!(h2.begin_ts(), 2); + assert_eq!(r.min_active_begin_ts(), Some(1)); + + // Closing the older transaction lifts the minimum. + drop(h1); + assert_eq!(r.min_active_begin_ts(), Some(2)); + + drop(h2); + assert_eq!(r.min_active_begin_ts(), None); + } + + #[test] + fn handles_carry_distinct_ids_and_unique_timestamps() { + let clock = MvccClock::new(0); + let r = ActiveTxRegistry::new(); + let h1 = r.register(&clock); + let h2 = r.register(&clock); + assert_ne!(h1.id(), h2.id()); + assert_ne!(h1.begin_ts(), h2.begin_ts()); + assert_eq!(r.active_count(), 2); + } + + #[test] + fn unregister_in_arbitrary_order_keeps_minimum_correct() { + let clock = MvccClock::new(0); + let r = ActiveTxRegistry::new(); + let h1 = r.register(&clock); // begin_ts = 1 + let h2 = r.register(&clock); // begin_ts = 2 + let h3 = r.register(&clock); // begin_ts = 3 + assert_eq!(r.min_active_begin_ts(), Some(1)); + + // Drop the middle one — minimum still h1. + drop(h2); + assert_eq!(r.min_active_begin_ts(), Some(1)); + + // Drop the oldest — minimum jumps to h3. + drop(h1); + assert_eq!(r.min_active_begin_ts(), Some(3)); + + drop(h3); + assert_eq!(r.min_active_begin_ts(), None); + } + + #[test] + fn registry_is_send_and_sync() { + // Compile-time check — required so the registry can be cloned + // into worker threads. + fn assert_send() {} + fn assert_sync() {} + assert_send::(); + assert_sync::(); + assert_send::(); + assert_sync::(); + } + + /// Many concurrent registrations — every begin_ts must be unique + /// and the registry's count must match the live handle count. + #[test] + fn concurrent_registrations_are_consistent() { + use std::thread; + const THREADS: usize = 8; + const PER_THREAD: usize = 100; + + let clock = Arc::new(MvccClock::new(0)); + let registry = ActiveTxRegistry::new(); + + let handles: Vec<_> = (0..THREADS) + .map(|_| { + let c = Arc::clone(&clock); + let r = registry.clone(); + thread::spawn(move || { + let mut held: Vec = Vec::with_capacity(PER_THREAD); + for _ in 0..PER_THREAD { + held.push(r.register(&c)); + } + // Don't drop here — return the handles so the + // outer thread sees them all simultaneously alive. + held + }) + }) + .collect(); + + let mut all: Vec = Vec::with_capacity(THREADS * PER_THREAD); + for h in handles { + all.extend(h.join().unwrap()); + } + + assert_eq!(registry.active_count(), THREADS * PER_THREAD); + let begins: std::collections::BTreeSet = all.iter().map(|h| h.begin_ts()).collect(); + assert_eq!( + begins.len(), + THREADS * PER_THREAD, + "every concurrent registration must allocate a unique begin_ts" + ); + + // Drop every handle — registry empties out cleanly. + drop(all); + assert_eq!(registry.active_count(), 0); + assert_eq!(registry.min_active_begin_ts(), None); + } + + #[test] + fn tx_id_displays_with_prefix() { + assert_eq!(format!("{}", TxId(7)), "tx7"); + } + + #[test] + fn tx_timestamp_or_id_round_trips() { + // Just exercise the Eq/Clone derives so a future refactor + // that changes the variants surfaces here. + let a = TxTimestampOrId::Timestamp(42); + let b = TxTimestampOrId::Id(TxId(42)); + assert_ne!(a, b); + assert_eq!(a, a); + assert_eq!(b, b); + } +} diff --git a/src/sql/pager/wal.rs b/src/sql/pager/wal.rs index ece6399..15ca42d 100644 --- a/src/sql/pager/wal.rs +++ b/src/sql/pager/wal.rs @@ -14,12 +14,20 @@ //! ```text //! byte 0..32 WAL header //! 0..8 magic "SQLRWAL\0" -//! 8..12 format version (u32 LE) = 1 +//! 8..12 format version (u32 LE) +//! v1: pre-Phase-11 +//! v2: Phase 11.2 — adds clock_high_water +//! in bytes 24..32 //! 12..16 page size (u32 LE) = 4096 //! 16..20 salt (u32 LE) — random on create, //! re-rolled per checkpoint //! 20..24 checkpoint seq (u32 LE) — bumps per checkpoint -//! 24..32 reserved / zero +//! 24..32 v2: clock_high_water (u64 LE) — last +//! persisted MVCC logical clock value; +//! `crate::mvcc::MvccClock::observe`'d on +//! reopen so timestamps don't reuse values +//! across restarts. +//! v1: reserved / zero (read as 0). //! //! byte 32.. sequence of frames, each `FRAME_SIZE` bytes: //! 0..4 page number (u32 LE) @@ -34,6 +42,14 @@ //! 16..16+PAGE_SIZE page bytes //! ``` //! +//! **Format version compatibility.** v1 WALs (written by pre-Phase-11 +//! builds) open cleanly: their reserved bytes are zero, which we +//! interpret as `clock_high_water = 0` — exactly what a fresh-from- +//! checkpoint clock would carry. The next time the WAL is rewritten +//! (any checkpoint, including the auto-checkpoint that fires past the +//! frame threshold) it lands on disk as v2. There's no "you must +//! upgrade your files" step. +//! //! **Checksum.** A rolling `rotate_left(1) + byte` sum over the //! concatenation of the frame's first 12 header bytes (page_num, //! commit-page-count, salt) and its PAGE_SIZE body. Catches bit flips @@ -60,7 +76,15 @@ use crate::sql::pager::pager::{AccessMode, acquire_lock}; pub const WAL_HEADER_SIZE: usize = 32; pub const WAL_MAGIC: &[u8; 8] = b"SQLRWAL\0"; -pub const WAL_FORMAT_VERSION: u32 = 1; +/// The version the engine writes today. Phase 11.2 bumped 1 → 2 to +/// introduce [`WalHeader::clock_high_water`] in bytes 24..32 of the +/// WAL header. `read_header` still accepts v1 files (treating those +/// bytes as zero); the next `truncate` rewrites them as v2. +pub const WAL_FORMAT_VERSION: u32 = 2; +/// Lowest format version we know how to open. v1 had the bytes that +/// now hold `clock_high_water` reserved-as-zero, which is identical +/// to "clock has never been persisted" and round-trips cleanly. +pub const WAL_FORMAT_VERSION_MIN_SUPPORTED: u32 = 1; pub const FRAME_HEADER_SIZE: usize = 16; pub const FRAME_SIZE: usize = FRAME_HEADER_SIZE + PAGE_SIZE; @@ -71,6 +95,13 @@ pub const FRAME_SIZE: usize = FRAME_HEADER_SIZE + PAGE_SIZE; pub struct WalHeader { pub salt: u32, pub checkpoint_seq: u32, + /// Phase 11.2 — the last MVCC logical-clock value persisted to + /// this WAL. Rewritten by `truncate` (= every checkpoint) so a + /// reopen-then-tick can never reuse a timestamp the previous run + /// already handed out. Always 0 for v1 WALs (the bytes were + /// reserved-zero before the bump); read back as 0 on any v1 file + /// that pre-existed this build. + pub clock_high_water: u64, } /// Parsed per-frame header (everything but the page body). @@ -131,6 +162,7 @@ impl Wal { let header = WalHeader { salt, checkpoint_seq: 0, + clock_high_water: 0, }; let mut wal = Self { file, @@ -193,6 +225,33 @@ impl Wal { self.last_commit_page_count } + /// Phase 11.2 — the MVCC logical-clock high-water mark persisted + /// in this WAL's header. Returns 0 for fresh-from-create WALs and + /// for v1 WALs (where the bytes were reserved-zero). The Pager + /// (Phase 11.3) seeds the in-memory `MvccClock` from this on open. + pub fn clock_high_water(&self) -> u64 { + self.header.clock_high_water + } + + /// Phase 11.2 — overrides the in-memory clock high-water value. + /// `truncate` writes whatever value the WAL is carrying when it + /// runs, so callers update this just before checkpoint to persist + /// the latest in-memory clock value. The setter rejects a + /// non-monotonic update (a value below the existing high-water + /// mark) — that would either be a bug in the caller or evidence + /// of a corrupted in-memory clock. Same value is a no-op. + pub fn set_clock_high_water(&mut self, value: u64) -> Result<()> { + if value < self.header.clock_high_water { + return Err(SQLRiteError::General(format!( + "WAL clock_high_water cannot move backwards: \ + attempted {value}, current {}", + self.header.clock_high_water + ))); + } + self.header.clock_high_water = value; + Ok(()) + } + /// Bulk-loads every committed page from the WAL into `dest`. Used by /// `Pager::open` to warm a WAL cache so subsequent reads don't have /// to seek back into the WAL file. Uncommitted frames are skipped @@ -298,7 +357,12 @@ impl Wal { buf[12..16].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes()); buf[16..20].copy_from_slice(&self.header.salt.to_le_bytes()); buf[20..24].copy_from_slice(&self.header.checkpoint_seq.to_le_bytes()); - // 24..32 zero + // Phase 11.2: bytes 24..32 carry the MVCC clock high-water + // mark (u64 LE). v1 WALs left this as zeros, which v2 reads + // as "no timestamps have been issued" — the same value a + // newly-created v2 WAL starts at. That's how the v1 → v2 + // upgrade stays seamless. + buf[24..32].copy_from_slice(&self.header.clock_high_water.to_le_bytes()); self.file.seek(SeekFrom::Start(0))?; self.file.write_all(&buf)?; Ok(()) @@ -416,9 +480,14 @@ fn read_header(file: &mut File) -> Result { )); } let version = u32::from_le_bytes(buf[8..12].try_into().unwrap()); - if version != WAL_FORMAT_VERSION { + // Phase 11.2 — accept v1 (clock bytes are reserved-zero) and v2 + // (clock bytes carry the high-water mark). Anything else is + // either a corrupt header or a forward-version we don't + // understand; reject with a clean error. + if !(WAL_FORMAT_VERSION_MIN_SUPPORTED..=WAL_FORMAT_VERSION).contains(&version) { return Err(SQLRiteError::General(format!( - "unsupported WAL format version {version}; this build understands {WAL_FORMAT_VERSION}" + "unsupported WAL format version {version}; this build reads \ + v{WAL_FORMAT_VERSION_MIN_SUPPORTED}..=v{WAL_FORMAT_VERSION}" ))); } let page_size = u32::from_le_bytes(buf[12..16].try_into().unwrap()) as usize; @@ -429,9 +498,14 @@ fn read_header(file: &mut File) -> Result { } let salt = u32::from_le_bytes(buf[16..20].try_into().unwrap()); let checkpoint_seq = u32::from_le_bytes(buf[20..24].try_into().unwrap()); + // v1 wrote zeros into bytes 24..32; v2 puts the clock high-water + // there. Either way, decoding as `u64::from_le_bytes` produces + // the right value — 0 for v1, the persisted clock for v2. + let clock_high_water = u64::from_le_bytes(buf[24..32].try_into().unwrap()); Ok(WalHeader { salt, checkpoint_seq, + clock_high_water, }) } @@ -653,6 +727,131 @@ mod tests { let _ = std::fs::remove_file(&p); } + // ----------------------------------------------------------------- + // Phase 11.2 — clock_high_water in the WAL header + // ----------------------------------------------------------------- + + #[test] + fn fresh_wal_starts_clock_at_zero() { + let p = tmp_wal("clock_fresh"); + let w = Wal::create(&p).unwrap(); + assert_eq!(w.clock_high_water(), 0); + drop(w); + let w2 = Wal::open(&p).unwrap(); + assert_eq!(w2.clock_high_water(), 0); + let _ = std::fs::remove_file(&p); + } + + /// Round-trip: setting the clock and triggering `truncate` + /// (= every checkpoint) must persist the high-water mark across a + /// reopen. This is the property Phase 11.6 GC relies on — without + /// it, two transactions on either side of a reopen could share a + /// timestamp and corrupt visibility. + #[test] + fn clock_high_water_round_trips_through_truncate() { + let p = tmp_wal("clock_truncate"); + let mut w = Wal::create(&p).unwrap(); + // Append a frame so truncate has something to drop. Doesn't + // matter what the body is — we're testing the header path. + w.append_frame(1, &page(0xaa), Some(1)).unwrap(); + w.set_clock_high_water(12_345).unwrap(); + w.truncate().unwrap(); + assert_eq!(w.clock_high_water(), 12_345); + drop(w); + + let w2 = Wal::open(&p).unwrap(); + assert_eq!(w2.clock_high_water(), 12_345); + let _ = std::fs::remove_file(&p); + } + + /// `truncate` rewrites the header. Bump-then-truncate-twice must + /// keep advancing the on-disk value as the in-memory clock moves. + #[test] + fn clock_high_water_is_monotonically_persisted_across_truncates() { + let p = tmp_wal("clock_monotonic_persist"); + let mut w = Wal::create(&p).unwrap(); + w.set_clock_high_water(100).unwrap(); + w.truncate().unwrap(); + w.set_clock_high_water(200).unwrap(); + w.truncate().unwrap(); + drop(w); + + let w2 = Wal::open(&p).unwrap(); + assert_eq!(w2.clock_high_water(), 200); + let _ = std::fs::remove_file(&p); + } + + /// Setter rejects a value below the current high-water mark with + /// a typed error. Same value is accepted as a no-op (test below). + /// Same-or-greater is the contract every consumer relies on; a + /// silent saturate-to-current would mask a real bug in the caller. + #[test] + fn set_clock_high_water_rejects_regressions() { + let p = tmp_wal("clock_no_regress"); + let mut w = Wal::create(&p).unwrap(); + w.set_clock_high_water(500).unwrap(); + let err = w.set_clock_high_water(499).unwrap_err(); + let msg = format!("{err}"); + assert!( + msg.contains("backwards") && msg.contains("499") && msg.contains("500"), + "expected typed regression error, got: {msg}" + ); + // Idempotent same-value update is fine. + w.set_clock_high_water(500).unwrap(); + assert_eq!(w.clock_high_water(), 500); + let _ = std::fs::remove_file(&p); + } + + /// Synthetic v1 WAL — exact byte layout the previous engine + /// version wrote. Opening it must succeed and report + /// `clock_high_water == 0` (the bytes were reserved-zero in v1). + /// This is the "graceful upgrade" contract. + #[test] + fn v1_wal_opens_with_zero_clock() { + let p = tmp_wal("v1_compat"); + // Hand-build a v1 WAL header. Frame body is intentionally + // omitted — we're testing header parsing, not frame replay. + let mut buf = vec![0u8; WAL_HEADER_SIZE]; + buf[0..8].copy_from_slice(WAL_MAGIC); + buf[8..12].copy_from_slice(&1u32.to_le_bytes()); // version 1 + buf[12..16].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes()); + buf[16..20].copy_from_slice(&0xdead_beef_u32.to_le_bytes()); // salt + buf[20..24].copy_from_slice(&7u32.to_le_bytes()); // checkpoint_seq + // bytes 24..32 left as zero — v1's reserved bytes. + std::fs::write(&p, &buf).unwrap(); + + let w = Wal::open(&p).unwrap(); + assert_eq!(w.header().salt, 0xdead_beef); + assert_eq!(w.header().checkpoint_seq, 7); + assert_eq!( + w.clock_high_water(), + 0, + "v1 reserved bytes must read as clock=0" + ); + let _ = std::fs::remove_file(&p); + } + + /// Forward-versions we don't understand must error cleanly rather + /// than silently misinterpreting bytes. Picks a version far above + /// our `WAL_FORMAT_VERSION` so the test stays valid even after + /// future bumps. + #[test] + fn unknown_future_version_is_rejected() { + let p = tmp_wal("unknown_version"); + let mut buf = vec![0u8; WAL_HEADER_SIZE]; + buf[0..8].copy_from_slice(WAL_MAGIC); + buf[8..12].copy_from_slice(&999u32.to_le_bytes()); + buf[12..16].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes()); + std::fs::write(&p, &buf).unwrap(); + let err = Wal::open(&p).unwrap_err(); + let msg = format!("{err}"); + assert!( + msg.contains("unsupported WAL format version") && msg.contains("999"), + "unexpected error shape: {msg}" + ); + let _ = std::fs::remove_file(&p); + } + #[test] fn partial_trailing_frame_is_ignored() { // Write one valid frame, then append a half-frame's worth of