diff --git a/.claude/board/entries/2026-11-10-dt64-map-table-lance.md b/.claude/board/entries/2026-11-10-dt64-map-table-lance.md new file mode 100644 index 000000000..7249e616e --- /dev/null +++ b/.claude/board/entries/2026-11-10-dt64-map-table-lance.md @@ -0,0 +1,51 @@ +# 2026-11-10 — DR map table made durable through the fire-and-forget cycle writer + +**Status:** VERIFIED-IN-CODE (12 tests; the round trip runs on a real +`LanceCycleWriter`). Answers the storage half of D-DT64-7 O5 / D10' +(D-LITHO Q5: "which table"). WAL choice: D-DR64-WAL-1 (EXO-225). + +> ⊘ **SUPERSEDED the same day:** the first version of this crate wrote its +> own Lance dataset synchronously, compared dataset versions to refuse a +> stale writer, and reported a post-commit `Race` / `ClockMoved`. That is a +> custom WAL with arbitration, which the persistence doctrine forbids +> (`.claude/v3/knowledge/d-mbx-a6-owner-consume-and-persistence.md`). It was +> rebuilt on the cycle writer. + +**What.** `crates/lance-graph-dt64-lance` (excluded, own workspace) makes +`dr::MapTable` durable. dt64 stays free of lance and arrow. + +- `MapLog::advance` claims the route version (`MapTable::claim`), turns the + advance into casts on behalf of the owning mailbox, submits them with + `CycleWriterHandle::try_submit` (cycle id = route version) and returns at + once. One background task owns the `LanceCycleWriter`. +- `MapLog::settle` applies an advance with `MapTable::advance_at` only when + its seal ticket is `Ok`. A failed seal applies nothing and its version is a + gap; every later advance is refused (`Poisoned`), matching the writer, + which seals nothing after a failure. +- One 512-byte cast per 64 ids: a W64 tile is one cast, a W256 tile four, a + removal one; each advance adds one marker cast (version, source, + relation-map revisions). +- `restore` reads every sealed cycle (`scan_sealed` + `scan_image`) and + re-applies each advance at its own route version. It refuses a landing on + behalf of another mailbox and any image that is not a valid log. +- There is no arbitration: one owner holds the `MapLog`, which owns its + table mutably. Two writers on one dataset are outside the contract + (`LanceCycleWriter` fences on its own cached head). + +**Tests.** 4 unit + 8 integration. Disable runs, each restored: +- applying without a sealed ticket → `a_failed_seal_applies_nothing_and_poisons_what_follows`; +- no owner check on restore → `every_landing_is_on_behalf_of_the_owner`; +- applying at the clock's current version instead of the claimed one → 4 tests, + including `a_clock_tick_between_claim_and_seal_lands_at_the_claimed_version`; +- no poison gate → the failed-seal test; +- no reserved-id check → `a_reserved_id_or_a_repeated_tile_is_refused_before_a_version_is_claimed`. + +**Open.** +- Frames are not restored: a frame is one table instance's witness, so a + frame from before a restore is `ForeignFrame` (pinned by a test). Issuing a + frame at a past version after a restore is a DR owner decision. +- A table whose clock is shared with a `RouteLog` needs the route log + durable too; only the map is stored here. +- The map table's compaction schedule (L's files and scan time grow with + advances; EXO-225). +- lance-graph does not build without `ndarray-hpc`; this crate enables it. diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md index c0ba84432..cd4852349 100644 --- a/.claude/board/entries/README.md +++ b/.claude/board/entries/README.md @@ -25,13 +25,14 @@ index row, (3) no duplicate entry id. Checks 1 and 2 are deliberately opposite directions; the stranding this convention prevents shows up in exactly one of them, never both. -594 entries, 2026-08-06 .. 2026-11-10. +595 entries, 2026-08-06 .. 2026-11-10. | date | entry id | finding | file | |---|---|---|---| | 2026-11-10 | `lab-resolve-borrow-all` | | [2026-11-10-lab-resolve-borrow-all.md](2026-11-10-lab-resolve-borrow-all.md) | | 2026-11-10 | `int8-direct-packed-b` | | [2026-11-10-int8-direct-packed-b.md](2026-11-10-int8-direct-packed-b.md) | | 2026-11-10 | `D-FOUNDRY-PARITY-1` | | [2026-11-10-foundry-parity-1-lab.md](2026-11-10-foundry-parity-1-lab.md) | +| 2026-11-10 | `dt64-map-table-lance` | | [2026-11-10-dt64-map-table-lance.md](2026-11-10-dt64-map-table-lance.md) | | 2026-11-10 | `dt64-lane-borrow-all` | | [2026-11-10-dt64-lane-borrow-all.md](2026-11-10-dt64-lane-borrow-all.md) | | 2026-11-10 | `dt64-int8-facet-stride` | | [2026-11-10-dt64-int8-facet-stride.md](2026-11-10-dt64-int8-facet-stride.md) | | 2026-11-10 | `D-DT64-GEMM-SHAPE-1` | | [2026-11-10-dt64-gemm-shape.md](2026-11-10-dt64-gemm-shape.md) | diff --git a/.claude/plans/d-dt64-7-behaviour-addressing-registration-v1.md b/.claude/plans/d-dt64-7-behaviour-addressing-registration-v1.md index 8c1b29771..94eeded85 100644 --- a/.claude/plans/d-dt64-7-behaviour-addressing-registration-v1.md +++ b/.claude/plans/d-dt64-7-behaviour-addressing-registration-v1.md @@ -120,6 +120,8 @@ Preserved throughout: the `0xC401` name-bound reading, the `FnIndex` opcode cont **Status:** core shipped (`dr::MapTable`, T1–T4 green, both disable runs red). Stored selections wait on O4. +**Storage (2026-11-10):** the versioned table is made durable by `crates/lance-graph-dt64-lance` (`MapLog`) through the fire-and-forget cycle writer over `LanceCycleWriter` (D-DR64-WAL-1). An advance claims its route version, is submitted as casts on behalf of the owning mailbox, and is applied with `advance_at` only after its seal ticket is `Ok`; a failed seal is a gap. `restore` replays every sealed cycle at its own route version. (⊘ An earlier same-day version wrote its own dataset synchronously with stale-writer/race checks; replaced.) OPEN: frames are not restored (a frame is one table instance's witness), so executing at a past version after a restore needs a DR decision; a table sharing its clock with a `RouteLog` needs the route log stored too. + O5 is a BindStore plan item (D-DT64-6 Amendment A). Litho is the mental model it came from, not a component. **What exists (VERIFIED-IN-CODE, `lance-graph-dt64/src/dr.rs`):** diff --git a/.claude/plans/wal-shootout-v1.md b/.claude/plans/wal-shootout-v1.md index 6feef49a9..76a9dd393 100644 --- a/.claude/plans/wal-shootout-v1.md +++ b/.claude/plans/wal-shootout-v1.md @@ -252,3 +252,37 @@ compactor) does to both curves. - **D-LNC-7, decided:** `LanceCycleWriter` stays the default with the capped session; MemWAL stays an option. The 60-minute soak is deferred. + +## 8. DR64 sizes (2026-11-10) — the WAL for DR/DT64 map advances + +A DR64 tile is 64 global ids × 8 B = one 512-byte cast, so a DR advance that +changes N tiles is one cycle of N casts. `examples/wal_shootout.rs dr64` runs +N = 1, 4, 16, 64 at 200 cycles each, on one AVX-512 VM, three separate +processes. The arms are the same as §1: L = `LanceCycleWriter`, M10 / M100 = +lancedb MemWAL flushed every 10 / 100 ms. Ranges below cover the three runs. + +| N tiles | L p50 | L p99 | M10 p50 | M10 p99 | M100 p50 | +|---|---|---|---|---|---| +| 1 | 2.2–4.4 ms | 4.0–7.1 ms | 8.1–8.3 ms | 10.7–11.2 ms | 97.7–98.2 ms | +| 4 | 2.8–5.7 ms | 5.6–9.2 ms | 7.5–8.3 ms | 10.8–11.8 ms | 97.2–98.3 ms | +| 16 | 3.4–5.4 ms | 5.2–9.0 ms | 7.8–8.3 ms | 10.9–11.1 ms | 97.4–97.5 ms | +| 64 | 4.3–5.4 ms | 6.2–9.3 ms | 7.4–7.9 ms | 11.0–11.2 ms | 97.0–97.4 ms | + +| N tiles | L MiB / files / versions / scan | M10 MiB / files / versions / scan | +|---|---|---| +| 1 | 2.7 / 601 / 200 / 23–28 ms | 1.0 / 209 / 3 / 6–7 ms | +| 64 | 15.6 / 601 / 200 / 23–25 ms | 14.9 / 209 / 3 / 8–9 ms | + +- **L is faster per advance at every N measured**, at p50 and at p99, in all + three runs. M's commit time is its flush interval, not its write. +- **L keeps one Lance version per advance** (200 versions for 200 cycles). M + folds 200 advances into 3 versions. DR reads the map table as of a route + version, so a version per advance is the shape DR needs, not overhead. +- **L's cost is files and scan time:** 601 files and a 23–32 ms scan after 200 + advances, against 209 files and 5–9 ms for M. Both grow with the number of + advances, and L's needs compaction (an owner-scheduled step; not part of the + advance path). + +**D-DR64-WAL-1, decided:** DR/DT64 map advances go through `LanceCycleWriter` +(L) behind `CycleWriter`, fire and forget. MemWAL is not used for DR. Open: +the compaction schedule for the map table. diff --git a/.github/workflows/dt64-map-table-lance.yml b/.github/workflows/dt64-map-table-lance.yml new file mode 100644 index 000000000..ba19bd1e6 --- /dev/null +++ b/.github/workflows/dt64-map-table-lance.yml @@ -0,0 +1,84 @@ +name: dt64 map table in Lance +# lance-graph-dt64-lance makes dt64's DR map table (dr::MapTable) durable +# through the supervisor's cycle writer over LanceCycleWriter. +# It is workspace-excluded (own [workspace] root), so no other job builds it. +# Without this one a change to dt64's dr module or a lance bump could break it +# unnoticed. +# +# lance's build compiles .proto files: `protoc` alone is not enough, the +# well-known includes (libprotobuf-dev) are needed too. +on: + push: + branches: + - main + pull_request: + paths: + - crates/lance-graph-contract/** + - crates/lance-graph-dt64/** + - crates/lance-graph-dt64-lance/** + # The map log is written through the cycle writer over LanceCycleWriter. + - crates/lance-graph-supervisor/** + - crates/lance-graph-planner/** + - crates/lance-graph/Cargo.toml + - crates/lance-graph/src/graph/cycle_sink.rs + - .github/workflows/dt64-map-table-lance.yml + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +env: + CARGO_TERM_COLOR: always + CARGO_INCREMENTAL: "0" + CARGO_PROFILE_DEV_DEBUG: "0" + CARGO_PROFILE_TEST_DEBUG: "0" + RUST_BACKTRACE: "1" + +jobs: + map-table: + name: DR map table in Lance + runs-on: ubuntu-24.04 + timeout-minutes: 45 + defaults: + run: + working-directory: lance-graph + steps: + - uses: actions/checkout@v4 + with: + path: lance-graph + persist-credentials: false + # lance-graph's ndarray-hpc feature and the planner path-depend on + # ../../../ndarray. + - name: Checkout AdaWorldAPI/ndarray (sibling dependency) + uses: actions/checkout@v4 + with: + repository: AdaWorldAPI/ndarray + path: ndarray + persist-credentials: false + - name: Install protoc and the protobuf includes (for lance) + run: sudo apt-get update && sudo apt-get install -y protobuf-compiler libprotobuf-dev + - name: Setup rust toolchain (pinned by rust-toolchain.toml) + run: rustup show + # See dir-sim-persist.yml: spell target-cpu=native out as this runner's + # feature list so the cache never reuses an artifact on another CPU. + - name: Resolve target-cpu=native for this runner + id: cpu + run: | + flags=$(.github/scripts/resolve-native-rustflags.sh) + echo "RUSTFLAGS=$flags" >> "$GITHUB_ENV" + hash=$(printf %s "$flags" | sha256sum | cut -c1-16) + echo "flags=$hash" >> "$GITHUB_OUTPUT" + - uses: Swatinem/rust-cache@v2 + with: + key: cpu-${{ steps.cpu.outputs.flags }} + workspaces: lance-graph/crates/lance-graph-dt64-lance + - name: fmt + run: cargo fmt --manifest-path crates/lance-graph-dt64-lance/Cargo.toml --check + - name: clippy + run: cargo clippy --manifest-path crates/lance-graph-dt64-lance/Cargo.toml --all-targets -- -D warnings + - name: test + run: cargo test --manifest-path crates/lance-graph-dt64-lance/Cargo.toml diff --git a/Cargo.toml b/Cargo.toml index d56fc05a5..0cb6f49c7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -176,6 +176,10 @@ exclude = [ # (ogar-dir-core / ogar-ad / ogar-dir-sim). Verify via # `cargo test --manifest-path crates/lance-graph-dir-sim/Cargo.toml`. "crates/lance-graph-dir-sim", + # DR's versioned global -> local map table stored in Lance (D-DT64-7 + # O5). Own [workspace], lance + arrow; dt64 stays lance-free. Verify via + # `cargo test --manifest-path crates/lance-graph-dt64-lance/Cargo.toml`. + "crates/lance-graph-dt64-lance", # Cognitive Compilation loop (claude/cognitive-compilation-lance-graph-h8sgym). # The NEW piece is the Elixir-shaped template — thinking styles, JITson, and # i4-32D thinking-style vectors already exist; the template was the gap. Four diff --git a/crates/lance-graph-dt64-lance/Cargo.toml b/crates/lance-graph-dt64-lance/Cargo.toml new file mode 100644 index 000000000..441276049 --- /dev/null +++ b/crates/lance-graph-dt64-lance/Cargo.toml @@ -0,0 +1,36 @@ +# lance-graph-dt64-lance — DR's versioned global -> local table (dt64 +# `dr::MapTable`) stored as a separate Lance-versioned dataset +# (D-DT64-7 O5 / D10', D-LITHO Q5). EXCLUDED from the lance-graph workspace +# (own [workspace] root), like lance-graph-dir-sim-lance: dt64 stays free of +# lance and arrow. +# +# Advances are fire and forget: each one is a cycle submitted to the +# supervisor's CycleWriter, whose single background task owns a +# LanceCycleWriter (D-DR64-WAL-1). The advance is applied to the table only +# after its seal ticket resolves Ok; a failed seal is a gap. +# +# Build/verify: +# cargo test --manifest-path crates/lance-graph-dt64-lance/Cargo.toml + +[package] +name = "lance-graph-dt64-lance" +version = "0.1.0" +edition = "2021" +publish = false +license = "Apache-2.0" +description = "DR's versioned global -> local map table made durable through the fire-and-forget cycle writer, replayed exactly at restore." + +[workspace] + +[dependencies] +lance-graph-dt64 = { path = "../lance-graph-dt64" } +lance-graph-contract = { path = "../lance-graph-contract" } +lance-graph-planner = { path = "../lance-graph-planner" } +lance-graph-supervisor = { path = "../lance-graph-supervisor", features = ["cycle-driver"] } +# LanceCycleWriter, the WAL (graph::cycle_sink, behind `planner`). `ndarray-hpc` +# is required: lance-graph does not build without it (its fallback is stale). +lance-graph = { path = "../lance-graph", default-features = false, features = ["planner", "ndarray-hpc"] } + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] } +tempfile = "3" diff --git a/crates/lance-graph-dt64-lance/src/lib.rs b/crates/lance-graph-dt64-lance/src/lib.rs new file mode 100644 index 000000000..3fc663503 --- /dev/null +++ b/crates/lance-graph-dt64-lance/src/lib.rs @@ -0,0 +1,859 @@ +//! DR's versioned global → local table ([`MapTable`]) made durable through the +//! fire-and-forget cycle writer (D-DT64-7 O5 / D10', D-DR64-WAL-1). +//! +//! The owner of a DR map table never waits for storage: +//! +//! ```text +//! owner: advance ── claim v ── casts ──try_submit(v, casts)──▶ returns v +//! writer: │ one background task owns the sink +//! ▼ +//! seal cycle v (LanceCycleWriter: one Lance version) +//! owner: settle ── ticket Ok ── advance_at(v) ticket Err ── v is a gap +//! ``` +//! +//! - **The WAL is `LanceCycleWriter` behind `CycleWriter`.** This crate writes +//! no dataset itself and keeps no confirmation ledger, replay queue or retry +//! loop. Measured at DR64 sizes it commits an advance in 2.2–5.7 ms p50 and +//! keeps one Lance version per advance (`.claude/plans/wal-shootout-v1.md` §8). +//! - **One advance is one cycle**, with cycle id = route version. The route +//! version is claimed from the table's clock when the advance is built +//! ([`MapTable::claim`]) and applied at that version only after the cycle's +//! seal ticket resolves `Ok` ([`MapTable::advance_at`]). A seal that fails +//! applies nothing: its version is a gap in the table and in the store. +//! - **Ownership travels in the casts.** Every cast is written on behalf of the +//! mailbox that owns the table (`SweepSlot::owner`). Restore refuses a +//! landing on behalf of any other mailbox. There is one writer by +//! construction: [`MapLog`] owns the table mutably and submits through the +//! writer handle it was given; it does not arbitrate between writers. +//! - **One 512-byte cast per 64 ids.** A DR64 tile's map is 64 global ids, +//! `8 × 64 = 512` bytes: exactly one artifact payload. A 256-wide tile is +//! four. Each advance also carries one marker cast with its route version, +//! source and relation-map revisions. +//! - **Restore replays, exactly.** [`restore`] reads every sealed cycle and +//! re-applies each advance at its own route version, so every version +//! answers as before. Rows that do not form a valid log are refused; nothing +//! is repaired. +//! +//! What a restore does **not** carry: frames. A [`Frame`] is a witness issued +//! by one table instance, so frames sealed before a restore are foreign to the +//! restored table (`Stale(ForeignFrame)`). Past versions stay readable through +//! [`MapTable::diagnose_at`]. Issuing frames at a past version after a restore +//! is a DR owner decision this crate does not make (OPEN). +//! +//! [`Frame`]: lance_graph_dt64::dr::Frame + +use std::collections::{BTreeMap, VecDeque}; + +use lance_graph::graph::cycle_sink::LanceCycleWriter; +use lance_graph_contract::collapse_gate::MailboxId; +use lance_graph_dt64::dr::{ + AdvanceRefusal, LocalMap, MapRefusal, MapTable, RouteVersion, TileId, VersionClock, Width, +}; +use lance_graph_planner::persist_sink::{CycleId, SweepSlot, WalSink, WriteFailed}; +use lance_graph_supervisor::cycle_driver::SealRecovery; +use lance_graph_supervisor::cycle_writer::{CycleWriterHandle, SealError, SealTicket, SubmitError}; + +/// Bytes in one cast payload (the cycle writer's artifact width). +pub const CAST_BYTES: usize = 512; +/// Global ids carried by one cast. +const IDS_PER_CAST: usize = CAST_BYTES / 8; +/// The image row of an advance's marker cast. +const MARKER_ROW: u64 = u64::MAX; +/// Tile part that marks a removal. +const REMOVED_PART: u64 = 7; +/// An unused id slot. A map holding this id is refused before anything is +/// claimed or submitted. +pub const RESERVED_ID: u64 = u64::MAX; + +/// Marker kinds. +const GENESIS: u8 = 0; +const ADVANCE: u8 = 1; + +/// The image row of `part` of `tile`. +fn tile_row(tile: TileId, part: u64) -> u64 { + (u64::from(tile) << 3) | part +} + +fn parts(width: Width) -> u64 { + match width { + Width::W64 => 1, + Width::W256 => 4, + } +} + +/// Why an advance was not submitted, or did not become durable. +#[derive(Debug)] +pub enum AdvanceError { + /// One advance named the same tile twice. Nothing was claimed. + DuplicateTile(TileId), + /// A map holds [`RESERVED_ID`]. Nothing was claimed. + ReservedId(TileId), + /// An earlier advance did not become durable, so this log submits + /// nothing further (the writer seals nothing after a failure either). + /// Nothing was claimed. + Poisoned { failed: RouteVersion }, + /// The writer refused the cycle. `version` was claimed and is a gap; the + /// casts come back inside `error`. + Submit { + version: RouteVersion, + error: SubmitError, + }, +} + +/// What happened to one submitted advance once its seal resolved. +#[derive(Debug)] +pub enum Settled { + /// The cycle is durable and the advance is applied at `version`. + Applied(RouteVersion), + /// The cycle did not become durable; nothing was applied and `version` is + /// a gap. A failed seal hands its casts back inside `error`. + NotDurable { + version: RouteVersion, + error: SealError, + }, + /// The seal failed without proving whether the cycle was published: an + /// ambiguous commit (`SealRecovery::ResubmitFrozen`), a hash conflict + /// (`Escalate`), or a writer that stopped before answering. The store may + /// hold a cycle the table lacks, so the log refuses every later advance + /// and [`MapLog::into_table`]; a restore replays what the store holds. + OutcomeUnknown { + version: RouteVersion, + error: SealError, + }, + /// The cycle is durable but the table refused to apply it. `MapLog` + /// applies its own advances in claimed order and owns its table, so its + /// API cannot produce this; it is reported rather than dropped so a + /// disagreement between store and table can never pass silently. The + /// store now holds a cycle the table lacks: the log refuses every later + /// advance ([`AdvanceError::Poisoned`]) and [`MapLog::into_table`]. A + /// restore replays the store. + Refused { + version: RouteVersion, + refusal: AdvanceRefusal, + }, +} + +/// Whether a failed seal leaves open that the cycle was published. +fn outcome_unknown(error: &SealError) -> bool { + match error { + SealError::Failed(f) => !matches!( + f.recovery(), + SealRecovery::Regenerate | SealRecovery::Permanent + ), + SealError::WriterGone => true, + SealError::Poisoned { .. } | SealError::OutOfOrder { .. } => false, + } +} + +/// An advance's `(source, relation_map, changes)`. +type Advance = (u64, u64, Vec<(TileId, Option)>); + +struct Pending { + version: RouteVersion, + /// `None` for the genesis cycle, which applies nothing. + advance: Option, + ticket: SealTicket, +} + +/// The durable owner side of one DR map table. +pub struct MapLog { + table: MapTable, + writer: CycleWriterHandle, + on_behalf: MailboxId, + next_position: u64, + pending: VecDeque, + failed: Option, + /// The first cycle the store may hold and the table lacks: refused while + /// durable ([`Settled::Refused`]) or of unknown outcome + /// ([`Settled::OutcomeUnknown`]). + diverged: Option, +} + +/// [`MapLog::into_table`] refused: the table is not the store's last sealed +/// state. Carries the log back so the caller can settle it, or restore. +pub struct Undrained { + log: Box, +} + +impl Undrained { + /// Submitted cycles not yet settled when the table was asked for. + #[must_use] + pub fn pending(&self) -> usize { + self.log.pending() + } + + /// The cycle the store may hold and the table lacks, if that is why. + #[must_use] + pub fn diverged(&self) -> Option { + self.log.diverged + } + + /// The log, to settle what is pending and ask again. + #[must_use] + pub fn into_log(self) -> MapLog { + *self.log + } +} + +impl std::fmt::Debug for Undrained { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Undrained") + .field("pending", &self.pending()) + .field("diverged", &self.diverged()) + .finish() + } +} + +impl MapLog { + /// Start a new durable table on an empty store: submits the genesis + /// cycle (the table's source and relation-map revisions at its current + /// version) and returns at once. `clock` may be shared with a `RouteLog`. + pub fn create( + writer: CycleWriterHandle, + on_behalf: MailboxId, + source: u64, + relation_map: u64, + clock: VersionClock, + ) -> Result { + let table = MapTable::with_clock(source, relation_map, clock); + let version = table.current_version(); + let mut log = MapLog { + table, + writer, + on_behalf, + next_position: 0, + pending: VecDeque::new(), + failed: None, + diverged: None, + }; + let casts = vec![log.marker(GENESIS, version, source, relation_map)]; + log.submit(version, None, casts)?; + Ok(log) + } + + /// Continue a restored table: further advances follow its last sealed + /// cycle. `writer` must be framed on the store's current head. + pub fn resume(restored: Restored, writer: CycleWriterHandle, on_behalf: MailboxId) -> MapLog { + MapLog { + table: restored.table, + writer, + on_behalf, + next_position: restored.next_position, + pending: VecDeque::new(), + failed: None, + diverged: None, + } + } + + /// The table as of the last applied (durable) advance. + pub fn table(&self) -> &MapTable { + &self.table + } + + /// Give up the log and keep its table, which must be drained: nothing + /// pending and no cycle the store may hold missing from the table. Otherwise the log comes back in + /// [`Undrained`], so a sealed cycle can never be dropped unapplied + /// (settle it, then ask again). Drops this log's writer handle on + /// success, so the writer stops once every other handle is gone too. + /// + /// # Errors + /// + /// [`Undrained`] while [`MapLog::pending`] is non-zero, or a cycle the + /// store may hold is missing from the table ([`Settled::Refused`], + /// [`Settled::OutcomeUnknown`]). + pub fn into_table(self) -> Result { + if self.pending.is_empty() && self.diverged.is_none() { + Ok(self.table) + } else { + Err(Undrained { + log: Box::new(self), + }) + } + } + + /// Submitted advances whose seal has not been settled yet. + pub fn pending(&self) -> usize { + self.pending.len() + } + + /// Submit an advance and return its claimed route version at once. It is + /// applied to [`MapLog::table`] only when [`MapLog::settle`] sees its + /// cycle sealed. + pub fn advance( + &mut self, + source: u64, + relation_map: u64, + changes: Vec<(TileId, Option)>, + ) -> Result { + if let Some(failed) = self.failed { + return Err(AdvanceError::Poisoned { failed }); + } + for (i, (tile, map)) in changes.iter().enumerate() { + if changes[..i].iter().any(|(t, _)| t == tile) { + return Err(AdvanceError::DuplicateTile(*tile)); + } + if let Some(m) = map { + if m.view().globals().contains(&RESERVED_ID) { + return Err(AdvanceError::ReservedId(*tile)); + } + } + } + let version = self.table.claim(); + let mut casts = vec![self.marker(ADVANCE, version, source, relation_map)]; + for (tile, map) in &changes { + casts.extend(self.tile_casts(version, *tile, map.as_ref())); + } + self.submit(version, Some((source, relation_map, changes)), casts)?; + Ok(version) + } + + /// Apply every advance whose seal has resolved, in submission order, + /// without waiting. Stops at the first cycle still being sealed. + pub fn settle(&mut self) -> Vec { + let mut out = Vec::new(); + while let Some(front) = self.pending.front_mut() { + let Some(result) = front.ticket.try_sealed() else { + break; + }; + let Pending { + version, advance, .. + } = self.pending.pop_front().expect("front exists"); + if let Some(s) = self.resolve(version, advance, result.map(|_| ())) { + out.push(s); + } + } + out + } + + /// Wait for the oldest submitted advance and apply it. `None` when + /// nothing is pending. Used at shutdown and in tests; the owner's loop + /// calls [`MapLog::settle`]. + pub async fn settle_next(&mut self) -> Option { + loop { + // The entry stays queued while waiting: if this future is + // cancelled, the advance is still pending and a later settle + // applies it. + let result = self.pending.front_mut()?.ticket.wait().await; + let Pending { + version, advance, .. + } = self.pending.pop_front().expect("front exists"); + if let Some(s) = self.resolve(version, advance, result.map(|_| ())) { + return Some(s); + } + } + } + + fn resolve( + &mut self, + version: RouteVersion, + advance: Option, + result: Result<(), SealError>, + ) -> Option { + match result { + Err(error) => { + self.failed.get_or_insert(version); + if outcome_unknown(&error) { + self.diverged.get_or_insert(version); + Some(Settled::OutcomeUnknown { version, error }) + } else { + Some(Settled::NotDurable { version, error }) + } + } + Ok(_) => { + let (source, relation_map, changes) = advance?; + Some( + match self + .table + .advance_at(version, source, relation_map, changes) + { + Ok(v) => Settled::Applied(v), + Err(refusal) => { + self.failed.get_or_insert(version); + self.diverged.get_or_insert(version); + Settled::Refused { version, refusal } + } + }, + ) + } + } + } + + fn submit( + &mut self, + version: RouteVersion, + advance: Option, + casts: Vec, + ) -> Result<(), AdvanceError> { + match self.writer.try_submit(CycleId(version.0), casts) { + Ok(ticket) => { + self.pending.push_back(Pending { + version, + advance, + ticket, + }); + Ok(()) + } + Err(error) => Err(AdvanceError::Submit { version, error }), + } + } + + fn cast(&mut self, version: RouteVersion, row: u64, payload: Vec) -> SweepSlot { + let stream_position = self.next_position; + self.next_position += 1; + SweepSlot { + cycle: CycleId(version.0), + stream_position, + owner: self.on_behalf, + row, + paired_move: None, + payload, + } + } + + fn marker(&mut self, kind: u8, version: RouteVersion, source: u64, rm: u64) -> SweepSlot { + let mut p = vec![0u8; CAST_BYTES]; + p[0] = kind; + p[8..16].copy_from_slice(&version.0.to_le_bytes()); + p[16..24].copy_from_slice(&source.to_le_bytes()); + p[24..32].copy_from_slice(&rm.to_le_bytes()); + self.cast(version, MARKER_ROW, p) + } + + fn tile_casts( + &mut self, + version: RouteVersion, + tile: TileId, + map: Option<&LocalMap>, + ) -> Vec { + let Some(map) = map else { + let p = vec![0u8; CAST_BYTES]; + return vec![self.cast(version, tile_row(tile, REMOVED_PART), p)]; + }; + let view = map.view(); + let globals = view.globals(); + (0..parts(map.width())) + .map(|part| { + let mut p = vec![0u8; CAST_BYTES]; + for slot in 0..IDS_PER_CAST { + let i = part as usize * IDS_PER_CAST + slot; + let id = globals.get(i).copied().unwrap_or(RESERVED_ID); + p[slot * 8..slot * 8 + 8].copy_from_slice(&id.to_le_bytes()); + } + self.cast(version, tile_row(tile, part), p) + }) + .collect() + } +} + +/// A sealed store that can hand back each cycle's coalesced image (the +/// payloads `WalSink::scan_sealed` deliberately does not carry). +#[allow(async_fn_in_trait)] +pub trait SealedImages: WalSink { + /// `row → payload` for one sealed cycle. + async fn sealed_image(&self, cycle: CycleId) -> Result>, WriteFailed>; +} + +impl SealedImages for LanceCycleWriter { + async fn sealed_image(&self, cycle: CycleId) -> Result>, WriteFailed> { + self.scan_image(cycle).await + } +} + +/// A table rebuilt from its store. +pub struct Restored { + pub table: MapTable, + /// The next cast stream position (one past the last stored landing). + pub next_position: u64, + /// The last sealed cycle (the last advance's route version). + pub last: RouteVersion, +} + +/// Why a restore failed. +#[derive(Debug)] +pub enum RestoreError { + /// The store could not be read. + Read(WriteFailed), + /// The stored cycles do not form a valid log. + Replay(ReplayError), +} + +/// Why stored cycles are not a valid log. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReplayError { + /// The store holds no cycles. + Empty, + /// The first cycle is not a genesis, or a second genesis follows. + Genesis { cycle: u64 }, + /// A cycle has no marker cast. + MissingMarker { cycle: u64 }, + /// A marker's route version is not its cycle id. + VersionMismatch { cycle: u64, marker: u64 }, + /// A marker kind this crate does not write. + UnknownKind { cycle: u64, kind: u8 }, + /// A landing was written on behalf of another mailbox. + ForeignOwner { cycle: u64, owner: MailboxId }, + /// A cast length other than [`CAST_BYTES`]. + PayloadLength { cycle: u64, row: u64, len: usize }, + /// A tile's parts are not one width's parts or one removal, or its ids + /// are not a prefix followed only by unused slots. + Positions { cycle: u64, tile: TileId }, + /// The ids do not form a map (duplicate id, too many). + Map { + cycle: u64, + tile: TileId, + refusal: MapRefusal, + }, + /// Re-applying the advance was refused by the table. + Apply { cycle: u64, refusal: AdvanceRefusal }, +} + +/// Rebuild a table from every sealed cycle in `sink`, re-applying each +/// advance at its own route version. Every landing must be on behalf of +/// `owner`. `clock` becomes the restored table's clock. +pub async fn restore( + sink: &S, + owner: MailboxId, + clock: VersionClock, +) -> Result { + let landings = sink.scan_sealed(None).await.map_err(RestoreError::Read)?; + let mut cycles: Vec = Vec::new(); + let mut next_position = 0; + for l in &landings { + if l.slot.owner != owner { + return Err(RestoreError::Replay(ReplayError::ForeignOwner { + cycle: l.cycle.0, + owner: l.slot.owner, + })); + } + next_position = next_position.max(l.slot.stream_position + 1); + if cycles.last() != Some(&l.cycle.0) { + cycles.push(l.cycle.0); + } + } + let mut images = Vec::with_capacity(cycles.len()); + for &c in &cycles { + let image = sink + .sealed_image(CycleId(c)) + .await + .map_err(RestoreError::Read)?; + images.push((c, image)); + } + let (table, last) = replay(&images, clock).map_err(RestoreError::Replay)?; + Ok(Restored { + table, + next_position, + last, + }) +} + +/// A decoded marker: `(kind, version, source, relation_map)`. +fn decode_marker( + cycle: u64, + image: &BTreeMap>, +) -> Result<(u8, u64, u64, u64), ReplayError> { + let p = image + .get(&MARKER_ROW) + .ok_or(ReplayError::MissingMarker { cycle })?; + let word = |at: usize| u64::from_le_bytes(p[at..at + 8].try_into().expect("8 bytes")); + Ok((p[0], word(8), word(16), word(24))) +} + +fn replay( + images: &[(u64, BTreeMap>)], + clock: VersionClock, +) -> Result<(MapTable, RouteVersion), ReplayError> { + for (cycle, image) in images { + for (row, p) in image { + if p.len() != CAST_BYTES { + return Err(ReplayError::PayloadLength { + cycle: *cycle, + row: *row, + len: p.len(), + }); + } + } + } + let ((first, genesis), rest) = images.split_first().ok_or(ReplayError::Empty)?; + let (kind, version, source, relation_map) = decode_marker(*first, genesis)?; + if kind != GENESIS || genesis.len() != 1 { + return Err(ReplayError::Genesis { cycle: *first }); + } + if version != *first { + return Err(ReplayError::VersionMismatch { + cycle: *first, + marker: version, + }); + } + while clock.current().0 < version { + clock.tick(); + } + // Start at the stored genesis, not at the clock: a shared clock may + // already be past it. + let mut table = + MapTable::with_clock_at(source, relation_map, clock.clone(), RouteVersion(version)) + .map_err(|refusal| ReplayError::Apply { + cycle: *first, + refusal, + })?; + let mut last = RouteVersion(version); + for (cycle, image) in rest { + let (kind, version, source, relation_map) = decode_marker(*cycle, image)?; + match kind { + ADVANCE => {} + GENESIS => return Err(ReplayError::Genesis { cycle: *cycle }), + other => { + return Err(ReplayError::UnknownKind { + cycle: *cycle, + kind: other, + }) + } + } + if version != *cycle { + return Err(ReplayError::VersionMismatch { + cycle: *cycle, + marker: version, + }); + } + let changes = changes_of(*cycle, image)?; + while clock.current().0 < version { + clock.tick(); + } + table + .advance_at(RouteVersion(version), source, relation_map, changes) + .map_err(|refusal| ReplayError::Apply { + cycle: *cycle, + refusal, + })?; + last = RouteVersion(version); + } + Ok((table, last)) +} + +/// The changes of one advance from its tile casts. +fn changes_of( + cycle: u64, + image: &BTreeMap>, +) -> Result)>, ReplayError> { + let mut by_tile: BTreeMap> = BTreeMap::new(); + for (&row, p) in image { + if row == MARKER_ROW { + continue; + } + let tile = TileId::try_from(row >> 3).map_err(|_| ReplayError::Positions { + cycle, + tile: TileId::MAX, + })?; + by_tile.entry(tile).or_default().push((row & 7, p)); + } + let mut changes = Vec::with_capacity(by_tile.len()); + for (tile, mut rows) in by_tile { + rows.sort_by_key(|(part, _)| *part); + let found: Vec = rows.iter().map(|(part, _)| *part).collect(); + let bad = ReplayError::Positions { cycle, tile }; + let width = match found.as_slice() { + [REMOVED_PART] => { + changes.push((tile, None)); + continue; + } + [0] => Width::W64, + [0, 1, 2, 3] => Width::W256, + _ => return Err(bad), + }; + let mut globals = Vec::new(); + let mut ended = false; + for (_, p) in &rows { + for slot in p.as_chunks::<8>().0 { + let id = u64::from_le_bytes(*slot); + if id == RESERVED_ID { + ended = true; + } else if ended { + return Err(bad); + } else { + globals.push(id); + } + } + } + let map = LocalMap::prepare(width, &globals).map_err(|refusal| ReplayError::Map { + cycle, + tile, + refusal, + })?; + changes.push((tile, Some(map))); + } + Ok(changes) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Accepts every cycle; enough to run a `MapLog` against a real writer. + struct AcceptAll { + head: u64, + } + + impl WalSink for AcceptAll { + async fn commit_cycle( + &mut self, + batch: lance_graph_planner::persist_sink::DetachedCycleBatch, + ) -> Result< + lance_graph_planner::persist_sink::CommitOutcome, + lance_graph_planner::persist_sink::CommitError, + > { + self.head += 1; + Ok( + lance_graph_planner::persist_sink::CommitOutcome::Committed { + version: lance_graph_contract::scheduler::DatasetVersion(self.head), + cycle: batch.frame.cycle, + batch_hash: batch.batch_hash, + }, + ) + } + + async fn scan_sealed( + &self, + _after: Option, + ) -> Result, WriteFailed> { + Ok(Vec::new()) + } + + async fn timeline( + &self, + ) -> Result, WriteFailed> { + Ok(Vec::new()) + } + } + + /// A durable cycle the table refuses poisons the log: no later advance, + /// no table handed out as if it were the store's state. `MapLog`'s API + /// cannot produce the refusal, so the test resolves an already-applied + /// version a second time, which the table refuses as `NotAfterLast`. + #[tokio::test] + async fn a_durable_cycle_the_table_refuses_poisons_the_log() { + let (writer, handle) = lance_graph_supervisor::cycle_writer::cycle_writer( + AcceptAll { head: 0 }, + lance_graph_contract::scheduler::DatasetVersion(0), + 8, + ); + let (_exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, 7, 1, 1, VersionClock::new()).unwrap(); + let v = log.advance(1, 1, vec![(1, None)]).unwrap(); + assert!(matches!(log.settle_next().await, Some(Settled::Applied(a)) if a == v)); + // Silence twin: a drained, agreeing log hands out its table. + assert_eq!(log.pending(), 0); + assert!(log.diverged.is_none()); + + let refused = log.resolve(v, Some((1, 1, vec![(1, None)])), Ok(())); + assert!( + matches!(refused, Some(Settled::Refused { version, .. }) if version == v), + "{refused:?}" + ); + assert!(matches!( + log.advance(1, 1, vec![]), + Err(AdvanceError::Poisoned { failed }) if failed == v + )); + let undrained = log + .into_table() + .expect_err("a diverged log keeps its table"); + assert_eq!(undrained.pending(), 0); + assert_eq!(undrained.diverged(), Some(v)); + }); + } + + fn image(entries: &[(u64, Vec)]) -> BTreeMap> { + entries.iter().cloned().collect() + } + + fn ids(list: &[u64]) -> Vec { + let mut p = vec![0u8; CAST_BYTES]; + for slot in 0..IDS_PER_CAST { + let id = list.get(slot).copied().unwrap_or(RESERVED_ID); + p[slot * 8..slot * 8 + 8].copy_from_slice(&id.to_le_bytes()); + } + p + } + + #[test] + fn a_w64_tile_is_one_cast_and_reads_back_as_its_ids() { + let c = changes_of(1, &image(&[(tile_row(5, 0), ids(&[9, 8, 7]))])).unwrap(); + assert_eq!(c.len(), 1); + let (tile, map) = &c[0]; + assert_eq!(*tile, 5); + let map = map.as_ref().unwrap(); + assert_eq!(map.width(), Width::W64); + assert_eq!(map.view().globals(), &[9, 8, 7]); + } + + #[test] + fn tile_parts_that_are_not_one_width_are_refused() { + // Two of a W256 tile's four parts. + let two = image(&[(tile_row(1, 0), ids(&[1])), (tile_row(1, 1), ids(&[]))]); + assert_eq!( + changes_of(3, &two).unwrap_err(), + ReplayError::Positions { cycle: 3, tile: 1 } + ); + // A removal next to a part. + let mixed = image(&[ + (tile_row(1, 0), ids(&[1])), + (tile_row(1, REMOVED_PART), vec![0; CAST_BYTES]), + ]); + assert!(changes_of(3, &mixed).is_err()); + } + + #[test] + fn an_id_after_an_unused_slot_is_refused() { + let mut p = ids(&[1, 2]); + p[5 * 8..6 * 8].copy_from_slice(&3u64.to_le_bytes()); + let err = changes_of(4, &image(&[(tile_row(2, 0), p)])).unwrap_err(); + assert_eq!(err, ReplayError::Positions { cycle: 4, tile: 2 }); + } + + #[test] + fn replay_refuses_a_log_without_genesis_or_marker() { + let advance = |v: u64| { + let mut p = vec![0u8; CAST_BYTES]; + p[0] = ADVANCE; + p[8..16].copy_from_slice(&v.to_le_bytes()); + p + }; + let genesis = { + let mut p = vec![0u8; CAST_BYTES]; + p[0] = GENESIS; + p + }; + let replay_of = |cycles: Vec<(u64, BTreeMap>)>| { + replay(&cycles, VersionClock::new()) + .map(|_| ()) + .unwrap_err() + }; + assert_eq!(replay_of(vec![]), ReplayError::Empty); + assert_eq!( + replay_of(vec![(1, image(&[(MARKER_ROW, advance(1))]))]), + ReplayError::Genesis { cycle: 1 } + ); + assert_eq!( + replay_of(vec![ + (0, image(&[(MARKER_ROW, genesis.clone())])), + (1, image(&[(tile_row(1, 0), ids(&[1]))])), + ]), + ReplayError::MissingMarker { cycle: 1 } + ); + assert_eq!( + replay_of(vec![ + (0, image(&[(MARKER_ROW, genesis.clone())])), + (2, image(&[(MARKER_ROW, advance(1))])), + ]), + ReplayError::VersionMismatch { + cycle: 2, + marker: 1 + } + ); + assert_eq!( + replay_of(vec![ + (0, image(&[(MARKER_ROW, genesis.clone())])), + (1, image(&[(MARKER_ROW, genesis)])), + ]), + ReplayError::Genesis { cycle: 1 } + ); + } +} diff --git a/crates/lance-graph-dt64-lance/tests/map_log.rs b/crates/lance-graph-dt64-lance/tests/map_log.rs new file mode 100644 index 000000000..2af48ebe8 --- /dev/null +++ b/crates/lance-graph-dt64-lance/tests/map_log.rs @@ -0,0 +1,570 @@ +//! A DR map table made durable through the fire-and-forget cycle writer. +//! +//! - The owner submits and returns at once; an advance is applied only after +//! its cycle seals, at the route version claimed when it was built. +//! - A failed seal applies nothing and leaves a gap. +//! - Every landing is on behalf of the owning mailbox. +//! - A restore answers every past version exactly. +//! +//! The interleavings are driven deterministically by a gated in-memory sink +//! (`GateWal`): commits wait on a semaphore the test opens, and a chosen cycle +//! fails. The round trip runs on the real `LanceCycleWriter`. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use lance_graph::graph::cycle_sink::LanceCycleWriter; +use lance_graph_contract::scheduler::DatasetVersion; +use lance_graph_dt64::dr::{ + FrameRefusal, LocalMap, MapRead, MapTable, RouteVersion, VersionClock, Width, +}; +use lance_graph_dt64_lance::{ + restore, AdvanceError, MapLog, ReplayError, RestoreError, SealedImages, Settled, RESERVED_ID, +}; +use lance_graph_planner::persist_sink::{ + CommitError, CommitOutcome, CycleId, DetachedCycleBatch, FrameMeta, LandedSlot, WalSink, + WriteFailed, +}; +use lance_graph_supervisor::cycle_writer::{cycle_writer, SealError}; +use tokio::sync::Semaphore; + +const OWNER: u32 = 42; +const TILES: [u32; 3] = [1, 2, 3]; + +fn map(w: Width, ids: &[u64]) -> Option { + Some(LocalMap::prepare(w, ids).unwrap()) +} + +/// What a table answers for `tile` at `version`: `(width, globals)`, or the +/// refusal. +fn answer(t: &MapTable, version: u64, tile: u32) -> Result<(u16, Vec), MapRead> { + let d = t.diagnose_at(RouteVersion(version), tile)?; + let width = match d.width() { + Width::W64 => 64, + Width::W256 => 256, + }; + let globals = (0..d.len()) + .map(|l| d.global(u8::try_from(l).unwrap()).unwrap()) + .collect(); + Ok((width, globals)) +} + +/// Every version up to `upto`, every tile: the two tables answer alike. +fn assert_same(a: &MapTable, b: &MapTable, upto: u64) { + for v in 0..=upto { + for tile in TILES { + assert_eq!( + answer(a, v, tile), + answer(b, v, tile), + "version {v} tile {tile}" + ); + } + } +} + +/// Six advances covering every kind of history event. +fn history() -> Vec)>> { + vec![ + vec![ + (1, map(Width::W64, &[10, 11, 12])), + (2, map(Width::W256, &[7, 3, 9, 1])), + ], + vec![(1, map(Width::W64, &[10, 11, 12, 13]))], // extension + vec![(1, map(Width::W64, &[13, 10, 11, 12]))], // replacement (reorder) + vec![(2, None)], // removal + vec![], // advance with no change + vec![(2, map(Width::W64, &[7])), (3, map(Width::W256, &[]))], // re-add, empty map + ] +} + +/// One sealed cycle as the fake keeps it: id, landings, coalesced image. +type SealedCycleRecord = (CycleId, Vec, BTreeMap>); + +/// One-writer in-memory sink. Fences on `base_version`, publishes head + 1, +/// keeps each cycle's landings and image, can hold commits behind a gate and +/// can fail a chosen cycle. Like the real writer, scanned landings carry no +/// payload. +#[derive(Default)] +struct GateWal { + head: u64, + cycles: Vec, + gate: Option>, + fail_on: Option, + /// Commit this cycle, then answer as if the acknowledgement was lost. + ambiguous_on: Option, +} + +impl WalSink for GateWal { + async fn commit_cycle( + &mut self, + batch: DetachedCycleBatch, + ) -> Result { + if let Some(gate) = &self.gate { + gate.acquire().await.expect("gate open").forget(); + } + if Some(batch.frame.cycle) == self.fail_on { + return Err(CommitError::Io(WriteFailed("injected".into()))); + } + if batch.frame.base_version.0 != self.head { + return Err(CommitError::Fenced { + current_head: DatasetVersion(self.head), + }); + } + self.head += 1; + let landings = batch + .landings + .iter() + .map(|s| { + let mut slot = s.clone(); + slot.payload = Vec::new(); + LandedSlot { + cycle: batch.frame.cycle, + slot, + } + }) + .collect(); + self.cycles + .push((batch.frame.cycle, landings, batch.image.clone())); + if Some(batch.frame.cycle) == self.ambiguous_on { + return Err(CommitError::Ambiguous { + cycle: batch.frame.cycle, + batch_hash: batch.batch_hash, + cause: "acknowledgement lost".into(), + }); + } + Ok(CommitOutcome::Committed { + version: DatasetVersion(self.head), + cycle: batch.frame.cycle, + batch_hash: batch.batch_hash, + }) + } + + async fn scan_sealed(&self, after: Option) -> Result, WriteFailed> { + Ok(self + .cycles + .iter() + .filter(|(c, _, _)| after.is_none_or(|a| *c > a)) + .flat_map(|(_, l, _)| l.iter().cloned()) + .collect()) + } + + async fn timeline(&self) -> Result, WriteFailed> { + Ok(Vec::new()) + } +} + +impl SealedImages for GateWal { + async fn sealed_image(&self, cycle: CycleId) -> Result>, WriteFailed> { + Ok(self + .cycles + .iter() + .find(|(c, _, _)| *c == cycle) + .map(|(_, _, i)| i.clone()) + .unwrap_or_default()) + } +} + +fn applied(s: Option) -> RouteVersion { + match s { + Some(Settled::Applied(v)) => v, + other => panic!("expected Applied, got {other:?}"), + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_restored_table_answers_every_past_version_exactly() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("map.lance"); + let sink = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + let head = sink.head(); + let (writer, handle) = cycle_writer(sink, head, 8); + + let (exit, live) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + for (k, changes) in history().into_iter().enumerate() { + let v = log.advance(1, 2 + k as u64, changes).unwrap(); + assert_eq!(v, RouteVersion(k as u64 + 1)); + } + for k in 1..=6 { + assert_eq!(applied(log.settle_next().await), RouteVersion(k)); + } + log.into_table().unwrap() + }); + let live = &live; + + // One Lance version per cycle: genesis plus six advances. + assert_eq!(exit.sink.head(), DatasetVersion(head.0 + 7)); + + let restored = restore(&exit.sink, OWNER, VersionClock::new()) + .await + .unwrap(); + assert_eq!(restored.last, RouteVersion(6)); + assert_same(live, &restored.table, 6); + for tile in TILES { + assert_eq!( + live.generation_count(tile), + restored.table.generation_count(tile) + ); + assert_eq!(live.event_count(tile), restored.table.event_count(tile)); + } +} + +#[tokio::test] +async fn the_owner_returns_before_the_seal_and_applies_only_after_it() { + let gate = Arc::new(Semaphore::new(0)); + let sink = GateWal { + gate: Some(gate.clone()), + ..GateWal::default() + }; + let (writer, handle) = cycle_writer(sink, DatasetVersion(0), 8); + + let (_exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + let v1 = log.advance(1, 1, vec![(1, map(Width::W64, &[5]))]).unwrap(); + let v2 = log + .advance(1, 1, vec![(1, map(Width::W64, &[5, 6]))]) + .unwrap(); + assert_eq!((v1, v2), (RouteVersion(1), RouteVersion(2))); + + // Nothing sealed yet: the table has not moved. + for _ in 0..100 { + tokio::task::yield_now().await; + } + assert!(log.settle().is_empty()); + assert_eq!(log.pending(), 3); + assert_eq!( + answer(log.table(), 2, 1), + Err(MapRead::NoSuchTile(1)), + "unsealed advances must not be applied" + ); + + gate.add_permits(3); + assert_eq!(applied(log.settle_next().await), v1); + assert_eq!(applied(log.settle_next().await), v2); + assert_eq!(answer(log.table(), 2, 1), Ok((64, vec![5, 6]))); + assert_eq!(answer(log.table(), 1, 1), Ok((64, vec![5]))); + }); +} + +#[tokio::test] +async fn a_clock_tick_between_claim_and_seal_lands_at_the_claimed_version() { + let gate = Arc::new(Semaphore::new(0)); + let sink = GateWal { + gate: Some(gate.clone()), + ..GateWal::default() + }; + let (writer, handle) = cycle_writer(sink, DatasetVersion(0), 8); + + let (exit, ()) = tokio::join!(writer.run(), async move { + let clock = VersionClock::new(); + let mut log = MapLog::create(handle, OWNER, 1, 1, clock.clone()).unwrap(); + let v = log.advance(1, 1, vec![(1, map(Width::W64, &[9]))]).unwrap(); + // Another holder of the clock (a RouteLog) ticks while the cycle seals. + clock.tick(); + clock.tick(); + gate.add_permits(2); + assert_eq!(applied(log.settle_next().await), v); + assert_eq!(log.table().current_version(), RouteVersion(v.0 + 2)); + assert_eq!(answer(log.table(), v.0, 1), Ok((64, vec![9]))); + assert_eq!(answer(log.table(), v.0 - 1, 1), Err(MapRead::NoSuchTile(1))); + }); + + let restored = restore(&exit.sink, OWNER, VersionClock::new()) + .await + .unwrap(); + assert_eq!(restored.last, RouteVersion(1)); + assert_eq!(answer(&restored.table, 1, 1), Ok((64, vec![9]))); +} + +#[tokio::test] +async fn a_failed_seal_applies_nothing_and_poisons_what_follows() { + let gate = Arc::new(Semaphore::new(0)); + let sink = GateWal { + gate: Some(gate.clone()), + fail_on: Some(CycleId(2)), + ..GateWal::default() + }; + let (writer, handle) = cycle_writer(sink, DatasetVersion(0), 8); + + let (exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + let v1 = log.advance(1, 1, vec![(1, map(Width::W64, &[1]))]).unwrap(); + let v2 = log + .advance(1, 1, vec![(1, map(Width::W64, &[1, 2]))]) + .unwrap(); + let v3 = log.advance(1, 1, vec![(2, map(Width::W64, &[3]))]).unwrap(); + gate.add_permits(4); + + assert_eq!(applied(log.settle_next().await), v1); + match log.settle_next().await { + Some(Settled::NotDurable { + version, + error: SealError::Failed(f), + }) => { + assert_eq!(version, v2); + assert_eq!(f.casts.len(), 2, "marker + one tile cast come back"); + } + other => panic!("expected the failed seal, got {other:?}"), + } + match log.settle_next().await { + Some(Settled::NotDurable { + version, + error: SealError::Poisoned { casts, .. }, + }) => { + assert_eq!(version, v3); + assert_eq!(casts.len(), 2); + } + other => panic!("expected the poisoned seal, got {other:?}"), + } + + // Neither v2 nor v3 was applied. + assert_eq!(answer(log.table(), v2.0, 1), Ok((64, vec![1]))); + assert_eq!(answer(log.table(), v3.0, 2), Err(MapRead::NoSuchTile(2))); + // And nothing more is submitted. + let before = log.table().current_version(); + match log.advance(1, 1, vec![]) { + Err(AdvanceError::Poisoned { failed }) => assert_eq!(failed, v2), + other => panic!("expected Poisoned, got {other:?}"), + } + assert_eq!(log.table().current_version(), before, "nothing was claimed"); + // A proven gap leaves store and table in agreement: the log drains. + log.into_table() + .expect("a proven-unpublished failure still drains"); + }); + + // The store holds genesis and v1 only, and replays to the same table. + let restored = restore(&exit.sink, OWNER, VersionClock::new()) + .await + .unwrap(); + assert_eq!(restored.last, RouteVersion(1)); + assert_eq!(answer(&restored.table, 1, 1), Ok((64, vec![1]))); + assert_eq!(answer(&restored.table, 1, 2), Err(MapRead::NoSuchTile(2))); +} + +#[tokio::test] +async fn every_landing_is_on_behalf_of_the_owner() { + let (writer, handle) = cycle_writer(GateWal::default(), DatasetVersion(0), 8); + let (exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + log.advance(1, 1, vec![(1, map(Width::W256, &[1, 2]))]) + .unwrap(); + while log.settle_next().await.is_some() {} + }); + let landings = exit.sink.scan_sealed(None).await.unwrap(); + // genesis marker + advance marker + four W256 parts + assert_eq!(landings.len(), 6); + assert!(landings.iter().all(|l| l.slot.owner == OWNER)); + assert!(landings.iter().all(|l| l.slot.paired_move.is_none())); + + match restore(&exit.sink, OWNER + 1, VersionClock::new()).await { + Err(RestoreError::Replay(ReplayError::ForeignOwner { owner, .. })) => { + assert_eq!(owner, OWNER) + } + Ok(_) => panic!("a restore on behalf of another mailbox must be refused"), + Err(e) => panic!("expected ForeignOwner, got {e:?}"), + } +} + +#[tokio::test] +async fn a_reserved_id_or_a_repeated_tile_is_refused_before_a_version_is_claimed() { + let (writer, handle) = cycle_writer(GateWal::default(), DatasetVersion(0), 8); + let (_exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + let before = log.table().current_version(); + let reserved = LocalMap::prepare(Width::W64, &[1, RESERVED_ID]).unwrap(); + assert!(matches!( + log.advance(1, 1, vec![(4, Some(reserved))]), + Err(AdvanceError::ReservedId(4)) + )); + assert!(matches!( + log.advance(1, 1, vec![(4, None), (4, map(Width::W64, &[1]))]), + Err(AdvanceError::DuplicateTile(4)) + )); + assert_eq!(log.table().current_version(), before); + // The silence twin: a valid advance is claimed and applied. + let v = log.advance(1, 1, vec![(4, map(Width::W64, &[1]))]).unwrap(); + assert_eq!(v.0, before.0 + 1); + while log.settle_next().await.is_some() {} + assert_eq!(answer(log.table(), v.0, 4), Ok((64, vec![1]))); + }); +} + +#[tokio::test] +async fn a_restored_log_continues_and_restores_again() { + let (writer, handle) = cycle_writer(GateWal::default(), DatasetVersion(0), 8); + let (exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + log.advance(1, 1, vec![(1, map(Width::W64, &[1]))]).unwrap(); + while log.settle_next().await.is_some() {} + }); + let sink = exit.sink; + let restored = restore(&sink, OWNER, VersionClock::new()).await.unwrap(); + let head = DatasetVersion(sink.head); + let (writer, handle) = cycle_writer(sink, head, 8); + let (exit, first) = tokio::join!(writer.run(), async move { + let mut log = MapLog::resume(restored, handle, OWNER); + let v = log.advance(1, 1, vec![(1, None)]).unwrap(); + assert_eq!(v, RouteVersion(2)); + assert_eq!(applied(log.settle_next().await), v); + log.into_table().unwrap() + }); + let again = restore(&exit.sink, OWNER, VersionClock::new()) + .await + .unwrap(); + assert_eq!(again.last, RouteVersion(2)); + assert_same(&first, &again.table, 2); +} + +#[tokio::test] +async fn a_frame_from_before_a_restore_is_foreign() { + let (writer, handle) = cycle_writer(GateWal::default(), DatasetVersion(0), 8); + let (exit, log) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + log.advance(1, 1, vec![(1, map(Width::W64, &[1]))]).unwrap(); + while log.settle_next().await.is_some() {} + log.into_table().unwrap() + }); + let frame = log.seal(); + let restored = restore(&exit.sink, OWNER, VersionClock::new()) + .await + .unwrap(); + assert!(matches!( + restored.table.map(&frame, 1), + Err(MapRead::Stale(FrameRefusal::ForeignFrame)) + )); + // The same version still answers through diagnose_at. + assert_eq!(answer(&restored.table, 1, 1), Ok((64, vec![1]))); +} + +#[tokio::test] +async fn a_restore_onto_a_clock_that_has_moved_on_replays_exactly() { + let (writer, handle) = cycle_writer(GateWal::default(), DatasetVersion(0), 8); + let (exit, live) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + log.advance(1, 1, vec![(1, map(Width::W64, &[4]))]).unwrap(); + log.advance(1, 1, vec![(1, map(Width::W64, &[4, 5]))]) + .unwrap(); + while log.settle_next().await.is_some() {} + log.into_table().unwrap() + }); + + // Another component sharing the clock was restored first and moved it on. + let shared = VersionClock::new(); + for _ in 0..5 { + shared.tick(); + } + let restored = restore(&exit.sink, OWNER, shared.clone()).await.unwrap(); + assert_eq!(restored.last, RouteVersion(2)); + assert_same(&live, &restored.table, 2); + assert_eq!(restored.table.current_version(), RouteVersion(5)); + + // Advances continue on the shared clock. + let head = DatasetVersion(exit.sink.head); + let (writer, handle) = cycle_writer(exit.sink, head, 8); + let ((), ()) = tokio::join!( + async { + let _ = writer.run().await; + }, + async move { + let mut log = MapLog::resume(restored, handle, OWNER); + let v = log.advance(1, 1, vec![(1, None)]).unwrap(); + assert_eq!(v, RouteVersion(6)); + assert_eq!(applied(log.settle_next().await), v); + } + ); +} + +#[tokio::test] +async fn a_cancelled_settle_keeps_the_advance_pending() { + let gate = Arc::new(Semaphore::new(0)); + let sink = GateWal { + gate: Some(gate.clone()), + ..GateWal::default() + }; + let (writer, handle) = cycle_writer(sink, DatasetVersion(0), 8); + let (_exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + let v = log.advance(1, 1, vec![(1, map(Width::W64, &[8]))]).unwrap(); + + // Waiting is cancelled before the seal lands (a shutdown timeout). + let waited = + tokio::time::timeout(std::time::Duration::from_millis(20), log.settle_next()).await; + assert!(waited.is_err(), "the seal is held, so the wait times out"); + assert_eq!(log.pending(), 2, "genesis and the advance are still queued"); + + gate.add_permits(2); + assert_eq!(applied(log.settle_next().await), v); + assert_eq!(answer(log.table(), v.0, 1), Ok((64, vec![8]))); + }); +} + +/// The drained-state contract: `into_table` never hands out a table while a +/// submitted cycle is unsettled, even one the store already sealed. The log +/// comes back, settles, and the second ask returns the applied table. +#[tokio::test] +async fn into_table_refuses_while_a_submitted_cycle_is_unsettled() { + let gate = Arc::new(Semaphore::new(0)); + let sink = GateWal { + gate: Some(gate.clone()), + ..GateWal::default() + }; + let (writer, handle) = cycle_writer(sink, DatasetVersion(0), 8); + + let (exit, table) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + let v = log.advance(1, 1, vec![(1, map(Width::W64, &[5]))]).unwrap(); + gate.add_permits(2); + // Let the writer seal both cycles without the log settling them. + for _ in 0..1000 { + tokio::task::yield_now().await; + } + + let undrained = log.into_table().expect_err("sealed but unsettled"); + assert_eq!(undrained.pending(), 2); + assert_eq!(undrained.diverged(), None); + let mut log = undrained.into_log(); + assert_eq!(applied(log.settle_next().await), v); + assert_eq!(log.pending(), 0); + log.into_table().expect("drained") + }); + assert_eq!(exit.sink.cycles.len(), 2, "both cycles were sealed"); + assert_eq!(answer(&table, 1, 1), Ok((64, vec![5]))); +} + +/// A seal whose outcome is unknown may have published the cycle. The log +/// must not hand out a table that may lack it; a restore shows the store did +/// hold it. +#[tokio::test] +async fn an_ambiguous_seal_blocks_the_table_until_a_restore() { + let sink = GateWal { + ambiguous_on: Some(CycleId(1)), + ..GateWal::default() + }; + let (writer, handle) = cycle_writer(sink, DatasetVersion(0), 8); + + let (exit, ()) = tokio::join!(writer.run(), async move { + let mut log = MapLog::create(handle, OWNER, 1, 1, VersionClock::new()).unwrap(); + let v1 = log.advance(1, 1, vec![(1, map(Width::W64, &[4]))]).unwrap(); + match log.settle_next().await { + Some(Settled::OutcomeUnknown { version, .. }) => assert_eq!(version, v1), + other => panic!("expected an unknown outcome, got {other:?}"), + } + assert_eq!(answer(log.table(), 1, 1), Err(MapRead::NoSuchTile(1))); + assert!(matches!( + log.advance(1, 1, vec![]), + Err(AdvanceError::Poisoned { failed }) if failed == v1 + )); + let undrained = log.into_table().expect_err("the table may lack v1"); + assert_eq!(undrained.pending(), 0); + assert_eq!(undrained.diverged(), Some(v1)); + }); + + // The store did hold v1: the restored table has it. + let restored = restore(&exit.sink, OWNER, VersionClock::new()) + .await + .unwrap(); + assert_eq!(restored.last, RouteVersion(1)); + assert_eq!(answer(&restored.table, 1, 1), Ok((64, vec![4]))); +} diff --git a/crates/lance-graph-dt64/src/dr.rs b/crates/lance-graph-dt64/src/dr.rs index 47cfe9dd2..50c7ae1dd 100644 --- a/crates/lance-graph-dt64/src/dr.rs +++ b/crates/lance-graph-dt64/src/dr.rs @@ -392,6 +392,16 @@ impl FrameStore { /// A store whose advances tick `clock`. pub fn with_clock(revisions: FrameRevisions, clock: VersionClock) -> Self { let version = clock.current(); + Self::with_clock_at(revisions, clock, version) + } + + /// A store whose current seal is at `version`, which the caller has + /// checked is not past `clock`'s current version. + fn with_clock_at( + revisions: FrameRevisions, + clock: VersionClock, + version: RouteVersion, + ) -> Self { Self { identity: Arc::new(()), current: Arc::new(revisions), @@ -659,6 +669,21 @@ fn fnv_ids(globals: &[u64]) -> u64 { /// A tile's name in a [`MapTable`], chosen by the caller. pub type TileId = u32; +/// Why [`MapTable::advance_at`] refused a claimed version. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AdvanceRefusal { + /// `version` is at or before `last`, the last version this table applied. + NotAfterLast { + last: RouteVersion, + version: RouteVersion, + }, + /// The clock has not issued `version` (its last issued is `current`). + NotIssued { + version: RouteVersion, + current: RouteVersion, + }, +} + /// Why a [`MapTable`] read was refused. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum MapRead { @@ -873,6 +898,39 @@ impl MapTable { } } + /// An empty table whose genesis is at `genesis`, an earlier version of + /// `clock` (a table rebuilt from storage while the clock it shares has + /// moved on). Its first advance must come after `genesis`. Refuses a + /// genesis the clock has not issued. + pub fn with_clock_at( + source: u64, + relation_map: u64, + clock: VersionClock, + genesis: RouteVersion, + ) -> Result { + let current = clock.current(); + if genesis > current { + return Err(AdvanceRefusal::NotIssued { + version: genesis, + current, + }); + } + let tiles = BTreeMap::new(); + let coordinates = Self::coordinates(&tiles); + Ok(Self { + frames: FrameStore::with_clock_at( + FrameRevisions { + source, + relation_map, + coordinates, + }, + clock, + genesis, + ), + tiles, + }) + } + /// The clock this table advances (a clone that shares the counter). pub fn clock(&self) -> VersionClock { self.frames.clock.clone() @@ -916,6 +974,52 @@ impl MapTable { changes: Vec<(TileId, Option)>, ) -> RouteVersion { let version = self.frames.clock.tick(); + self.apply(version, source, relation_map, changes); + version + } + + /// Take the next version from the (possibly shared) clock for an advance + /// that will be applied later with [`MapTable::advance_at`], once its + /// rows are durable. A claimed version that is never applied is a gap: + /// no advance, no frame, nothing to replay. + pub fn claim(&self) -> RouteVersion { + self.frames.clock.tick() + } + + /// Apply an advance at a version claimed earlier with [`MapTable::claim`], + /// exactly as [`MapTable::advance`] would have at that version. Other + /// holders of the clock may have ticked since the claim; the advance still + /// lands at `version`. + /// + /// Refuses, changing nothing, a version at or before the last one this + /// table applied (its event logs would leave version order) and a version + /// the clock has not issued. + pub fn advance_at( + &mut self, + version: RouteVersion, + source: u64, + relation_map: u64, + changes: Vec<(TileId, Option)>, + ) -> Result { + let current = self.frames.clock.current(); + if version > current { + return Err(AdvanceRefusal::NotIssued { version, current }); + } + let last = self.frames.version; + if version <= last { + return Err(AdvanceRefusal::NotAfterLast { last, version }); + } + self.apply(version, source, relation_map, changes); + Ok(version) + } + + fn apply( + &mut self, + version: RouteVersion, + source: u64, + relation_map: u64, + changes: Vec<(TileId, Option)>, + ) { for (tile, map) in changes { match map { Some(m) => self.tiles.entry(tile).or_default().append(version, m), @@ -935,7 +1039,6 @@ impl MapTable { }, version, ); - version } /// Seal the current frame. diff --git a/crates/lance-graph-dt64/tests/dr.rs b/crates/lance-graph-dt64/tests/dr.rs index cea3036f8..19188b3b0 100644 --- a/crates/lance-graph-dt64/tests/dr.rs +++ b/crates/lance-graph-dt64/tests/dr.rs @@ -344,8 +344,8 @@ mod locate { mod vmap { use super::map; use lance_graph_dt64::dr::{ - Axis, FrameRefusal, MapRead, MapRefusal, MapTable, RouteLog, RouteVersion, VersionClock, - Width, + AdvanceRefusal, Axis, FrameRefusal, MapRead, MapRefusal, MapTable, RouteLog, RouteVersion, + VersionClock, Width, }; use std::alloc::{GlobalAlloc, Layout, System}; use std::cell::Cell; @@ -553,6 +553,70 @@ mod vmap { assert!(copy.current_version() > log.current_version()); } + /// A version claimed before a write is applied exactly, even when another + /// holder of the shared clock ticks between the claim and the apply (a + /// cast is sealed by the WAL while a RouteLog keeps routing). Fails if the + /// apply takes a fresh tick instead of the claimed version. + #[test] + fn a_claimed_version_is_applied_exactly_after_the_clock_moves() { + let clock = VersionClock::new(); + let mut log = RouteLog::with_clock(clock.clone()); + let mut t = MapTable::with_clock(1, 1, clock.clone()); + let id = Axis::new(Width::W64, 1); + let v = t.claim(); + let moved = log.route(3, 9, id).unwrap(); + assert!( + moved > v, + "anti-vacuity: the clock really moved past the claim" + ); + assert_eq!( + t.advance_at(v, 1, 1, vec![(1, Some(map(&[10, 11])))]), + Ok(v) + ); + assert_eq!(t.seal().version(), v); + let d = t.diagnose_at(v, 1).unwrap(); + assert_eq!((d.len(), d.global(0), d.global(1)), (2, Some(10), Some(11))); + assert_eq!( + t.diagnose_at(RouteVersion(v.0 - 1), 1).unwrap_err(), + MapRead::NoSuchTile(1) + ); + // Silence twin: plain advance still takes the next tick. + let w = t.advance(1, 1, vec![(1, Some(map(&[10, 11, 12])))]); + assert!(w > moved); + } + + /// advance_at refuses, and changes nothing, for a version at or before + /// the last one applied (events would leave version order) and for a + /// version the clock never issued (a fabricated version). + #[test] + fn advance_at_refuses_an_old_or_unissued_version() { + let mut t = MapTable::new(1, 1); + let early = t.claim(); + let late = t.claim(); + assert_eq!( + t.advance_at(late, 1, 1, vec![(1, Some(map(&[5])))]), + Ok(late) + ); + assert_eq!( + t.advance_at(early, 1, 1, vec![(1, Some(map(&[6])))]), + Err(AdvanceRefusal::NotAfterLast { + last: late, + version: early + }) + ); + let forged = RouteVersion(t.current_version().0 + 5); + assert_eq!( + t.advance_at(forged, 1, 1, vec![(1, Some(map(&[7])))]), + Err(AdvanceRefusal::NotIssued { + version: forged, + current: t.current_version() + }) + ); + let d = t.diagnose_at(late, 1).unwrap(); + assert_eq!((d.len(), d.global(0)), (1, Some(5))); + assert_eq!(t.seal().version(), late); + } + /// Q2: a version past the end is a typed refusal, not a clamp. #[test] fn diagnose_at_refuses_a_version_past_the_end() { @@ -784,3 +848,35 @@ fn an_unchanged_space_survives_an_advance_of_another() { assert_eq!(t.code(&f, &priority(), &vk("high")), Ok(1)); assert_eq!(t.key(&f, &priority(), 1), Ok(&vk("high"))); } + +mod genesis_at { + use lance_graph_dt64::dr::{AdvanceRefusal, MapTable, RouteVersion, VersionClock}; + + #[test] + fn a_table_can_start_at_a_past_version_of_its_clock() { + let clock = VersionClock::new(); + for _ in 0..4 { + clock.tick(); + } + let mut t = MapTable::with_clock_at(1, 1, clock.clone(), RouteVersion(1)).unwrap(); + // The first advance after genesis applies, though the clock is at 4. + assert_eq!( + t.advance_at(RouteVersion(2), 1, 1, vec![]), + Ok(RouteVersion(2)) + ); + // Not at or before genesis. + let mut t = MapTable::with_clock_at(1, 1, clock.clone(), RouteVersion(3)).unwrap(); + assert_eq!( + t.advance_at(RouteVersion(3), 1, 1, vec![]), + Err(AdvanceRefusal::NotAfterLast { + last: RouteVersion(3), + version: RouteVersion(3) + }) + ); + // A genesis the clock has not issued is refused. + assert!(matches!( + MapTable::with_clock_at(1, 1, clock, RouteVersion(9)), + Err(AdvanceRefusal::NotIssued { .. }) + )); + } +} diff --git a/crates/lance-graph-supervisor/src/cycle_writer.rs b/crates/lance-graph-supervisor/src/cycle_writer.rs index 67e033d01..add57c72e 100644 --- a/crates/lance-graph-supervisor/src/cycle_writer.rs +++ b/crates/lance-graph-supervisor/src/cycle_writer.rs @@ -144,6 +144,14 @@ impl SealTicket { self.rx.await.unwrap_or(Err(SealError::WriterGone)) } + /// Wait for this cycle's seal without consuming the ticket, so a caller + /// that is cancelled while waiting still holds it and can wait again. + /// After this returns, the ticket is spent: call neither `wait` nor + /// `try_sealed` on it again. + pub async fn wait(&mut self) -> Result { + (&mut self.rx).await.unwrap_or(Err(SealError::WriterGone)) + } + /// The seal if it has already resolved, without waiting. `None` while the /// writer is still on it. After this returns `Some`, the ticket is spent. pub fn try_sealed(&mut self) -> Option> { diff --git a/crates/lance-graph/examples/wal_shootout.rs b/crates/lance-graph/examples/wal_shootout.rs index a229936c4..00f003b48 100644 --- a/crates/lance-graph/examples/wal_shootout.rs +++ b/crates/lance-graph/examples/wal_shootout.rs @@ -4,6 +4,8 @@ //! ```text //! # cost table (D-WAL-2) //! cargo run --release -p lance-graph --features lancedb-sdk --example wal_shootout -- latency +//! # DR/DT64 64x64: N = 1, 4, 16, 64 changed tiles per advance +//! cargo run --release -p lance-graph --features lancedb-sdk --example wal_shootout -- dr64 //! # soak, one process per arm (D-WAL-3) //! cargo run --release -p lance-graph --features lancedb-sdk --example wal_shootout -- \ //! soak @@ -101,13 +103,21 @@ impl Arm { } } -async fn latency(root: &str) { +/// The SoA-512 sizes the shoot-out was run at (D-WAL-2). +const SOA512: [(usize, u64); 3] = [(64, 200), (1024, 200), (8192, 40)]; + +/// DR/DT64 64x64: one cast per changed DR64 tile (64 global ids x 8 bytes = +/// one 512-byte payload). N is the number of tiles one advance changes: a +/// single tile up to the whole 64-tile cube. +const DR64: [(usize, u64); 4] = [(1, 200), (4, 200), (16, 200), (64, 200)]; + +async fn latency(root: &str, sizes: &[(usize, u64)]) { println!( "tier: avx512f={} | host I/O bound: numbers are per cycle (µs) and per stored row (ns)", cfg!(target_feature = "avx512f") ); println!("arm N cycles p50_us p99_us ns/row rows MiB files versions scan_ms"); - for (n, cycles) in [(64usize, 200u64), (1024, 200), (8192, 40)] { + for &(n, cycles) in sizes { for kind in ["L", "M10", "M100"] { let dir = format!("{root}/lat-{kind}-{n}"); let _ = std::fs::remove_dir_all(&dir); @@ -197,7 +207,8 @@ async fn soak(kind: &str, dir: &str, minutes: u64, n: usize, interval_ms: u64, c async fn main() { let args: Vec = std::env::args().collect(); match args.get(1).map(String::as_str) { - Some("latency") => latency(args.get(2).expect("scratch dir")).await, + Some("latency") => latency(args.get(2).expect("scratch dir"), &SOA512).await, + Some("dr64") => latency(args.get(2).expect("scratch dir"), &DR64).await, Some("soak") => { soak( &args[2],