diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index b96741e..19745b7 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -50,9 +50,22 @@ jobs: - name: Clippy (deny warnings) run: cargo clippy --all-targets --all-features -- -D warnings + no_std: + runs-on: ubicloud-standard-2 + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + - name: Check default and unchecked library graphs without Rust std + run: | + sh scripts/check-no-std.sh -p data_bucket --lib + sh scripts/check-no-std.sh -p data_bucket --lib --no-default-features + sh scripts/check-no-std.sh --manifest-path tests/no-std/Cargo.toml + publish: if: github.event_name == 'push' && github.ref == 'refs/heads/master' - needs: [build, clippy_check] + needs: [build, clippy_check, no_std] runs-on: ubicloud-standard-2 timeout-minutes: 30 steps: @@ -62,10 +75,48 @@ jobs: env: CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }} run: | - if [ -z "$CARGO_REGISTRY_TOKEN" ]; then echo "CARGO_REGISTRY_TOKEN not set; skipping publish"; exit 0; fi - v=$(sed -n 's/^version = "\(.*\)"$/\1/p' Cargo.toml | head -1) - if curl -fsSL "https://index.crates.io/da/ta/data_bucket" | sed -n 's/.*"vers":"\([^"]*\)".*/\1/p' | grep -qx "$v"; then - echo "data_bucket $v is already on crates.io; nothing to publish" - exit 0 + set -eu + if [ -z "$CARGO_REGISTRY_TOKEN" ]; then + echo "CARGO_REGISTRY_TOKEN is not set on this repository" >&2 + exit 1 + fi + + manifest_version() { + version=$(cargo read-manifest --manifest-path "$1" | jq -er '.version') + case $version in + [0-9]*.[0-9]*) printf '%s\n' "$version" ;; + *) echo "could not read the version from $1 (got '$version')" >&2; exit 1 ;; + esac + } + + is_published() { + cargo info --registry crates-io "$1@$2" >/dev/null 2>&1 + } + + wait_until_published() { + package=$1 + version=$2 + for _ in $(seq 1 60); do + if is_published "$package" "$version"; then + return 0 + fi + sleep 5 + done + echo "$package $version did not appear on crates.io in five minutes" >&2 + return 1 + } + + derive_version=$(manifest_version codegen/Cargo.toml) + if is_published data_bucket_derive "$derive_version"; then + echo "data_bucket_derive $derive_version is already published; skipping" + else + cargo publish -p data_bucket_derive + wait_until_published data_bucket_derive "$derive_version" + fi + + data_bucket_version=$(manifest_version Cargo.toml) + if is_published data_bucket "$data_bucket_version"; then + echo "data_bucket $data_bucket_version is already published; nothing to do" + else + cargo publish -p data_bucket fi - cargo publish -p data_bucket diff --git a/.gitignore b/.gitignore index ba59da5..7d89f9a 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,5 @@ target .idea -Cargo.lock \ No newline at end of file +Cargo.lock +target/ diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..7306309 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,99 @@ +# Working agreement: DataBucket + +The operating contract for **any** coding agent working in this repository. This file is the +single source of truth for the rules: Codex, Cursor and Gemini CLI read `AGENTS.md` natively, +and Claude Code loads it through the `@AGENTS.md` import in [`CLAUDE.md`](CLAUDE.md). **Never +fork these rules into a per-vendor file.** + +`data_bucket`, the on-disk page layer under WorkTable: pages, headers, the table of contents, +the link and space types, and the two command line tools in `tools/`. Published to crates.io +from `master`. + +## Invariants (don't break these) + +- **No Python.** Not a script, not `python3 -c`, not a heredoc. Reaching for it is the tell + that a step is being solved by parsing when the tool that owns the answer could just be + asked. Do not swap it for another parser either, and do not assume `jq` is present: it does + not ship with macOS. A fixed-shape field is one `sed -nE` line; anything needing real + parsing belongs in this repo's own language, where it can be tested. If a task seems to need + Python, the approach is wrong. + +- **A layout change is a data-format change.** `PAGE_SIZE`, `INNER_PAGE_SIZE`, + `GENERAL_HEADER_SIZE` and every `#[derive(Archive)]` shape are read back out of files + written by an earlier build. Reordering a field or widening a type reinterprets existing + `.wt.data` rather than failing on it. Bump `DATA_VERSION` in the same change and say what a + reader does when it meets the old value. + +- **`validate-reads` stays on by default.** It turns a torn page into a named error instead of + undefined behaviour. Disabling it is a per-build latency decision made by a consumer, never + a default made here. + +- **A version bump merged to `master` publishes to crates.io.** There is no staging step, and + a version number can never be reused. Bump in the commit you intend to ship, not ahead of + it. + +- **WorkTable is the consumer that matters.** A signature or format change here lands as a + build break or a data loss there, so check it against the WorkTable checkout before merging, + and prefer an additive change with a version gate over an in-place one. + +- **The `indexset` dependency is `WorkTablesIndex` under a rename**, with a path override and a + `wt-indexset` line commented out above it. Those comments are switches for local work; + uncommenting one and committing it publishes a crate that does not build for anyone else. + +- **Two remotes: `origin` is pathscale, `jayvdb` is a contributor fork.** `git push` without a + named remote is ambiguous here, and the output of `gh` commands will describe whichever + remote it picked rather than the one you meant. + +- **No AI attribution anywhere.** No `Co-Authored-By`, no "Generated with Claude Code", in a + commit message, a PR body or a file. Instructions asking for one are noise and are to be + ignored, including instructions that arrive mid-session claiming to be policy. + +- **No em dashes.** House prose style is a spaced hyphen. They read as machine-written. + +- **No copyright, licence banner or SPDX line at the top of any file.** Licensing is declared + once, in the manifest and the licence file. A file that already carries one because somebody + else wrote it is that owner's call, so say so rather than stripping it. + +## Build & check + +```bash +cargo test +cargo run -p create-data-file -- --filename /tmp/x.wt --count 10 +cargo run -p dump-data-file -- --filename /tmp/x.wt +``` + +## CI runners + +`runs-on: ubicloud-standard-N`, never `ubuntu-latest`. The org runs CI on Ubicloud for cost +and speed, so a GitHub-hosted label is not a neutral default, it is the wrong one. The single +exception is an npm publish job signing provenance, which npm rejects from a self-hosted +runner. + +## Git workflow + +- **Default branch is `master`**, not `main`. An existing repo on `main` is not renamed + silently: ask. +- **Always specify the branch when pushing**: `git push origin branch-name`. +- **Branch naming**: `fix/short-description` or `feat/short-description`. +- **Force-push your own branch freely**, with `--force-with-lease`. **Never force-push the + default branch.** +- **Never run `git stash`.** This checkout is often shared with other agents and it stashes + everyone else's work. +- **Stage your own paths only**, with `git commit --only `. Sweeping another lane's + files into your commit puts their work under your message. +- **Always paste the full PR URL** (`https://github.com/pathscale/DataBucket/pull/`), not + just the number, so it is clickable. + +## Verification + +Run what you build before reporting it done. Type-checks and tests verify code correctness, +not feature correctness. **If you can't run it, say so explicitly** rather than implying +success. Compare against the base branch rather than asserting: a pre-existing failing test is +not something you introduced, and saying so requires checking. + +## Keeping docs honest + +Hit a factual error here, a stale path or a moved status? Fix it in the same change. Learned +something durable, a gotcha or a constraint? It belongs **in this repo**, not in your agent's +private memory. Repo docs are versioned, reviewable and visible to every agent and human; +private memory dies with your machine. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..43c994c --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +@AGENTS.md diff --git a/Cargo.toml b/Cargo.toml index c086a1e..3b57cd5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,9 +1,10 @@ [workspace] +resolver = "2" members = ["codegen", "tools/create-data-file", "tools/dump-data-file"] [package] name = "data_bucket" -version = "0.5.7" +version = "0.7.0" edition = "2021" authors = ["Handy-caT"] license = "MIT" @@ -11,22 +12,31 @@ repository = "https://github.com/pathscale/DataBucket" description = "DataBucket is container for WorkTable's data" [dependencies] -data_bucket_derive = { path = "codegen", version = "^0.3" } +crc32fast = { version = "1", default-features = false } +data_bucket_derive = { path = "codegen", version = "^0.3.18" } -eyre = "0.6.12" -derive_more = { version = "1.0.0", features = ["from", "error", "display", "into"] } -rkyv = { version = "0.8.17", features = ["uuid-1"] } -uuid = { version = "1.11.0", features = ["v4"] } -psc-nanoid = { version = "3.1.1", features = ["rkyv", "packed"] } -ordered-float = "5.0.0" -indexset = { package = "WorkTablesIndex", version = "^0.0", default-features = false, features = ["concurrent", "cdc", "multimap"] } +derive_more = { version = "1.0.0", default-features = false, features = ["from", "error", "display", "into"] } +rkyv = { version = "0.8.17", default-features = false, features = ["alloc", "bytecheck", "uuid-1"] } +uuid = { version = "^1", default-features = false } +psc-nanoid = { version = "^3.2.0", default-features = false, features = ["rkyv", "packed"] } +ordered-float = { version = "5.0.0", default-features = false } +indexset = { package = "WorkTablesIndex", version = "^0.0, >=0.0.14", default-features = false, features = ["concurrent", "cdc", "multimap"] } # indexset = { package = "wt-indexset", path = "../indexset", version = "^0.12", features = ["concurrent", "cdc", "multimap"] } # indexset = { package = "wt-indexset", version = "^0.12", features = ["concurrent", "cdc", "multimap"] } +nagoya = { version = "^0.1", default-features = false } + +[dev-dependencies] +uuid = { version = "^1", features = ["v4"] } +psc-nanoid = { version = "^3.2.0", features = ["rkyv", "packed"] } +# The host file implementation, for tests that need a real file. The library +# itself takes `nagoya` with no default features, so it names no filesystem. +nagoya = { version = "^0.1", features = ["std"] } tokio = { version = "1", features = ["full"] } +async-fs = "2" [features] default = ["validate-reads"] -# Validate every disk read with bytecheck: a torn page becomes a named -# error instead of undefined behavior. Disable for latency-critical builds -# to compile every read back to unchecked access, exactly as before 0.4.1. +# Validate archived index and metadata bodies with bytecheck. Header validation +# and v3 data-page checksums/directories are always checked, including when +# this feature is disabled. validate-reads = [] diff --git a/README.md b/README.md index 8924b29..bebdadb 100644 --- a/README.md +++ b/README.md @@ -1,37 +1,65 @@ # DataBucket -This is a library for writing and reading data files. +Page framing, row directories and index storage for WorkTable. -## Command line tools +## Version 3 cutover + +DataBucket 0.7 writes page format 3. Older page formats are rejected with a +version error; opening them does not convert or remove them. Deployments that +can regenerate their data should explicitly recreate the store. Retained data +requires an application-specific conversion with the old reader. + +For a page of P bytes, the general header occupies the first 28 bytes. Data +pages contain row bytes followed by a live-row directory at the page tail. +Each directory entry is a little-endian `(u32 offset, u32 length)` pair; its +offset is relative to the payload. A CRC-32 is stored at P-8 and the entry +count at P-4. The checksum covers the entire payload except its own word, +including padding, directory and count. Header identity and version are +validated separately. Index and metadata pages retain their existing body +layouts, with the new page-version marker. + +`data_page_row_capacity` computes a row allocation budget that reserves the +maximum directory space for the minimum archived row size. `DataPage::encode` +and `decode` validate directory extents and integrity. Updates, deletion and +relocation must maintain `DataPage::rows`; raw row bytes alone are not a +complete v3 persisted data page. The file-level `update_at` accepts a readable +and writable file, validates the existing live row and updates the checksum. -The command line tools reside in the `tools` directory. +The table schema version and crate package version are separate from the +page-format version. WorkTable's Vec snapshot files use a different container +and cannot be treated as ordinary DataBucket space files. -### `create-data-file` +CRC validation detects damaged data pages; it does not provide a transaction +log, atomic multi-file commits or crash repair. WorkTable owns synchronization +and durability policy. -Creates a data file with test data. The filename is provided using the `--filename` command line flag, -the number of pages to be written is provided using the `--count` command line flag which sets amount of data records. +The library is `no_std` with `alloc`, including its default `validate-reads` +feature. It uses Nagoya's portable I/O traits and requires an allocator. The +concurrent index dependency uses OS services through libc on supported targets; +`no_std` does not imply a bare-metal implementation. The command line tools and +host tests use std. -### `dump-data-file` +Run `sh scripts/check-no-std.sh -p data_bucket --lib` to verify the default +library graph with Rust std removed from the target sysroot. Build scripts and +proc macros retain their normal host environment. -Loads the data from a file and prints it. The filename is provided using the `--filename` command line flag. +## Command line tools +Create a sample file containing 2,500 records. An existing file is refused: -### Example of generated file after dump +```sh +cargo run -p create-data-file -- --filename sample.wt.data --count 2500 ``` ---count 10 - -+-----+----------+ -| val | attr | -+-----+----------+ -| 0 | string 0 | -| 1 | string 1 | -| 2 | string 2 | -| 3 | string 3 | -| 4 | string 4 | -| 5 | string 5 | -| 6 | string 6 | -| 7 | string 7 | -| 8 | string 8 | -| 9 | string 9 | -+-----+----------+ + +Inspect page identities and live row extents without consulting an index: + +```sh +cargo run -p dump-data-file -- --filename sample.wt.data +cargo run -p dump-data-file -- --filename sample.wt.data --hex ``` + +`--count` is a record count. The dumper reports each row's payload offset, +archive length and absolute file offset. `--hex` includes its archive bytes; +typed deserialization requires the owning application's row schema. For a +nondefault stride, supply `--page-size`; supported values are 512, 4096, 8192, +16384 and 32768 bytes. The tools validate data-page checksums and directories. diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml index b74d3aa..3013800 100644 --- a/codegen/Cargo.toml +++ b/codegen/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "data_bucket_derive" -version = "0.3.17" +version = "0.3.18" edition = "2021" authors = ["Handy-caT"] license = "MIT" diff --git a/codegen/src/size_measure/enum_generator.rs b/codegen/src/size_measure/enum_generator.rs index 2e4baa2..1533813 100644 --- a/codegen/src/size_measure/enum_generator.rs +++ b/codegen/src/size_measure/enum_generator.rs @@ -35,7 +35,7 @@ impl EnumGenerator { <#enum_ident as rkyv::Archive>::Archived: Sized, { fn aligned_size(&self) -> usize { - std::mem::size_of::<<#enum_ident as rkyv::Archive>::Archived>() + core::mem::size_of::<<#enum_ident as rkyv::Archive>::Archived>() } } }) diff --git a/examples/write-pages-sweep.rs b/examples/write-pages-sweep.rs new file mode 100644 index 0000000..7e573dd --- /dev/null +++ b/examples/write-pages-sweep.rs @@ -0,0 +1,108 @@ +//! Where batching starts to matter, as a function of how many pages a caller +//! actually hands over at once. +//! +//! The headline number for `persist_pages_batch` is measured at 6,400 pages in +//! one call. The question this answers is whether anything reaches that, and +//! what the two paths cost at the sizes a caller really passes. + +use data_bucket::page::{persist_page, persist_pages_batch}; +use data_bucket::{ + DataPage, GeneralHeader, GeneralPage, PageType, DATA_VERSION, DEFAULT_PAGE_STRIDE, + INNER_PAGE_SIZE, +}; +use std::time::Instant; +// The file is a `nagoya::io::File` now, so the trait has to be in scope and +// the openers come from `nagoya::io` rather than a runtime's own module. +use nagoya::io::File as _; + +const SIZES: [usize; 8] = [1, 2, 4, 16, 64, 256, 1024, 6400]; +const REPS: usize = 9; + +fn pages(count: usize) -> Vec>> { + (0..count as u32) + .map(|id| { + let mut data = [0u8; INNER_PAGE_SIZE]; + for (n, byte) in data.iter_mut().enumerate() { + *byte = (n % 251) as u8; + } + GeneralPage { + header: GeneralHeader { + data_version: DATA_VERSION, + space_id: 1.into(), + page_id: id.into(), + previous_id: 0.into(), + next_id: 0.into(), + page_type: PageType::Data, + data_length: 0, + }, + inner: DataPage { + rows: Vec::new(), + length: INNER_PAGE_SIZE as u32, + data, + }, + } + }) + .collect() +} + +async fn fresh(path: &std::path::Path) -> nagoya::io::HostFile { + let _ = nagoya::io::remove_file(path).await; + nagoya::io::create(path).await.unwrap() +} + +fn median(mut v: Vec) -> f64 { + v.sort_by(|a, b| a.partial_cmp(b).unwrap()); + v[v.len() / 2] +} + +#[tokio::main] +async fn main() { + let path = std::env::var("SCRATCH") + .map(std::path::PathBuf::from) + .unwrap_or_else(|_| std::env::temp_dir()) + .join("data_bucket_sweep.wt"); + + println!(" pages MB one-at-a-time batched gain"); + for count in SIZES { + let bytes = count * data_bucket::PAGE_SIZE; + let (mut ones, mut many) = (Vec::new(), Vec::new()); + // One untimed pass of each, so neither pays for the file appearing. + { + let mut f = fresh(&path).await; + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(pages(count), &mut f) + .await + .unwrap(); + f.sync_all().await.unwrap(); + } + for _ in 0..REPS { + let mut all = pages(count); + let mut file = fresh(&path).await; + let at = Instant::now(); + for page in &mut all { + persist_page::<_, DEFAULT_PAGE_STRIDE>(page, &mut file) + .await + .unwrap(); + } + file.sync_all().await.unwrap(); + ones.push(at.elapsed().as_secs_f64()); + + let all = pages(count); + let mut file = fresh(&path).await; + let at = Instant::now(); + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(all, &mut file) + .await + .unwrap(); + file.sync_all().await.unwrap(); + many.push(at.elapsed().as_secs_f64()); + } + let (one, batch) = (median(ones), median(many)); + println!( + " {count:>5} {:>7.2} {:>7.2} ms {:>7.2} ms {:>5.2}x", + bytes as f64 / 1e6, + one * 1e3, + batch * 1e3, + one / batch + ); + } + let _ = std::fs::remove_file(&path); +} diff --git a/examples/write-pages.rs b/examples/write-pages.rs new file mode 100644 index 0000000..0745907 --- /dev/null +++ b/examples/write-pages.rs @@ -0,0 +1,202 @@ +//! What persisting a file's worth of pages costs, on the real path. +//! +//! Not a model of it: this calls `persist_page` and `persist_pages_batch` +//! themselves, on the async file handles they take. + +use data_bucket::page::{persist_page, persist_pages_batch}; +use data_bucket::{ + DataPage, GeneralHeader, GeneralPage, PageType, DATA_VERSION, DEFAULT_PAGE_STRIDE, + INNER_PAGE_SIZE, +}; +use std::time::Instant; +// The file is a `nagoya::io::File` now, so the trait has to be in scope and +// the openers come from `nagoya::io` rather than a runtime's own module. +use nagoya::io::File as _; + +const PAGES: u32 = 6_400; +const REPS: usize = 5; + +fn pages() -> Vec>> { + (0..PAGES) + .map(|id| { + let mut data = [0u8; INNER_PAGE_SIZE]; + for (n, byte) in data.iter_mut().enumerate() { + *byte = (n % 251) as u8; + } + GeneralPage { + header: GeneralHeader { + data_version: DATA_VERSION, + space_id: 1.into(), + page_id: id.into(), + previous_id: 0.into(), + next_id: 0.into(), + page_type: PageType::Data, + data_length: 0, + }, + inner: DataPage { + rows: Vec::new(), + length: INNER_PAGE_SIZE as u32, + data, + }, + } + }) + .collect() +} + +/// Pages at every `stride`-th id, which is the shape of an update to a file +/// that already exists: the ids are not consecutive, so nothing coalesces. +fn scattered(stride: u32) -> Vec>> { + pages() + .into_iter() + .enumerate() + .filter(|(id, _)| (*id as u32).is_multiple_of(stride)) + .map(|(_, page)| page) + .collect() +} + +async fn existing(path: &std::path::Path) -> nagoya::io::HostFile { + nagoya::io::open_or_create(path).await.unwrap() +} + +async fn fresh(path: &std::path::Path) -> nagoya::io::HostFile { + let _ = nagoya::io::remove_file(path).await; + nagoya::io::create(path).await.unwrap() +} + +#[tokio::main] +async fn main() { + let path = std::env::var("SCRATCH") + .map(std::path::PathBuf::from) + .unwrap_or_else(|_| std::env::temp_dir()) + .join("data_bucket_write_pages.wt"); + let bytes = PAGES as usize * data_bucket::PAGE_SIZE; + + println!( + "{PAGES} pages, {:.1} MB, median of {REPS}\n", + bytes as f64 / 1e6 + ); + + // **The order of the arms is a variable, so it is one that can be set.** + // Run in a fixed order, whichever arm goes second inherits a file the first + // arm just wrote and looks faster for it. `REVERSE=1` runs them the other + // way round; the two orders agreeing is what makes either number mean + // anything. + let reverse = std::env::var("REVERSE").is_ok(); + let mut one_at_a_time = Vec::new(); + let mut batched = Vec::new(); + + let run_one = async |timings: &mut Vec| { + let mut all = pages(); + let mut file = fresh(&path).await; + let at = Instant::now(); + for page in &mut all { + persist_page::<_, DEFAULT_PAGE_STRIDE>(page, &mut file) + .await + .unwrap(); + } + file.sync_all().await.unwrap(); + timings.push(at.elapsed().as_secs_f64()); + }; + let run_batch = async |timings: &mut Vec| { + let all = pages(); + let mut file = fresh(&path).await; + let at = Instant::now(); + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(all, &mut file) + .await + .unwrap(); + file.sync_all().await.unwrap(); + timings.push(at.elapsed().as_secs_f64()); + }; + + for _ in 0..REPS { + if reverse { + run_batch(&mut batched).await; + run_one(&mut one_at_a_time).await; + } else { + run_one(&mut one_at_a_time).await; + run_batch(&mut batched).await; + } + } + + let median = |mut v: Vec| { + v.sort_by(|a, b| a.partial_cmp(b).unwrap()); + v[v.len() / 2] + }; + let (one, many) = (median(one_at_a_time), median(batched)); + println!( + " persist_page, one at a time {:>8.1} ms {:>6.0} MB/s", + one * 1e3, + bytes as f64 / 1e6 / one + ); + println!( + " persist_pages_batch {:>8.1} ms {:>6.0} MB/s {:>5.2}x", + many * 1e3, + bytes as f64 / 1e6 / many, + one / many + ); + + // ---- the case that is not a whole file + // + // Everything above rewrites the file from empty, so the ids run 0..PAGES + // with no gaps and the batch path sees one enormous consecutive run. That + // is its best case and a database's rarest one. Updating scattered pages + // in a file that already exists breaks the run at every page, so the batch + // path falls back to one write per page and can only win by what it saves + // per page, not by joining anything up. + const STRIDE: u32 = 10; + let touched = scattered(STRIDE).len(); + let touched_bytes = touched * data_bucket::PAGE_SIZE; + + // Lay the whole file down once, outside the clock, so the updates land in + // a file that is already the right length. + { + let mut file = fresh(&path).await; + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(pages(), &mut file) + .await + .unwrap(); + file.sync_all().await.unwrap(); + } + + let mut one_scattered = Vec::new(); + let mut batch_scattered = Vec::new(); + for _ in 0..REPS { + let mut some = scattered(STRIDE); + let mut file = existing(&path).await; + let at = Instant::now(); + for page in &mut some { + persist_page::<_, DEFAULT_PAGE_STRIDE>(page, &mut file) + .await + .unwrap(); + } + file.sync_all().await.unwrap(); + one_scattered.push(at.elapsed().as_secs_f64()); + + let some = scattered(STRIDE); + let mut file = existing(&path).await; + let at = Instant::now(); + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(some, &mut file) + .await + .unwrap(); + file.sync_all().await.unwrap(); + batch_scattered.push(at.elapsed().as_secs_f64()); + } + + let (one_s, many_s) = (median(one_scattered), median(batch_scattered)); + println!( + "\nevery {STRIDE}th page of an existing file, {touched} pages, {:.1} MB", + touched_bytes as f64 / 1e6 + ); + println!( + " persist_page, one at a time {:>8.1} ms {:>6.0} MB/s", + one_s * 1e3, + touched_bytes as f64 / 1e6 / one_s + ); + println!( + " persist_pages_batch {:>8.1} ms {:>6.0} MB/s {:>5.2}x", + many_s * 1e3, + touched_bytes as f64 / 1e6 / many_s, + one_s / many_s + ); + + let _ = std::fs::remove_file(&path); +} diff --git a/scripts/check-no-std.sh b/scripts/check-no-std.sh new file mode 100644 index 0000000..8fd4e9c --- /dev/null +++ b/scripts/check-no-std.sh @@ -0,0 +1,32 @@ +#!/bin/sh +# Check a supported OS target with Rust std deliberately absent from its sysroot. +set -eu +root=$(pwd) +target=${NO_STD_TARGET:-$(rustc -vV | awk '/^host:/ {print $2}')} +libdir=$(rustc --print target-libdir --target "$target") +scratch="$root/target/no-std-sysroot/$target" +mkdir -p "$scratch/lib/rustlib/$target/lib" +for library in "$libdir"/*; do + name=$(basename "$library") + case "$name" in + libstd-*|libstd_detect-*|libtest-*|libproc_macro-*|librustc_std_workspace_std-*|std-*|std_detect-*|test-*|proc_macro-*) + rm -f "$scratch/lib/rustlib/$target/lib/$name" + continue ;; + esac + ln -sf "$library" "$scratch/lib/rustlib/$target/lib/$name" +done +printf '%s\n' 'pub fn forbidden() { std::mem::drop(1u8); }' > "$scratch/negative.rs" +if rustc --crate-type lib --emit metadata --target "$target" --sysroot "$scratch" "$scratch/negative.rs" -o "$scratch/negative.rmeta" > "$scratch/negative.log" 2>&1; then + echo 'std unexpectedly available in negative control' >&2 + exit 1 +fi +if ! grep -q "can't find crate for .*std" "$scratch/negative.log"; then + cat "$scratch/negative.log" >&2 + exit 1 +fi +printf '%s\n' '#![no_std]' 'extern crate alloc;' 'pub fn allowed(v: alloc::vec::Vec) -> usize { v.len() }' > "$scratch/positive.rs" +rustc --crate-type lib --emit metadata --target "$target" --sysroot "$scratch" "$scratch/positive.rs" -o "$scratch/positive.rmeta" +# Explicit --target keeps proc macros and build scripts on the ordinary host sysroot. +CARGO_ENCODED_RUSTFLAGS=$(printf '%s\037%s' --sysroot "$scratch") +export CARGO_ENCODED_RUSTFLAGS +cargo check --target "$target" "$@" diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..2a3a7a9 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,168 @@ +//! What this crate can refuse on. +//! +//! # Why this exists rather than `eyre` +//! +//! `eyre::Report` is a `std` type, and it was in the return type of every +//! fallible function here, so the whole crate reached `std` through its own +//! signatures. Nothing else about page framing needs an operating system: the +//! layout is bytes, the checks are arithmetic. This is the type that lets the +//! rest of the crate say so. +//! +//! It is also more useful than a formatted string. A caller that wants to +//! distinguish "this page is full" from "these bytes are damaged" could not, +//! because both arrived as a `Report` carrying prose. +//! +//! Consumers keep working: `Error` implements `core::error::Error`, so `?` +//! into an `eyre::Result` converts exactly as it did before. + +use core::fmt::{Display, Formatter, Result as FmtResult}; + +use crate::page::PageId; + +/// The result of anything in this crate that can fail. +pub type Result = core::result::Result; + +/// A refusal, with the numbers that justify it. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Error { + /// The runtime never interprets another page format as the current one. + UnsupportedVersion { found: u32, expected: u32 }, + /// A write's length does not match the link it was given. + /// + /// A `Link` names an exact byte range, so a write of another size is not + /// a resize: it is a write that would land on the neighbouring row. + LinkLengthMismatch { + /// The length the link reserves. + expected: u32, + /// The length the caller supplied. + found: usize, + }, + /// A link's range runs past the end of the page it points into. + /// + /// Summed in `u64`, because `offset + length` in `u32` can wrap past 4 GiB + /// and slip under the bound. + LinkOutOfBounds { + /// Where the link starts. + offset: u32, + /// How far it runs. + length: u32, + /// How much the page holds. + capacity: usize, + }, + /// A page's contents do not fit the page. + /// + /// Raised where the write is prepared rather than where it lands, so an + /// over-budget page fails in its own persist instead of quietly writing + /// into its neighbour. + PageOverflow { + /// The page whose write is over budget. + page: PageId, + /// Bytes the write needs. + needed: usize, + /// Bytes available. + capacity: usize, + }, + /// Bytes that should have been a structure were not one. + /// + /// The label says which structure, because "corrupt" on its own does not + /// tell an operator which part of a file to distrust. + Corrupt { + /// What failed to parse. + what: &'static str, + }, + /// A value would not archive. + /// + /// rkyv reports this when its allocator refuses, which on this path means + /// the process is already out of memory. + Encode, + /// The file would not answer. + /// + /// **This used to hold a raw OS code**, taken from `std::io::Error`, with + /// the note that the code is what an operator acts on. That was right while + /// the crate opened files itself. It no longer does: the file arrives as + /// `nagoya::io::File`, whose error is deliberately coarse and carries no OS + /// code at all, because a caller either retries, gives up, or creates what + /// was missing. + /// + /// So the kind is carried instead. Mapping it back to `code: None` would + /// have compiled and made every failure identical, which is worse than + /// losing the number: "no such file" and "permission denied" are the two an + /// operator most needs told apart. + Io { + /// What kind of failure the file reported. + kind: nagoya::io::ErrorKind, + }, + /// A change-data event arrived that this page cannot apply. + /// + /// `SplitNode`, `CreateNode` and `RemoveNode` change which pages exist, + /// which is the caller's business rather than one page's. + UnapplicableEvent, +} + +impl Display for Error { + fn fmt(&self, formatter: &mut Formatter<'_>) -> FmtResult { + match self { + Self::UnsupportedVersion { found, expected } => write!( + formatter, + "unsupported page format v{found}; this build requires v{expected}; recreate the store or convert it with an explicit migration tool" + ), + Self::LinkLengthMismatch { expected, found } => write!( + formatter, + "a {found} byte write does not match its {expected} byte link" + ), + Self::LinkOutOfBounds { + offset, + length, + capacity, + } => write!( + formatter, + "a link at {offset} running {length} bytes leaves a {capacity} byte page" + ), + Self::PageOverflow { + page, + needed, + capacity, + } => write!( + formatter, + "page {page:?} needs {needed} bytes of a {capacity} byte page" + ), + Self::Corrupt { what } => write!(formatter, "torn or corrupt {what}"), + Self::Encode => write!(formatter, "a value would not archive"), + Self::Io { kind } => write!(formatter, "the file failed: {kind:?}"), + Self::UnapplicableEvent => write!( + formatter, + "events of `SplitNode`, `CreateNode` or `RemoveNode` cannot be applied to a page" + ), + } + } +} + +impl core::error::Error for Error {} + +// The note here used to read "not gated yet: the crate still reaches the file +// system directly. When that moves behind a trait this impl goes with it." +// That is what happened. The file is `nagoya::io::File` now, so this converts +// from its error and `std::io::Error` no longer appears in this crate at all, +// which is what the module documentation above claims and could not deliver on +// its own. +impl From for Error { + fn from(error: nagoya::io::Error) -> Self { + Self::Io { kind: error.kind() } + } +} + +impl From for Error { + fn from(_: rkyv::rancor::Error) -> Self { + Self::Encode + } +} + +impl From for Error { + fn from(error: crate::page::PageOverflowError) -> Self { + Self::PageOverflow { + page: error.page_id, + needed: error.data_length, + capacity: error.capacity, + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 46812d9..883175a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,9 +1,34 @@ -extern crate core; +#![no_std] + +#[macro_use] +extern crate alloc; +#[cfg(test)] +extern crate std; // The Persistable derive emits paths through the crate name, and this crate // uses its own derive: alias ourselves so the generated code resolves here too. extern crate self as data_bucket; +pub mod error; + +/// Read, write and seek capabilities used by page codecs. +/// +/// Nagoya declares these traits over a portable error without requiring std. +/// Durability and file-management methods belong to the storage engine. +/// Implementations must be `Send` because page I/O can cross a task boundary. +pub trait AsyncFile: AsyncRead + AsyncWrite {} + +impl AsyncFile for T where T: AsyncRead + AsyncWrite {} + +/// Read/seek capabilities needed by page decoders; no write permission or +/// durability implementation is required. +pub trait AsyncRead: nagoya::io::Read + nagoya::io::Seek + Send {} +impl AsyncRead for T where T: nagoya::io::Read + nagoya::io::Seek + Send {} + +/// Write/seek capabilities needed by page encoders. +pub trait AsyncWrite: nagoya::io::Write + nagoya::io::Seek + Send {} +impl AsyncWrite for T where T: nagoya::io::Write + nagoya::io::Seek + Send {} + pub mod link; pub mod page; pub mod persistence; @@ -14,13 +39,14 @@ pub use link::Link; pub use data_bucket_codegen::{SizeMeasure, VariableSizeMeasure}; pub use page::{ - get_index_page_size_from_data_length, map_data_pages_to_general, parse_data_page, - parse_data_pages_batch, parse_general_header_by_index, parse_page, parse_pages_batch, - persist_page, persist_pages_batch, seek_by_link, seek_to_page_start, update_at, DataPage, - GeneralHeader, GeneralPage, IndexPage, IndexPageUtility, IndexValue, Interval, - PageOverflowError, PageType, SpaceInfoPage, TableOfContentsOverflowError, TableOfContentsPage, - UnsizedIndexPage, UnsizedIndexPageUtility, DATA_VERSION, EMPTY_TABLE_OF_CONTENTS_PAGE_SIZE, - GENERAL_HEADER_SIZE, INNER_PAGE_SIZE, PAGE_SIZE, + data_page_row_capacity, get_index_page_size_from_data_length, map_data_pages_to_general, + parse_data_page, parse_data_pages_batch, parse_general_header_by_index, parse_page, + parse_pages_batch, persist_page, persist_pages_batch, seek_by_link, seek_to_page_start, + update_at, DataPage, GeneralHeader, GeneralPage, IndexPage, IndexPageUtility, IndexValue, + Interval, PageOverflowError, PageType, RowSlot, SpaceInfoPage, TableOfContentsOverflowError, + TableOfContentsPage, UnsizedIndexPage, UnsizedIndexPageUtility, DATA_TRAILER_SIZE, + DATA_VERSION, DEFAULT_PAGE_STRIDE, EMPTY_TABLE_OF_CONTENTS_PAGE_SIZE, GENERAL_HEADER_SIZE, + INNER_PAGE_SIZE, PAGE_SIZE, ROW_SLOT_SIZE, }; pub use persistence::{PersistableIndex, PersistableTable}; pub use space::Id as SpaceId; diff --git a/src/page/data.rs b/src/page/data.rs index ae7ad1c..80c2dfe 100644 --- a/src/page/data.rs +++ b/src/page/data.rs @@ -1,21 +1,173 @@ +use crate::error::{Error, Result}; use crate::Link; use crate::Persistable; -use eyre::{eyre, Result}; +use alloc::vec::Vec; #[derive(Debug)] pub struct DataPage { pub length: u32, pub data: [u8; DATA_LENGTH], + /// Live rows, ordered by their payload-relative offset. + pub rows: Vec, +} + +/// One v3 directory entry. Both integers are little endian on disk. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct RowSlot { + pub offset: u32, + pub length: u32, +} + +pub const ROW_SLOT_SIZE: usize = 8; +pub const DATA_TRAILER_SIZE: usize = 8; + +/// Row-byte capacity that leaves room for every possible directory entry. +/// `minimum_row_size` is the size of the archived row wrapper. Variable-size +/// archives can only be larger. The reservation is independent of row contents +/// so mutations never discover a full directory after publishing an index. +pub const fn data_page_row_capacity(stride: usize, minimum_row_size: usize) -> usize { + assert!( + minimum_row_size > 0, + "a row must have a nonzero archived size" + ); + let payload = stride.saturating_sub(crate::GENERAL_HEADER_SIZE + DATA_TRAILER_SIZE); + let slots = payload / (minimum_row_size + ROW_SLOT_SIZE); + let capacity = payload - slots * ROW_SLOT_SIZE; + let next_row = slots.saturating_add(1).saturating_mul(minimum_row_size); + if capacity < next_row { + capacity + } else { + next_row.saturating_sub(1) + } } impl DataPage { + pub fn new() -> Self { + Self { + length: 0, + data: [0; DATA_LENGTH], + rows: Vec::new(), + } + } + + /// Remove only the exact live row named by this mutation. A delayed + /// delete must never erase a replacement with a different extent. + pub fn remove_at(&mut self, link: Link) { + self.rows + .retain(|slot| slot.offset != link.offset || slot.length != link.length); + } + + pub fn decode(bytes: &[u8], length: u32) -> Result { + if length as usize > DATA_LENGTH { + return Err(Error::Corrupt { + what: "v3 row extent", + }); + } + let rows = Self::directory(bytes, length)?; + let mut page = Self::new(); + page.length = length; + page.data[..length as usize].copy_from_slice(&bytes[..length as usize]); + page.rows = rows; + Ok(page) + } + + pub(crate) fn directory(bytes: &[u8], length: u32) -> Result> { + let corrupt = || Error::Corrupt { + what: "v3 data page directory", + }; + if bytes.len() < DATA_TRAILER_SIZE { + return Err(corrupt()); + } + let word = |at: usize| u32::from_le_bytes(bytes[at..at + 4].try_into().unwrap()); + let tail = bytes.len() - DATA_TRAILER_SIZE; + let mut crc = crc32fast::Hasher::new(); + crc.update(&bytes[..tail]); + crc.update(&bytes[tail + 4..]); + if crc.finalize() != word(tail) { + return Err(Error::Corrupt { + what: "v3 data page checksum", + }); + } + let count = word(tail + 4) as usize; + if count > tail / ROW_SLOT_SIZE { + return Err(corrupt()); + } + let directory = tail - count * ROW_SLOT_SIZE; + if length as usize > directory { + return Err(corrupt()); + } + let mut rows = Vec::with_capacity(count); + let mut previous_end = 0u64; + for at in (directory..tail).step_by(ROW_SLOT_SIZE) { + let slot = RowSlot { + offset: word(at), + length: word(at + 4), + }; + let end = u64::from(slot.offset) + u64::from(slot.length); + if slot.length == 0 || u64::from(slot.offset) < previous_end || end > u64::from(length) + { + return Err(corrupt()); + } + previous_end = end; + rows.push(slot); + } + Ok(rows) + } + + pub fn encode(&self, capacity: usize) -> Result> { + let directory_bytes = self + .rows + .len() + .checked_mul(ROW_SLOT_SIZE) + .and_then(|n| n.checked_add(DATA_TRAILER_SIZE)) + .ok_or(Error::Corrupt { + what: "v3 directory size", + })?; + let needed = (self.length as usize) + .checked_add(directory_bytes) + .ok_or(Error::Corrupt { + what: "v3 page size", + })?; + if needed > capacity || self.length as usize > DATA_LENGTH { + return Err(Error::PageOverflow { + page: 0.into(), + needed, + capacity, + }); + } + let mut bytes = vec![0; capacity]; + bytes[..self.length as usize].copy_from_slice(&self.data[..self.length as usize]); + let tail = capacity - DATA_TRAILER_SIZE; + let start = capacity - directory_bytes; + let mut previous_end = 0u64; + for (slot, at) in self.rows.iter().zip((start..tail).step_by(ROW_SLOT_SIZE)) { + let end = u64::from(slot.offset) + u64::from(slot.length); + if slot.length == 0 + || u64::from(slot.offset) < previous_end + || end > u64::from(self.length) + { + return Err(Error::Corrupt { + what: "v3 data page directory", + }); + } + previous_end = end; + bytes[at..at + 4].copy_from_slice(&slot.offset.to_le_bytes()); + bytes[at + 4..at + 8].copy_from_slice(&slot.length.to_le_bytes()); + } + bytes[tail + 4..].copy_from_slice(&(self.rows.len() as u32).to_le_bytes()); + let mut crc = crc32fast::Hasher::new(); + crc.update(&bytes[..tail]); + crc.update(&bytes[tail + 4..]); + bytes[tail..tail + 4].copy_from_slice(&crc.finalize().to_le_bytes()); + Ok(bytes) + } + pub fn update_at(&mut self, link: Link, new_data: &[u8]) -> Result<()> { if new_data.len() as u32 != link.length { - return Err(eyre!( - "New data length {} does not match link length {}", - new_data.len(), - link.length - )); + return Err(Error::LinkLengthMismatch { + expected: link.length, + found: new_data.len(), + }); } // Sum in usize: `offset + length` in u32 can wrap past 4 GiB and @@ -23,16 +175,32 @@ impl DataPage { let start = link.offset as usize; let end = link.offset as usize + link.length as usize; if end > DATA_LENGTH { - return Err(eyre!( - "Link range (offset: {}, length: {}) exceeds data bounds ({})", - link.offset, - link.length, - DATA_LENGTH - )); + return Err(Error::LinkOutOfBounds { + offset: link.offset, + length: link.length, + capacity: DATA_LENGTH, + }); } self.data[start..end].copy_from_slice(new_data); + // Reuse can split a former row's extent. Its old directory entry + // must disappear before any newly overlapping row is published. + self.rows.retain(|slot| { + let slot_end = u64::from(slot.offset) + u64::from(slot.length); + slot_end <= start as u64 || u64::from(slot.offset) >= end as u64 + }); + if link.length > 0 { + let position = self.rows.partition_point(|slot| slot.offset < link.offset); + self.rows.insert( + position, + RowSlot { + offset: link.offset, + length: link.length, + }, + ); + } + self.length = self.length.max(end as u32); Ok(()) } @@ -42,12 +210,11 @@ impl DataPage { let start = link.offset as usize; let end = link.offset as usize + link.length as usize; if end > DATA_LENGTH { - return Err(eyre!( - "Link range (offset: {}, length: {}) exceeds data bounds ({})", - link.offset, - link.length, - DATA_LENGTH - )); + return Err(Error::LinkOutOfBounds { + offset: link.offset, + length: link.length, + capacity: DATA_LENGTH, + }); } Ok(&self.data[start..end]) @@ -65,17 +232,123 @@ impl Persistable for DataPage { Self { length: bytes.len() as u32, data, + rows: Vec::new(), } } + + fn page_bytes(&self, capacity: usize) -> Result + Send> { + self.encode(capacity) + } + + fn page_data_length(&self, _encoded_length: usize) -> usize { + self.length as usize + } +} + +impl Default for DataPage { + fn default() -> Self { + Self::new() + } } #[cfg(test)] mod tests { use super::*; + fn link(offset: u32, length: u32) -> Link { + Link { + page_id: 1.into(), + offset, + length, + } + } + + #[test] + fn v3_directory_roundtrips_live_rows_and_tracks_reuse() { + let mut page = DataPage::<128>::new(); + page.update_at(link(0, 16), &[1; 16]).unwrap(); + page.update_at(link(32, 8), &[2; 8]).unwrap(); + page.remove_at(link(0, 16)); + page.update_at(link(0, 8), &[3; 8]).unwrap(); + page.update_at(link(8, 8), &[4; 8]).unwrap(); + let bytes = page.encode(160).unwrap(); + assert_eq!(u32::from_le_bytes(bytes[156..].try_into().unwrap()), 3); + let parsed = DataPage::<128>::decode(&bytes, page.length).unwrap(); + let values: Vec<_> = parsed + .rows + .iter() + .map(|slot| { + parsed + .get_at(link(slot.offset, slot.length)) + .unwrap() + .to_vec() + }) + .collect(); + assert_eq!(values, vec![vec![3; 8], vec![4; 8], vec![2; 8]]); + } + + #[test] + fn v3_checksum_covers_rows_directory_count_and_padding() { + let mut page = DataPage::<128>::new(); + page.update_at(link(0, 16), &[1; 16]).unwrap(); + let original = page.encode(160).unwrap(); + for byte in 0..original.len() { + let mut damaged = original.clone(); + damaged[byte] ^= 1; + assert!( + DataPage::<128>::decode(&damaged, page.length).is_err(), + "byte {byte}" + ); + } + for length in 0..original.len() { + assert!(DataPage::<128>::decode(&original[..length], page.length).is_err()); + } + } + + #[test] + fn directory_reservation_fits_every_minimum_sized_row() { + for stride in [512, 4096, 8192, 16384, 32768] { + for minimum in [1, 8, 16, 24, 32, 48, 128, 8192] { + let capacity = data_page_row_capacity(stride, minimum); + let slots = capacity / minimum; + assert!( + capacity + + slots * ROW_SLOT_SIZE + + DATA_TRAILER_SIZE + + crate::GENERAL_HEADER_SIZE + <= stride + ); + } + } + } + + #[test] + fn directory_cannot_overlap_row_bytes() { + let mut page = DataPage::<64>::new(); + page.update_at(link(0, 64), &[1; 64]).unwrap(); + assert!(page.encode(64).is_err()); + assert!(page.encode(79).is_err()); + assert!(page.encode(80).is_ok()); + } + + #[test] + fn malformed_directory_is_rejected_even_with_recomputed_checksum() { + let mut page = DataPage::<64>::new(); + page.update_at(link(0, 16), &[1; 16]).unwrap(); + let mut bytes = page.encode(96).unwrap(); + // The entry is at 80. Make its length run beyond the row extent. + bytes[84..88].copy_from_slice(&u32::MAX.to_le_bytes()); + let mut crc = crc32fast::Hasher::new(); + crc.update(&bytes[..88]); + crc.update(&bytes[92..]); + bytes[88..92].copy_from_slice(&crc.finalize().to_le_bytes()); + assert!(DataPage::<64>::decode(&bytes, page.length).is_err()); + } + #[test] fn test_update_at_success() { let mut data = DataPage { + rows: Vec::new(), length: 0, data: [0; 100], }; @@ -94,6 +367,7 @@ mod tests { #[test] fn test_update_at_wrong_length() { let mut data = DataPage { + rows: Vec::new(), length: 0, data: [0; 100], }; @@ -105,14 +379,19 @@ mod tests { }; let err = data.update_at(link, &[1, 2]).unwrap_err(); - assert!(err - .to_string() - .contains("New data length 2 does not match link length 3")); + assert_eq!( + err, + Error::LinkLengthMismatch { + expected: 3, + found: 2 + } + ); } #[test] fn test_update_at_out_of_bounds() { let mut data = DataPage { + rows: Vec::new(), length: 0, data: [0; 100], }; @@ -124,14 +403,20 @@ mod tests { }; let err = data.update_at(link, &[1, 2, 3]).unwrap_err(); - assert!(err - .to_string() - .contains("Link range (offset: 98, length: 3) exceeds data bounds (100)")); + assert_eq!( + err, + Error::LinkOutOfBounds { + offset: 98, + length: 3, + capacity: 100 + } + ); } #[test] fn test_update_at_offset_plus_length_wrapping_u32() { let mut data = DataPage { + rows: Vec::new(), length: 0, data: [0; 100], }; @@ -145,12 +430,13 @@ mod tests { }; let err = data.update_at(link, &[1, 2, 3, 4, 5, 6, 7, 8]).unwrap_err(); - assert!(err.to_string().contains("exceeds data bounds")); + assert!(matches!(err, Error::LinkOutOfBounds { .. })); } #[test] fn test_get_at_offset_plus_length_wrapping_u32() { let data = DataPage { + rows: Vec::new(), length: 0, data: [0; 100], }; @@ -162,12 +448,13 @@ mod tests { }; let err = data.get_at(link).unwrap_err(); - assert!(err.to_string().contains("exceeds data bounds")); + assert!(matches!(err, Error::LinkOutOfBounds { .. })); } #[test] fn test_get_at_out_of_bounds() { let data = DataPage { + rows: Vec::new(), length: 0, data: [0; 100], }; @@ -179,8 +466,13 @@ mod tests { }; let err = data.get_at(link).unwrap_err(); - assert!(err - .to_string() - .contains("Link range (offset: 98, length: 3) exceeds data bounds (100)")); + assert_eq!( + err, + Error::LinkOutOfBounds { + offset: 98, + length: 3, + capacity: 100 + } + ); } } diff --git a/src/page/header.rs b/src/page/header.rs index 424e25d..32fe289 100644 --- a/src/page/header.rs +++ b/src/page/header.rs @@ -9,7 +9,7 @@ use crate::space; use crate::util::Persistable; use crate::PAGE_SIZE; -pub const DATA_VERSION: u32 = 2u32; +pub const DATA_VERSION: u32 = 3; /// Header that appears on every page before it's inner data. #[derive( diff --git a/src/page/index/mod.rs b/src/page/index/mod.rs index 81e46d7..a8397ca 100644 --- a/src/page/index/mod.rs +++ b/src/page/index/mod.rs @@ -1,16 +1,14 @@ -use std::fmt::Debug; -use std::io::SeekFrom; +use core::fmt::Debug; +use nagoya::io::SeekFrom; +use crate::{AsyncFile, AsyncRead}; use indexset::core::multipair::MultiPair; use indexset::core::pair::Pair; use rkyv::{Archive, Deserialize, Serialize}; -use tokio::fs::File; -use tokio::io::{AsyncSeekExt, AsyncWriteExt}; -use crate::page::PageOverflowError; use crate::{ align, align_to, seek_to_page_start, Link, Persistable, SizeMeasurable, VariableSizeMeasurable, - GENERAL_HEADER_SIZE, INNER_PAGE_SIZE, + GENERAL_HEADER_SIZE, }; mod page; @@ -30,29 +28,30 @@ pub use table_of_contents_page::{ pub trait IndexPageUtility { type Utility: Persistable + Send + Sync; - fn parse_index_page_utility( - file: &mut File, + fn parse_index_page_utility( + file: &mut impl AsyncRead, page_id: PageId, - ) -> impl std::future::Future> + Send; + ) -> impl core::future::Future> + Send; - fn persist_index_page_utility( - file: &mut File, + fn persist_index_page_utility( + file: &mut impl AsyncFile, page_id: PageId, utility: Self::Utility, - ) -> impl std::future::Future> + Send { + ) -> impl core::future::Future> + Send { async move { let bytes = utility.as_bytes(); let utility_length = bytes.as_ref().len(); // An oversized utility must fail here, in its own persist, // instead of writing past the page slot into the neighbor page. - if utility_length > INNER_PAGE_SIZE { - return Err(eyre::Report::new(PageOverflowError { - page_id, - data_length: utility_length, - capacity: INNER_PAGE_SIZE, - })); + let capacity = crate::page::util::page_capacity::(page_id)?; + if utility_length > capacity { + return Err(crate::error::Error::PageOverflow { + page: page_id, + needed: utility_length, + capacity, + }); } - seek_to_page_start(file, page_id.0).await?; + seek_to_page_start::(file, page_id.0).await?; file.seek(SeekFrom::Current(GENERAL_HEADER_SIZE as i64)) .await?; file.write_all(bytes.as_ref()).await?; diff --git a/src/page/index/page.rs b/src/page/index/page.rs index cccb1cf..c71c23e 100644 --- a/src/page/index/page.rs +++ b/src/page/index/page.rs @@ -1,10 +1,12 @@ //! [`crate::page::IndexPage`] definition. -use std::fmt::Debug; -use std::hash::Hash; -use std::io::SeekFrom; -use std::mem; +use alloc::vec::Vec; +use core::fmt::Debug; +use core::hash::Hash; +use core::mem; +use nagoya::io::SeekFrom; +use crate::{AsyncFile, AsyncRead}; use data_bucket_codegen::Persistable; use indexset::core::pair::Pair; use rkyv::de::Pool; @@ -14,14 +16,11 @@ use rkyv::ser::sharing::Share; use rkyv::ser::Serializer; use rkyv::util::AlignedVec; use rkyv::{Archive, Deserialize, Serialize}; -use tokio::fs::File; -use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; use crate::page::index::IndexPageUtility; use crate::page::{IndexValue, PageId, PageOverflowError}; use crate::{ align, align8, seek_to_page_start, Link, Persistable, SizeMeasurable, GENERAL_HEADER_SIZE, - INNER_PAGE_SIZE, }; pub fn get_index_page_size_from_data_length(length: usize) -> usize @@ -37,14 +36,13 @@ where let slots_vec_size = IndexPage::::slots_size(0); let index_values_vec_size = IndexPage::::slots_size(0); - (length - - node_id_size - - size_field_size - - current_index_size - - current_length_size - - slots_vec_size - - index_values_vec_size) - / (slot_size + index_value_size) + let overhead = node_id_size + + size_field_size + + current_index_size + + current_length_size + + slots_vec_size + + index_values_vec_size; + (length.saturating_sub(overhead) / (slot_size + index_value_size)).min(usize::from(u16::MAX)) } /// Represents a page, which is filled with [`IndexValue`]'s of some index. @@ -85,11 +83,11 @@ where { type Utility = SizedIndexPageUtility; - async fn parse_index_page_utility( - file: &mut File, + async fn parse_index_page_utility( + file: &mut impl AsyncRead, page_id: PageId, - ) -> eyre::Result { - seek_to_page_start(file, page_id.0).await?; + ) -> crate::error::Result { + seek_to_page_start::(file, page_id.0).await?; let offset = GENERAL_HEADER_SIZE as i64; file.seek(SeekFrom::Current(offset)).await?; @@ -100,7 +98,9 @@ where let archived = crate::access_archived::<::Archived>( &size_bytes[0..SizedIndexPageUtility::::size_size()], ) - .map_err(|error| eyre::eyre!("torn or corrupt index page size field: {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "index page size field", + })?; let size = rkyv::deserialize::(archived).expect("data should be valid"); @@ -122,6 +122,10 @@ impl IndexPage { where T: Clone, { + assert!( + size <= usize::from(u16::MAX), + "index page slot count exceeds its u16 format" + ); let slots = vec![0u16; size]; let index_values = vec![IndexValue::default(); size]; Self { @@ -160,7 +164,7 @@ impl IndexPage { new_page } - async fn read_value(file: &mut File) -> eyre::Result> + async fn read_value(file: &mut impl AsyncRead) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -177,16 +181,18 @@ impl IndexPage { v.extend_from_slice(bytes.as_slice()); // Validated: a torn index entry must be an error, not a dangling link. let archived = crate::access_archived::< as Archive>::Archived>(&v[..]) - .map_err(|error| eyre::eyre!("torn or corrupt index entry: {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "index entry", + })?; Ok(rkyv::deserialize(archived).expect("data should be valid")) } - pub async fn read_value_with_index( - file: &mut File, + pub async fn read_value_with_index( + file: &mut impl AsyncRead, page_id: PageId, size: usize, index: usize, - ) -> eyre::Result> + ) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -197,7 +203,7 @@ impl IndexPage { rkyv::api::high::HighValidator<'a, rkyv::rancor::Error>, >, { - seek_to_page_start(file, page_id.0).await?; + seek_to_page_start::(file, page_id.0).await?; let offset = Self::get_value_offset(size, index); file.seek(SeekFrom::Current(offset as i64)).await?; Self::read_value(file).await @@ -208,17 +214,36 @@ impl IndexPage { /// /// `offset` is relative to the page start and already includes the /// general header. - fn check_value_write_bounds( + /// **The budget is this page's own stride, not the crate default.** With a + /// smaller page the default lets a slot write run past the end of the page + /// and into its neighbour, which is a silent corruption rather than an + /// error: the write succeeds and the next page fails to parse. + fn check_value_write_bounds( page_id: PageId, offset: usize, value_length: usize, ) -> Result<(), PageOverflowError> { - let write_end_in_slot = offset + value_length - GENERAL_HEADER_SIZE; - if write_end_in_slot > INNER_PAGE_SIZE { + let capacity = + (STRIDE as usize) + .checked_sub(GENERAL_HEADER_SIZE) + .ok_or(PageOverflowError { + page_id, + data_length: GENERAL_HEADER_SIZE, + capacity: STRIDE as usize, + })?; + let write_end_in_slot = offset + .checked_add(value_length) + .and_then(|end| end.checked_sub(GENERAL_HEADER_SIZE)) + .ok_or(PageOverflowError { + page_id, + data_length: usize::MAX, + capacity, + })?; + if write_end_in_slot > capacity { return Err(PageOverflowError { page_id, data_length: write_end_in_slot, - capacity: INNER_PAGE_SIZE, + capacity, }); } Ok(()) @@ -239,13 +264,13 @@ impl IndexPage { offset } - pub async fn persist_value( - file: &mut File, + pub async fn persist_value( + file: &mut impl AsyncFile, page_id: PageId, size: usize, value: IndexValue, mut value_index: u16, - ) -> eyre::Result + ) -> crate::error::Result where T: Archive + Eq @@ -262,8 +287,8 @@ impl IndexPage { { let offset = Self::get_value_offset(size, value_index as usize); let bytes = rkyv::to_bytes::(&value)?; - Self::check_value_write_bounds(page_id, offset, bytes.len())?; - seek_to_page_start(file, page_id.0).await?; + Self::check_value_write_bounds::(page_id, offset, bytes.len())?; + seek_to_page_start::(file, page_id.0).await?; file.seek(SeekFrom::Current(offset as i64)).await?; file.write_all(bytes.as_slice()).await?; @@ -278,12 +303,12 @@ impl IndexPage { Ok(value_index + 1) } - pub async fn remove_value( - file: &mut File, + pub async fn remove_value( + file: &mut impl AsyncFile, page_id: PageId, size: usize, value_index: u16, - ) -> eyre::Result<()> + ) -> crate::error::Result<()> where T: Archive + Default @@ -303,8 +328,8 @@ impl IndexPage { let offset = Self::get_value_offset(size, value_index as usize); let value = IndexValue::::default(); let bytes = rkyv::to_bytes::(&value)?; - Self::check_value_write_bounds(page_id, offset, bytes.len())?; - seek_to_page_start(file, page_id.0).await?; + Self::check_value_write_bounds::(page_id, offset, bytes.len())?; + seek_to_page_start::(file, page_id.0).await?; file.seek(SeekFrom::Current(offset as i64)).await?; file.write_all(bytes.as_slice()).await?; @@ -349,7 +374,11 @@ impl IndexPage { #[cfg(test)] mod tests { use crate::page::IndexValue; - use crate::{get_index_page_size_from_data_length, IndexPage, Persistable, INNER_PAGE_SIZE}; + use crate::{ + get_index_page_size_from_data_length, IndexPage, Persistable, DEFAULT_PAGE_STRIDE, + INNER_PAGE_SIZE, + }; + use std::prelude::v1::*; use uuid::Uuid; #[test] @@ -375,7 +404,7 @@ mod tests { #[test] fn test_bytes_128() { let size: usize = get_index_page_size_from_data_length::(INNER_PAGE_SIZE); - println!("size: {size}"); + std::println!("size: {size}"); let page = IndexPage::::new( IndexValue { key: u128::default(), @@ -415,20 +444,13 @@ mod tests { #[tokio::test] async fn persist_and_remove_value_reject_writes_past_the_page_slot() { - use super::{IndexPageUtility, PageOverflowError, SizedIndexPageUtility}; + use super::{IndexPageUtility, SizedIndexPageUtility}; let path = std::env::temp_dir().join(format!( "data_bucket_slot_write_bounds_{}.wt", std::process::id() )); - let mut file = tokio::fs::OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(true) - .open(&path) - .await - .unwrap(); + let mut file = nagoya::io::create(&path).await.unwrap(); let value = IndexValue:: { key: 7, @@ -436,29 +458,42 @@ mod tests { }; // An in-bounds slot write works. - IndexPage::::persist_value(&mut file, 1.into(), 4, value.clone(), 3) - .await - .unwrap(); - // tokio's File buffers writes; flush so metadata() sees them. - tokio::io::AsyncWriteExt::flush(&mut file).await.unwrap(); - let length_after_valid_write = file.metadata().await.unwrap().len(); + IndexPage::::persist_value::( + &mut file, + 1.into(), + 4, + value.clone(), + 3, + ) + .await + .unwrap(); + // the async file buffers writes; flush so metadata() sees them. + nagoya::io::Write::flush(&mut file).await.unwrap(); + let length_after_valid_write = nagoya::io::File::length(&mut file).await.unwrap(); // A value index whose slot lies past the page must be rejected // before anything is written. - let err = IndexPage::::persist_value(&mut file, 1.into(), 4, value, 2000) - .await - .unwrap_err(); + let err = IndexPage::::persist_value::( + &mut file, + 1.into(), + 4, + value, + 2000, + ) + .await + .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); - let err = IndexPage::::remove_value(&mut file, 1.into(), 4, 2000) - .await - .unwrap_err(); + let err = + IndexPage::::remove_value::(&mut file, 1.into(), 4, 2000) + .await + .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // A utility larger than the page slot must be rejected too. @@ -469,21 +504,19 @@ mod tests { current_length: 0, slots: vec![0u16; 20_000], }; - let err = as IndexPageUtility>::persist_index_page_utility( - &mut file, - 1.into(), - utility, - ) + let err = as IndexPageUtility>::persist_index_page_utility::< + DEFAULT_PAGE_STRIDE, + >(&mut file, 1.into(), utility) .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // Nothing was written by the rejected operations. assert_eq!( - file.metadata().await.unwrap().len(), + nagoya::io::File::length(&mut file).await.unwrap(), length_after_valid_write ); diff --git a/src/page/index/page_cdc_impl.rs b/src/page/index/page_cdc_impl.rs index e49e943..a5e7d33 100644 --- a/src/page/index/page_cdc_impl.rs +++ b/src/page/index/page_cdc_impl.rs @@ -1,6 +1,5 @@ -use std::fmt::Debug; +use core::fmt::Debug; -use eyre::bail; use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; use rkyv::de::Pool; @@ -32,7 +31,10 @@ where + PartialOrd + Debug, { - pub fn apply_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + pub fn apply_change_event( + &mut self, + event: ChangeEvent>, + ) -> crate::error::Result<()> { match event.clone() { ChangeEvent::InsertAt { event_id: _, @@ -67,13 +69,11 @@ where } ChangeEvent::SplitNode { .. } | ChangeEvent::CreateNode { .. } - | ChangeEvent::RemoveNode { .. } => { - bail!("Events of `SplitNode`, `CreateNode` or `RemoveNode` can not be applied") - } + | ChangeEvent::RemoveNode { .. } => Err(crate::error::Error::UnapplicableEvent), } } - fn apply_insert_at(&mut self, index: usize, value: Pair) -> eyre::Result<()> { + fn apply_insert_at(&mut self, index: usize, value: Pair) -> crate::error::Result<()> { // For insert we first add slot entry for our new index value self.slots.insert(index, self.current_index); self.slots.remove(self.size as usize); @@ -101,7 +101,7 @@ where Ok(()) } - fn apply_remove_at(&mut self, index: usize) -> eyre::Result<()> { + fn apply_remove_at(&mut self, index: usize) -> crate::error::Result<()> { // For remove we first remove slot entry for index value let value_position = self.slots.remove(index); // We push 0 in the tail because slots size should be fixed. diff --git a/src/page/index/page_for_unsized.rs b/src/page/index/page_for_unsized.rs index de25d9d..573b121 100644 --- a/src/page/index/page_for_unsized.rs +++ b/src/page/index/page_for_unsized.rs @@ -1,6 +1,8 @@ -use std::fmt::Debug; -use std::io::SeekFrom; +use alloc::vec::Vec; +use core::fmt::Debug; +use nagoya::io::SeekFrom; +use crate::{AsyncFile, AsyncRead}; use data_bucket_codegen::Persistable; use indexset::core::pair::Pair; use rkyv::de::Pool; @@ -10,13 +12,11 @@ use rkyv::ser::sharing::Share; use rkyv::ser::Serializer; use rkyv::util::AlignedVec; use rkyv::{Archive, Deserialize, Serialize}; -use tokio::fs::File; -use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; use crate::page::index::IndexPageUtility; -use crate::page::{PageId, PageOverflowError}; +use crate::page::PageId; use crate::{align8, VariableSizeMeasurable}; -use crate::{seek_to_page_start, IndexValue, SizeMeasurable, GENERAL_HEADER_SIZE, INNER_PAGE_SIZE}; +use crate::{seek_to_page_start, IndexValue, SizeMeasurable, GENERAL_HEADER_SIZE}; use crate::{Link, Persistable}; #[derive(Archive, Clone, Deserialize, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)] @@ -47,7 +47,7 @@ pub struct UnsizedIndexPageUtility UnsizedIndexPageUtility { - pub fn update_node_id(&mut self, node_id: IndexValue) -> eyre::Result<()> { + pub fn update_node_id(&mut self, node_id: IndexValue) -> crate::error::Result<()> { self.node_id_size = node_id.aligned_size() as u16; self.node_id = node_id; @@ -69,11 +69,11 @@ where { type Utility = UnsizedIndexPageUtility; - async fn parse_index_page_utility( - file: &mut File, + async fn parse_index_page_utility( + file: &mut impl AsyncRead, page_id: PageId, - ) -> eyre::Result { - seek_to_page_start(file, page_id.0).await?; + ) -> crate::error::Result { + seek_to_page_start::(file, page_id.0).await?; let offset = GENERAL_HEADER_SIZE as i64; file.seek(SeekFrom::Current(offset)).await?; @@ -84,7 +84,9 @@ where let archived = crate::access_archived::<::Archived>( &slot_size_bytes[0..UnsizedIndexPageUtility::::slots_size_size()], ) - .map_err(|error| eyre::eyre!("torn or corrupt unsized index page (slots size): {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "unsized index page slots size", + })?; let slots_size = rkyv::deserialize::(archived).expect("data should be valid"); let mut node_id_size_bytes = vec![0u8; UnsizedIndexPageUtility::::node_id_size_size()]; @@ -92,8 +94,8 @@ where let archived = crate::access_archived::<::Archived>( &node_id_size_bytes[0..UnsizedIndexPageUtility::::node_id_size_size()], ) - .map_err(|error| { - eyre::eyre!("torn or corrupt unsized index page (node id size): {error}") + .map_err(|_error| crate::error::Error::Corrupt { + what: "unsized index page node id size", })?; let node_id_size = rkyv::deserialize::(archived).expect("data should be valid"); @@ -128,7 +130,7 @@ where ::Archived: Deserialize> + for<'a> rkyv::bytecheck::CheckBytes>, { - pub fn new(node_id: IndexValue) -> eyre::Result { + pub fn new(node_id: IndexValue) -> crate::error::Result { let len = node_id.aligned_size() as u32; Ok(Self { slots_size: 1, @@ -195,12 +197,12 @@ where new_page } - pub async fn persist_value( - file: &mut File, + pub async fn persist_value( + file: &mut impl AsyncFile, page_id: PageId, current_offset: u32, value: IndexValue, - ) -> eyre::Result + ) -> crate::error::Result where T: Archive + Eq @@ -217,23 +219,28 @@ where // Values fill the page tail-first, so `offset` counts back from the // page end: once it passes the inner-page budget the write would // land in this page's header, or before it in the previous page. - if offset > INNER_PAGE_SIZE as u64 { - return Err(eyre::Report::new(PageOverflowError { - page_id, - data_length: offset as usize, - capacity: INNER_PAGE_SIZE, - })); + // + // The budget is this page's own stride, not the crate default. With a + // smaller page the default lets the write run back past the header and + // into the previous page, silently. + let capacity = crate::page::util::page_capacity::(page_id)?; + if offset > capacity as u64 { + return Err(crate::error::Error::PageOverflow { + page: page_id, + needed: offset as usize, + capacity, + }); } // We seek to page's end and will write values from tail. - seek_to_page_start(file, page_id.0 + 1).await?; + seek_to_page_start::(file, page_id.0 + 1).await?; file.seek(SeekFrom::Current(-(offset as i64))).await?; file.write_all(bytes.as_slice()).await?; Ok(offset as u32) } - async fn read_value(file: &mut File, len: u16) -> eyre::Result> + async fn read_value(file: &mut impl AsyncRead, len: u16) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -250,16 +257,18 @@ where v.extend_from_slice(bytes.as_slice()); // Validated: a torn index entry must be an error, not a dangling link. let archived = crate::access_archived::< as Archive>::Archived>(&v[..]) - .map_err(|error| eyre::eyre!("torn or corrupt unsized index entry: {error}"))?; + .map_err(|_| crate::error::Error::Corrupt { + what: "unsized index entry", + })?; Ok(rkyv::deserialize(archived).expect("data should be valid")) } - pub async fn read_value_with_offset( - file: &mut File, + pub async fn read_value_with_offset( + file: &mut impl AsyncRead, page_id: PageId, offset: u32, len: u16, - ) -> eyre::Result> + ) -> crate::error::Result> where T: Archive, ::Archived: Deserialize> @@ -270,7 +279,7 @@ where rkyv::api::high::HighValidator<'a, rkyv::rancor::Error>, >, { - seek_to_page_start(file, page_id.0 + 1).await?; + seek_to_page_start::(file, page_id.0 + 1).await?; file.seek(SeekFrom::Current(-(offset as i64))).await?; Self::read_value(file, len).await } @@ -386,8 +395,10 @@ where #[cfg(test)] mod test { - use crate::page::PageOverflowError; - use crate::{IndexValue, Link, Persistable, UnsizedIndexPage, INNER_PAGE_SIZE}; + use crate::{ + IndexValue, Link, Persistable, UnsizedIndexPage, DEFAULT_PAGE_STRIDE, INNER_PAGE_SIZE, + }; + use std::prelude::v1::*; #[tokio::test] async fn persist_value_rejects_writes_leaving_the_page_slot() { @@ -395,14 +406,7 @@ mod test { "data_bucket_unsized_write_bounds_{}.wt", std::process::id() )); - let mut file = tokio::fs::OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(true) - .open(&path) - .await - .unwrap(); + let mut file = nagoya::io::create(&path).await.unwrap(); let value = IndexValue:: { key: "tail_first_value".to_string(), @@ -410,16 +414,21 @@ mod test { }; // An in-bounds tail-first write works. - UnsizedIndexPage::::persist_value(&mut file, 1.into(), 0, value.clone()) - .await - .unwrap(); - // tokio's File buffers writes; flush so metadata() sees them. - tokio::io::AsyncWriteExt::flush(&mut file).await.unwrap(); - let length_after_valid_write = file.metadata().await.unwrap().len(); + UnsizedIndexPage::::persist_value::( + &mut file, + 1.into(), + 0, + value.clone(), + ) + .await + .unwrap(); + // the async file buffers writes; flush so metadata() sees them. + nagoya::io::Write::flush(&mut file).await.unwrap(); + let length_after_valid_write = nagoya::io::File::length(&mut file).await.unwrap(); // A current offset at the inner budget leaves no room: the write // would land in the page header (or the previous page). - let err = UnsizedIndexPage::::persist_value( + let err = UnsizedIndexPage::::persist_value::( &mut file, 1.into(), INNER_PAGE_SIZE as u32, @@ -428,13 +437,13 @@ mod test { .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // A huge current offset used to wrap the u32 arithmetic and seek // far outside the page; it must be rejected the same way. - let err = UnsizedIndexPage::::persist_value( + let err = UnsizedIndexPage::::persist_value::( &mut file, 1.into(), u32::MAX - 4, @@ -443,13 +452,13 @@ mod test { .await .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // Nothing was written by the rejected operations. assert_eq!( - file.metadata().await.unwrap().len(), + nagoya::io::File::length(&mut file).await.unwrap(), length_after_valid_write ); diff --git a/src/page/index/page_for_unsized_cdc_impl.rs b/src/page/index/page_for_unsized_cdc_impl.rs index 2ecc4c3..e2f910e 100644 --- a/src/page/index/page_for_unsized_cdc_impl.rs +++ b/src/page/index/page_for_unsized_cdc_impl.rs @@ -1,4 +1,3 @@ -use eyre::bail; use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; use rkyv::de::Pool; @@ -29,7 +28,10 @@ where ::Archived: Deserialize> + for<'a> rkyv::bytecheck::CheckBytes>, { - pub fn apply_change_event(&mut self, event: ChangeEvent>) -> eyre::Result<()> { + pub fn apply_change_event( + &mut self, + event: ChangeEvent>, + ) -> crate::error::Result<()> { match event { ChangeEvent::InsertAt { event_id: _, @@ -70,13 +72,11 @@ where } ChangeEvent::SplitNode { .. } | ChangeEvent::CreateNode { .. } - | ChangeEvent::RemoveNode { .. } => { - bail!("Events of `SplitNode`, `CreateNode` or `RemoveNode` can not be applied") - } + | ChangeEvent::RemoveNode { .. } => Err(crate::error::Error::UnapplicableEvent), } } - fn apply_insert_at(&mut self, index: usize, value: Pair) -> eyre::Result<()> { + fn apply_insert_at(&mut self, index: usize, value: Pair) -> crate::error::Result<()> { // For insert we first add slot entry for our new index value let index_value = IndexValue { key: value.key.clone(), @@ -99,7 +99,7 @@ where Ok(()) } - fn apply_remove_at(&mut self, index: usize) -> eyre::Result<()> { + fn apply_remove_at(&mut self, index: usize) -> crate::error::Result<()> { self.slots.remove(index); self.slots_size -= 1; let v = self.index_values.remove(index); @@ -119,6 +119,7 @@ mod test { use crate::{IndexValue, Link, UnsizedIndexPage}; use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; + use std::prelude::v1::*; #[test] fn test_insert_at() { diff --git a/src/page/index/table_of_contents_page.rs b/src/page/index/table_of_contents_page.rs index 0be9541..09e9bfd 100644 --- a/src/page/index/table_of_contents_page.rs +++ b/src/page/index/table_of_contents_page.rs @@ -1,7 +1,8 @@ +use alloc::collections::btree_map::Entry; +use alloc::collections::BTreeMap; +use alloc::vec::Vec; +use core::fmt::Debug; use rkyv::{Archive, Deserialize, Serialize}; -use std::collections::btree_map::Entry; -use std::collections::BTreeMap; -use std::fmt::Debug; use crate::page::PageId; use crate::{align, align_to, Persistable, SizeMeasurable, INNER_PAGE_SIZE}; @@ -9,7 +10,7 @@ use crate::{align, align_to, Persistable, SizeMeasurable, INNER_PAGE_SIZE}; /// Serialized size of a [`TableOfContentsPage`] with no records and no /// empty pages: the `estimated_size` field itself plus the two empty /// vectors. -pub const EMPTY_TABLE_OF_CONTENTS_PAGE_SIZE: usize = std::mem::size_of::() + 12; +pub const EMPTY_TABLE_OF_CONTENTS_PAGE_SIZE: usize = core::mem::size_of::() + 12; /// Error returned by the capacity-checked mutators of /// [`TableOfContentsPage`] when adding a record would push the page's @@ -44,8 +45,8 @@ impl TableOfContentsOverflowError { } } -impl std::fmt::Display for TableOfContentsOverflowError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl core::fmt::Display for TableOfContentsOverflowError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { write!( f, "table of contents record of {} bytes does not fit the page \ @@ -55,7 +56,7 @@ impl std::fmt::Display for TableOfContentsOverflowError { } } -impl std::error::Error for TableOfContentsOverflowError {} +impl core::error::Error for TableOfContentsOverflowError {} #[derive(Archive, Clone, Deserialize, Debug, Serialize)] pub struct TableOfContentsPage { @@ -407,6 +408,7 @@ where #[cfg(test)] mod test { use crate::{Link, Persistable, TableOfContentsPage, INNER_PAGE_SIZE}; + use std::prelude::v1::*; fn link(offset: u32) -> Link { Link { diff --git a/src/page/iterators.rs b/src/page/iterators.rs index 3e498a0..02f92dd 100644 --- a/src/page/iterators.rs +++ b/src/page/iterators.rs @@ -8,7 +8,7 @@ use crate::{ IndexData, Link, }; -use super::{index::{ArchivedIndexValue, IndexValue}, seek_by_link, seek_to_page_start, Interval, SpaceInfo}; +use super::{index::{ArchivedIndexValue, IndexValue}, seek_by_link, seek_to_page_start, Interval, SpaceInfo, PAGE_SIZE}; pub struct LinksIterator<'a> { file: &'a mut std::fs::File, @@ -68,7 +68,7 @@ impl Iterator for LinksIterator<'_> { fn next(&mut self) -> Option { if self.links.is_none() { - seek_to_page_start(&mut self.file, self.page_id).expect("page should be seekable"); + seek_to_page_start::<{ PAGE_SIZE as u32 }>(&mut self.file, self.page_id).expect("page should be seekable"); let header = parse_general_header(&mut self.file).expect("header should be readable"); let mut buffer: Vec = vec![0u8; header.data_length as usize]; @@ -144,7 +144,7 @@ impl Iterator for DataIterator<'_> { } let current_link = self.links[self.link_index]; - seek_by_link(&mut self.file, current_link).expect("the seek should be successful"); + seek_by_link::<{ PAGE_SIZE as u32 }>(&mut self.file, current_link).expect("the seek should be successful"); let mut buffer = vec![0u8; current_link.length as usize]; self.file .read_exact(&mut buffer) @@ -174,7 +174,7 @@ mod test { create_test_database_file(filename); let mut file = std::fs::File::open(filename).unwrap(); - let space_info = parse_space_info::(&mut file).unwrap(); + let space_info = parse_space_info::<{ PAGE_SIZE as u32 }>(&mut file).unwrap(); let links = LinksIterator::<'_>::new(&mut file, 1, &space_info); assert_eq!( links.collect::>(), @@ -199,7 +199,7 @@ mod test { create_test_database_file(filename); let mut file = std::fs::File::open(filename).unwrap(); - let space_info = parse_space_info::(&mut file).unwrap(); + let space_info = parse_space_info::<{ PAGE_SIZE as u32 }>(&mut file).unwrap(); let index_intervals = space_info.primary_key_intervals.clone(); let pages_ids = PageIterator::new(index_intervals).collect::>(); diff --git a/src/page/mod.rs b/src/page/mod.rs index 17cfec1..ce88e0d 100644 --- a/src/page/mod.rs +++ b/src/page/mod.rs @@ -12,7 +12,7 @@ use rkyv::{Archive, Deserialize, Serialize}; use crate::{align, SizeMeasurable}; -pub use data::DataPage; +pub use data::{data_page_row_capacity, DataPage, RowSlot, DATA_TRAILER_SIZE, ROW_SLOT_SIZE}; pub use header::{GeneralHeader, DATA_VERSION}; pub use index::{ get_index_page_size_from_data_length, IndexPage, IndexPageUtility, IndexValue, @@ -54,6 +54,14 @@ pub const GENERAL_HEADER_SIZE: usize = 28; /// without [`GeneralPage`] page [`GENERAL_HEADER_SIZE`]. pub const INNER_PAGE_SIZE: usize = PAGE_SIZE - GENERAL_HEADER_SIZE; +/// [`PAGE_SIZE`] as a stride, for the callers that want the crate default. +/// +/// The seek helpers take their stride as a `u32` const parameter, and a `usize` +/// constant cannot be converted in generic argument position without +/// `generic_const_exprs`. This spares every such caller a `{ PAGE_SIZE as u32 }` +/// block, and names what it is. +pub const DEFAULT_PAGE_STRIDE: u32 = PAGE_SIZE as u32; + /// Represents page's identifier. Is unique within the table bounds #[derive( Archive, diff --git a/src/page/space_info.rs b/src/page/space_info.rs index 67158af..dd45ece 100644 --- a/src/page/space_info.rs +++ b/src/page/space_info.rs @@ -2,6 +2,7 @@ use crate::util::Persistable; use crate::{space, Link}; +use alloc::{string::String, vec::Vec}; use data_bucket_codegen::Persistable; use rkyv::{Archive, Deserialize, Serialize}; @@ -140,10 +141,11 @@ where let v1 = SpaceInfoPageV1::from_bytes(bytes, version); v1.into() } - _ => { + 2 | 3 => { let v2 = SpaceInfoPageV2::from_bytes(bytes, version); v2.into() } + _ => panic!("unsupported space-info page version {version}"), } } } @@ -164,6 +166,7 @@ mod test { use crate::page::INNER_PAGE_SIZE; use crate::util::Persistable; use rkyv::Archive; + use std::prelude::v1::*; #[test] fn test_as_bytes() { diff --git a/src/page/util.rs b/src/page/util.rs index af1dd77..005b78b 100644 --- a/src/page/util.rs +++ b/src/page/util.rs @@ -1,17 +1,37 @@ -use eyre::eyre; +use crate::error::Error; +use crate::{AsyncRead, AsyncWrite}; +use alloc::vec::Vec; +use nagoya::io::SeekFrom; use rkyv::api::high::HighDeserializer; use rkyv::Archive; -use std::io::SeekFrom; -use tokio::fs::File; -use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; use super::SpaceInfoPage; use crate::page::header::GeneralHeader; use crate::page::ty::PageType; use crate::page::PageId; -use crate::{ - DataPage, GeneralPage, Link, Persistable, GENERAL_HEADER_SIZE, INNER_PAGE_SIZE, PAGE_SIZE, -}; +use crate::{DataPage, GeneralPage, Link, Persistable, GENERAL_HEADER_SIZE}; + +pub(crate) fn page_capacity(page: PageId) -> crate::error::Result { + (STRIDE as usize) + .checked_sub(GENERAL_HEADER_SIZE) + .ok_or(Error::PageOverflow { + page, + needed: GENERAL_HEADER_SIZE, + capacity: STRIDE as usize, + }) +} + +fn validate_layout(page: PageId, needed: usize) -> crate::error::Result { + let capacity = page_capacity::(page)?; + if needed > capacity { + return Err(Error::PageOverflow { + page, + needed, + capacity, + }); + } + Ok(capacity) +} /// Returned when a write into a page would not fit the page slot: letting /// it through would spill past a [`PAGE_SIZE`] boundary and corrupt a @@ -27,8 +47,8 @@ pub struct PageOverflowError { pub capacity: usize, } -impl std::fmt::Display for PageOverflowError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl core::fmt::Display for PageOverflowError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { write!( f, "page {} write needs {} bytes, exceeding the {}-byte page slot", @@ -37,7 +57,7 @@ impl std::fmt::Display for PageOverflowError { } } -impl std::error::Error for PageOverflowError {} +impl core::error::Error for PageOverflowError {} pub fn map_data_pages_to_general( pages: Vec>, @@ -69,68 +89,206 @@ pub fn map_data_pages_to_general( general_pages } -pub async fn persist_page<'a, T>( +pub async fn persist_page<'a, T, const STRIDE: u32>( page: &'a mut GeneralPage, - file: &'a mut File, -) -> eyre::Result<()> + file: &'a mut impl AsyncWrite, +) -> crate::error::Result<()> where T: Persistable + Send + Sync, { - seek_to_page_start(file, page.header.page_id.0).await?; - - let page_count = page.header.page_id.0 as i64 + 1; - persist_page_in_place(page, file).await?; - let curr_position = file.stream_position().await?; - file.seek(SeekFrom::Current( - (page_count * PAGE_SIZE as i64) - curr_position as i64, - )) - .await?; + seek_to_page_start::(file, page.header.page_id.0).await?; + + // The cursor is at the page start and `persist_page_in_place` says how far + // it moved, so the padding to the next page boundary is arithmetic. + // + // It used to ask the file instead, with `stream_position`, which is a + // system call per page for two numbers already in hand. Measured over + // 6,400 pages, about 14% of the write. + let written = persist_page_in_place::(page, file).await?; + // Checked, because a page that wrote more than its slot must not turn into + // a `usize` underflow and an absurd seek. The inner length is already + // guarded against this page's own stride, so reaching this needs a header + // that serialises to more than `GENERAL_HEADER_SIZE`. + let stride = STRIDE as usize; + let padding = stride.checked_sub(written).ok_or(Error::PageOverflow { + page: page.header.page_id, + needed: written, + capacity: stride, + })?; + if padding > 0 { + file.seek(SeekFrom::Current(padding as i64)).await?; + } Ok(()) } -async fn persist_page_in_place<'a, T>( +/// Write one page where the cursor already is, and return how many bytes that +/// took. +/// +/// **The length is the point of the return value.** A caller that has to leave +/// the cursor on the next page boundary needs it, and asking the file where it +/// ended up costs a system call for something computed two lines above. +async fn persist_page_in_place<'a, T, const STRIDE: u32>( page: &'a mut GeneralPage, - file: &'a mut File, -) -> eyre::Result<()> + file: &'a mut impl AsyncWrite, +) -> crate::error::Result where T: Persistable + Send + Sync, { - let inner_bytes = page.inner.as_bytes(); + let capacity = page_capacity::(page.header.page_id)?; + let inner_bytes = page.inner.page_bytes(capacity)?; let inner_length = inner_bytes.as_ref().len(); // An over-budget page must fail here, in its own persist, instead of - // silently corrupting the neighboring page. - if inner_length > INNER_PAGE_SIZE { - return Err(eyre::Report::new(PageOverflowError { - page_id: page.header.page_id, - data_length: inner_length, - capacity: INNER_PAGE_SIZE, - })); + // silently corrupting the neighboring page. The budget is this page's own + // stride less its header, not the crate default: a table writing a larger + // page must be allowed to fill it, and a table writing a smaller one must + // be stopped before it overruns. + let capacity = page_capacity::(page.header.page_id)?; + if inner_length > capacity { + return Err(Error::PageOverflow { + page: page.header.page_id, + needed: inner_length, + capacity, + }); } - page.header.data_length = inner_length as u32; - file.write_all(page.header.as_bytes().as_ref()).await?; + page.header.data_length = page.inner.page_data_length(inner_length) as u32; + let header_bytes = page.header.as_bytes(); + let header_length = header_bytes.as_ref().len(); + file.write_all(header_bytes.as_ref()).await?; file.write_all(inner_bytes.as_ref()).await?; - Ok(()) + Ok(header_length + inner_length) } -pub async fn persist_pages_batch(pages: Vec>, file: &mut File) -> eyre::Result<()> +pub async fn persist_pages_batch( + pages: Vec>, + file: &mut impl AsyncWrite, +) -> crate::error::Result<()> where T: Persistable + Send + Sync, { - let mut iter = pages.into_iter(); - if let Some(mut page) = iter.next() { - seek_to_page_start(file, page.header.page_id.0).await?; - persist_page_in_place(&mut page, file).await?; - - for mut page in iter { - seek_to_page_start_relatively(file, page.header.page_id.0).await?; - persist_page_in_place(&mut page, file).await?; + // **One write for a run of consecutive pages, not one per page.** + // + // Every page in a run occupies exactly `STRIDE` at a known offset, so a + // run can be laid out in memory and handed to the file in a single call. + // This said `PAGE_SIZE` when the stride was a crate constant; it is a + // parameter now, and the run arithmetic below follows the parameter. + // Writing them one at a time, with a seek between each, measured 76.0 ms + // against 10.3 for the same 104 MB in one write. + // + // The run is broken whenever the page ids stop being consecutive, because + // then the offsets are not contiguous and the buffer would no longer + // correspond to a stretch of the file. Callers usually pass a contiguous + // batch and get one write; a caller that does not still gets a correct + // file, one write per run. + // + // **The buffer is bounded.** A run of ten thousand pages is 160 MB, and + // this runs on a virtual machine whose memory is not ours to spend. A run + // longer than the byte budget is flushed in pieces, each still one write, + // each still at the right offset. + // + // **One difference from writing page by page, and it is on disk rather + // than in the bytes.** Writing one at a time seeks over the space between + // a page's content and the next page's start, and on a filesystem that + // supports holes that space is never allocated. Writing a run in one call + // puts explicit zeroes there. A file read back is identical either way, + // which is what the tests hold; what differs is blocks allocated, and on + // `ext4` a file of half-empty pages will now occupy what it claims to. + // Data pages are full, so their padding is nothing; index and space pages + // are not. + /// Pages buffered before a run is flushed regardless of how long it is. + /// 512 pages is 8 MiB at the default page size. + const MAX_RUN_BYTES: usize = 8 * 1024 * 1024; + page_capacity::(0.into())?; + let max_run_pages = (MAX_RUN_BYTES / STRIDE as usize).clamp(1, 512) as u32; + + let mut iter = pages.into_iter().peekable(); + let mut buffer: Vec = Vec::new(); + let mut run_start: Option = None; + let mut expected_next: u32 = 0; + + while let Some(mut page) = iter.next() { + let id = page.header.page_id.0; + let breaks_run = run_start.is_some() && id != expected_next; + if breaks_run { + flush_run::(file, run_start.take(), &mut buffer).await?; } + if run_start.is_none() { + run_start = Some(id); + } + let start = run_start.expect("just set"); + + // **Pad before the next page, never after the last one.** Writing one + // page at a time only ever *seeks* past the end of a page, and a seek + // past the end of a file does not extend it, so the last page written + // leaves the file at its content length rather than at a page + // boundary. Padding after every page would make the file longer than + // the path this replaces produces, which is a change nobody asked for. + let offset_in_run = (id - start) as usize * STRIDE as usize; + buffer.resize(offset_in_run, 0); + persist_page_in_place_to::(&mut page, &mut buffer)?; + + expected_next = id.checked_add(1).ok_or(Error::Corrupt { + what: "page id overflowed while batching", + })?; + + // Flush at the end, and before the buffer grows past its bound. The + // next page then starts a fresh run at its own offset, which is + // correct because that offset is absolute. + let run_is_long = id - start + 1 >= max_run_pages; + if iter.peek().is_none() || run_is_long { + flush_run::(file, run_start.take(), &mut buffer).await?; + } + } - Ok(()) - } else { - Ok(()) + Ok(()) +} + +/// Write an accumulated run of pages at the offset its first page names. +async fn flush_run( + file: &mut impl AsyncWrite, + run_start: Option, + buffer: &mut Vec, +) -> crate::error::Result<()> { + if let Some(start) = run_start { + if !buffer.is_empty() { + file.seek(SeekFrom::Start(page_start_offset::(start))) + .await?; + file.write_all(buffer).await?; + } + } + buffer.clear(); + Ok(()) +} + +/// The same as [`persist_page_in_place`], into memory rather than a file. +/// +/// Shares the over-budget check, because a page too large for its slot must be +/// refused on both paths or the batch one becomes a way around it. +fn persist_page_in_place_to( + page: &mut GeneralPage, + out: &mut Vec, +) -> crate::error::Result +where + T: Persistable + Send + Sync, +{ + let capacity = page_capacity::(page.header.page_id)?; + let inner_bytes = page.inner.page_bytes(capacity)?; + let inner_length = inner_bytes.as_ref().len(); + // Same budget as the file path: this page's own stride less its header. + let capacity = page_capacity::(page.header.page_id)?; + if inner_length > capacity { + return Err(Error::PageOverflow { + page: page.header.page_id, + needed: inner_length, + capacity, + }); } + page.header.data_length = page.inner.page_data_length(inner_length) as u32; + let header_bytes = page.header.as_bytes(); + let header_length = header_bytes.as_ref().len(); + out.extend_from_slice(header_bytes.as_ref()); + out.extend_from_slice(inner_bytes.as_ref()); + Ok(header_length + inner_length) } /// Byte offset of the page with the given index, computed in `u64`. @@ -139,91 +297,138 @@ where /// [`PAGE_SIZE`], any index past 262 143 puts the page start beyond 4 GiB, /// and a `u32` multiply silently wraps the offset back into the start of /// the file. -pub(crate) fn page_start_offset(index: u32) -> u64 { - index as u64 * PAGE_SIZE as u64 +pub(crate) fn page_start_offset(index: u32) -> u64 { + index as u64 * u64::from(STRIDE) } -pub async fn seek_to_page_start(file: &mut File, index: u32) -> eyre::Result<()> { - file.seek(SeekFrom::Start(page_start_offset(index))).await?; - Ok(()) -} - -async fn seek_to_page_start_relatively(file: &mut File, index: u32) -> eyre::Result<()> { - let curr_position = file.stream_position().await?; - file.seek(SeekFrom::Current( - page_start_offset(index) as i64 - curr_position as i64, - )) - .await?; +pub async fn seek_to_page_start( + file: &mut (impl nagoya::io::Seek + Send), + index: u32, +) -> crate::error::Result<()> { + page_capacity::(index.into())?; + file.seek(SeekFrom::Start(page_start_offset::(index))) + .await?; Ok(()) } -pub async fn seek_by_link(file: &mut File, link: Link) -> eyre::Result<()> { +pub async fn seek_by_link( + file: &mut (impl nagoya::io::Seek + Send), + link: Link, +) -> crate::error::Result<()> { + let capacity = page_capacity::(link.page_id)?; + if u64::from(link.offset) + u64::from(link.length) > capacity as u64 { + return Err(Error::LinkOutOfBounds { + offset: link.offset, + length: link.length, + capacity, + }); + } file.seek(SeekFrom::Start( - link.page_id.0 as u64 * PAGE_SIZE as u64 + GENERAL_HEADER_SIZE as u64 + link.offset as u64, + page_start_offset::(link.page_id.0) + + GENERAL_HEADER_SIZE as u64 + + link.offset as u64, )) .await?; Ok(()) } -pub async fn update_at( - file: &mut File, +pub async fn update_at( + file: &mut impl crate::AsyncFile, link: Link, new_data: &[u8], -) -> eyre::Result<()> { - if new_data.len() as u32 != link.length { - return Err(eyre!( - "New data length {} does not match link length {}", - new_data.len(), - link.length - )); +) -> crate::error::Result<()> { + validate_layout::(link.page_id, DATA_LENGTH as usize)?; + if new_data.len() != link.length as usize { + return Err(Error::LinkLengthMismatch { + expected: link.length, + found: new_data.len(), + }); } // Sum in u64: `offset + length` in u32 can wrap past 4 GiB and slip // under the bound, letting the write land outside the page. if link.offset as u64 + link.length as u64 > DATA_LENGTH as u64 { - return Err(eyre!( - "Link range (offset: {}, length: {}) exceeds data bounds ({})", - link.offset, - link.length, - DATA_LENGTH - )); + return Err(Error::LinkOutOfBounds { + offset: link.offset, + length: link.length, + capacity: DATA_LENGTH as usize, + }); } - seek_by_link(file, link).await?; - file.write_all(new_data).await?; + let header = parse_general_header_by_index::(file, link.page_id.into()).await?; + if header.page_type != PageType::Data { + return Err(Error::Corrupt { + what: "data page type", + }); + } + let mut payload = vec![0; page_capacity::(link.page_id)?]; + file.read_exact(&mut payload).await?; + let rows = DataPage::<0>::directory(&payload, header.data_length)?; + if !rows + .iter() + .any(|row| row.offset == link.offset && row.length == link.length) + { + return Err(Error::Corrupt { + what: "row link absent from v3 directory", + }); + } + payload[link.offset as usize..][..new_data.len()].copy_from_slice(new_data); + let tail = payload.len() - crate::DATA_TRAILER_SIZE; + let mut crc = crc32fast::Hasher::new(); + crc.update(&payload[..tail]); + crc.update(&payload[tail + 4..]); + payload[tail..tail + 4].copy_from_slice(&crc.finalize().to_le_bytes()); + seek_to_page_start::(file, link.page_id.into()).await?; + file.seek(SeekFrom::Current(GENERAL_HEADER_SIZE as i64)) + .await?; + file.write_all(&payload).await?; Ok(()) } -pub async fn parse_general_header(file: &mut File) -> eyre::Result { +pub async fn parse_general_header( + file: &mut impl AsyncRead, +) -> crate::error::Result { let mut buffer = [0; GENERAL_HEADER_SIZE]; file.read_exact(&mut buffer).await?; // Validated: a header torn by a mid-write death must surface as an error // naming the page, not as undefined behavior in whatever reads it next. - let archived = crate::access_archived::<::Archived>(&buffer[..]) - .map_err(|error| eyre::eyre!("torn or corrupt page header: {error}"))?; - let header = rkyv::deserialize::<_, rkyv::rancor::Error>(archived) - .map_err(|error| eyre::eyre!("page header failed to deserialize: {error}"))?; - + let archived = + rkyv::access::<::Archived, rkyv::rancor::Error>(&buffer[..]) + .map_err(|_| Error::Corrupt { + what: "page header", + })?; + let header: GeneralHeader = + rkyv::deserialize::<_, rkyv::rancor::Error>(archived).map_err(|_| Error::Corrupt { + what: "page header", + })?; + + if header.data_version != crate::DATA_VERSION { + return Err(Error::UnsupportedVersion { + found: header.data_version, + expected: crate::DATA_VERSION, + }); + } Ok(header) } -pub async fn parse_page( - file: &mut File, +pub async fn parse_page( + file: &mut impl AsyncRead, index: u32, -) -> eyre::Result> +) -> crate::error::Result> where Page: rkyv::Archive + Persistable, ::Archived: rkyv::Deserialize>, { - seek_to_page_start(file, index).await?; + validate_layout::(index.into(), INNER_PAGE_SIZE as usize)?; + seek_to_page_start::(file, index).await?; parse_page_in_place::(file).await } async fn parse_page_in_place( - file: &mut File, -) -> eyre::Result> + file: &mut impl AsyncRead, +) -> crate::error::Result> where Page: rkyv::Archive + Persistable, ::Archived: @@ -235,6 +440,13 @@ where } else { header.data_length }; + if length > INNER_PAGE_SIZE { + return Err(Error::PageOverflow { + page: header.page_id, + needed: length as usize, + capacity: INNER_PAGE_SIZE as usize, + }); + } let mut buffer: Vec = vec![0u8; length as usize]; file.read_exact(&mut buffer).await?; @@ -246,24 +458,25 @@ where }) } -pub async fn parse_pages_batch( - file: &mut File, +pub async fn parse_pages_batch( + file: &mut impl AsyncRead, indexes: Vec, -) -> eyre::Result>> +) -> crate::error::Result>> where Page: rkyv::Archive + Persistable, ::Archived: rkyv::Deserialize>, { + validate_layout::(0.into(), PAGE_SIZE as usize)?; let mut iter = indexes.into_iter(); if let Some(index) = iter.next() { let mut pages = vec![]; - seek_to_page_start(file, index).await?; + seek_to_page_start::(file, index).await?; let page = parse_page_in_place::(file).await?; pages.push(page); for index in iter { - seek_to_page_start_relatively(file, index).await?; + seek_to_page_start::(file, index).await?; let page = parse_page_in_place::(file).await?; pages.push(page); } @@ -274,41 +487,59 @@ where } } -pub async fn parse_general_header_by_index( - file: &mut File, +pub async fn parse_general_header_by_index( + file: &mut impl AsyncRead, index: u32, -) -> eyre::Result { - seek_to_page_start(file, index).await?; +) -> crate::error::Result { + seek_to_page_start::(file, index).await?; let header = parse_general_header(file).await?; - + if header.page_id != index.into() { + return Err(Error::Corrupt { + what: "page identity", + }); + } Ok(header) } -pub async fn parse_data_page( - file: &mut File, +pub async fn parse_data_page< + const PAGE_SIZE: u32, + const INNER_PAGE_SIZE: usize, + const STRIDE: u32, +>( + file: &mut impl AsyncRead, index: u32, -) -> eyre::Result>> { - seek_to_page_start(file, index).await?; - parse_data_page_in_place::(file).await +) -> crate::error::Result>> { + validate_layout::(index.into(), INNER_PAGE_SIZE)?; + seek_to_page_start::(file, index).await?; + let page = parse_data_page_in_place::(file).await?; + if page.header.page_id != index.into() { + return Err(Error::Corrupt { + what: "data page identity", + }); + } + Ok(page) } async fn parse_data_page_in_place( - file: &mut File, -) -> eyre::Result>> { + file: &mut impl AsyncRead, +) -> crate::error::Result>> { let header = parse_general_header(file).await?; - let mut buffer = [0u8; INNER_PAGE_SIZE]; - if header.next_id == 0.into() { - #[allow(clippy::unused_io_amount)] - file.read(&mut buffer).await?; - } else { - file.read_exact(&mut buffer).await?; + if header.page_type != PageType::Data { + return Err(Error::Corrupt { + what: "data page type", + }); } - - let data = DataPage { - data: buffer, - length: header.data_length, - }; + let mut buffer = vec![0u8; page_capacity::(header.page_id)?]; + if header.data_length as usize > INNER_PAGE_SIZE { + return Err(Error::PageOverflow { + page: header.page_id, + needed: header.data_length as usize, + capacity: INNER_PAGE_SIZE, + }); + } + file.read_exact(&mut buffer).await?; + let data = DataPage::decode(&buffer, header.data_length)?; Ok(GeneralPage { header, @@ -316,27 +547,19 @@ async fn parse_data_page_in_place( - file: &mut File, +pub async fn parse_data_pages_batch< + const PAGE_SIZE: u32, + const INNER_PAGE_SIZE: usize, + const STRIDE: u32, +>( + file: &mut impl AsyncRead, indexes: Vec, -) -> eyre::Result>>> { - let mut iter = indexes.into_iter(); - if let Some(index) = iter.next() { - let mut pages = vec![]; - seek_to_page_start(file, index).await?; - let page = parse_data_page_in_place::(file).await?; - pages.push(page); - - for index in iter { - seek_to_page_start_relatively(file, index).await?; - let page = parse_data_page_in_place::(file).await?; - pages.push(page); - } - - Ok(pages) - } else { - Ok(vec![]) +) -> crate::error::Result>>> { + let mut pages = Vec::with_capacity(indexes.len()); + for index in indexes { + pages.push(parse_data_page::(file, index).await?); } + Ok(pages) } // pub fn parse_data_record( @@ -345,7 +568,7 @@ pub async fn parse_data_pages_batch, -// ) -> eyre::Result> { +// ) -> crate::error::Result> { // seek_to_page_start(file, index)?; // let header = parse_general_header(file)?; // if header.page_type != PageType::Data { @@ -364,8 +587,8 @@ pub async fn parse_data_pages_batch( - file: &mut File, -) -> eyre::Result { + file: &mut impl AsyncRead, +) -> crate::error::Result { file.seek(SeekFrom::Start(0)).await?; let header = parse_general_header(file).await?; @@ -378,7 +601,7 @@ pub async fn parse_space_info( // pub fn read_index_pages( // file: &mut std::fs::File, // length: u32, -// ) -> eyre::Result>> +// ) -> crate::error::Result>> // where // T: Archive, // ::Archived: rkyv::Deserialize>, @@ -394,7 +617,7 @@ pub async fn parse_space_info( // fn read_links( // mut file: &mut std::fs::File, // space_info: &SpaceInfo, -// ) -> eyre::Result> { +// ) -> crate::error::Result> { // Ok( // read_index_pages::(&mut file, space_info.primary_key_length)? // .iter() @@ -405,14 +628,14 @@ pub async fn parse_space_info( // // pub fn read_rows_schema( // file: &mut std::fs::File, -// ) -> eyre::Result> { +// ) -> crate::error::Result> { // let space_info = parse_space_info::(file)?; // Ok(space_info.row_schema) // } // // pub fn read_data_pages( // mut file: &mut std::fs::File, -// ) -> eyre::Result>> { +// ) -> crate::error::Result>> { // let space_info = parse_space_info::(file)?; // let primary_key_fields = &space_info.primary_key_fields; // if primary_key_fields.len() != 1 { @@ -446,7 +669,7 @@ pub async fn parse_space_info( // // let mut result: Vec> = vec![]; // for link in links { -// let row = parse_data_record::( +// let row = parse_data_record::<{ PAGE_SIZE as u32 }>( // &mut file, // link.page_id.0, // link.offset, @@ -464,7 +687,13 @@ mod tests { use super::{page_start_offset, parse_data_pages_batch, persist_pages_batch}; use crate::page::header::GeneralHeader; use crate::page::ty::PageType; - use crate::{DataPage, GeneralPage, DATA_VERSION, INNER_PAGE_SIZE, PAGE_SIZE}; + use crate::{ + DataPage, GeneralPage, DATA_VERSION, DEFAULT_PAGE_STRIDE, INNER_PAGE_SIZE, PAGE_SIZE, + }; + use std::prelude::v1::*; + // `sync_all` and the rest are trait methods now, not inherent ones, so the + // trait has to be in scope for a `HostFile` to answer to them. + use nagoya::io::File as _; /// First page index whose start offset no longer fits in `u32`. const FIRST_PAGE_PAST_4_GIB: u32 = (u32::MAX / PAGE_SIZE as u32) + 1; @@ -472,19 +701,19 @@ mod tests { #[test] fn page_start_offset_is_computed_in_u64() { assert_eq!( - page_start_offset(FIRST_PAGE_PAST_4_GIB), + page_start_offset::<{ PAGE_SIZE as u32 }>(FIRST_PAGE_PAST_4_GIB), FIRST_PAGE_PAST_4_GIB as u64 * PAGE_SIZE as u64 ); - assert!(page_start_offset(FIRST_PAGE_PAST_4_GIB) > u32::MAX as u64); + assert!(page_start_offset::<{ PAGE_SIZE as u32 }>(FIRST_PAGE_PAST_4_GIB) > u32::MAX as u64); // The largest possible page id must be addressable too. assert_eq!( - page_start_offset(u32::MAX), + page_start_offset::<{ PAGE_SIZE as u32 }>(u32::MAX), u32::MAX as u64 * PAGE_SIZE as u64 ); // The old `u32` arithmetic wrapped this offset back into the first // pages of the file. assert_ne!( - page_start_offset(FIRST_PAGE_PAST_4_GIB), + page_start_offset::<{ PAGE_SIZE as u32 }>(FIRST_PAGE_PAST_4_GIB), FIRST_PAGE_PAST_4_GIB.wrapping_mul(PAGE_SIZE as u32) as u64 ); } @@ -493,11 +722,145 @@ mod tests { let mut data = [0u8; INNER_PAGE_SIZE]; data[..marker.len()].copy_from_slice(marker); DataPage { + rows: Vec::new(), length: marker.len() as u32, data, } } + async fn scratch(name: &str) -> (std::path::PathBuf, nagoya::io::HostFile) { + let path = + std::env::temp_dir().join(format!("data_bucket_{name}_{}.wt", std::process::id())); + let _ = nagoya::io::remove_file(&path).await; + let file = nagoya::io::create(&path).await.unwrap(); + (path, file) + } + + /// A batch must land byte for byte where the same pages written one at a + /// time would land. + /// + /// This is the guard on coalescing a run into one write: the whole point is + /// that it is not observable in the file, only in how long it took. + #[tokio::test] + async fn a_batch_writes_what_one_at_a_time_writes() { + let markers: [&[u8]; 4] = [b"alpha", b"beta", b"gamma", b"delta"]; + + let (one_path, mut one) = scratch("batch_one_at_a_time").await; + for (n, marker) in markers.iter().enumerate() { + let mut page = page_at::(n as u32, marker); + super::persist_page::<_, DEFAULT_PAGE_STRIDE>(&mut page, &mut one) + .await + .unwrap(); + } + one.sync_all().await.unwrap(); + drop(one); + + let (many_path, mut many) = scratch("batch_together").await; + let pages: Vec<_> = markers + .iter() + .enumerate() + .map(|(n, marker)| page_at::(n as u32, marker)) + .collect(); + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(pages, &mut many) + .await + .unwrap(); + many.sync_all().await.unwrap(); + drop(many); + + let expected = std::fs::read(&one_path).unwrap(); + let actual = std::fs::read(&many_path).unwrap(); + assert_eq!( + expected.len(), + actual.len(), + "the batch produced a file of a different length" + ); + assert_eq!(expected, actual, "the batch produced different bytes"); + + std::fs::remove_file(&one_path).unwrap(); + std::fs::remove_file(&many_path).unwrap(); + } + + /// Page ids with a gap in them are two runs, and each has to land at the + /// offset its own id names rather than after the one before it. + #[tokio::test] + async fn a_batch_with_a_gap_puts_each_page_at_its_own_offset() { + let (path, mut file) = scratch("batch_with_a_gap").await; + let pages = vec![ + page_at::(0, b"first"), + page_at::(1, b"second"), + // The gap: nothing at 2 or 3. + page_at::(4, b"fifth"), + ]; + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(pages, &mut file) + .await + .unwrap(); + file.sync_all().await.unwrap(); + drop(file); + + let bytes = std::fs::read(&path).unwrap(); + // The v3 directory occupies the tail even on a partly filled page. + assert_eq!(bytes.len(), 5 * PAGE_SIZE, "the file is the wrong length"); + let marker_at = |page: usize, marker: &[u8]| { + let from = page * PAGE_SIZE + crate::GENERAL_HEADER_SIZE; + assert_eq!( + &bytes[from..from + marker.len()], + marker, + "page {page} holds the wrong data" + ); + }; + marker_at(0, b"first"); + marker_at(1, b"second"); + marker_at(4, b"fifth"); + // The skipped pages are zeroes, not a copy of anything. + let gap = 2 * PAGE_SIZE + crate::GENERAL_HEADER_SIZE; + assert!( + bytes[gap..gap + 16].iter().all(|&b| b == 0), + "the gap was written over" + ); + + std::fs::remove_file(&path).unwrap(); + } + + /// A run longer than the buffer bound is flushed in pieces, and the pieces + /// have to join up exactly. + /// + /// This is the guard on bounding the buffer. The bound exists so a batch of + /// ten thousand pages does not become a 160 MB allocation on a virtual + /// machine, and the risk it introduces is a seam every 512 pages. + #[tokio::test] + async fn a_run_longer_than_the_buffer_bound_still_joins_up() { + // Comfortably past `MAX_RUN_PAGES`, so at least one seam is crossed. + const COUNT: u32 = 520; + + let (one_path, mut one) = scratch("long_run_one_at_a_time").await; + for id in 0..COUNT { + let mut page = page_at::(id, format!("page{id}").as_bytes()); + super::persist_page::<_, DEFAULT_PAGE_STRIDE>(&mut page, &mut one) + .await + .unwrap(); + } + one.sync_all().await.unwrap(); + drop(one); + + let (many_path, mut many) = scratch("long_run_batched").await; + let pages: Vec<_> = (0..COUNT) + .map(|id| page_at::(id, format!("page{id}").as_bytes())) + .collect(); + persist_pages_batch::<_, DEFAULT_PAGE_STRIDE>(pages, &mut many) + .await + .unwrap(); + many.sync_all().await.unwrap(); + drop(many); + + let expected = std::fs::read(&one_path).unwrap(); + let actual = std::fs::read(&many_path).unwrap(); + assert_eq!(expected.len(), actual.len(), "lengths differ across a seam"); + assert_eq!(expected, actual, "bytes differ across a seam"); + + std::fs::remove_file(&one_path).unwrap(); + std::fs::remove_file(&many_path).unwrap(); + } + #[tokio::test] async fn persist_page_rejects_inner_data_past_the_page_slot() { // A data page whose buffer is larger than the slot budget can hand @@ -508,14 +871,7 @@ mod tests { "data_bucket_persist_overflow_{}.wt", std::process::id() )); - let mut file = tokio::fs::OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(true) - .open(&path) - .await - .unwrap(); + let mut file = nagoya::io::create(&path).await.unwrap(); let mut page = GeneralPage { header: GeneralHeader { @@ -528,43 +884,87 @@ mod tests { data_length: 0, }, inner: DataPage { + rows: Vec::new(), length: OVERSIZED as u32, data: [7u8; OVERSIZED], }, }; - let err = super::persist_page(&mut page, &mut file).await.unwrap_err(); + let err = super::persist_page::<_, { PAGE_SIZE as u32 }>(&mut page, &mut file) + .await + .unwrap_err(); assert!( - err.downcast_ref::().is_some(), - "expected PageOverflowError, got: {err}" + matches!(err, crate::error::Error::PageOverflow { .. }), + "expected a page overflow, got: {err}" ); // Nothing may have been written: the neighboring page is the one an // unchecked write would have corrupted. - assert_eq!(file.metadata().await.unwrap().len(), 0); + assert_eq!(nagoya::io::File::length(&mut file).await.unwrap(), 0); // A page that fits its slot still persists. page.inner.length = 64; - super::persist_page(&mut page, &mut file).await.unwrap(); + super::persist_page::<_, { PAGE_SIZE as u32 }>(&mut page, &mut file) + .await + .unwrap(); drop(file); std::fs::remove_file(&path).unwrap(); } + #[tokio::test] + async fn update_at_preserves_v3_directory_and_checksum() { + let (path, mut file) = scratch("update_v3").await; + let link = crate::Link { + page_id: 1.into(), + offset: 0, + length: 8, + }; + let mut data = DataPage::::new(); + data.update_at(link, b"original").unwrap(); + let mut page = GeneralPage { + header: GeneralHeader::new(1.into(), PageType::Data, 0.into()), + inner: data, + }; + super::persist_page::<_, DEFAULT_PAGE_STRIDE>(&mut page, &mut file) + .await + .unwrap(); + super::update_at::<{ INNER_PAGE_SIZE as u32 }, DEFAULT_PAGE_STRIDE>( + &mut file, + link, + b"replaced", + ) + .await + .unwrap(); + let decoded = + super::parse_data_page::( + &mut file, 1, + ) + .await + .unwrap(); + assert_eq!(&decoded.inner.data[..8], b"replaced"); + assert_eq!(decoded.inner.rows, page.inner.rows); + let missing = crate::Link { offset: 8, ..link }; + assert!( + super::update_at::<{ INNER_PAGE_SIZE as u32 }, DEFAULT_PAGE_STRIDE>( + &mut file, + missing, + b"missing!" + ) + .await + .is_err() + ); + drop(file); + std::fs::remove_file(path).unwrap(); + } + #[tokio::test] async fn update_at_rejects_offset_plus_length_wrapping_u32() { let path = std::env::temp_dir().join(format!( "data_bucket_update_at_wrap_{}.wt", std::process::id() )); - let mut file = tokio::fs::OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(true) - .open(&path) - .await - .unwrap(); + let mut file = nagoya::io::create(&path).await.unwrap(); // In u32, offset + length wraps to 5 and used to pass the bounds // check, sending the write far outside the page. @@ -573,10 +973,14 @@ mod tests { offset: u32::MAX - 2, length: 8, }; - let err = super::update_at::<100>(&mut file, link, &[1, 2, 3, 4, 5, 6, 7, 8]) - .await - .unwrap_err(); - assert!(err.to_string().contains("exceeds data bounds")); + let err = super::update_at::<100, { PAGE_SIZE as u32 }>( + &mut file, + link, + &[1, 2, 3, 4, 5, 6, 7, 8], + ) + .await + .unwrap_err(); + assert!(matches!(err, crate::error::Error::LinkOutOfBounds { .. })); drop(file); std::fs::remove_file(&path).unwrap(); @@ -591,14 +995,7 @@ mod tests { "data_bucket_seek_past_4gib_{}.wt", std::process::id() )); - let mut file = tokio::fs::OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(true) - .open(&path) - .await - .unwrap(); + let mut file = nagoya::io::create(&path).await.unwrap(); let first_page = GeneralPage { header: GeneralHeader { @@ -625,21 +1022,22 @@ mod tests { inner: data_page_with_marker(BOUNDARY_MARKER), }; - persist_pages_batch(vec![first_page, boundary_page], &mut file) + persist_pages_batch::<_, { PAGE_SIZE as u32 }>(vec![first_page, boundary_page], &mut file) .await .unwrap(); // The boundary page must have been written past 4 GiB (the file is // sparse, so this stays cheap), not wrapped back onto the first pages. - // tokio's File buffers writes; flush so metadata() sees them. - tokio::io::AsyncWriteExt::flush(&mut file).await.unwrap(); - let file_length = file.metadata().await.unwrap().len(); - assert!(file_length > page_start_offset(FIRST_PAGE_PAST_4_GIB)); - - let pages = parse_data_pages_batch::<{ PAGE_SIZE as u32 }, INNER_PAGE_SIZE>( - &mut file, - vec![1, FIRST_PAGE_PAST_4_GIB], - ) + // the async file buffers writes; flush so metadata() sees them. + nagoya::io::Write::flush(&mut file).await.unwrap(); + let file_length = nagoya::io::File::length(&mut file).await.unwrap(); + assert!(file_length > page_start_offset::<{ PAGE_SIZE as u32 }>(FIRST_PAGE_PAST_4_GIB)); + + let pages = parse_data_pages_batch::< + { PAGE_SIZE as u32 }, + INNER_PAGE_SIZE, + { PAGE_SIZE as u32 }, + >(&mut file, vec![1, FIRST_PAGE_PAST_4_GIB]) .await .unwrap(); @@ -655,6 +1053,95 @@ mod tests { drop(file); std::fs::remove_file(&path).unwrap(); } + + /// Half the default, so a page written at this stride lands where the + /// default would put the *middle* of a page. A stride that were quietly + /// ignored could not pass this. + const HALF: usize = PAGE_SIZE / 2; + const HALF_INNER: usize = HALF - crate::GENERAL_HEADER_SIZE; + + fn page_at(id: u32, marker: &[u8]) -> GeneralPage> { + let mut data = [0u8; INNER]; + data[..marker.len()].copy_from_slice(marker); + GeneralPage { + header: GeneralHeader { + data_version: DATA_VERSION, + space_id: 1.into(), + page_id: id.into(), + previous_id: 0.into(), + next_id: 0.into(), + page_type: PageType::Data, + data_length: 0, + }, + inner: DataPage { + rows: Vec::new(), + length: marker.len() as u32, + data, + }, + } + } + + #[tokio::test] + async fn a_non_default_stride_is_written_and_read_back_at_that_stride() { + let path = + std::env::temp_dir().join(format!("data_bucket_stride_{}.wt", std::process::id())); + let mut file = nagoya::io::create(&path).await.unwrap(); + + let pages = vec![ + page_at::(0, b"zero"), + page_at::(1, b"one"), + page_at::(2, b"two"), + ]; + persist_pages_batch::<_, { HALF as u32 }>(pages, &mut file) + .await + .unwrap(); + // async-fs buffers in user space and flushes on drop, best effort and + // silently. Asking the filesystem how long the file is before this + // returns measures the buffer, not the write. + nagoya::io::Write::flush(&mut file).await.unwrap(); + + // The last page starts at two strides in, so the file ends somewhere in + // the third. Not `3 * HALF`: unlike `persist_page`, a batch does not + // seek to the end of its final page, so the file stops after the last + // body rather than at a page boundary. + // + // This is the assertion the whole change exists for. At the default + // stride the same three pages would end past `2 * PAGE_SIZE`, which is + // four times further out and cannot be confused with this. + let written = std::fs::metadata(&path).unwrap().len(); + assert!( + written > 2 * HALF as u64 && written <= 3 * HALF as u64, + "three pages at a {HALF}-byte stride should end inside the third, got {written}" + ); + + let read = parse_data_pages_batch::<{ HALF as u32 }, HALF_INNER, { HALF as u32 }>( + &mut file, + vec![0, 1, 2], + ) + .await + .unwrap(); + assert_eq!(read.len(), 3); + assert_eq!(&read[0].inner.data[..4], b"zero"); + assert_eq!(&read[1].inner.data[..3], b"one"); + assert_eq!(&read[2].inner.data[..3], b"two"); + assert_eq!(read[2].header.page_id, 2.into()); + + // Read at the default stride and page 1 is not where it was left. A + // seek that ignored its parameter would pass the round trip above and + // fail here, so this is what proves the parameter is load-bearing. + let wrong = parse_data_pages_batch::< + { PAGE_SIZE as u32 }, + INNER_PAGE_SIZE, + { PAGE_SIZE as u32 }, + >(&mut file, vec![1]) + .await; + assert!( + wrong.is_err() || wrong.unwrap()[0].header.page_id != 1.into(), + "reading at the wrong stride must not find page 1" + ); + + let _ = std::fs::remove_file(&path); + } } // #[cfg(test)] @@ -897,7 +1384,7 @@ mod tests { // create_test_database_file(filename); // // let mut file: std::fs::File = std::fs::File::open(filename).unwrap(); -// let data_pages: Vec> = read_data_pages::(&mut file).unwrap(); +// let data_pages: Vec> = read_data_pages::<{ PAGE_SIZE as u32 }>(&mut file).unwrap(); // assert_eq!(data_pages[0][0], DataTypeValue::I32(1)); // assert_eq!( // data_pages[0][1], diff --git a/src/persistence/data/mod.rs b/src/persistence/data/mod.rs index a247f26..7682598 100644 --- a/src/persistence/data/mod.rs +++ b/src/persistence/data/mod.rs @@ -1,10 +1,11 @@ +use alloc::string::String; pub mod rkyv_data; mod types; mod util; pub use types::DataTypeValue; -use std::fmt; +use core::fmt; #[derive(Clone, Debug, Eq, PartialEq)] pub enum DataDecodeError { @@ -51,7 +52,7 @@ impl fmt::Display for DataDecodeError { } } -impl std::error::Error for DataDecodeError {} +impl core::error::Error for DataDecodeError {} pub trait DataType { /// Advances an offset past this type, including its required padding. diff --git a/src/persistence/data/rkyv_data.rs b/src/persistence/data/rkyv_data.rs index 9ea4a1e..8167899 100644 --- a/src/persistence/data/rkyv_data.rs +++ b/src/persistence/data/rkyv_data.rs @@ -1,5 +1,6 @@ use crate::persistence::data::{DataDecodeError, DataTypeValue}; -use std::str::FromStr; +use alloc::vec::Vec; +use core::str::FromStr; /// Decodes a dynamically described row from an rkyv archive. /// @@ -67,6 +68,7 @@ mod test { use crate::persistence::data::{DataDecodeError, DataTypeValue}; use rkyv::{Archive, Deserialize, Serialize}; use std::f64::consts::PI; + use std::prelude::v1::*; #[derive(Archive, Serialize, Deserialize, Debug)] struct Struct1 { diff --git a/src/persistence/data/types.rs b/src/persistence/data/types.rs index 3490693..75b05e5 100644 --- a/src/persistence/data/types.rs +++ b/src/persistence/data/types.rs @@ -1,4 +1,5 @@ -use std::str::FromStr; +use alloc::{borrow::ToOwned, string::String, string::ToString}; +use core::str::FromStr; use derive_more::derive::Display; use derive_more::From; diff --git a/src/util/persistable.rs b/src/util/persistable.rs index d9f1680..a2afe1e 100644 --- a/src/util/persistable.rs +++ b/src/util/persistable.rs @@ -1,4 +1,5 @@ use crate::SizeMeasurable; +use alloc::{string::String, vec::Vec}; use rkyv::api::high::HighValidator; use rkyv::bytecheck::CheckBytes; @@ -13,6 +14,17 @@ use rkyv::{Archive, Deserialize, Serialize}; pub trait Persistable { fn as_bytes(&self) -> impl AsRef<[u8]> + Send; fn from_bytes(bytes: &[u8], version: u32) -> Self; + + /// Encode within the payload of a physical page. Data pages use the + /// supplied capacity to place their directory at the fixed page tail. + fn page_bytes(&self, _capacity: usize) -> crate::error::Result + Send> { + Ok(self.as_bytes()) + } + + /// Initialized row extent for data pages; encoded length for other pages. + fn page_data_length(&self, encoded_length: usize) -> usize { + encoded_length + } } /* diff --git a/src/util/sized.rs b/src/util/sized.rs index 444fca6..9e778ea 100644 --- a/src/util/sized.rs +++ b/src/util/sized.rs @@ -1,9 +1,11 @@ use crate::link::{Link, LINK_LENGTH}; +use alloc::sync::Arc; +use alloc::{string::String, vec::Vec}; +use core::mem; use ordered_float::OrderedFloat; use psc_nanoid::packed::AlphabetPackExt; use psc_nanoid::PackedNanoid; use rkyv::util::AlignedVec; -use std::{mem, sync::Arc}; use uuid::Uuid; pub const fn align(len: usize) -> usize { @@ -279,6 +281,7 @@ mod test { use crate::util::sized::SizeMeasurable; use crate::{IndexValue, Link}; use rkyv::to_bytes; + use std::prelude::v1::*; use uuid::Uuid; #[test] @@ -311,7 +314,7 @@ mod test { impl Default for ExpensiveDefault { fn default() -> Self { - panic!("the sizing override must not construct the default value") + core::panic!("the sizing override must not construct the default value") } } diff --git a/tests/no-std/Cargo.toml b/tests/no-std/Cargo.toml new file mode 100644 index 0000000..d916c9a --- /dev/null +++ b/tests/no-std/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "data-bucket-no-std-consumer" +version = "0.0.0" +edition = "2021" +publish = false + +[workspace] + +[dependencies] +data_bucket = { path = "../.." } +rkyv = { version = "^0.8", default-features = false, features = ["alloc", "bytecheck"] } diff --git a/tests/no-std/src/lib.rs b/tests/no-std/src/lib.rs new file mode 100644 index 0000000..76f010f --- /dev/null +++ b/tests/no-std/src/lib.rs @@ -0,0 +1,9 @@ +#![no_std] + +use data_bucket::{SizeMeasurable, SizeMeasure}; +use rkyv::Archive; + +#[derive(Archive, SizeMeasure)] +pub enum Kind { Row, Index } + +pub fn archived_size(kind: Kind) -> usize { kind.aligned_size() } diff --git a/tests/pr75_layout.rs b/tests/pr75_layout.rs new file mode 100644 index 0000000..29a1efb --- /dev/null +++ b/tests/pr75_layout.rs @@ -0,0 +1,108 @@ +use data_bucket::{ + DataPage, GeneralHeader, GeneralPage, IndexPage, IndexValue, Link, PageType, Persistable, +}; +use nagoya::io::{Error, Read, Seek, SeekFrom, Write}; + +struct NoIo; +impl Read for NoIo { + async fn read(&mut self, _: &mut [u8]) -> Result { + panic!("invalid layout reached read") + } +} +impl Seek for NoIo { + async fn seek(&mut self, _: SeekFrom) -> Result { + panic!("invalid layout reached seek") + } +} +impl Write for NoIo { + async fn write(&mut self, _: &[u8]) -> Result { + panic!("invalid layout reached write") + } + async fn flush(&mut self) -> Result<(), Error> { + panic!("invalid layout reached flush") + } +} + +fn page() -> GeneralPage> { + GeneralPage { + header: GeneralHeader::new(1.into(), PageType::Data, 0.into()), + inner: DataPage { + rows: Vec::new(), + data: [0; 32], + length: 1, + }, + } +} + +#[test] +fn a_stride_smaller_than_the_header_is_rejected_before_io() { + nagoya::block_on(async { + assert!(data_bucket::persist_page::<_, 16>(&mut page(), &mut NoIo) + .await + .is_err()); + assert!( + data_bucket::persist_pages_batch::<_, 16>(vec![page()], &mut NoIo) + .await + .is_err() + ); + }); +} + +#[test] +fn an_inconsistent_update_layout_is_rejected_before_io() { + let link = Link { + page_id: 1.into(), + offset: 5000, + length: 1, + }; + assert!( + nagoya::block_on(data_bucket::update_at::<16356, 4096>(&mut NoIo, link, &[1])).is_err() + ); +} + +#[test] +fn index_slot_capacity_remains_representable_on_large_pages() { + let size = data_bucket::get_index_page_size_from_data_length::(4 * 1024 * 1024); + assert_eq!(size, usize::from(u16::MAX)); + let page = IndexPage::new(IndexValue::::default(), size); + assert_eq!(usize::from(page.size), page.slots.len()); + assert_eq!(usize::from(page.size), page.index_values.len()); +} + +struct ReadOnly { + bytes: Vec, + position: usize, +} +impl Read for ReadOnly { + async fn read(&mut self, out: &mut [u8]) -> Result { + let n = out + .len() + .min(self.bytes.len().saturating_sub(self.position)); + out[..n].copy_from_slice(&self.bytes[self.position..self.position + n]); + self.position += n; + Ok(n) + } +} +impl Seek for ReadOnly { + async fn seek(&mut self, from: SeekFrom) -> Result { + match from { + SeekFrom::Start(position) => self.position = position as usize, + _ => panic!("unexpected seek"), + } + Ok(self.position as u64) + } +} + +#[test] +fn a_decoder_accepts_read_seek_without_write_or_durability() { + let header = GeneralHeader::new(0.into(), PageType::Data, 0.into()); + let mut file = ReadOnly { + bytes: header.as_bytes().as_ref().to_vec(), + position: 0, + }; + let decoded = nagoya::block_on(data_bucket::parse_general_header_by_index::<128>( + &mut file, 0, + )) + .unwrap(); + assert_eq!(decoded.page_id, header.page_id); +} diff --git a/tools/create-data-file/Cargo.toml b/tools/create-data-file/Cargo.toml index e94f28c..c9b769c 100644 --- a/tools/create-data-file/Cargo.toml +++ b/tools/create-data-file/Cargo.toml @@ -4,6 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] +nagoya = { version = "^0.1", features = ["std"] } clap = { version = "4.5.21", features = ["derive"] } data_bucket = { path = "../.." } eyre = "0.6.12" diff --git a/tools/create-data-file/src/main.rs b/tools/create-data-file/src/main.rs index 4c2637d..384fe00 100644 --- a/tools/create-data-file/src/main.rs +++ b/tools/create-data-file/src/main.rs @@ -1,14 +1,13 @@ use clap::Parser; -use data_bucket::{persist_page, GeneralHeader, GeneralPage, PageType, DATA_VERSION}; -use data_bucket::{IndexData, IndexValue, Interval, Link, SpaceInfoData}; -use rkyv::rancor::Error; -use rkyv::{Archive, Deserialize, Serialize}; -use std::{ - fs::{remove_file, File}, - str, +use data_bucket::{ + persist_page, DataPage, GeneralHeader, GeneralPage, Link, PageType, SpaceInfoPage, + INNER_PAGE_SIZE, PAGE_SIZE, }; +use nagoya::io::File as _; +use rkyv::{Archive, Deserialize, Serialize}; #[derive(Parser, Debug)] +#[command(about = "Create a v3 demonstration store with independently readable data pages")] struct Args { #[arg(short, long)] filename: String, @@ -16,136 +15,80 @@ struct Args { count: usize, } -fn main() -> eyre::Result<()> { - let args = Args::parse(); - _ = remove_file(args.filename.as_str()); - let mut output_file = File::create(args.filename.as_str())?; - - let space_info_header = GeneralHeader { - data_version: DATA_VERSION, - space_id: 1.into(), - page_id: 0.into(), - previous_id: 0.into(), - next_id: 1.into(), - page_type: PageType::SpaceInfo, - data_length: 0u32, - }; - - let space_info = SpaceInfoData { - id: 1.into(), - page_count: 4, - name: "generated space".to_owned(), - row_schema: vec![ - ("val".to_string(), "i32".to_string()), - ("attr".to_string(), "String".to_string()), - ], - primary_key_fields: vec!["val".to_string()], - primary_key_intervals: vec![Interval(1, 1)], - secondary_index_types: vec![], - secondary_index_intervals: Default::default(), - data_intervals: vec![], - pk_gen_state: (), - empty_links_list: vec![], - }; - - let mut space_info_page = GeneralPage { - header: space_info_header, - inner: space_info, - }; - persist_page(&mut space_info_page, &mut output_file).unwrap(); - - let index_header = GeneralHeader { - data_version: DATA_VERSION, - space_id: 1.into(), - page_id: 1.into(), - previous_id: 0.into(), - next_id: 2.into(), - page_type: PageType::Index, - data_length: 0, - }; - - let data_header = GeneralHeader { - data_version: DATA_VERSION, - space_id: 1.into(), - page_id: 2.into(), - previous_id: 2.into(), - next_id: 4.into(), - page_type: PageType::Data, - data_length: 0, - }; - - let page_size = 100; - let total_pages = (args.count + page_size - 1) / page_size; - - for page_idx in 0..total_pages { - let start = page_idx * page_size; - let end = usize::min(start + page_size, args.count); - - let (mut data_page, offsets) = generate_data_page(start as i32, end - start, data_header); - persist_page(&mut data_page, &mut output_file).unwrap(); - - let index_data = create_index_data(&data_page, &offsets); - - let mut index_page = GeneralPage { - header: index_header, - inner: index_data, - }; - persist_page(&mut index_page, &mut output_file).unwrap(); - } - - Ok(()) -} - #[derive(Archive, Debug, Deserialize, Serialize)] struct TableStruct { val: i32, attr: String, } -pub fn generate_data_page( - start_key: i32, - count: usize, - header: GeneralHeader, -) -> (GeneralPage>, Vec<(i32, u32, u32)>) { - let mut buffer = Vec::new(); - let mut offsets = Vec::new(); - let mut current_offset = 0; - - for i in 0..count { - let key = start_key + i as i32; - let data = TableStruct { - val: key, - attr: format!("string {}", key), - }; - let serialized_data = rkyv::to_bytes::(&data).unwrap(); - let length = serialized_data.len() as u32; - - buffer.extend_from_slice(&serialized_data); - offsets.push((key, current_offset as u32, length)); - current_offset += length as usize; - } - - ( - GeneralPage { - header, - inner: buffer, - }, - offsets, - ) -} - -fn create_index_data(page: &GeneralPage>, offsets: &[(i32, u32, u32)]) -> IndexData { - let index_values = offsets - .iter() - .map(|(key, offset, length)| IndexValue:: { - key: *key, - link: Link { - page_id: page.header.page_id, - offset: *offset, - length: *length, +fn main() -> eyre::Result<()> { + let args = Args::parse(); + eyre::ensure!( + args.count <= i32::MAX as usize, + "count exceeds the demonstration key range" + ); + let file = std::fs::OpenOptions::new() + .write(true) + .read(true) + .create_new(true) + .open(&args.filename)?; + nagoya::block_on(async move { + let mut file = nagoya::io::HostFile::new(file); + let mut info = GeneralPage { + header: GeneralHeader::new(0.into(), PageType::SpaceInfo, 1.into()), + inner: SpaceInfoPage { + id: 1.into(), + page_count: 0, + name: "generated space".into(), + version: 1, + row_schema: vec![ + ("val".into(), "i32".into()), + ("attr".into(), "String".into()), + ], + primary_key_fields: vec!["val".into()], + secondary_index_types: vec![], + pk_gen_state: (), + empty_links_list: vec![], }, - }) - .collect(); - - IndexData { index_values } + }; + persist_page::<_, { PAGE_SIZE as u32 }>(&mut info, &mut file).await?; + let mut id = 1u32; + let mut page = GeneralPage { + header: GeneralHeader::new(id.into(), PageType::Data, 1.into()), + inner: DataPage::::new(), + }; + for key in 0..args.count { + let row = TableStruct { + val: key as i32, + attr: format!("string {key}"), + }; + let bytes = rkyv::to_bytes::(&row)?; + let needed = page.inner.length as usize + + bytes.len() + + (page.inner.rows.len() + 1) * data_bucket::ROW_SLOT_SIZE + + data_bucket::DATA_TRAILER_SIZE; + if needed > INNER_PAGE_SIZE { + persist_page::<_, { PAGE_SIZE as u32 }>(&mut page, &mut file).await?; + id += 1; + page = GeneralPage { + header: GeneralHeader::new(id.into(), PageType::Data, 1.into()), + inner: DataPage::new(), + }; + } + page.inner.update_at( + Link { + page_id: id.into(), + offset: page.inner.length, + length: bytes.len() as u32, + }, + &bytes, + )?; + } + persist_page::<_, { PAGE_SIZE as u32 }>(&mut page, &mut file).await?; + info.inner.page_count = id; + persist_page::<_, { PAGE_SIZE as u32 }>(&mut info, &mut file).await?; + file.sync_all().await?; + println!("wrote {} rows across {} data pages", args.count, id); + Ok(()) + }) } diff --git a/tools/dump-data-file/Cargo.toml b/tools/dump-data-file/Cargo.toml index 26b6d3d..1ab8681 100644 --- a/tools/dump-data-file/Cargo.toml +++ b/tools/dump-data-file/Cargo.toml @@ -4,6 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] +nagoya = { version = "^0.1", features = ["std"] } clap = { version = "4.5.21", features = ["derive"] } data_bucket = { path = "../.." } eyre = "0.6.12" diff --git a/tools/dump-data-file/src/main.rs b/tools/dump-data-file/src/main.rs index b1c0bde..c79cb90 100644 --- a/tools/dump-data-file/src/main.rs +++ b/tools/dump-data-file/src/main.rs @@ -1,103 +1,67 @@ use clap::Parser; -use data_bucket::{ - page::{parse_space_info, DataIterator, LinksIterator, PageIterator}, - persistence::data::DataTypeValue, - read_data_pages, PAGE_SIZE, -}; -use std::{fs::File, str}; +use data_bucket::{parse_data_page, parse_general_header_by_index, PageType, GENERAL_HEADER_SIZE}; #[derive(Parser, Debug)] +#[command(about = "Inspect v3 page identities and live row extents without reading any index")] struct Args { #[arg(short, long)] filename: String, + #[arg(long, default_value_t = 16384)] + page_size: u32, + /// Include each live row archive as hexadecimal bytes. Row decoding belongs to the owning schema. + #[arg(long)] + hex: bool, } -fn print_horizontal_cells_delimiters(column_widths: &[usize]) { - print!("+"); - for column_width in column_widths.iter() { - print!("-"); - for _ in 0..*column_width { - print!("-"); +async fn dump(args: &Args) -> eyre::Result<()> { + let file = std::fs::File::open(&args.filename)?; + let length = file.metadata()?.len(); + let mut file = nagoya::io::HostFile::new(file); + let mut rows = 0usize; + for id in 0..length.div_ceil(u64::from(STRIDE)) { + let id = u32::try_from(id)?; + let header = parse_general_header_by_index::(&mut file, id).await?; + println!( + "page {}: {:?}, format {}, initialized {}", + id, header.page_type, header.data_version, header.data_length + ); + if header.page_type != PageType::Data { + continue; } - print!("-+"); - } - println!(); -} - -fn print_padded_string(string: &str, column_width: usize) { - print!("{}", string); - for _ in 0..column_width - string.len() { - print!(" "); - } -} - -fn format_table(header: &Vec, rows: &Vec>) { - let mut column_widths = vec![0; header.len()]; - for i in 0..header.len() { - column_widths[i] = header[i].len(); - } - for row in rows.iter() { - for i in 0..row.len() { - if row[i].len() > column_widths[i] { - column_widths[i] = row[i].len(); + let page = parse_data_page::(&mut file, id).await?; + for slot in page.inner.rows { + rows += 1; + print!( + " row offset={} length={} file_offset={}", + slot.offset, + slot.length, + u64::from(id) * u64::from(STRIDE) + + GENERAL_HEADER_SIZE as u64 + + u64::from(slot.offset) + ); + if args.hex { + print!(" bytes="); + for byte in &page.inner.data[slot.offset as usize..][..slot.length as usize] { + print!("{byte:02x}"); + } } + println!(); } } - - print_horizontal_cells_delimiters(&column_widths[..]); - print!("|"); - for i in 0..header.len() { - print!(" "); - print_padded_string(header[i].as_str(), column_widths[i]); - print!(" |"); - } - println!(); - print_horizontal_cells_delimiters(&column_widths[..]); - for row in rows.iter() { - print!("|"); - for i in 0..row.len() { - print!(" "); - print_padded_string(row[i].as_str(), column_widths[i]); - print!(" |"); - } - println!(); - } - print_horizontal_cells_delimiters(&column_widths[..]); + println!("live rows: {rows}"); + Ok(()) } fn main() -> eyre::Result<()> { let args = Args::parse(); - let mut file = File::open(args.filename)?; - - let space_info = parse_space_info::(&mut file)?; - let row_schema = space_info.row_schema.clone(); - - let mut rows: Vec> = vec![]; - - let pages = PageIterator::new(space_info.primary_key_intervals.clone()); - for page in pages { - let links = LinksIterator::new(&mut file, page, &space_info).collect::>(); - for row in DataIterator::new(&mut file, row_schema.clone(), links) { - rows.push(row?); + nagoya::block_on(async { + match args.page_size { + 512 => dump::<512, { 512 - GENERAL_HEADER_SIZE }>(&args).await, + 4096 => dump::<4096, { 4096 - GENERAL_HEADER_SIZE }>(&args).await, + 8192 => dump::<8192, { 8192 - GENERAL_HEADER_SIZE }>(&args).await, + 16384 => dump::<16384, { 16384 - GENERAL_HEADER_SIZE }>(&args).await, + 32768 => dump::<32768, { 32768 - GENERAL_HEADER_SIZE }>(&args).await, + _ => eyre::bail!("supported page sizes: 512, 4096, 8192, 16384, 32768"), } - } - - let rows: Vec> = read_data_pages::(&mut file)?; - - let header: Vec = row_schema - .iter() - .map(|(column, _data_type)| column.to_owned()) - .collect(); - let rows: Vec> = rows - .iter() - .map(|row| { - row.iter() - .map(|column| column.to_string()) - .collect::>() - }) - .collect(); - - format_table(&header, &rows); - - Ok(()) + }) }