diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 9755258..b3901b7 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -262,6 +262,41 @@ jobs: - name: cargo check --lib run: cargo check -p pagedb --target ${{ matrix.target }} --lib ${{ matrix.features }} + # ───────────────────────────────────────────────────────────────────────── + # Runtime wasm coverage. The `wasm` job above only compiles: a wall-clock + # read inside the txn layer compiles fine on wasm32 and panics at the first + # commit ("time not implemented on this platform"), which a check cannot see. + # This job opens, commits, and reads a store under node on + # `wasm32-unknown-unknown`: the target's own clock, and the pager's retry + # backoff, which has no Tokio time driver there. + wasm-tests: + name: WASM / node smoke (wasm32) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Install Rust + target + uses: dtolnay/rust-toolchain@4cda84d5c5c54efe2404f9d843567869ab1699d4 # stable + with: + targets: wasm32-unknown-unknown + + - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 + with: + prefix-key: wasm-smoke + + # The runner must match the wasm-bindgen version the lockfile resolves, + # so read it from `cargo metadata` instead of pinning a number here. + - name: Install wasm-bindgen-test-runner + run: | + version=$(cargo metadata --format-version 1 --locked \ + | python3 -c "import json,sys; d=json.load(sys.stdin); print(next(p['version'] for p in d['packages'] if p['name']=='wasm-bindgen'))") + cargo install wasm-bindgen-cli --version "$version" --locked + + - name: Run wasm smoke tests (node) + env: + CARGO_TARGET_WASM32_UNKNOWN_UNKNOWN_RUNNER: wasm-bindgen-test-runner + run: cargo test -p pagedb-wasm-smoke --target wasm32-unknown-unknown --test commit_smoke + # ───────────────────────────────────────────────────────────────────────── features: name: Feature matrix (${{ matrix.flags }}) diff --git a/Cargo.lock b/Cargo.lock index 2ba4527..5fb991b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1278,6 +1278,15 @@ dependencies = [ "tokio", ] +[[package]] +name = "pagedb-wasm-smoke" +version = "0.0.0" +dependencies = [ + "pagedb", + "tokio", + "wasm-bindgen-test", +] + [[package]] name = "parking_lot" version = "0.12.5" diff --git a/Cargo.toml b/Cargo.toml index 113dfce..cb37cc2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["benchmarks/engine-comparison"] +members = ["benchmarks/engine-comparison", "wasm-smoke"] default-members = ["."] resolver = "3" diff --git a/src/btree/tree/core.rs b/src/btree/tree/core.rs index daa85a8..81bf698 100644 --- a/src/btree/tree/core.rs +++ b/src/btree/tree/core.rs @@ -565,7 +565,7 @@ mod tests { let mut tree = fresh_tree(None).await; tree.free_page(9); assert_eq!(tree.allocate_page(), 9); - assert!(tree.drain_freed().is_empty()); + assert_eq!(tree.drain_freed(), [] as [u64; 0]); } /// Eligibility is settled at the moment of the free and does not change diff --git a/src/pager/core.rs b/src/pager/core.rs index d11b264..0046bef 100644 --- a/src/pager/core.rs +++ b/src/pager/core.rs @@ -1028,7 +1028,8 @@ impl Pager { let file_handle = self.open_file_handle(file).await?; // Observer-mode retry loop: on AEAD failure retry up to - // `observer_retry_count` times (10 ms backoff) to absorb torn reads + // `observer_retry_count` times (10 ms backoff, a yield where no time + // driver exists; see the per-target split below) to absorb torn reads // from a concurrent writer. In non-observer mode (retry_count == 0) // the loop body executes exactly once and any AEAD failure is a hard // corruption signal. @@ -1044,7 +1045,17 @@ impl Pager { let mut last_envelope: Option = None; for attempt in 0..max_attempts { if attempt > 0 { + // Yield the executor on `wasm32-unknown-unknown` rather than + // sleeping: this loop runs on an embedder's single-threaded + // executor, which has no Tokio time driver, and + // `tokio::time::sleep` panics without one. That panic would + // replace the corruption this loop exists to report with a + // crash. Every other target sleeps, so a torn read still gets + // its backoff. + #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] tokio::time::sleep(std::time::Duration::from_millis(10)).await; + #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] + tokio::task::yield_now().await; } let mut buf = vec![0u8; page_size]; { diff --git a/src/recovery/provenance/probe.rs b/src/recovery/provenance/probe.rs index e66dd34..64b35d0 100644 --- a/src/recovery/provenance/probe.rs +++ b/src/recovery/provenance/probe.rs @@ -327,7 +327,7 @@ mod tests { let provenance = db.page_provenance(page_id).await.unwrap(); assert_eq!(provenance.standing, PageStanding::Free); assert_eq!(provenance.freed_by_commit, Some(commit_id)); - assert!(provenance.reachable_from.is_empty()); + assert_eq!(provenance.reachable_from, [] as [&str; 0]); assert!(!provenance.is_double_owned()); } diff --git a/src/txn/db/catalog/history.rs b/src/txn/db/catalog/history.rs index 443cba6..2bf32ae 100644 --- a/src/txn/db/catalog/history.rs +++ b/src/txn/db/catalog/history.rs @@ -454,4 +454,53 @@ mod tests { assert!(matches!(err, PagedbError::Corruption(_))); } } + + /// Age retention must prune, which it can only do against a real clock. + /// + /// The threshold is `now - duration`, so a clock that answers `0` makes + /// every threshold `0`: no recorded timestamp is older than it, the walk + /// ends at its first row, and `Age` behaves as `Unbounded` while still + /// reporting itself as age-based. Nothing in the API reports that absence, + /// the store simply keeps everything. + /// + /// `Age(ZERO)` prunes every entry older than the current second, so the wait + /// only has to cross one second boundary, and a commit never prunes the row + /// it just inserted. + #[tokio::test(flavor = "current_thread")] + async fn age_retention_prunes_entries_older_than_its_threshold() { + use std::time::Duration; + + let db = Db::open_internal_with_options( + MemVfs::new(), + [7u8; 32], + PAGE, + REALM, + OpenOptions::default().with_commit_history_retain(RetainPolicy::Age(Duration::ZERO)), + ) + .await + .unwrap(); + + let oldest = db.begin_write().await.unwrap().commit().await.unwrap(); + assert!( + db.begin_read_at(oldest).await.is_ok(), + "the only commit so far must be readable" + ); + + // Integer-second timestamps: 1.2 s crosses a boundary whichever + // fraction of a second the first commit landed on. + std::thread::sleep(Duration::from_millis(1200)); + let newest = db.begin_write().await.unwrap().commit().await.unwrap(); + + assert!( + db.begin_read_at(newest).await.is_ok(), + "a commit must not prune the row it inserts" + ); + let Err(pruned) = db.begin_read_at(oldest).await else { + panic!("an entry older than the age threshold must be pruned"); + }; + assert!( + matches!(pruned, PagedbError::CommitGone { .. }), + "a pruned commit is gone, not corruption and not a stall: {pruned:?}" + ); + } } diff --git a/src/vfs/native.rs b/src/vfs/native.rs index f5ea787..73ee7f9 100644 --- a/src/vfs/native.rs +++ b/src/vfs/native.rs @@ -299,7 +299,7 @@ mod tests { assert_eq!(vfs.list_dir("/seg").await.unwrap(), vec!["renamed"]); vfs.sync_dir("/seg").await.unwrap(); vfs.remove("/seg/renamed").await.unwrap(); - assert!(vfs.list_dir("/seg").await.unwrap().is_empty()); + assert_eq!(vfs.list_dir("/seg").await.unwrap(), [] as [String; 0]); std::fs::remove_dir_all(&dir).ok(); } diff --git a/tests/durability/unpublished_commit.rs b/tests/durability/unpublished_commit.rs index 57687ac..b5e60ff 100644 --- a/tests/durability/unpublished_commit.rs +++ b/tests/durability/unpublished_commit.rs @@ -1,7 +1,7 @@ //! A failed post-header segment reconciliation poisons only the active handle. use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, Ordering}; use pagedb::vfs::memory::{MemFile, MemLockHandle, MemVfs}; use pagedb::vfs::{OpenMode, Vfs}; @@ -16,7 +16,7 @@ struct RenameFaultVfs { inner: MemVfs, fail_renames: Arc, fail_sync_dirs: Arc, - failures_remaining: Arc, + fail_rename_once: Arc, } impl RenameFaultVfs { @@ -25,7 +25,7 @@ impl RenameFaultVfs { inner: MemVfs::new(), fail_renames: Arc::new(AtomicBool::new(false)), fail_sync_dirs: Arc::new(AtomicBool::new(false)), - failures_remaining: Arc::new(AtomicUsize::new(0)), + fail_rename_once: Arc::new(AtomicBool::new(false)), } } @@ -34,7 +34,7 @@ impl RenameFaultVfs { } fn fail_next_rename(&self) { - self.failures_remaining.store(1, Ordering::SeqCst); + self.fail_rename_once.store(true, Ordering::SeqCst); } fn fail_sync_dirs(&self, fail: bool) { @@ -55,12 +55,7 @@ impl Vfs for RenameFaultVfs { } async fn rename(&self, from: &str, to: &str) -> pagedb::Result<()> { - let fail_once = self - .failures_remaining - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { - remaining.checked_sub(1) - }) - .is_ok(); + let fail_once = self.fail_rename_once.swap(false, Ordering::SeqCst); if self.fail_renames.load(Ordering::SeqCst) || fail_once { return Err(PagedbError::Io(std::io::Error::other( "injected segment reconciliation failure", diff --git a/wasm-smoke/Cargo.toml b/wasm-smoke/Cargo.toml new file mode 100644 index 0000000..b6eb3e9 --- /dev/null +++ b/wasm-smoke/Cargo.toml @@ -0,0 +1,19 @@ +# wasm32 smoke tests for pagedb. +# +# Separate crate on purpose: the pagedb dev-dependencies (tokio +# rt-multi-thread, tempfile) do not compile for wasm32, and Cargo builds every +# dev-dependency for a crate's test targets. This crate depends on pagedb with +# the `opfs` feature only. +[package] +name = "pagedb-wasm-smoke" +version = "0.0.0" +edition = "2024" +publish = false + +[lib] +path = "src/lib.rs" + +[target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies] +wasm-bindgen-test = "0.3" +tokio = { version = "1", features = ["rt", "macros", "sync", "io-util", "time"] } +pagedb = { path = "..", features = ["opfs"] } diff --git a/wasm-smoke/src/lib.rs b/wasm-smoke/src/lib.rs new file mode 100644 index 0000000..a596e85 --- /dev/null +++ b/wasm-smoke/src/lib.rs @@ -0,0 +1 @@ +//! wasm32 smoke-test crate; see tests/. diff --git a/wasm-smoke/tests/commit_smoke.rs b/wasm-smoke/tests/commit_smoke.rs new file mode 100644 index 0000000..84febb8 --- /dev/null +++ b/wasm-smoke/tests/commit_smoke.rs @@ -0,0 +1,164 @@ +//! wasm32 commit smoke tests, on the build an embedder actually ships. +//! +//! The invariants these protect, all on `wasm32-unknown-unknown`: +//! +//! - Nothing on the commit path may call `SystemTime::now()`, which panics with +//! "time not implemented on this platform". The clock is reached only through +//! `clock::unix_seconds()`. +//! - A corrupted page must be reported as `PagedbError::Corruption`, not turned +//! into a panic by the read loop's backoff. +//! +//! This crate depends on `pagedb` with the `opfs` feature, the build an +//! embedded consumer ships, and pins the succeeding path end to end: open, +//! commit, and commit under age retention. The clock those paths read is +//! `clock::unix_seconds()`, which takes the host clock from `web-time` on +//! `wasm32-unknown-unknown` and from std elsewhere, so age retention works on +//! every wasm build rather than only the `opfs` one. The runtime is built +//! without a Tokio time driver on purpose, so the corrupted-page test exercises +//! the retry loop's `yield_now` backoff for real. +//! +//! Not asserted anywhere: the pruned row count. Neither `Db` nor `DbStats` +//! exposes the commit history, and these tests see only the public API. +//! +//! Run with `wasm-pack test --node wasm-smoke` or the CI job's +//! `wasm-bindgen-test-runner` invocation. + +#![cfg(all(target_arch = "wasm32", target_os = "unknown"))] + +use pagedb::vfs::memory::MemVfs; +use pagedb::vfs::{OpenMode, Vfs, VfsFile}; +use pagedb::{Db, OpenOptions, PagedbError, RealmId, RetainPolicy}; +use wasm_bindgen_test::*; + +wasm_bindgen_test_configure!(run_in_node_experimental); + +const PAGE: usize = 4096; +const KEK: [u8; 32] = [7u8; 32]; +const REALM: RealmId = RealmId::new([1u8; 16]); + +/// A current-thread runtime keeps the async plumbing identical to native. +/// The memory VFS never yields to the JS event loop, so `block_on` completes +/// without blocking a host callback. +/// +/// Deliberately built **without** `enable_time()`. An embedder driving these +/// futures through `wasm-bindgen-futures` has no Tokio runtime at all, so a +/// test that quietly depended on a time driver would pass here and panic +/// there. The page-read retry loop yields instead of sleeping for the same +/// reason. +fn block_on(future: F) -> F::Output { + tokio::runtime::Builder::new_current_thread() + .build() + .expect("current-thread runtime") + .block_on(future) +} + +async fn commit_once(policy: &RetainPolicy) -> pagedb::Result<()> { + let opts = OpenOptions::default().with_commit_history_retain(policy.clone()); + let db = Db::open(MemVfs::new(), KEK, PAGE, REALM, opts).await?; + let mut w = db.begin_write().await.expect("begin_write"); + w.put(b"k", b"v").await.expect("put"); + w.commit().await.expect("commit"); + Ok(()) +} + +/// Opening and committing must not reach the std clock on this target. +#[wasm_bindgen_test] +fn commit_does_not_read_the_std_clock() { + block_on(async { + commit_once(&RetainPolicy::Unbounded) + .await + .expect("Unbounded consults no clock, so it must open and commit"); + }); +} + +/// Every clock-free policy must clear a commit, not just the one a default +/// configuration happens to use. +#[wasm_bindgen_test] +fn commit_clears_every_clock_free_policy() { + block_on(async { + for policy in [RetainPolicy::Unbounded, RetainPolicy::Count(4)] { + commit_once(&policy) + .await + .unwrap_or_else(|e| panic!("a clock-free policy must commit: {e}")); + } + }); +} + +/// Age retention reads the clock on every commit — its pruning threshold is +/// `now - duration` — so it is the policy whose commitment to a wall clock has +/// to be met on the build an embedder ships, not just on the one the default +/// configuration happens to use. +#[wasm_bindgen_test] +fn commit_under_age_retention_does_not_read_the_std_clock() { + block_on(async { + commit_once(&RetainPolicy::Age(std::time::Duration::from_secs(60))) + .await + .expect("a build with a clock must serve age retention"); + }); +} + +/// Corrupted pages must be reported, not panicked on. +/// +/// This is the case the page-read retry loop exists for: a read-only handle +/// retries an AEAD failure `observer_retry_count` times before reporting +/// corruption, and the backoff between those attempts is the one platform +/// split in this change — `tokio::time::sleep` where a time driver exists, +/// `yield_now` on `wasm32-unknown-unknown`, where an embedder's executor has +/// none and sleeping would panic with "time not implemented on this platform". +/// A panic here would replace a reportable `Corruption` with a crash, so this +/// test is what makes that split load-bearing rather than cosmetic. +#[wasm_bindgen_test] +fn a_corrupted_page_is_reported_as_corruption_not_a_panic() { + block_on(async { + let vfs = MemVfs::new(); + let db = Db::open(vfs.clone(), KEK, PAGE, REALM, OpenOptions::default()) + .await + .expect("open"); + let mut w = db.begin_write().await.expect("begin_write"); + w.put(b"k", b"v").await.expect("put"); + w.commit().await.expect("commit"); + drop(db); + + // Pages 0 and 1 hold the A/B header slots, whose MAC covers the slot; + // corrupting those would fail *open* with a header error instead of + // exercising a verified page read. Every page after them is flipped, so + // whichever page the read path reaches is corrupt. The flip lands + // mid-page: the page's first bytes carry its cleartext cipher id, and + // overwriting that reports `Unsupported` before the authenticated read + // ever fails. + let mut file = vfs + .open("/main.db", OpenMode::ReadWrite) + .await + .expect("raw open"); + let len = file.len().await.expect("len") as usize; + let mut byte = [0u8; 1]; + let mut offset = 2 * PAGE; + while offset + PAGE <= len { + let mid = (offset + PAGE / 2) as u64; + file.read_at(mid, &mut byte).await.expect("read"); + byte[0] ^= 0xFF; + file.write_at(mid, &byte).await.expect("write"); + offset += PAGE; + } + assert!( + offset > 2 * PAGE, + "the store must have pages past the two header slots" + ); + + // `Observer` is the mode whose retry budget turns this into four + // attempts on the corrupted page; `open_observer` on defaults is what a + // reader on a live store does. + let outcome: pagedb::Result<()> = async { + let db = Db::open_observer(vfs, KEK, PAGE, REALM, OpenOptions::default()).await?; + let txn = db.begin_read().await?; + txn.get(b"k").await.map(|_| ()) + } + .await; + + let err = outcome.expect_err("a corrupted page must not read cleanly"); + assert!( + matches!(err, PagedbError::Corruption(_)), + "a corrupted page must be reported as Corruption, got {err:?}" + ); + }); +}