diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 85a6bac0b..b8d7eb230 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,6 +10,12 @@ # - Actions → CI → Run workflow, against any branch. # # While the label is on, every subsequent push to the PR re-runs the suite. +# +# A nightly schedule probes the same suites against `main` itself. Nothing +# else reports a red `main`: the label gate keeps upstream churn out of PR +# runs, so without this the first run to notice a broken main is the next PR +# a maintainer labels. GitHub sends failure notifications for a scheduled run +# to whoever last edited the cron line. name: CI @@ -18,6 +24,8 @@ on: branches: [main] types: [labeled, synchronize, reopened, ready_for_review] workflow_dispatch: + schedule: + - cron: "0 4 * * *" # 04:00 UTC daily; staggered from NodeDB-Lite's 03:00 concurrency: group: ci-${{ github.ref }} diff --git a/.github/workflows/static-gates.yml b/.github/workflows/static-gates.yml index 610ba7f82..ea24e1b5c 100644 --- a/.github/workflows/static-gates.yml +++ b/.github/workflows/static-gates.yml @@ -25,6 +25,10 @@ jobs: timeout-minutes: 10 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + - name: Scheduled-main and wasm-probe gate + run: | + python3 scripts/ci/check_wasm_probe.py --self-test + python3 scripts/ci/check_wasm_probe.py - name: Warm-storage ObjectStore gate run: bash scripts/ci/check_warm_storage_gate.sh - name: Format-path .expect() gate diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 2a8aae67e..81a3345c4 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -56,6 +56,10 @@ jobs: run: cargo fmt --all -- --check - name: Run clippy run: cargo clippy --workspace --all-targets --all-features --profile ci -- -D warnings + - name: Scheduled-main and wasm-probe gate + run: | + python3 scripts/ci/check_wasm_probe.py --self-test + python3 scripts/ci/check_wasm_probe.py - name: Warm-storage ObjectStore gate run: bash scripts/ci/check_warm_storage_gate.sh - name: Format-path .expect() gate @@ -93,6 +97,87 @@ jobs: - name: Dependency audit run: cargo deny check + # Compile *and run* the shared crates for a wasm target. The server, cluster + # and raft crates are not wasm targets; these are the crates NodeDB-Lite + # compiles into its browser/wasm build, so a wasm-incompatible dependency or a + # target-specific cfg mistake fails here instead of in the Lite repo. + # + # A check is not enough, which is the whole point of the job: wasm-only code + # that compiles and then traps at run time is invisible to `cargo check`. The + # crates carry `#[cfg(target_arch = "wasm32")]` arms — the ruzstd decoder in + # `nodedb-codec`, the wasm SIMD filter and aggregate kernels in `nodedb-query`, + # the raw-IO shims in `nodedb-wal` — and until this job existed none of them + # was ever executed anywhere. `wasmtime` runs the test binaries, so a panic in + # one of those arms fails the run. + # + # `wasm32-wasip1` on purpose: `wasm32-unknown-unknown` needs a getrandom + # backend the workspace does not select, so the job would fail on the random + # source rather than on the code under change, and it has no way to execute a + # libtest binary. + wasm: + name: WASM tests (wasip1) + runs-on: ubuntu-24.04-arm + env: + # Mirrors the `RUST_MIN_STACK` ceiling the native test job sets: these are + # unoptimized `ci`-profile test binaries, and wasmtime's own default + # (512 KiB) is far below what tokio's and libtest's frames need. Without + # it the run dies on a stack overflow rather than on a test failure. + CARGO_TARGET_WASM32_WASIP1_RUNNER: wasmtime -W max-wasm-stack=33554432 --dir=. + # Clear the workflow-level `RUSTFLAGS`, which points the linker at mold + # (`-C link-arg=-fuse-ld=mold`). Mold is a native linker; this target links + # through the `wasm32-wasip1` spec with `rust-lld`, which rejects that + # argument outright — so every test binary that has to *link* fails with + # "rust-lld: error: unknown argument: -fuse-ld=mold". The compile-only job + # this replaces never linked, which is why it went unnoticed until tests + # were actually executed here. + # + # This has to be `RUSTFLAGS` and not `CARGO_TARGET_WASM32_WASIP1_RUSTFLAGS`: + # cargo gives the un-prefixed variable precedence, verified by building a + # scratch crate with both set — the mold argument still reached rustc. A + # job-level `env` overrides the workflow-level one, so the native jobs keep + # mold and this job gets a clean linker command. + RUSTFLAGS: "" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + + - name: Install Rust + wasm target + uses: dtolnay/rust-toolchain@4cda84d5c5c54efe2404f9d843567869ab1699d4 # stable + with: + targets: wasm32-wasip1 + + - name: Install wasmtime + # v1 of the collection, pinned like every other action here. The version + # is explicit rather than `latest`: the runner's default stack size is + # part of what the command above depends on. + uses: bytecodealliance/actions/wasmtime/setup@9152e710e9f7182e4c29ad218e4f335a7b203613 # v1 + with: + version: "35.0.0" + + - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + with: + prefix-key: wasm-wasip1 + + - name: Run tests (shared crates) + # `top1_recall_on_training_set` is the one test skipped by name. It + # asserts an approximate-recall floor of 0.70 on a ten-vector synthetic + # set; it measures 0.60 on this target and passes natively. The training + # path is deliberately seeded, so the gap is float behaviour on a + # different codegen path, not a random draw — and the dataset is too + # small for the assertion to survive it. It is skipped rather than + # weakened, and it still runs on every PR in the native suite. + # + # nodedb-codec's other native-only tests — the eight that roundtrip + # through `encode()` or drive `ZstdEncoder` — are ignored in-source by + # `#[cfg_attr(target_arch = "wasm32", ignore = ...)]`, because there the + # reason is categorical: this target has no zstd encoder. + run: | + cargo test --profile ci --target wasm32-wasip1 \ + -p nodedb-types -p nodedb-codec -p nodedb-physical -p nodedb-mem \ + -p nodedb-crdt -p nodedb-columnar -p nodedb-fts -p nodedb-graph \ + -p nodedb-spatial -p nodedb-strict -p nodedb-sql -p nodedb-query \ + -p nodedb-array -p nodedb-vector -p nodedb-client -p nodedb-wal \ + -- --skip top1_recall + # Single test job: runs the fast suite first, then the 3-node cluster # suite (`nodedb-cluster-tests`) on the same runner. Combining them # avoids a second cold compile of the whole workspace; the cluster diff --git a/Cargo.lock b/Cargo.lock index c47be165a..f09eae813 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4683,6 +4683,7 @@ dependencies = [ "bytemuck", "crc32c", "getrandom 0.4.3", + "js-sys", "nanoid", "nodedb-codec", "rand 0.10.2", diff --git a/docs/wasm.md b/docs/wasm.md index 4a718edfc..8ea638ca3 100644 --- a/docs/wasm.md +++ b/docs/wasm.md @@ -4,7 +4,10 @@ NodeDB-Lite compiles to WebAssembly and exposes the same `NodeDb` trait you use in native Lite. To talk to an Origin cluster from the browser, use Lite-WASM locally and replicate via CRDT sync over WebSocket — never run Origin in the browser. -**Status: Experimental.** Lite-WASM support is feature-complete for all eight engines. Testing and CI integration are ongoing; treat the build as preview-quality. Report issues via GitHub. +**Status: Experimental.** Lite-WASM support is feature-complete for all eight engines. The shared +crates are compiled and their test suites executed for a WASI target in CI, so a target-specific +build or runtime failure is caught here rather than in the Lite repo; treat the build as +preview-quality and report issues via GitHub. ## Building for WASM @@ -198,6 +201,29 @@ See [NodeDB-Lite](https://github.com/NodeDB-Lab/nodedb-lite) for full CRDT sync - **Module size** — measure your own build; gzip before serving - **Browser compatibility** — Requires WebAssembly support (all modern browsers + Node.js 14+) +### The WAL on a WASI target + +The shared crates are compiled and their test suites executed for `wasm32-wasip1` in CI, so the +coverage below is measured rather than assumed. Three guarantees the native WAL provides are weaker +or absent there: + +- **Directory fsync does not exist.** WASI preview 1 has no directory fsync — `File::sync_all()` on a + directory returns `EBADF`, and there is no weaker syscall with the same effect. File creation, + rename and file fsync all still work; what is lost is the guarantee that a rename survives a host + crash. `fsync_directory` is a documented no-op on this target. +- **Positional writes do not exist.** There is no `pwrite`, so the writer seeks to its offset and + writes sequentially instead. This is equivalent for an append-only log, which is what the WAL is. +- **The double-write buffer is unavailable.** It depends on positional reads as well as writes, and + `recover_record` is a stub on this target, so a slot written here could never be read back. + `raw_io::pwrite_all` therefore returns `WalError::Unsupported` for DWB Direct mode rather than + falling back to a seek-and-write: a fallback would mirror the slot, let the writer report + `DwbProtection::Active`, and deliver no protection at all. As it stands the append still succeeds — + DWB failures are deliberately not fatal to the WAL — while `dwb_protection()` reports + `Degraded(WriteFailed)` and every record is counted as unprotected. + +`std::thread::spawn` is also unavailable, so any WAL coordination that spawns a thread is +native-only. + ## Deployment ### Browser (Web App) diff --git a/nodedb-array/src/sync/hlc.rs b/nodedb-array/src/sync/hlc.rs index 665985fe9..7635f13fd 100644 --- a/nodedb-array/src/sync/hlc.rs +++ b/nodedb-array/src/sync/hlc.rs @@ -13,7 +13,6 @@ //! spec's u48 intent while keeping byte alignment simple. use std::sync::Mutex; -use std::time::{SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; @@ -165,11 +164,10 @@ impl HlcGenerator { /// Return the current wall-clock milliseconds since Unix epoch. fn now_ms() -> ArrayResult { - SystemTime::now() - .duration_since(UNIX_EPOCH) + nodedb_types::clock::since_epoch() .map(|d| d.as_millis() as u64) - .map_err(|e| ArrayError::InvalidHlc { - detail: format!("system clock before Unix epoch: {e}"), + .ok_or_else(|| ArrayError::InvalidHlc { + detail: "system clock before Unix epoch".to_owned(), }) } diff --git a/nodedb-client/Cargo.toml b/nodedb-client/Cargo.toml index e036aca22..5ab230c83 100644 --- a/nodedb-client/Cargo.toml +++ b/nodedb-client/Cargo.toml @@ -36,4 +36,23 @@ tokio-rustls = { workspace = true, optional = true } rustls-pemfile = { workspace = true, optional = true } [dev-dependencies] -tokio = { workspace = true, features = ["rt", "macros"] } +# The only default-feature tokio use is `#[tokio::test]` in +# src/traits/core/trait_def.rs (8 tests, default current-thread flavour), which +# needs `macros` + `rt`. This narrow set is also the only shape that is legal on +# wasm: `tokio = { workspace = true }` inherits `features = ["full"]`, whose +# fs/io-std/net/process/rt-multi-thread/signal arms trip tokio's wasm +# compile_error. The native/remote feature-gated modules do reach +# net/spawn/io-util/time/sync, but enabling those features re-enables the +# optional production `tokio` dependency below, which carries `full` again. +tokio = { version = "1", default-features = false, features = [ + "io-util", + "macros", + "rt", + "sync", + "time", +] } + +# Restores the pre-split native resolution exactly: off wasm the union of this +# entry and the narrow one above is `full`, as it was before. +[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] +tokio = { workspace = true } diff --git a/nodedb-codec/src/bounds.rs b/nodedb-codec/src/bounds.rs index 278d0d525..d6e0a7f53 100644 --- a/nodedb-codec/src/bounds.rs +++ b/nodedb-codec/src/bounds.rs @@ -63,13 +63,27 @@ pub(crate) fn decoded_len(value: usize, codec: &str) -> Result Result { - let bytes = checked_mul(count, element_size, context)?; - decoded_len(bytes, context)?; + let bytes = (count as u64).saturating_mul(element_size as u64); + if bytes > MAX_DECODED_BYTES as u64 { + return Err(CodecError::ResourceLimit { + resource: format!("{context} decoded bytes"), + requested: usize::try_from(bytes).unwrap_or(usize::MAX), + limit: MAX_DECODED_BYTES, + }); + } Ok(count) } @@ -102,8 +116,21 @@ mod tests { } #[test] - fn checked_capacity_rejects_usize_overflow() { - let error = checked_capacity(usize::MAX, 2, "test"); - assert!(matches!(error, Err(CodecError::Corrupt { .. }))); + fn checked_capacity_rejects_oversized_element_count() { + // `usize::MAX` elements of eight bytes each. Multiplying those in + // `usize` overflows, and on a 32-bit target even `u32::MAX` elements + // do; the classification is the resource limit either way, on every + // target the workspace builds for. + let error = checked_capacity(usize::MAX, 8, "test"); + assert!(matches!(error, Err(CodecError::ResourceLimit { .. }))); + } + + #[test] + fn checked_capacity_rejects_wide_element_count_on_every_target() { + // The 32-bit case in the large: `u32::MAX` elements of eight bytes is + // 32 GiB at an element size that cannot be expressed in a 32-bit + // `usize`. + let error = checked_capacity(u32::MAX as usize, 8, "test"); + assert!(matches!(error, Err(CodecError::ResourceLimit { .. }))); } } diff --git a/nodedb-codec/src/delta.rs b/nodedb-codec/src/delta.rs index 86f3381cc..a19b3e43a 100644 --- a/nodedb-codec/src/delta.rs +++ b/nodedb-codec/src/delta.rs @@ -20,9 +20,7 @@ use std::mem::size_of; -use crate::bounds::{ - checked_add, checked_capacity, checked_mul, decoded_len, encode_input_len, u32_to_usize, -}; +use crate::bounds::{checked_add, checked_capacity, checked_mul, encode_input_len, u32_to_usize}; use crate::error::CodecError; // --------------------------------------------------------------------------- @@ -198,8 +196,10 @@ fn encode_value_count(len: usize, codec: &str) -> Result { } fn validate_value_count(count: usize, codec: &str) -> Result<(), CodecError> { - let bytes = checked_mul(count, size_of::(), "integer decoded bytes")?; - decoded_len(bytes, codec)?; + // See `checked_capacity`: the byte count is scaled in 64-bit arithmetic so + // a 32-bit target reports the resource limit rather than a `usize` + // overflow for a count it simply cannot express. + checked_capacity(count, size_of::(), codec)?; Ok(()) } diff --git a/nodedb-codec/src/double_delta.rs b/nodedb-codec/src/double_delta.rs index e42cec162..a507c38b1 100644 --- a/nodedb-codec/src/double_delta.rs +++ b/nodedb-codec/src/double_delta.rs @@ -31,9 +31,7 @@ use std::mem::size_of; -use crate::bounds::{ - checked_add, checked_capacity, checked_mul, decoded_len, encode_input_len, u32_to_usize, -}; +use crate::bounds::{checked_add, checked_capacity, encode_input_len, u32_to_usize}; use crate::error::CodecError; // --------------------------------------------------------------------------- @@ -334,8 +332,10 @@ fn encode_value_count(len: usize, codec: &str) -> Result { } fn validate_value_count(count: usize, codec: &str) -> Result<(), CodecError> { - let bytes = checked_mul(count, size_of::(), "integer decoded bytes")?; - decoded_len(bytes, codec)?; + // See `checked_capacity`: the byte count is scaled in 64-bit arithmetic so + // a 32-bit target reports the resource limit rather than a `usize` + // overflow for a count it simply cannot express. + checked_capacity(count, size_of::(), codec)?; Ok(()) } diff --git a/nodedb-codec/src/fastlanes/header.rs b/nodedb-codec/src/fastlanes/header.rs index bb5079a45..54b0f3303 100644 --- a/nodedb-codec/src/fastlanes/header.rs +++ b/nodedb-codec/src/fastlanes/header.rs @@ -4,7 +4,7 @@ use std::mem::size_of; -use crate::bounds::{checked_mul, checked_range, decoded_len, u32_to_usize}; +use crate::bounds::{checked_capacity, checked_mul, checked_range, u32_to_usize}; use crate::error::CodecError; use super::block::skip_block; @@ -23,8 +23,9 @@ pub(super) fn parse_header(data: &[u8]) -> Result<(usize, usize), CodecError> { u32::from_le_bytes([header[0], header[1], header[2], header[3]]), "FastLanes value count", )?; - let decoded_bytes = checked_mul(total_count, size_of::(), "FastLanes decoded bytes")?; - decoded_len(decoded_bytes, "FastLanes")?; + // Rejects a hostile count before it is used for block math; see + // `checked_capacity` for why the scaling is done in 64-bit arithmetic. + checked_capacity(total_count, size_of::(), "FastLanes")?; let block_count = usize::from(u16::from_le_bytes([header[4], header[5]])); let expected_blocks = total_count.div_ceil(BLOCK_SIZE); if block_count != expected_blocks { diff --git a/nodedb-codec/src/zstd_codec.rs b/nodedb-codec/src/zstd_codec.rs index d97222e02..48d8eb7ff 100644 --- a/nodedb-codec/src/zstd_codec.rs +++ b/nodedb-codec/src/zstd_codec.rs @@ -431,6 +431,20 @@ impl ZstdDecoder { mod tests { use super::*; + // Zstd *encoding* does not exist on wasm: `compress_native` returns + // `CompressFailed` there (the target decodes with `ruzstd` and encodes with + // LZ4). Every test below that roundtrips through `encode`, or that drives + // `ZstdEncoder`, asserts a property of an encoder this target does not have, + // so on wasm32 it is ignored by name and by reason rather than weakened. + // The decoder-only tests -- the hostile-frame and truncation checks -- carry + // no such attribute and still run on wasm, because decoding is the half the + // target does implement. + const _: () = (); + + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn empty_data() { let encoded = encode(&[]).unwrap(); @@ -438,6 +452,10 @@ mod tests { assert!(decoded.is_empty()); } + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn small_data_roundtrip() { let data = b"hello world, zstd compression test"; @@ -446,6 +464,10 @@ mod tests { assert_eq!(decoded, data); } + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn large_data_roundtrip() { let line = "2024-01-15 ERROR database connection timeout host=db-prod-01 retry=3\n"; @@ -461,6 +483,10 @@ mod tests { ); } + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn high_compression_level() { let data: Vec = (0..10_000).map(|i| (i % 256) as u8).collect(); @@ -475,6 +501,10 @@ mod tests { assert_eq!(decode(&high_encoded).unwrap(), data); } + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn header_metadata() { let data = vec![42u8; 1000]; @@ -484,6 +514,10 @@ mod tests { assert_eq!(compression_level(&encoded).unwrap(), 7); } + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn better_ratio_than_lz4() { // Structured data where Zstd should beat LZ4. @@ -551,6 +585,10 @@ mod tests { )); } + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn streaming_encoder() { let parts: Vec<&[u8]> = vec![b"part one ", b"part two ", b"part three"]; @@ -584,6 +622,10 @@ mod tests { assert_eq!(encoder.len(), MAX_DECODED_BYTES); } + #[cfg_attr( + target_arch = "wasm32", + ignore = "zstd encoding is not available on wasm; decode-only target" + )] #[test] fn level_clamping() { let data = b"test data for clamping"; diff --git a/nodedb-crdt/Cargo.toml b/nodedb-crdt/Cargo.toml index e5e100bf7..a3bbbd806 100644 --- a/nodedb-crdt/Cargo.toml +++ b/nodedb-crdt/Cargo.toml @@ -27,5 +27,10 @@ subtle = { workspace = true } getrandom = { workspace = true } [dev-dependencies] -tokio = { workspace = true } serde_json = { workspace = true } + +# Native-only: the workspace `tokio` entry inherits `features = ["full"]`, which +# tokio's wasm guard rejects. This crate has no tokio call sites (the dev-dep was +# unused), so native test builds keep it unchanged and wasm drops it entirely. +[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] +tokio = { workspace = true } diff --git a/nodedb-crdt/src/dead_letter.rs b/nodedb-crdt/src/dead_letter.rs index 7d29f55ab..146f49301 100644 --- a/nodedb-crdt/src/dead_letter.rs +++ b/nodedb-crdt/src/dead_letter.rs @@ -17,7 +17,6 @@ //! 3. A machine-readable `CompensationHint` suggesting how to fix it. use std::collections::VecDeque; -use std::time::{SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; @@ -91,8 +90,9 @@ pub struct DeadLetter { /// Timestamp when the rejection occurred (unix millis). /// - /// Node-local and NON-DETERMINISTIC: sourced from `SystemTime::now()` at - /// enqueue time, so two replicas that reject the same delta will record + /// Node-local and NON-DETERMINISTIC: sourced from + /// `nodedb_types::clock::since_epoch()` at enqueue time, so two replicas + /// that reject the same delta will record /// different values. Safe today because the DLQ is a per-replica in-memory /// structure that is never replicated or compared across nodes. If the DLQ /// is ever replicated, snapshotted into shared state, or diffed between @@ -158,8 +158,7 @@ impl DeadLetterQueue { let id = self.next_id; self.next_id += 1; - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) + let now = nodedb_types::clock::since_epoch() .unwrap_or_default() .as_millis() as u64; diff --git a/nodedb-crdt/src/validator/policy_dispatch.rs b/nodedb-crdt/src/validator/policy_dispatch.rs index cfbde7a0a..8094e6645 100644 --- a/nodedb-crdt/src/validator/policy_dispatch.rs +++ b/nodedb-crdt/src/validator/policy_dispatch.rs @@ -121,8 +121,7 @@ impl Validator { max_retries, ttl_secs, } => { - let now_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) + let now_ms = nodedb_types::clock::since_epoch() .unwrap_or_default() .as_millis() as u64; @@ -286,10 +285,7 @@ mod tests { fields: vec![("email".into(), LoroValue::String("a@b.com".into()))], }; - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as u64; + let now = nodedb_types::clock::since_epoch().unwrap().as_millis() as u64; let resolution = validator .validate_with_policy( @@ -342,10 +338,7 @@ mod tests { ], }; - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as u64; + let now = nodedb_types::clock::since_epoch().unwrap().as_millis() as u64; let resolution = validator .validate_with_policy( @@ -387,10 +380,7 @@ mod tests { fields: vec![("author_id".into(), LoroValue::String("u1".into()))], }; - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as u64; + let now = nodedb_types::clock::since_epoch().unwrap().as_millis() as u64; let resolution = validator .validate_with_policy( @@ -435,10 +425,7 @@ mod tests { fields: vec![("email".into(), LoroValue::String("a@b.com".into()))], }; - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as u64; + let now = nodedb_types::clock::since_epoch().unwrap().as_millis() as u64; let resolution = validator .validate_with_policy( diff --git a/nodedb-crdt/src/validator/validate.rs b/nodedb-crdt/src/validator/validate.rs index 1870bad1f..f9ec21c4a 100644 --- a/nodedb-crdt/src/validator/validate.rs +++ b/nodedb-crdt/src/validator/validate.rs @@ -133,8 +133,7 @@ impl Validator { // Check auth expiry: agents that accumulated deltas offline must // re-authenticate before syncing. if auth.auth_expires_at > 0 { - let now_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) + let now_ms = nodedb_types::clock::since_epoch() .unwrap_or_default() .as_millis() as u64; if now_ms > auth.auth_expires_at { @@ -147,8 +146,7 @@ impl Validator { self.verify_delta_auth(&change.collection, &auth, &delta_bytes)?; - let hlc_timestamp = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) + let hlc_timestamp = nodedb_types::clock::since_epoch() .unwrap_or_default() .as_millis() as u64; diff --git a/nodedb-mem/Cargo.toml b/nodedb-mem/Cargo.toml index 4ba61e85b..fb6fa7607 100644 --- a/nodedb-mem/Cargo.toml +++ b/nodedb-mem/Cargo.toml @@ -23,7 +23,13 @@ tikv-jemalloc-ctl = { workspace = true } libc = { workspace = true } [dev-dependencies] -tokio = { workspace = true } -fluxbench = { workspace = true } tracing-subscriber = { workspace = true } tempfile = { workspace = true } + +# Native-only: see nodedb-wal/Cargo.toml. Both entries resolve tokio with +# `features = ["full"]` (the workspace entry directly, fluxbench through its own +# normal dependency), which tokio rejects on wasm, and neither is referenced by +# any test target in this crate. Native test builds keep them, unchanged. +[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] +tokio = { workspace = true } +fluxbench = { workspace = true } diff --git a/nodedb-mem/src/budget.rs b/nodedb-mem/src/budget.rs index 8624a145e..75669ee67 100644 --- a/nodedb-mem/src/budget.rs +++ b/nodedb-mem/src/budget.rs @@ -320,6 +320,10 @@ mod tests { assert_eq!(budget.allocated(), 0); } + // wasm32 has no thread spawn: `std::thread::spawn` fails with + // `Unsupported`, and wasm targets abort on panic, so this test cannot run + // there. It still runs in full on native. + #[cfg(not(target_arch = "wasm32"))] #[test] fn concurrent_reserves() { use std::sync::Arc; diff --git a/nodedb-mem/src/governor/reserve.rs b/nodedb-mem/src/governor/reserve.rs index 1b3cd5dfb..7550dd6bc 100644 --- a/nodedb-mem/src/governor/reserve.rs +++ b/nodedb-mem/src/governor/reserve.rs @@ -167,10 +167,16 @@ impl MemoryGovernor { #[cfg(test)] mod tests { + // `Arc`, `DatabaseId` and `TenantId` are used only by the thread test that + // wasm32 cannot run, so they are gated with it to keep the wasm test build + // warning-free. + #[cfg(not(target_arch = "wasm32"))] use std::sync::Arc; use std::sync::atomic::Ordering; + #[cfg(not(target_arch = "wasm32"))] use std::thread; + #[cfg(not(target_arch = "wasm32"))] use nodedb_types::{DatabaseId, TenantId}; use super::*; @@ -345,6 +351,10 @@ mod tests { // ── Concurrent reserves ─────────────────────────────────────────────────── + // wasm32 has no thread spawn: `std::thread::spawn` fails with + // `Unsupported`, and wasm targets abort on panic, so this test cannot run + // there. It still runs in full on native. + #[cfg(not(target_arch = "wasm32"))] #[test] fn concurrent_reserves_never_exceed_cap() { let limits = EngineLimits::zeroed().with(EngineId::Vector, 10_000); diff --git a/nodedb-query/src/msgpack_scan/field.rs b/nodedb-query/src/msgpack_scan/field.rs index 02ed1f908..7186e9fd8 100644 --- a/nodedb-query/src/msgpack_scan/field.rs +++ b/nodedb-query/src/msgpack_scan/field.rs @@ -5,7 +5,7 @@ //! Given a `&[u8]` containing a MessagePack map, extract the byte range //! of a value for a given key — without allocating or decoding. -use crate::msgpack_scan::reader::{map_header, skip_value, str_bounds}; +use crate::msgpack_scan::reader::{checked_advance_len, map_header, skip_value, str_bounds}; /// A byte range `(start, end)` within a MessagePack buffer, pointing to /// a complete value (tag + payload). Use `read_f64`, `read_i64`, `read_str` @@ -28,10 +28,16 @@ pub fn extract_field(buf: &[u8], offset: usize, field: &str) -> Option buf - .get(start..start + len) - .map(|kb| kb == field_bytes) - .unwrap_or(false), + Some((start, len)) => { + // Hostile-length addition: see `checked_advance_len`. + match checked_advance_len(buf, start, 0, len) { + Some(end) => buf + .get(start..end) + .map(|kb| kb == field_bytes) + .unwrap_or(false), + None => false, + } + } None => false, }; diff --git a/nodedb-query/src/msgpack_scan/index.rs b/nodedb-query/src/msgpack_scan/index.rs index 2d6a30619..15a35fc34 100644 --- a/nodedb-query/src/msgpack_scan/index.rs +++ b/nodedb-query/src/msgpack_scan/index.rs @@ -9,7 +9,7 @@ //! Uses a flat array with linear search for small docs (≤ 16 fields) to //! avoid HashMap allocation overhead. Falls back to HashMap for large docs. -use crate::msgpack_scan::reader::{map_header, skip_value, str_bounds}; +use crate::msgpack_scan::reader::{checked_advance_len, map_header, skip_value, str_bounds}; /// Threshold: docs with more fields than this use HashMap, otherwise flat array. const HASH_THRESHOLD: usize = 16; @@ -38,7 +38,9 @@ impl FieldIndex { let mut entries = Vec::with_capacity(count); for _ in 0..count { let key_str = if let Some((start, len)) = str_bounds(buf, pos) { - std::str::from_utf8(buf.get(start..start + len)?).ok() + // Hostile-length addition: see `checked_advance_len`. + let end = checked_advance_len(buf, start, 0, len)?; + std::str::from_utf8(buf.get(start..end)?).ok() } else { None }; @@ -61,7 +63,9 @@ impl FieldIndex { let mut offsets = std::collections::HashMap::with_capacity(cap); for _ in 0..count { let key_str = if let Some((start, len)) = str_bounds(buf, pos) { - std::str::from_utf8(buf.get(start..start + len)?).ok() + // Hostile-length addition: see `checked_advance_len`. + let end = checked_advance_len(buf, start, 0, len)?; + std::str::from_utf8(buf.get(start..end)?).ok() } else { None }; diff --git a/nodedb-query/src/msgpack_scan/reader/mod.rs b/nodedb-query/src/msgpack_scan/reader/mod.rs index 1a19f8c94..dcf2c9b17 100644 --- a/nodedb-query/src/msgpack_scan/reader/mod.rs +++ b/nodedb-query/src/msgpack_scan/reader/mod.rs @@ -16,4 +16,5 @@ pub use scalar::{ read_str_advance, read_u32_advance, }; pub use skip::skip_value; +pub(crate) use tags::checked_advance_len; pub use value::read_value; diff --git a/nodedb-query/src/msgpack_scan/reader/scalar.rs b/nodedb-query/src/msgpack_scan/reader/scalar.rs index 1067b7e04..e4bf0598c 100644 --- a/nodedb-query/src/msgpack_scan/reader/scalar.rs +++ b/nodedb-query/src/msgpack_scan/reader/scalar.rs @@ -65,7 +65,9 @@ pub fn read_i64(buf: &[u8], offset: usize) -> Option { /// or invalid UTF-8. pub fn read_str(buf: &[u8], offset: usize) -> Option<&str> { let (start, len) = str_bounds(buf, offset)?; - let bytes = buf.get(start..start + len)?; + // `start + len` is a hostile-length addition; see `checked_advance_len`. + let end = checked_advance_len(buf, start, 0, len)?; + let bytes = buf.get(start..end)?; str::from_utf8(bytes).ok() } @@ -73,9 +75,10 @@ pub fn read_str(buf: &[u8], offset: usize) -> Option<&str> { /// Returns `None` for non-string types, invalid UTF-8, or truncated input. pub fn read_str_advance<'a>(buf: &'a [u8], off: &mut usize) -> Option<&'a str> { let (start, len) = str_bounds(buf, *off)?; - let bytes = buf.get(start..start + len)?; + let end = checked_advance_len(buf, start, 0, len)?; + let bytes = buf.get(start..end)?; let s = str::from_utf8(bytes).ok()?; - *off = start + len; + *off = end; Some(s) } @@ -91,7 +94,7 @@ pub fn read_bin_advance<'a>(buf: &'a [u8], off: &mut usize) -> Option<&'a [u8]> _ => return None, }; let start = *off + header; - let end = start + len; + let end = checked_advance_len(buf, start, 0, len)?; let data = buf.get(start..end)?; *off = end; Some(data) diff --git a/nodedb-query/src/msgpack_scan/reader/skip.rs b/nodedb-query/src/msgpack_scan/reader/skip.rs index 9f63ee35b..79d1cc35f 100644 --- a/nodedb-query/src/msgpack_scan/reader/skip.rs +++ b/nodedb-query/src/msgpack_scan/reader/skip.rs @@ -57,33 +57,33 @@ fn skip_value_depth(buf: &[u8], offset: usize, depth: u16) -> Option { // fixstr (0xa0..=0xbf) 0xa0..=0xbf => { let len = (tag & 0x1f) as usize; - checked_advance(buf, offset, 1 + len) + checked_advance_len(buf, offset, 1, len) } STR8 => { let len = get(buf, offset + 1)? as usize; - checked_advance(buf, offset, 2 + len) + checked_advance_len(buf, offset, 2, len) } STR16 => { let len = read_u16_be(buf, offset + 1)? as usize; - checked_advance(buf, offset, 3 + len) + checked_advance_len(buf, offset, 3, len) } STR32 => { let len = read_u32_be(buf, offset + 1)? as usize; - checked_advance(buf, offset, 5 + len) + checked_advance_len(buf, offset, 5, len) } // bin BIN8 => { let len = get(buf, offset + 1)? as usize; - checked_advance(buf, offset, 2 + len) + checked_advance_len(buf, offset, 2, len) } BIN16 => { let len = read_u16_be(buf, offset + 1)? as usize; - checked_advance(buf, offset, 3 + len) + checked_advance_len(buf, offset, 3, len) } BIN32 => { let len = read_u32_be(buf, offset + 1)? as usize; - checked_advance(buf, offset, 5 + len) + checked_advance_len(buf, offset, 5, len) } // fixed-width numerics (bounds-check against buffer length) @@ -102,15 +102,15 @@ fn skip_value_depth(buf: &[u8], offset: usize, depth: u16) -> Option { FIXEXT16 => checked_advance(buf, offset, 18), EXT8 => { let len = get(buf, offset + 1)? as usize; - checked_advance(buf, offset, 3 + len) + checked_advance_len(buf, offset, 3, len) } EXT16 => { let len = read_u16_be(buf, offset + 1)? as usize; - checked_advance(buf, offset, 4 + len) + checked_advance_len(buf, offset, 4, len) } EXT32 => { let len = read_u32_be(buf, offset + 1)? as usize; - checked_advance(buf, offset, 6 + len) + checked_advance_len(buf, offset, 6, len) } // 0xc1 is never used in the spec diff --git a/nodedb-query/src/msgpack_scan/reader/tags.rs b/nodedb-query/src/msgpack_scan/reader/tags.rs index 4b1daf45c..5277ee5c0 100644 --- a/nodedb-query/src/msgpack_scan/reader/tags.rs +++ b/nodedb-query/src/msgpack_scan/reader/tags.rs @@ -44,27 +44,50 @@ pub(super) fn get(buf: &[u8], pos: usize) -> Option { #[inline(always)] pub(super) fn read_u16_be(buf: &[u8], pos: usize) -> Option { - let bytes = buf.get(pos..pos + 2)?; + let bytes = buf.get(pos..pos.checked_add(2)?)?; Some(u16::from_be_bytes([bytes[0], bytes[1]])) } #[inline(always)] pub(super) fn read_u32_be(buf: &[u8], pos: usize) -> Option { - let bytes = buf.get(pos..pos + 4)?; + let bytes = buf.get(pos..pos.checked_add(4)?)?; Some(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])) } #[inline(always)] pub(super) fn read_u64_be(buf: &[u8], pos: usize) -> Option { - let bytes = buf.get(pos..pos + 8)?; + let bytes = buf.get(pos..pos.checked_add(8)?)?; Some(u64::from_be_bytes([ bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7], ])) } /// Return `Some(offset + size)` only if the buffer has enough bytes. +/// +/// The addition is checked. `size` reaches this function as a tag's header +/// width plus a payload length read from the input, so on a 32-bit target it +/// can exceed `usize` before the buffer is ever consulted: `5 + 0xffff_ffff` +/// panics rather than returning `None`. Checking here keeps a hostile length +/// field a decode failure on every target instead of an abort on 32-bit. #[inline(always)] pub(super) fn checked_advance(buf: &[u8], offset: usize, size: usize) -> Option { - let end = offset + size; + checked_advance_len(buf, offset, size, 0) +} + +/// Return `Some(offset + header + len)` only if the buffer has enough bytes, +/// with both additions checked. +/// +/// Every length-prefixed tag (`str`, `bin`, `ext`) reaches this shape, and +/// `len` is read straight from the input as up to a `u32`. Folding it into the +/// header width first — `header + len` — is what overflows a 32-bit `usize` +/// before any bounds check can reject the value. +#[inline(always)] +pub(crate) fn checked_advance_len( + buf: &[u8], + offset: usize, + header: usize, + len: usize, +) -> Option { + let end = offset.checked_add(header)?.checked_add(len)?; if end <= buf.len() { Some(end) } else { None } } diff --git a/nodedb-sql/src/parser/preprocess/temporal.rs b/nodedb-sql/src/parser/preprocess/temporal.rs index 7645da315..e219f298a 100644 --- a/nodedb-sql/src/parser/preprocess/temporal.rs +++ b/nodedb-sql/src/parser/preprocess/temporal.rs @@ -251,8 +251,7 @@ fn parse_temporal_expr(token: &str) -> Result { // NOW() — case-insensitive if t.to_uppercase() == "NOW()" { - let ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) + let ms = nodedb_types::clock::since_epoch() .map(|d| d.as_millis() as i64) .unwrap_or(0); return Ok(ms); @@ -579,17 +578,11 @@ mod tests { #[test] fn as_of_system_time_now() { - let before = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as i64; + let before = nodedb_types::clock::since_epoch().unwrap().as_millis() as i64; let ex = extract("SELECT * FROM array_slice('g', '{}') AS OF SYSTEM TIME NOW()") .unwrap() .unwrap(); - let after = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as i64; + let after = nodedb_types::clock::since_epoch().unwrap().as_millis() as i64; let ts = ex .temporal .system_time diff --git a/nodedb-sql/src/planner/defaults/kind.rs b/nodedb-sql/src/planner/defaults/kind.rs index 2e1b2deab..36f96eb00 100644 --- a/nodedb-sql/src/planner/defaults/kind.rs +++ b/nodedb-sql/src/planner/defaults/kind.rs @@ -64,9 +64,7 @@ pub(super) fn cuid2_with_length(len: usize) -> crate::Result { /// Render the current wall-clock instant the way `DEFAULT NOW()` stores it. fn now_rfc3339() -> String { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default(); + let now = nodedb_types::clock::since_epoch().unwrap_or_default(); chrono::DateTime::from_timestamp_millis(now.as_millis() as i64) .map(|dt| dt.to_rfc3339()) .unwrap_or_else(|| now.as_millis().to_string()) diff --git a/nodedb-types/Cargo.toml b/nodedb-types/Cargo.toml index 575d693db..d01a78746 100644 --- a/nodedb-types/Cargo.toml +++ b/nodedb-types/Cargo.toml @@ -41,3 +41,5 @@ sha2 = { workspace = true } [dev-dependencies] toml = { workspace = true } +[target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies] +js-sys = "0.3" diff --git a/nodedb-types/src/clock.rs b/nodedb-types/src/clock.rs new file mode 100644 index 000000000..0d90a2f80 --- /dev/null +++ b/nodedb-types/src/clock.rs @@ -0,0 +1,55 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Wall-clock access, one target split. +//! +//! `wasm32-unknown-unknown` has no std clock: `SystemTime::now()` panics with +//! "time not implemented on this platform". Builds for that target read the +//! host clock through `js_sys` instead; every other target, `wasm32-wasip1` +//! included, uses std. +//! +//! Callers keep their own conversion, saturation, and pre-epoch fallback — +//! this module only answers "how long since the Unix epoch". + +use std::time::Duration; + +/// Time elapsed since the Unix epoch. +/// +/// `None` when the system clock reads earlier than the epoch. On +/// `wasm32-unknown-unknown` the value comes from `Date.now()` and therefore +/// has millisecond resolution. +#[cfg(all(target_arch = "wasm32", target_os = "unknown"))] +pub fn since_epoch() -> Option { + Some(Duration::from_millis(js_sys::Date::now() as u64)) +} + +#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] +pub fn since_epoch() -> Option { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() +} + +#[cfg(test)] +mod tests { + use super::since_epoch; + + /// The helper exists so callers never reach `SystemTime::now()` on a + /// target that has no clock. It must answer on every target it builds for. + #[test] + fn since_epoch_is_available() { + let elapsed = since_epoch().expect("clock reads after the Unix epoch"); + assert!( + elapsed.as_secs() > 1_600_000_000, + "clock reads a date after 2020: {elapsed:?}" + ); + } + + /// Callers assume the clock does not go backwards between two reads + /// (HLC monotonicity, retry stamps, auth expiry). + #[test] + fn since_epoch_does_not_go_backwards() { + let first = since_epoch().expect("clock"); + let second = since_epoch().expect("clock"); + assert!(second >= first, "{second:?} is before {first:?}"); + } +} diff --git a/nodedb-types/src/datetime/timestamp.rs b/nodedb-types/src/datetime/timestamp.rs index c88daade6..d8609466b 100644 --- a/nodedb-types/src/datetime/timestamp.rs +++ b/nodedb-types/src/datetime/timestamp.rs @@ -74,20 +74,18 @@ impl NdbDateTime { /// `i64::MAX` (year ~292,277 CE) rather than wrapping — clocks that far /// in the future simply report the maximum representable timestamp. pub fn now() -> Self { - let dur = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_else(|_| { - use std::sync::atomic::{AtomicBool, Ordering}; - static LOGGED: AtomicBool = AtomicBool::new(false); - if !LOGGED.swap(true, Ordering::Relaxed) { - tracing::error!( - module = module_path!(), - "system clock is before UNIX_EPOCH; using 0 (epoch) \ + let dur = crate::clock::since_epoch().unwrap_or_else(|| { + use std::sync::atomic::{AtomicBool, Ordering}; + static LOGGED: AtomicBool = AtomicBool::new(false); + if !LOGGED.swap(true, Ordering::Relaxed) { + tracing::error!( + module = module_path!(), + "system clock is before UNIX_EPOCH; using 0 (epoch) \ — check NTP/RTC configuration" - ); - } - std::time::Duration::ZERO - }); + ); + } + std::time::Duration::ZERO + }); Self { micros: i64::try_from(dur.as_micros()).unwrap_or(i64::MAX), } diff --git a/nodedb-types/src/hlc.rs b/nodedb-types/src/hlc.rs index e500bcdc2..bed5204ff 100644 --- a/nodedb-types/src/hlc.rs +++ b/nodedb-types/src/hlc.rs @@ -12,7 +12,6 @@ use std::cmp::Ordering; use std::sync::Mutex; -use std::time::{SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; @@ -189,10 +188,9 @@ impl HlcClock { } fn wall_now_ns() -> u64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) + crate::clock::since_epoch() .map(|d| d.as_nanos() as u64) - .unwrap_or_else(|_| { + .unwrap_or_else(|| { use std::sync::atomic::{AtomicBool, Ordering}; static LOGGED: AtomicBool = AtomicBool::new(false); if !LOGGED.swap(true, Ordering::Relaxed) { diff --git a/nodedb-types/src/lib.rs b/nodedb-types/src/lib.rs index 68f14647e..06ed16131 100644 --- a/nodedb-types/src/lib.rs +++ b/nodedb-types/src/lib.rs @@ -10,6 +10,7 @@ //! crate. pub mod audit_dml; +pub mod clock; pub mod config; pub mod quota; diff --git a/nodedb-types/src/temporal/ordinal.rs b/nodedb-types/src/temporal/ordinal.rs index 1c005746e..ec6f16f69 100644 --- a/nodedb-types/src/temporal/ordinal.rs +++ b/nodedb-types/src/temporal/ordinal.rs @@ -23,7 +23,6 @@ //! exposed here so that wiring has a single canonical API to call. use std::sync::Mutex; -use std::time::{SystemTime, UNIX_EPOCH}; /// Nanoseconds in one millisecond. pub const NANOS_PER_MS: i64 = 1_000_000; @@ -87,8 +86,7 @@ pub fn ordinal_to_ms(ordinal: i64) -> i64 { } fn wall_now_ns() -> i64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) + crate::clock::since_epoch() .map(|d| { let ns = d.as_nanos(); if ns > i64::MAX as u128 { @@ -97,7 +95,7 @@ fn wall_now_ns() -> i64 { ns as i64 } }) - .unwrap_or_else(|_| { + .unwrap_or_else(|| { use std::sync::atomic::{AtomicBool, Ordering}; static LOGGED: AtomicBool = AtomicBool::new(false); if !LOGGED.swap(true, Ordering::Relaxed) { diff --git a/nodedb-types/src/wire_time.rs b/nodedb-types/src/wire_time.rs index a558c39b8..8d52132f3 100644 --- a/nodedb-types/src/wire_time.rs +++ b/nodedb-types/src/wire_time.rs @@ -13,8 +13,6 @@ //! Using these aliases in field declarations makes the intended semantics //! self-documenting and ensures a single canonical width per concept. -use std::time::{SystemTime, UNIX_EPOCH}; - /// A wall-clock instant expressed as signed milliseconds since the Unix epoch /// (UTC). Negative values represent dates before 1970-01-01. /// @@ -34,13 +32,14 @@ pub type DurMs = u64; /// easier to detect than `i64::MAX`. Logs the condition once per process via /// `tracing::error!` to alert operators. /// -/// # TODO(post-launch): funnel direct `SystemTime::now()` callers through this -/// helper so all inline clock-read sites also get the once-per-process log. +/// Inline clock reads go through `crate::clock::since_epoch()`, which owns the +/// target split. This helper keeps its own contract — substitute `0` and log +/// once per process — for callers that want a value rather than a `Result`; +/// callers that must surface a pre-epoch clock keep their own error mapping. pub fn current_wall_ms() -> WallMs { - SystemTime::now() - .duration_since(UNIX_EPOCH) + crate::clock::since_epoch() .map(|d| d.as_millis().min(i64::MAX as u128) as i64) - .unwrap_or_else(|_| { + .unwrap_or_else(|| { use std::sync::atomic::{AtomicBool, Ordering}; static LOGGED: AtomicBool = AtomicBool::new(false); if !LOGGED.swap(true, Ordering::Relaxed) { diff --git a/nodedb-vector/src/builder.rs b/nodedb-vector/src/builder.rs index 372119f9b..afa054e6a 100644 --- a/nodedb-vector/src/builder.rs +++ b/nodedb-vector/src/builder.rs @@ -47,6 +47,9 @@ fn builder_loop(core_id: usize, rx: mpsc::Receiver, tx: mpsc::Send "building HNSW index" ); + // The build log carries its duration; wasm32-unknown-unknown has no + // clock, so the timer is skipped there and the log reports 0. + #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] let start = std::time::Instant::now(); let mut index = HnswIndex::with_seed( req.dim, @@ -60,13 +63,16 @@ fn builder_loop(core_id: usize, rx: mpsc::Receiver, tx: mpsc::Send .unwrap_or_else(|e| tracing::error!(error = %e, "HNSW insert failed")); } - let elapsed = start.elapsed(); + #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] + let elapsed_ms = start.elapsed().as_millis() as u64; + #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] + let elapsed_ms = 0u64; info!( core_id, key = %req.key, segment_id = req.segment_id, vectors = index.len(), - elapsed_ms = elapsed.as_millis() as u64, + elapsed_ms, "HNSW index built" ); diff --git a/nodedb-vector/src/hnsw/graph/index/state.rs b/nodedb-vector/src/hnsw/graph/index/state.rs index 0743dd12b..64ea53913 100644 --- a/nodedb-vector/src/hnsw/graph/index/state.rs +++ b/nodedb-vector/src/hnsw/graph/index/state.rs @@ -255,6 +255,9 @@ mod tests { /// Rerank must see the vector for a narrow dtype too. Before the `Cow` return /// this yielded `None` for F16/BF16, which made the FP32 rerank path fail with /// "fetch_vector returned None" for every candidate in the collection. + /// + /// Native-only: `get_vector_or_backing` is gated with the backing type. + #[cfg(not(target_arch = "wasm32"))] #[test] fn get_vector_or_backing_serves_narrow_dtypes() { for dtype in [VectorStorageDtype::F16, VectorStorageDtype::BF16] { @@ -307,6 +310,10 @@ mod tests { /// backing is the worst case: the graph looks healthy, so search proceeds and /// then scores a node that has no vector — which is what made one poisoned /// segment panic the daemon on every query. + /// + /// Native-only: the trait under test lives in `crate::segment_backing`, + /// which is gated off on wasm32 (mmap is the only implementation). + #[cfg(not(target_arch = "wasm32"))] #[test] fn with_backing_refuses_a_backing_that_cannot_serve_the_index() { use crate::segment_backing::VectorSegmentBacking; diff --git a/nodedb-vector/tests/vector_suite/main.rs b/nodedb-vector/tests/vector_suite/main.rs index 0b2131900..55b239a83 100644 --- a/nodedb-vector/tests/vector_suite/main.rs +++ b/nodedb-vector/tests/vector_suite/main.rs @@ -9,6 +9,13 @@ //! //! Every case keeps its own module, so a test's path stays //! `cases::::` and names never collide across files. +//! +//! Native-only: these cases drive `VectorCollection`, `mmap_segment` and their +//! `libc::madvise` path, none of which exists on wasm32 — `src/lib.rs` gates all +//! three off there. They are skipped on that target rather than rewritten, +//! because there is no wasm equivalent to assert against. + +#![cfg(not(target_arch = "wasm32"))] mod cases; mod support; diff --git a/nodedb-wal/Cargo.toml b/nodedb-wal/Cargo.toml index c6e084769..feef713c8 100644 --- a/nodedb-wal/Cargo.toml +++ b/nodedb-wal/Cargo.toml @@ -63,8 +63,6 @@ getrandom_03 = { workspace = true } getrandom_02 = { workspace = true } [dev-dependencies] -tokio = { workspace = true } -fluxbench = { workspace = true } tracing-subscriber = { workspace = true } # Named again here so the diagnostics integration tests can act as the host — # call `init` and read reports back. Same published version as the optional @@ -73,6 +71,20 @@ faultbox = { workspace = true } tempfile = { workspace = true } rand = { workspace = true } +# Native-only dev-dependencies. `tokio = { workspace = true }` inherits +# `features = ["full"]`, and `fluxbench` declares its own normal +# `tokio = { features = ["full"] }`; tokio's wasm guard +# (tokio/src/lib.rs: `Only features sync,macros,io-util,rt,time are supported +# on wasm.`) rejects fs/io-std/net/process/rt-multi-thread/signal, and Cargo +# unifies features across the whole `cargo test` invocation, so either entry +# breaks every wasm test target in the build. Neither is reachable from a wasm +# test: this crate has no tokio call sites, and `fluxbench` is used only by +# benches/wal_throughput.rs, which `cargo test` does not build. Native test and +# bench builds keep both, unchanged. +[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] +tokio = { workspace = true } +fluxbench = { workspace = true } + [target.'cfg(target_os = "linux")'.dev-dependencies] io-uring = { workspace = true } diff --git a/nodedb-wal/src/crypto.rs b/nodedb-wal/src/crypto.rs index a0fc4b456..b1e0ce545 100644 --- a/nodedb-wal/src/crypto.rs +++ b/nodedb-wal/src/crypto.rs @@ -563,7 +563,7 @@ mod tests { use std::io::Write as _; use std::os::unix::fs::PermissionsExt as _; - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("key.bin"); let mut f = std::fs::File::create(&path).unwrap(); f.write_all(&[0x42u8; 32]).unwrap(); @@ -579,7 +579,7 @@ mod tests { use std::io::Write as _; use std::os::unix::fs::PermissionsExt as _; - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("key.bin"); let mut f = std::fs::File::create(&path).unwrap(); f.write_all(&[0x42u8; 32]).unwrap(); @@ -595,7 +595,7 @@ mod tests { use std::io::Write as _; use std::os::unix::fs::PermissionsExt as _; - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("key.bin"); let mut f = std::fs::File::create(&path).unwrap(); f.write_all(&[0x42u8; 32]).unwrap(); @@ -619,7 +619,7 @@ mod tests { use std::io::Write as _; use std::os::unix::fs::PermissionsExt as _; - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let target = dir.path().join("target.bin"); let mut f = std::fs::File::create(&target).unwrap(); f.write_all(&[0x42u8; 32]).unwrap(); diff --git a/nodedb-wal/src/double_write/buffer.rs b/nodedb-wal/src/double_write/buffer.rs index 54c50bce7..d493f0ee6 100644 --- a/nodedb-wal/src/double_write/buffer.rs +++ b/nodedb-wal/src/double_write/buffer.rs @@ -287,6 +287,13 @@ mod tests { use crate::double_write::metrics::wal_dwb_bytes_written_total; use crate::record::{RecordType, WalRecordArgs}; + // `DoubleWriteBuffer::recover_record` is a deliberate no-op on wasm32 + // (`double_write/recover.rs`: the only arm that compiles there is + // `Ok(None)`, because slot recovery needs `pread`/O_DIRECT). Every + // assertion below that expects a record *back* from it is therefore gated + // to native, while writes, flushes and the skip accounting still run on + // wasm. + fn open_buffered(path: &Path) -> DoubleWriteBuffer { DoubleWriteBuffer::open(path, DwbMode::Buffered).unwrap() } @@ -313,9 +320,10 @@ mod tests { assert_eq!(dwb.write_record_deferred(rec).unwrap(), DwbMirror::Mirrored); } + #[cfg(not(target_arch = "wasm32"))] #[test] fn write_and_recover() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("test.dwb")); mirror(&mut dwb, &record(42, b"hello double-write")); @@ -325,16 +333,18 @@ mod tests { assert_eq!(rec.payload, b"hello double-write"); } + #[cfg(not(target_arch = "wasm32"))] #[test] fn recover_nonexistent_returns_none() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("test2.dwb")); assert!(dwb.recover_record(999).unwrap().is_none()); } + #[cfg(not(target_arch = "wasm32"))] #[test] fn survives_reopen() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("reopen.dwb"); { @@ -349,7 +359,7 @@ mod tests { #[test] fn batch_deferred_writes_and_flush() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("batch.dwb")); for lsn in 1..=5u64 { @@ -360,6 +370,7 @@ mod tests { dwb.flush().unwrap(); assert!(!dwb.dirty); + #[cfg(not(target_arch = "wasm32"))] for lsn in 1..=5u64 { let recovered = dwb.recover_record(lsn).unwrap().expect("recoverable"); assert_eq!(recovered.payload, format!("batch-{lsn}").into_bytes()); @@ -368,7 +379,7 @@ mod tests { #[test] fn flush_is_idempotent() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("idem.dwb")); dwb.flush().unwrap(); @@ -382,7 +393,7 @@ mod tests { #[test] fn oversized_record_is_reported_as_skipped() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("oversized.dwb")); let payload = vec![0xabu8; DWB_SLOT_RECORD_MAX]; @@ -395,12 +406,13 @@ mod tests { max: DWB_SLOT_RECORD_MAX, }) ); + #[cfg(not(target_arch = "wasm32"))] assert!(dwb.recover_record(11).unwrap().is_none()); } #[test] fn bytes_written_counter_increments() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let before = wal_dwb_bytes_written_total(); let mut dwb = open_buffered(&dir.path().join("counter.dwb")); diff --git a/nodedb-wal/src/double_write/raw_io.rs b/nodedb-wal/src/double_write/raw_io.rs index 11a3f570f..7833ece91 100644 --- a/nodedb-wal/src/double_write/raw_io.rs +++ b/nodedb-wal/src/double_write/raw_io.rs @@ -30,6 +30,16 @@ pub(crate) fn full_capacity_slice(buf: &AlignedBuf) -> &[u8] { } /// `pwrite`-retry helper that handles short writes. +/// +/// Reports [`WalError::Unsupported`] on wasm32 rather than falling back to a +/// seek-and-write. A fallback would mirror the slot successfully and the writer +/// would then report `DwbProtection::Active`, but `recover_record` is a stub on +/// that target (slot recovery needs `pread`), so nothing could ever read the +/// slot back. The protection would be claimed and unusable. Failing here keeps +/// the writer's own degradation signal honest: the append still succeeds +/// (`DoubleWriteBuffer` failures are deliberately not fatal to the WAL), while +/// `dwb_protection()` reports `Degraded(WriteFailed)` and every record is +/// counted as unprotected. pub(crate) fn pwrite_all(file: &File, data: &[u8], offset: u64) -> Result<()> { #[cfg(not(target_arch = "wasm32"))] { diff --git a/nodedb-wal/src/double_write/recover.rs b/nodedb-wal/src/double_write/recover.rs index 70600e69a..e6b295262 100644 --- a/nodedb-wal/src/double_write/recover.rs +++ b/nodedb-wal/src/double_write/recover.rs @@ -160,7 +160,11 @@ fn decode_record(bytes: &[u8]) -> Option { }) } -#[cfg(test)] +// Every test here asserts what `recover_record` returns, and that is a +// deliberate `Ok(None)` stub on wasm32 — the only arm that compiles there, +// because slot recovery needs `pread`/O_DIRECT. `scan_max_seq`'s wasm arm is +// still exercised indirectly by the write path. +#[cfg(all(test, not(target_arch = "wasm32")))] mod tests { use std::path::Path; @@ -198,7 +202,7 @@ mod tests { #[test] fn recover_after_wraparound() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("wrap.dwb")); let total = DWB_CAPACITY as u64 + 5; @@ -226,7 +230,7 @@ mod tests { /// claim it. The one written last is the one the WAL committed. #[test] fn reused_lsn_recovers_the_newer_copy() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("reuse.dwb")); mirror(&mut dwb, &record(9, b"never-acknowledged")); @@ -246,7 +250,7 @@ mod tests { /// would get wrong. #[test] fn reused_lsn_across_a_wrap_recovers_the_newer_copy() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let mut dwb = open_buffered(&dir.path().join("reuse_wrap.dwb")); // Older copy lands in the last slot of the ring. @@ -266,7 +270,7 @@ mod tests { /// restart still has to outrank the stale one already on disk. #[test] fn reused_lsn_across_a_reopen_recovers_the_newer_copy() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("reuse_reopen.dwb"); { diff --git a/nodedb-wal/src/lazy_reader.rs b/nodedb-wal/src/lazy_reader.rs index 3e175dd72..64c8c963a 100644 --- a/nodedb-wal/src/lazy_reader.rs +++ b/nodedb-wal/src/lazy_reader.rs @@ -383,7 +383,7 @@ mod tests { #[test] fn lazy_read_all_payloads() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); { @@ -409,7 +409,7 @@ mod tests { #[test] fn lazy_skip_non_matching() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); { @@ -449,7 +449,7 @@ mod tests { #[test] fn oversized_payload_header_is_a_typed_error_before_skip_or_allocation() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("oversized.wal"); let header = RecordHeader { magic: crate::record::WAL_MAGIC, @@ -474,7 +474,7 @@ mod tests { #[test] fn skip_payload_rejects_truncated_tail_without_seeking_beyond_eof() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("truncated.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); writer @@ -505,7 +505,7 @@ mod tests { #[test] fn replay_all_segments_lazy_works() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); // Use proper segment filename pattern: wal-{lsn:020}.seg let path = dir.path().join("wal-00000000000000000001.seg"); @@ -528,7 +528,7 @@ mod tests { #[test] fn empty_wal_no_records() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("empty.wal"); { diff --git a/nodedb-wal/src/lib.rs b/nodedb-wal/src/lib.rs index f61e167e8..cebbf116f 100644 --- a/nodedb-wal/src/lib.rs +++ b/nodedb-wal/src/lib.rs @@ -73,3 +73,24 @@ pub use temporal_purge::{TemporalPurgeEngine, TemporalPurgePayload}; pub use tombstone::{CollectionTombstonePayload, MAX_COLLECTION_NAME_LEN}; pub use torn_tail::{TailVerdict, verify_committed_prefix}; pub use writer::WalWriter; + +/// Create a temporary directory that also works under `wasm32-wasip1`. +/// +/// `tempfile::tempdir()` aborts there: `std::env::temp_dir()` is +/// `unimplemented!("not supported by WASI yet")` in std +/// (`library/std/src/sys/paths/wasi.rs`), so the call fails before it reaches +/// the filesystem and neither a preopen nor `TMPDIR` can help. The wasm test +/// runner preopens only the working directory (`wasmtime --dir=.`), so on wasm +/// the directory is created under it. Native keeps `tempfile::tempdir()` +/// exactly as it was. +#[cfg(test)] +pub(crate) fn test_tempdir() -> std::io::Result { + #[cfg(target_arch = "wasm32")] + { + tempfile::Builder::new().tempdir_in(".") + } + #[cfg(not(target_arch = "wasm32"))] + { + tempfile::tempdir() + } +} diff --git a/nodedb-wal/src/mmap_reader/reader.rs b/nodedb-wal/src/mmap_reader/reader.rs index 0ca585489..623020ab9 100644 --- a/nodedb-wal/src/mmap_reader/reader.rs +++ b/nodedb-wal/src/mmap_reader/reader.rs @@ -415,7 +415,7 @@ mod tests { #[test] fn mmap_reader_basic() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); // Write some records with the standard writer. @@ -445,7 +445,7 @@ mod tests { #[test] fn mmap_reader_empty_file() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("empty.wal"); std::fs::write(&path, []).unwrap(); @@ -456,7 +456,7 @@ mod tests { #[test] fn mmap_reader_truncated_header() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("truncated.wal"); // Write 10 bytes — not enough for a header (30 bytes). std::fs::write(&path, [0u8; 10]).unwrap(); @@ -468,7 +468,7 @@ mod tests { #[test] fn mmap_reader_truncated_payload_stops_at_committed_prefix() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("truncated-payload.wal"); let header = RecordHeader { magic: WAL_MAGIC, @@ -490,7 +490,7 @@ mod tests { #[test] fn oversized_payload_header_is_a_typed_error_before_allocation() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("oversized.wal"); let header = RecordHeader { magic: WAL_MAGIC, diff --git a/nodedb-wal/src/mmap_reader/replay.rs b/nodedb-wal/src/mmap_reader/replay.rs index 7c4e172fb..c7204d1ca 100644 --- a/nodedb-wal/src/mmap_reader/replay.rs +++ b/nodedb-wal/src/mmap_reader/replay.rs @@ -239,7 +239,7 @@ mod tests { #[test] fn replay_mmap_from_lsn() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); std::fs::create_dir_all(&wal_dir).unwrap(); @@ -270,7 +270,7 @@ mod tests { fn replay_mmap_below_the_retained_floor_is_rejected() { use crate::error::WalError; - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let config = crate::segmented::SegmentedWalConfig { diff --git a/nodedb-wal/src/reader.rs b/nodedb-wal/src/reader.rs index edaade37c..7b4291f8a 100644 --- a/nodedb-wal/src/reader.rs +++ b/nodedb-wal/src/reader.rs @@ -374,7 +374,7 @@ mod tests { #[test] fn write_then_read_roundtrip() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); // Write records. @@ -413,7 +413,7 @@ mod tests { #[test] fn empty_wal_yields_no_records() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("empty.wal"); // Create an empty file. @@ -429,7 +429,7 @@ mod tests { #[test] fn truncated_file_stops_at_committed_prefix() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("truncated.wal"); // Write records. @@ -460,7 +460,7 @@ mod tests { #[test] fn oversized_payload_header_is_a_typed_error_before_allocation() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("oversized.wal"); let header = RecordHeader { magic: crate::record::WAL_MAGIC, @@ -499,7 +499,7 @@ mod tests { const UNKNOWN_OPTIONAL: u32 = 99; // no 0x8000 bit → optional, not in enum const SKIP_COUNT: usize = 50_000; - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("many_unknown.wal"); { diff --git a/nodedb-wal/src/recovery.rs b/nodedb-wal/src/recovery.rs index 59ffa96df..63fdbd712 100644 --- a/nodedb-wal/src/recovery.rs +++ b/nodedb-wal/src/recovery.rs @@ -106,7 +106,7 @@ mod tests { #[test] fn recover_empty_wal() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("empty.wal"); // Create empty WAL. @@ -123,7 +123,7 @@ mod tests { #[test] fn recover_nonexistent_file() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("nonexistent.wal"); let info = recover(&path).unwrap(); @@ -133,7 +133,7 @@ mod tests { #[test] fn recover_with_records() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); { @@ -159,7 +159,7 @@ mod tests { #[test] fn recover_truncated_wal() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("truncated.wal"); { diff --git a/nodedb-wal/src/segment/atomic_io.rs b/nodedb-wal/src/segment/atomic_io.rs index 5162579c6..538549bb7 100644 --- a/nodedb-wal/src/segment/atomic_io.rs +++ b/nodedb-wal/src/segment/atomic_io.rs @@ -48,6 +48,10 @@ use crate::error::{Result, WalError}; /// before the directory entry is persisted causes the file to "disappear" /// on reboot. Calling fsync on the directory fd ensures the metadata /// (filename, inode pointer) is on stable storage. +/// +/// The one exception is `wasm32-wasip1`: wasi preview1 has no directory +/// fsync, so on that target this is a documented no-op (see below) and the +/// durability guarantee above does not hold there. pub fn fsync_directory(dir: &Path) -> Result<()> { // Crash injection: the directory entry never reaches stable storage. // Every caller must treat this as a durability failure, not a warning. @@ -55,9 +59,23 @@ pub fn fsync_directory(dir: &Path) -> Result<()> { std::io::Error::other(format!("failpoint wal::fsync_directory: {detail}")) )); - let dir_file = fs::File::open(dir).map_err(WalError::Io)?; - dir_file.sync_all().map_err(WalError::Io)?; - Ok(()) + #[cfg(target_arch = "wasm32")] + { + // wasi preview1 has no directory fsync: `File::sync_all` on a directory + // returns EBADF and there is no weaker syscall with the same guarantee. + // The rename itself still succeeds; what is lost on this target is the + // guarantee that the directory entry survives a host crash. Native + // callers keep it, and the failpoint injection above still fires here. + let _ = dir; + Ok(()) + } + + #[cfg(not(target_arch = "wasm32"))] + { + let dir_file = fs::File::open(dir).map_err(WalError::Io)?; + dir_file.sync_all().map_err(WalError::Io)?; + Ok(()) + } } fn invalid_input(detail: String) -> WalError { @@ -204,7 +222,7 @@ mod tests { #[test] fn atomic_write_fsync_roundtrip() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); atomic_write_fsync(dir.path(), "payload.ckpt", b"hello world").unwrap(); @@ -220,7 +238,7 @@ mod tests { #[test] fn atomic_write_fsync_overwrites() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); atomic_write_fsync(dir.path(), "payload.ckpt", b"v1").unwrap(); atomic_write_fsync(dir.path(), "payload.ckpt", b"v2").unwrap(); @@ -238,7 +256,7 @@ mod tests { #[test] fn atomic_swap_dirs_fsync_swaps() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let live = dir.path().join("live"); let backup = dir.path().join("backup"); let staged = dir.path().join("staged"); @@ -259,7 +277,7 @@ mod tests { /// helper takes. Without the check each of these escapes `dir`. #[test] fn atomic_write_fsync_rejects_a_name_that_is_not_one_component() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); fs::create_dir(dir.path().join("sub")).unwrap(); for name in BAD_NAMES { @@ -286,7 +304,7 @@ mod tests { fn atomic_swap_dirs_fsync_rejects_a_bad_name_in_each_position() { for position in 0..3 { for name in BAD_NAMES { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let live = dir.path().join("live"); fs::create_dir(&live).unwrap(); fs::write(live.join("marker"), b"old").unwrap(); @@ -312,7 +330,7 @@ mod tests { #[test] fn read_checkpoint_dontneed_returns_bytes() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("ckpt"); fs::write(&path, b"checkpoint bytes").unwrap(); diff --git a/nodedb-wal/src/segment/checkpoint_frame.rs b/nodedb-wal/src/segment/checkpoint_frame.rs index fec6ff4e5..dc40d8fe8 100644 --- a/nodedb-wal/src/segment/checkpoint_frame.rs +++ b/nodedb-wal/src/segment/checkpoint_frame.rs @@ -103,7 +103,7 @@ mod tests { #[test] fn roundtrip() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let dst = dir.path().join("payload.ckpt"); write_checkpoint_framed(dir.path(), "payload.ckpt", b"hello framed world").unwrap(); @@ -113,7 +113,7 @@ mod tests { #[test] fn corruption_detected() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let dst = dir.path().join("payload.ckpt"); write_checkpoint_framed(dir.path(), "payload.ckpt", b"integrity matters").unwrap(); @@ -135,7 +135,7 @@ mod tests { #[test] fn truncation_detected() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let dst = dir.path().join("payload.ckpt"); write_checkpoint_framed(dir.path(), "payload.ckpt", b"a longer payload body here").unwrap(); @@ -155,7 +155,7 @@ mod tests { #[test] fn legacy_unframed_accepted() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let dst = dir.path().join("payload.ckpt"); // Raw pre-framing bytes: first 4 bytes are NOT the NCKF magic. diff --git a/nodedb-wal/src/segment/discovery.rs b/nodedb-wal/src/segment/discovery.rs index bee77f84c..ddeda519c 100644 --- a/nodedb-wal/src/segment/discovery.rs +++ b/nodedb-wal/src/segment/discovery.rs @@ -45,7 +45,7 @@ mod tests { #[test] fn discover_empty_dir() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let segments = discover_segments(dir.path()).unwrap(); assert!(segments.is_empty()); } @@ -58,7 +58,7 @@ mod tests { #[test] fn discover_segments_sorted() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); fs::write(dir.path().join("wal-00000000000000000050.seg"), b"seg3").unwrap(); fs::write(dir.path().join("wal-00000000000000000001.seg"), b"seg1").unwrap(); diff --git a/nodedb-wal/src/segment/truncate.rs b/nodedb-wal/src/segment/truncate.rs index 3466230ea..3101060dc 100644 --- a/nodedb-wal/src/segment/truncate.rs +++ b/nodedb-wal/src/segment/truncate.rs @@ -105,7 +105,7 @@ mod tests { #[test] fn truncate_deletes_old_segments() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); fs::write(dir.path().join("wal-00000000000000000001.seg"), b"data1").unwrap(); fs::write(dir.path().join("wal-00000000000000000001.dwb"), b"dwb1").unwrap(); @@ -124,7 +124,7 @@ mod tests { #[test] fn truncate_never_deletes_active_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); fs::write(dir.path().join("wal-00000000000000000001.seg"), b"data").unwrap(); @@ -137,7 +137,7 @@ mod tests { #[test] fn truncate_no_segments_below_checkpoint() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); fs::write(dir.path().join("wal-00000000000000000100.seg"), b"data").unwrap(); fs::write(dir.path().join("wal-00000000000000000200.seg"), b"data").unwrap(); diff --git a/nodedb-wal/src/segmented.rs b/nodedb-wal/src/segmented.rs index 1ee5bbe4f..a5c958632 100644 --- a/nodedb-wal/src/segmented.rs +++ b/nodedb-wal/src/segmented.rs @@ -430,7 +430,7 @@ mod tests { #[test] fn create_and_append() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let mut wal = SegmentedWal::open(test_config(&wal_dir)).unwrap(); @@ -449,7 +449,7 @@ mod tests { #[test] fn replay_after_close() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); // Write records. @@ -476,7 +476,7 @@ mod tests { #[test] fn automatic_segment_rollover() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); // Use a tiny segment target to force rollover. @@ -519,7 +519,7 @@ mod tests { #[test] fn truncation_removes_old_segments() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let config = SegmentedWalConfig { @@ -560,7 +560,7 @@ mod tests { #[test] fn replay_from_checkpoint_lsn() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let mut wal = SegmentedWal::open(test_config(&wal_dir)).unwrap(); @@ -615,7 +615,7 @@ mod tests { #[test] fn replay_from_the_retained_floor_returns_the_whole_suffix() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let (wal, floor) = truncated_wal(&dir.path().join("wal")); let records = wal.replay_from(floor).expect("floor is fully retained"); @@ -628,7 +628,7 @@ mod tests { /// request starting exactly there is complete, not short. #[test] fn replay_from_exactly_the_floor_is_not_an_error() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let (wal, floor) = truncated_wal(&dir.path().join("wal")); assert!(wal.replay_from(floor).is_ok()); @@ -638,7 +638,7 @@ mod tests { #[test] fn replay_below_the_retained_floor_is_a_typed_error_naming_both_lsns() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let (wal, floor) = truncated_wal(&dir.path().join("wal")); match wal.replay_from(1) { @@ -666,7 +666,7 @@ mod tests { #[test] fn total_size_bytes() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let mut wal = SegmentedWal::open(test_config(&wal_dir)).unwrap(); @@ -680,7 +680,7 @@ mod tests { #[test] fn reopen_continues_lsn() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); { @@ -713,7 +713,7 @@ mod tests { // 3. Reopen a new reader and replay — records are marked encrypted. // 4. For each segment, open the reader, read its preamble epoch, and // decrypt each record using that epoch. Verify payloads match. - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let key_bytes = [42u8; 32]; @@ -793,7 +793,7 @@ mod tests { /// is read from the on-disk preamble rather than the current key's epoch. #[test] fn wal_encrypted_restart_roundtrip() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let key_bytes = [0xABu8; 32]; @@ -856,7 +856,7 @@ mod tests { /// Decryption must fail because the preamble is part of the AAD. #[test] fn epoch_tamper_rejected() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let key_bytes = [0x55u8; 32]; @@ -902,7 +902,7 @@ mod tests { #[test] fn replay_from_limit_basic() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let config = test_config(dir.path()); let mut wal = SegmentedWal::open(config).unwrap(); @@ -927,7 +927,7 @@ mod tests { #[test] fn replay_from_limit_with_lsn_filter() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let config = test_config(dir.path()); let mut wal = SegmentedWal::open(config).unwrap(); @@ -950,7 +950,7 @@ mod tests { #[test] fn replay_from_limit_empty() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let config = test_config(dir.path()); let mut wal = SegmentedWal::open(config).unwrap(); @@ -965,7 +965,7 @@ mod tests { #[test] fn replay_from_limit_across_segments() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let config = test_config(dir.path()); let mut wal = SegmentedWal::open(config).unwrap(); diff --git a/nodedb-wal/src/torn_tail.rs b/nodedb-wal/src/torn_tail.rs index a13689499..049494063 100644 --- a/nodedb-wal/src/torn_tail.rs +++ b/nodedb-wal/src/torn_tail.rs @@ -230,7 +230,7 @@ mod tests { #[test] fn truncated_tail_is_a_torn_write() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("torn.wal"); write_segment(&path, 4); @@ -246,7 +246,7 @@ mod tests { #[test] fn garbage_at_the_end_is_a_torn_write() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("garbage_tail.wal"); write_segment(&path, 3); @@ -258,7 +258,7 @@ mod tests { #[test] fn damage_with_committed_records_behind_it_is_mid_file_corruption() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("hole.wal"); write_segment(&path, 6); @@ -279,7 +279,7 @@ mod tests { #[test] fn verify_rejects_a_hole_and_accepts_a_torn_tail() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let hole_path = dir.path().join("verify_hole.wal"); write_segment(&hole_path, 6); @@ -306,7 +306,7 @@ mod tests { #[test] fn clean_eof_needs_no_scan() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("clean.wal"); write_segment(&path, 2); assert!(verify_committed_prefix(&path, Some(StopReason::Eof), 2).is_ok()); diff --git a/nodedb-wal/src/uring_writer.rs b/nodedb-wal/src/uring_writer.rs index 6affceb8a..b783222f3 100644 --- a/nodedb-wal/src/uring_writer.rs +++ b/nodedb-wal/src/uring_writer.rs @@ -385,7 +385,7 @@ mod tests { #[test] fn uring_write_and_read_roundtrip() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("uring.wal"); { @@ -413,7 +413,7 @@ mod tests { #[test] fn uring_group_commit_many_records() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("uring_batch.wal"); { @@ -438,7 +438,7 @@ mod tests { #[test] fn uring_reopen_continues_lsn() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("uring_reopen.wal"); { diff --git a/nodedb-wal/src/writer/core.rs b/nodedb-wal/src/writer/core.rs index 9dde0f137..dd2df86b2 100644 --- a/nodedb-wal/src/writer/core.rs +++ b/nodedb-wal/src/writer/core.rs @@ -378,7 +378,7 @@ mod tests { #[test] fn write_and_sync_single_record() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -389,11 +389,20 @@ mod tests { writer.sync().unwrap(); assert!(writer.file_offset() > 0); + // `file_offset` alone cannot prove the flush wrote anything: wasm32 + // used to advance it, clear the buffer and report success without a + // single byte reaching the file (`flush_buffer` had no wasi arm). Pin + // the count on disk to the count the writer claims. + assert_eq!( + std::fs::metadata(&path).unwrap().len(), + writer.file_offset(), + "bytes on disk must equal bytes flushed" + ); } #[test] fn lsn_increments() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -415,7 +424,7 @@ mod tests { #[test] fn sealed_writer_rejects_writes() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -430,11 +439,15 @@ mod tests { /// The shape the durability barrier actually uses: many threads append /// through one shared writer and a single `sync()` covers the whole batch. /// Every record must survive, exactly once, with no LSN reused. + /// + /// Not run on wasm32: `std::thread::spawn` fails with `Unsupported` there, + /// and wasm targets abort on panic. It still runs in full on native. + #[cfg(not(target_arch = "wasm32"))] #[test] fn concurrent_appends_through_a_shared_writer_all_land() { use std::sync::{Arc, Mutex}; - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("test.wal"); let writer = Arc::new(Mutex::new( diff --git a/nodedb-wal/src/writer/durability.rs b/nodedb-wal/src/writer/durability.rs index 72ac9f8a3..3c6a4d382 100644 --- a/nodedb-wal/src/writer/durability.rs +++ b/nodedb-wal/src/writer/durability.rs @@ -126,7 +126,7 @@ mod tests { #[test] fn successful_fsync_clears_the_marker() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let file = File::create(dir.path().join("f")).unwrap(); let mut state = DurabilityState::new(); state.record_flush(); diff --git a/nodedb-wal/src/writer/flush.rs b/nodedb-wal/src/writer/flush.rs index aac2c1cdb..142c54e9b 100644 --- a/nodedb-wal/src/writer/flush.rs +++ b/nodedb-wal/src/writer/flush.rs @@ -1,9 +1,10 @@ // SPDX-License-Identifier: Apache-2.0 use crate::error::Result; -// Only the unix `pwrite` path constructs an error value directly; elsewhere -// failures propagate as `Result` from calls that build their own. -#[cfg(unix)] +// The unix `pwrite` path and the wasm seek+write path both construct an error +// value directly; elsewhere failures propagate as `Result` from calls that +// build their own. +#[cfg(any(unix, target_arch = "wasm32"))] use crate::error::WalError; use super::core::WalWriter; @@ -42,8 +43,12 @@ impl WalWriter { self.buffer.as_slice() }; - // Use pwrite to write at the exact offset, retrying on short writes. - #[cfg(unix)] + // Write at the exact offset. The unix path uses `pwrite` and retries + // short writes. `cfg(unix)` is false on wasm32-wasip1 (its + // `target_family` is "wasm", not "unix"), so wasi needs its own arm or + // this function would advance `file_offset`, clear the buffer and + // report the flush with nothing written at all. + #[cfg(all(unix, not(target_arch = "wasm32")))] { use std::os::unix::io::AsRawFd; let fd = self.file.as_raw_fd(); @@ -79,6 +84,21 @@ impl WalWriter { } } + // wasi has no positional write, and the file is only ever appended to, + // so seeking to the current end and writing is equivalent. The shared + // `file_offset` advance below accounts for the bytes; nothing is + // advanced here. + #[cfg(target_arch = "wasm32")] + { + use std::io::{Seek as _, SeekFrom, Write as _}; + self.file + .seek(SeekFrom::Start(self.file_offset)) + .map_err(WalError::Io)?; + self.file.write_all(data).map_err(|_| { + write_error("WAL segment append", self.file_offset, data.len() as u64) + })?; + } + self.file_offset += data.len() as u64; self.buffer.clear(); @@ -98,10 +118,11 @@ impl WalWriter { /// where the batch stalled and how much of it never reached the file, which is /// what a report needs to describe the write that could not complete. /// -/// Gated to match its only call site: the `pwrite` loop is unix-only, so on -/// other targets (wasm32) this would be dead code and a `-D warnings` build -/// would reject it. -#[cfg(unix)] +/// Gated to match its call sites: the `pwrite` loop is unix-only and the wasm +/// seek+write arm also classifies its failure through here. The ENOSPC branch +/// itself stays unix-only — on wasi `raw_os_error()` is `None`, so the generic +/// `WalError::Io` branch is what wasm takes. +#[cfg(any(unix, target_arch = "wasm32"))] fn write_error(context: &'static str, offset: u64, pending: u64) -> WalError { let err = std::io::Error::last_os_error(); #[cfg(unix)] diff --git a/nodedb-wal/src/writer/open.rs b/nodedb-wal/src/writer/open.rs index 8766f57f7..bbf7b0908 100644 --- a/nodedb-wal/src/writer/open.rs +++ b/nodedb-wal/src/writer/open.rs @@ -126,7 +126,7 @@ mod tests { #[test] fn buffered_open_creates_the_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("seg.wal"); let file = open_segment_file(&path, false).unwrap(); drop(file); @@ -143,7 +143,7 @@ mod tests { #[cfg(target_os = "linux")] #[test] fn classifying_a_direct_io_refusal_leaves_no_files_behind() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("seg.wal"); let err = std::io::Error::from_raw_os_error(libc::EINVAL); @@ -172,7 +172,7 @@ mod tests { /// about the filesystem's `O_DIRECT` support. #[test] fn unrelated_open_failure_is_not_reported_as_direct_io_unsupported() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::test_tempdir().unwrap(); let path = dir.path().join("absent").join("seg.wal"); let err = open_segment_file(&path, true).unwrap_err(); assert!( diff --git a/nodedb-wal/tests/faultbox_single_emit.rs b/nodedb-wal/tests/faultbox_single_emit.rs index bc122ce5f..b682ddd5a 100644 --- a/nodedb-wal/tests/faultbox_single_emit.rs +++ b/nodedb-wal/tests/faultbox_single_emit.rs @@ -21,6 +21,20 @@ use nodedb_wal::record::{HEADER_SIZE, RecordType, WAL_MAGIC}; use nodedb_wal::segmented::replay_all_segments; use nodedb_wal::{SegmentedWal, SegmentedWalConfig, WalError}; +/// Create a temporary directory that also works under `wasm32-wasip1`, where +/// `std::env::temp_dir()` is `unimplemented!()` (see the same helper in +/// `tests/wal_suite/main.rs`). Native keeps `tempfile::tempdir()`. +fn tempdir() -> std::io::Result { + #[cfg(target_arch = "wasm32")] + { + tempfile::Builder::new().tempdir_in(".") + } + #[cfg(not(target_arch = "wasm32"))] + { + tempfile::tempdir() + } +} + fn init_recorder(reports_dir: &Path) { assert!( faultbox::init( @@ -47,10 +61,10 @@ fn second_record_offset(path: &Path) -> u64 { #[test] fn one_damaged_segment_files_one_report() { - let reports = tempfile::tempdir().expect("reports temp dir"); + let reports = tempdir().expect("reports temp dir"); init_recorder(reports.path()); - let dir = tempfile::tempdir().expect("wal temp dir"); + let dir = tempdir().expect("wal temp dir"); let wal_dir = dir.path().join("wal"); { diff --git a/nodedb-wal/tests/wal_suite/cases/crash_injection.rs b/nodedb-wal/tests/wal_suite/cases/crash_injection.rs index 79654c3ce..40c16b0c3 100644 --- a/nodedb-wal/tests/wal_suite/cases/crash_injection.rs +++ b/nodedb-wal/tests/wal_suite/cases/crash_injection.rs @@ -32,7 +32,7 @@ fn read_lsns(path: &Path) -> Vec { #[test] fn out_of_space_on_sync_retains_the_batch_for_retry() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("enospc_sync.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -82,7 +82,7 @@ fn out_of_space_on_sync_retains_the_batch_for_retry() { #[test] fn out_of_space_on_append_does_not_burn_an_lsn() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("enospc_append.wal"); // A small buffer makes `append` itself flush once it fills up, which is @@ -140,7 +140,7 @@ fn out_of_space_on_append_does_not_burn_an_lsn() { #[test] fn crash_before_dwb_flush_keeps_the_acknowledged_prefix() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("crash_dwb.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -195,7 +195,7 @@ fn crash_before_dwb_flush_keeps_the_acknowledged_prefix() { #[test] fn crash_before_wal_fsync_never_resurrects_records_from_the_dwb() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("crash_fsync.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -225,7 +225,7 @@ fn crash_before_wal_fsync_never_resurrects_records_from_the_dwb() { #[test] fn crash_mid_truncate_leaves_a_replayable_log() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let mut wal = SegmentedWal::open(SegmentedWalConfig { diff --git a/nodedb-wal/tests/wal_suite/cases/crash_recovery.rs b/nodedb-wal/tests/wal_suite/cases/crash_recovery.rs index 7a3bc44a4..41aa58d63 100644 --- a/nodedb-wal/tests/wal_suite/cases/crash_recovery.rs +++ b/nodedb-wal/tests/wal_suite/cases/crash_recovery.rs @@ -39,7 +39,7 @@ fn read_all(path: &std::path::Path) -> Vec { #[test] fn crash_before_sync_loses_buffered_records() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("test.wal"); // Write and sync 3 records. @@ -64,7 +64,7 @@ fn crash_before_sync_loses_buffered_records() { #[test] fn torn_write_mid_header() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("test.wal"); write_records(&path, 5); @@ -89,7 +89,7 @@ fn torn_write_mid_header() { #[test] fn torn_write_mid_payload() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("test.wal"); write_records(&path, 3); @@ -126,7 +126,7 @@ fn torn_write_mid_payload() { #[test] fn corrupted_checksum_mid_file_fails_recovery_instead_of_truncating() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("test.wal"); write_records(&path, 5); @@ -189,9 +189,15 @@ fn corrupted_checksum_mid_file_fails_recovery_instead_of_truncating() { } } +// Needs the DWB splice-back, i.e. `DoubleWriteBuffer::recover_record`, which is +// a deliberate `Ok(None)` stub on wasm32 (`double_write/recover.rs`: slot +// recovery needs `pread`/O_DIRECT). Without it the reader stops at the torn +// record, so this is native-only; the surrounding recovery tests still run on +// wasm. +#[cfg(not(target_arch = "wasm32"))] #[test] fn dwb_recovery_mid_segment_keeps_every_later_record() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("dwb_mid.wal"); write_records(&path, 5); @@ -239,7 +245,7 @@ fn dwb_recovery_mid_segment_keeps_every_later_record() { #[test] fn corrupted_tail_record_is_a_torn_write() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("torn_tail.wal"); write_records(&path, 5); @@ -267,7 +273,7 @@ fn corrupted_tail_record_is_a_torn_write() { #[test] fn reopen_after_crash_continues_correctly() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("test.wal"); // Session 1: write 5 records. @@ -305,7 +311,7 @@ fn reopen_after_crash_continues_correctly() { #[test] fn idempotent_replay() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("test.wal"); write_records(&path, 10); @@ -323,7 +329,7 @@ fn idempotent_replay() { #[test] fn many_records_roundtrip() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("test.wal"); let count = 10_000u32; diff --git a/nodedb-wal/tests/wal_suite/cases/durability_fsync.rs b/nodedb-wal/tests/wal_suite/cases/durability_fsync.rs index d323cd862..c3537648e 100644 --- a/nodedb-wal/tests/wal_suite/cases/durability_fsync.rs +++ b/nodedb-wal/tests/wal_suite/cases/durability_fsync.rs @@ -39,7 +39,7 @@ fn append(wal: &mut SegmentedWal, payload: &[u8]) -> WalResult { /// not mistake emptiness for durability. #[test] fn retry_after_a_failed_fsync_never_reports_durability() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("failed_fsync.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -86,7 +86,7 @@ fn retry_after_a_failed_fsync_never_reports_durability() { /// if the fsync were issued, this would return an error. #[test] fn sync_with_nothing_outstanding_skips_the_fsync() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("noop_sync.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -104,7 +104,7 @@ fn sync_with_nothing_outstanding_skips_the_fsync() { /// record is acknowledged into an inode whose name may never reach disk. #[test] fn rollover_propagates_a_failed_directory_fsync() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let mut wal = SegmentedWal::open(rolling_config(dir.path().join("wal"))).unwrap(); append(&mut wal, b"first").unwrap(); @@ -124,7 +124,7 @@ fn rollover_propagates_a_failed_directory_fsync() { /// installed and unsealed, so the next append retries the roll and lands. #[test] fn a_failed_rollover_leaves_the_wal_writable() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let mut wal = SegmentedWal::open(rolling_config(dir.path().join("wal"))).unwrap(); append(&mut wal, b"first").unwrap(); @@ -168,7 +168,7 @@ fn a_failed_rollover_leaves_the_wal_writable() { /// back after a crash and replay below the checkpoint. #[test] fn truncation_propagates_a_failed_directory_fsync() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let mut wal = SegmentedWal::open(rolling_config(dir.path().join("wal"))).unwrap(); for i in 0..5u32 { @@ -189,7 +189,7 @@ fn truncation_propagates_a_failed_directory_fsync() { /// segment creation, rollover, and truncation all still work end to end. #[test] fn rollover_and_truncation_succeed_when_the_directory_fsync_works() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let mut wal = SegmentedWal::open(rolling_config(dir.path().join("wal"))).unwrap(); for i in 0..5u32 { diff --git a/nodedb-wal/tests/wal_suite/cases/durability_stress.rs b/nodedb-wal/tests/wal_suite/cases/durability_stress.rs index 5d4a598e5..30b6ff247 100644 --- a/nodedb-wal/tests/wal_suite/cases/durability_stress.rs +++ b/nodedb-wal/tests/wal_suite/cases/durability_stress.rs @@ -12,6 +12,12 @@ //! - All recovered records pass the built-in checksum check (WalReader //! already enforces this; stopping before a torn record is enough). +// The case spawns 8 writer threads, and wasm32 has no thread spawn: +// `std::thread::scope` fails with `Unsupported` there and wasm targets abort on +// panic. The whole file is one native-only case, so it is gated as a unit — +// equivalent to the `mod` gate in cases/mod.rs. +#![cfg(not(target_arch = "wasm32"))] + use std::time::Instant; use nodedb_wal::Result; @@ -84,7 +90,7 @@ fn parallel_writers_survive_truncation() { // Each thread gets its own tempdir so there's no shared state. let dirs: Vec = (0..WRITER_COUNT) - .map(|_| tempfile::tempdir().unwrap()) + .map(|_| crate::tempdir().unwrap()) .collect(); let paths: Vec = dirs.iter().map(|d| d.path().join("writer.wal")).collect(); diff --git a/nodedb-wal/tests/wal_suite/cases/dwb_protection.rs b/nodedb-wal/tests/wal_suite/cases/dwb_protection.rs index 87917a746..60e582f7b 100644 --- a/nodedb-wal/tests/wal_suite/cases/dwb_protection.rs +++ b/nodedb-wal/tests/wal_suite/cases/dwb_protection.rs @@ -33,7 +33,7 @@ fn replayed_payloads(path: &std::path::Path) -> Vec> { /// simply counts every append. #[test] fn mirrored_records_report_full_protection() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("healthy.wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -53,7 +53,7 @@ fn mirrored_records_report_full_protection() { /// was covered. #[test] fn oversized_record_is_reported_as_unprotected() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("oversized.wal"); let before = wal_dwb_unprotected_records_total(); @@ -85,7 +85,7 @@ fn oversized_record_is_reported_as_unprotected() { /// show up as one. #[test] fn a_disabled_dwb_is_not_reported_as_degraded() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("dwb_off.wal"); let mut writer = WalWriter::open( @@ -118,7 +118,7 @@ mod injected { /// pretend protection is still in place. #[test] fn dwb_write_failure_is_observable_and_the_append_still_succeeds() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("dwb_write_failure.wal"); let unprotected_before = wal_dwb_unprotected_records_total(); diff --git a/nodedb-wal/tests/wal_suite/cases/encrypted_replay.rs b/nodedb-wal/tests/wal_suite/cases/encrypted_replay.rs index 1b890a232..36a1a1215 100644 --- a/nodedb-wal/tests/wal_suite/cases/encrypted_replay.rs +++ b/nodedb-wal/tests/wal_suite/cases/encrypted_replay.rs @@ -10,6 +10,10 @@ //! the behaviour where consumers actually observe it. use nodedb_wal::crypto::{KeyRing, WalEncryptionKey}; +// `mmap_reader` is gated off on wasm32 (src/lib.rs:32-33), so the import and +// the one test that drives it are gated together. The other four cases here use +// only wasm-safe readers and stay in the wasm run. +#[cfg(not(target_arch = "wasm32"))] use nodedb_wal::mmap_reader::replay_segments_mmap; use nodedb_wal::record::RecordType; use nodedb_wal::segmented::{SegmentedWal, SegmentedWalConfig, replay_all_segments}; @@ -46,7 +50,7 @@ fn reopen_with_key(wal_dir: &std::path::Path) -> SegmentedWal { #[test] fn replay_of_encrypted_wal_returns_plaintext() { - let dir = tempfile::tempdir().expect("tempdir"); + let dir = crate::tempdir().expect("tempdir"); let wal_dir = dir.path().join("wal"); let payloads: Vec> = (0..8) @@ -75,9 +79,10 @@ fn replay_of_encrypted_wal_returns_plaintext() { } } +#[cfg(not(target_arch = "wasm32"))] #[test] fn mmap_replay_of_encrypted_wal_returns_plaintext() { - let dir = tempfile::tempdir().expect("tempdir"); + let dir = crate::tempdir().expect("tempdir"); let wal_dir = dir.path().join("wal"); let payloads: Vec> = (0..4).map(|i| format!("mmap-{i}").into_bytes()).collect(); @@ -93,7 +98,7 @@ fn mmap_replay_of_encrypted_wal_returns_plaintext() { #[test] fn tombstones_are_extracted_from_an_encrypted_wal() { - let dir = tempfile::tempdir().expect("tempdir"); + let dir = crate::tempdir().expect("tempdir"); let wal_dir = dir.path().join("wal"); let tombstone = CollectionTombstonePayload::new("users", 42) @@ -115,7 +120,7 @@ fn tombstones_are_extracted_from_an_encrypted_wal() { #[test] fn encrypted_replay_without_a_key_is_a_typed_error() { - let dir = tempfile::tempdir().expect("tempdir"); + let dir = crate::tempdir().expect("tempdir"); let wal_dir = dir.path().join("wal"); write_encrypted(&wal_dir, RecordType::Put as u32, &[b"secret".to_vec()]); @@ -133,7 +138,7 @@ fn encrypted_replay_without_a_key_is_a_typed_error() { #[test] fn unencrypted_replay_is_unchanged() { - let dir = tempfile::tempdir().expect("tempdir"); + let dir = crate::tempdir().expect("tempdir"); let wal_dir = dir.path().join("wal"); let payloads: Vec> = (0..5).map(|i| format!("plain-{i}").into_bytes()).collect(); diff --git a/nodedb-wal/tests/wal_suite/cases/faultbox_corruption_report.rs b/nodedb-wal/tests/wal_suite/cases/faultbox_corruption_report.rs index fddb307ed..164bdcc96 100644 --- a/nodedb-wal/tests/wal_suite/cases/faultbox_corruption_report.rs +++ b/nodedb-wal/tests/wal_suite/cases/faultbox_corruption_report.rs @@ -24,7 +24,7 @@ static REPORTS: OnceLock = OnceLock::new(); fn reports_dir() -> &'static Path { REPORTS .get_or_init(|| { - let dir = tempfile::tempdir().expect("reports temp dir"); + let dir = crate::tempdir().expect("reports temp dir"); faultbox::init( faultbox::Config::new("nodedb-wal-test", "0.0.0", dir.path()) // The test harness owns panic reporting; a chained hook here @@ -102,7 +102,7 @@ fn group_for(kind: &str) -> faultbox::reader::Group { #[test] fn recovery_reports_mid_file_corruption_and_preserves_the_segment() { let _ = reports_dir(); - let dir = tempfile::tempdir().expect("wal temp dir"); + let dir = crate::tempdir().expect("wal temp dir"); let (path, hole) = corrupt_segment(dir.path(), "hole.wal"); let err = recover(&path).expect_err("a hole must fail recovery"); diff --git a/nodedb-wal/tests/wal_suite/cases/faultbox_disabled.rs b/nodedb-wal/tests/wal_suite/cases/faultbox_disabled.rs index c43d89583..d4086e890 100644 --- a/nodedb-wal/tests/wal_suite/cases/faultbox_disabled.rs +++ b/nodedb-wal/tests/wal_suite/cases/faultbox_disabled.rs @@ -45,7 +45,7 @@ fn second_record_offset(path: &Path) -> u64 { #[test] fn a_hole_still_fails_recovery_with_the_same_error() { - let dir = tempfile::tempdir().expect("temp dir"); + let dir = crate::tempdir().expect("temp dir"); let path = dir.path().join("hole.wal"); write_segment(&path, 6); @@ -76,7 +76,7 @@ fn a_hole_still_fails_recovery_with_the_same_error() { #[test] fn a_healthy_segment_recovers_untouched() { - let dir = tempfile::tempdir().expect("temp dir"); + let dir = crate::tempdir().expect("temp dir"); let path = dir.path().join("clean.wal"); write_segment(&path, 4); diff --git a/nodedb-wal/tests/wal_suite/cases/mod.rs b/nodedb-wal/tests/wal_suite/cases/mod.rs index 515e7ce35..b045df12d 100644 --- a/nodedb-wal/tests/wal_suite/cases/mod.rs +++ b/nodedb-wal/tests/wal_suite/cases/mod.rs @@ -18,6 +18,13 @@ mod mmap_reader_madvise; mod o_direct_alignment; #[cfg(target_os = "linux")] mod o_direct_replay; +// Both reader cases are gated inside their own files with +// `#![cfg(not(target_arch = "wasm32"))]` rather than here: every test in them +// drives `nodedb_wal::mmap_reader`, which is gated off on wasm32, so the whole +// module is native-only and a wasm build must not resolve those imports at all. +// `encrypted_replay` is the opposite case and is gated inside itself for the +// same reason: four of its five tests use wasm-safe readers and keep running on +// wasm, only the one mmap replay case is native-only. mod reader_ciphertext_invariant; mod reader_preamble_and_continuity; mod segment_lsn_integrity; diff --git a/nodedb-wal/tests/wal_suite/cases/reader_ciphertext_invariant.rs b/nodedb-wal/tests/wal_suite/cases/reader_ciphertext_invariant.rs index 880b3f26e..8ca7fcc6c 100644 --- a/nodedb-wal/tests/wal_suite/cases/reader_ciphertext_invariant.rs +++ b/nodedb-wal/tests/wal_suite/cases/reader_ciphertext_invariant.rs @@ -9,6 +9,13 @@ //! walks an encrypted segment with no key at all, because `WalWriter::open` //! resumes a segment for appending and has no key ring to give it. +// `MmapWalReader` drives `nodedb_wal::mmap_reader`, which `src/lib.rs` gates +// off on wasm32, and every case here compares the mmap reader against the file +// reader. Gating the whole case keeps the wasm build from resolving the import +// at all — the equivalent of the `mod` gate in cases/mod.rs, which this file's +// target cannot use because that file is held elsewhere. +#![cfg(not(target_arch = "wasm32"))] + use std::path::{Path, PathBuf}; use nodedb_wal::{ @@ -51,7 +58,7 @@ fn write_segment(wal_dir: &Path, ring: Option) -> PathBuf { #[test] fn keyed_wal_reader_yields_plaintext_from_an_encrypted_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = write_segment(&dir.path().join("wal"), Some(test_ring())); let ring = test_ring(); @@ -74,7 +81,7 @@ fn keyed_wal_reader_yields_plaintext_from_an_encrypted_segment() { #[test] fn keyed_wal_reader_without_a_key_refuses_an_encrypted_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = write_segment(&dir.path().join("wal"), Some(test_ring())); let mut reader = WalReader::open(&path, None).unwrap(); @@ -87,7 +94,7 @@ fn keyed_wal_reader_without_a_key_refuses_an_encrypted_segment() { #[test] fn keyed_mmap_reader_yields_plaintext_from_an_encrypted_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = write_segment(&dir.path().join("wal"), Some(test_ring())); let ring = test_ring(); @@ -110,7 +117,7 @@ fn keyed_mmap_reader_yields_plaintext_from_an_encrypted_segment() { #[test] fn keyed_mmap_reader_without_a_key_refuses_an_encrypted_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = write_segment(&dir.path().join("wal"), Some(test_ring())); let mut reader = MmapWalReader::open(&path, None).unwrap(); @@ -122,7 +129,7 @@ fn keyed_mmap_reader_without_a_key_refuses_an_encrypted_segment() { #[test] fn recovery_scans_an_encrypted_segment_with_no_key_available() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let encrypted = write_segment(&dir.path().join("enc"), Some(test_ring())); let plain = write_segment(&dir.path().join("plain"), None); @@ -146,7 +153,7 @@ fn recovery_scans_an_encrypted_segment_with_no_key_available() { #[test] fn an_encrypted_wal_can_still_be_reopened_for_writing() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let path = write_segment(&wal_dir, Some(test_ring())); @@ -184,7 +191,7 @@ fn an_encrypted_wal_can_still_be_reopened_for_writing() { #[test] fn unencrypted_segments_read_the_same_through_both_constructors() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = write_segment(&dir.path().join("wal"), None); let keyed: Vec<_> = WalReader::open(&path, None) diff --git a/nodedb-wal/tests/wal_suite/cases/reader_preamble_and_continuity.rs b/nodedb-wal/tests/wal_suite/cases/reader_preamble_and_continuity.rs index 2cb7c944c..a1bafd47b 100644 --- a/nodedb-wal/tests/wal_suite/cases/reader_preamble_and_continuity.rs +++ b/nodedb-wal/tests/wal_suite/cases/reader_preamble_and_continuity.rs @@ -8,6 +8,13 @@ //! missing from the middle of the log) must be an error rather than a quiet //! truncation. +// `MmapWalReader` drives `nodedb_wal::mmap_reader`, which `src/lib.rs` gates +// off on wasm32, and the corrupted-segment cases compare it against the lazy +// reader. Gating the whole case keeps the wasm build from resolving the import +// at all — the equivalent of the `mod` gate in cases/mod.rs, which this file's +// target cannot use because that file is held elsewhere. +#![cfg(not(target_arch = "wasm32"))] + use std::path::{Path, PathBuf}; use nodedb_wal::{ @@ -61,7 +68,7 @@ fn write_plain_segment(wal_dir: &Path, payloads: &[&[u8]]) -> PathBuf { #[test] fn mmap_reader_replays_every_record_of_an_encrypted_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let payloads: Vec<&[u8]> = vec![b"enc-mmap-0", b"enc-mmap-1", b"enc-mmap-2"]; let path = write_encrypted_segment(&dir.path().join("wal"), &payloads); @@ -92,7 +99,7 @@ fn mmap_reader_replays_every_record_of_an_encrypted_segment() { #[test] fn lazy_reader_replays_every_record_of_an_encrypted_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let payloads: Vec<&[u8]> = vec![b"enc-lazy-0", b"enc-lazy-1", b"enc-lazy-2"]; let path = write_encrypted_segment(&dir.path().join("wal"), &payloads); @@ -127,7 +134,7 @@ fn lazy_reader_replays_every_record_of_an_encrypted_segment() { #[test] fn both_readers_still_replay_an_unencrypted_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let payloads: Vec<&[u8]> = vec![b"plain-0", b"plain-1", b"plain-2"]; let path = write_plain_segment(&dir.path().join("wal"), &payloads); @@ -182,7 +189,7 @@ fn record_header_offset(path: &Path, n: usize) -> u64 { #[test] fn mmap_replay_rejects_mid_file_corruption_instead_of_truncating() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let payloads: Vec<&[u8]> = vec![b"row-0", b"row-1", b"row-2", b"row-3", b"row-4", b"row-5"]; let path = write_plain_segment(&wal_dir, &payloads); @@ -225,7 +232,7 @@ fn write_many_segments(wal_dir: &Path, count: u32) { #[test] fn mmap_replay_rejects_a_missing_middle_segment() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); write_many_segments(&wal_dir, 20); @@ -249,7 +256,7 @@ fn mmap_replay_rejects_a_missing_middle_segment() { #[test] fn mmap_replay_accepts_a_truncated_prefix() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); write_many_segments(&wal_dir, 20); diff --git a/nodedb-wal/tests/wal_suite/cases/segment_lsn_integrity.rs b/nodedb-wal/tests/wal_suite/cases/segment_lsn_integrity.rs index 31c62ca59..19df44258 100644 --- a/nodedb-wal/tests/wal_suite/cases/segment_lsn_integrity.rs +++ b/nodedb-wal/tests/wal_suite/cases/segment_lsn_integrity.rs @@ -82,7 +82,7 @@ fn build_multi_segment_wal(wal_dir: &Path, records: usize) -> Vec { #[test] fn empty_rolled_segment_does_not_reuse_lsns() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let written = { @@ -120,7 +120,7 @@ fn empty_rolled_segment_does_not_reuse_lsns() { #[test] fn torn_last_segment_does_not_reuse_lsns() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); { @@ -162,7 +162,7 @@ fn torn_last_segment_does_not_reuse_lsns() { #[test] fn missing_middle_segment_fails_replay() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let paths = build_multi_segment_wal(&wal_dir, 20); @@ -189,7 +189,7 @@ fn missing_middle_segment_fails_replay() { #[test] fn missing_oldest_segments_replay_cleanly() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let paths = build_multi_segment_wal(&wal_dir, 20); @@ -208,7 +208,7 @@ fn missing_oldest_segments_replay_cleanly() { #[test] fn limited_replay_stopping_early_reports_no_gap() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let paths = build_multi_segment_wal(&wal_dir, 20); diff --git a/nodedb-wal/tests/wal_suite/cases/wal_collection_tombstone.rs b/nodedb-wal/tests/wal_suite/cases/wal_collection_tombstone.rs index 08feb85f8..8628fe60e 100644 --- a/nodedb-wal/tests/wal_suite/cases/wal_collection_tombstone.rs +++ b/nodedb-wal/tests/wal_suite/cases/wal_collection_tombstone.rs @@ -36,7 +36,7 @@ fn read_all(path: &std::path::Path) -> Vec { #[test] fn tombstone_record_roundtrips_through_wal() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -59,7 +59,7 @@ fn tombstone_record_roundtrips_through_wal() { #[test] fn extract_and_shadow_writes_before_purge_lsn() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -124,7 +124,7 @@ fn extract_and_shadow_writes_before_purge_lsn() { fn multiple_tombstones_keep_highest_purge_lsn() { // Simulates drop → recreate → drop on the same collection: the // second tombstone's purge_lsn shadows more history than the first. - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); @@ -167,7 +167,7 @@ fn multiple_tombstones_keep_highest_purge_lsn() { #[test] fn extract_ignores_unrelated_record_types() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let path = dir.path().join("wal"); let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); diff --git a/nodedb-wal/tests/wal_suite/cases/wal_encryption.rs b/nodedb-wal/tests/wal_suite/cases/wal_encryption.rs index 4cb23c39e..38f919cee 100644 --- a/nodedb-wal/tests/wal_suite/cases/wal_encryption.rs +++ b/nodedb-wal/tests/wal_suite/cases/wal_encryption.rs @@ -15,7 +15,7 @@ use nodedb_wal::{ #[test] fn wal_encryption_restart_roundtrip() { - let dir = tempfile::tempdir().unwrap(); + let dir = crate::tempdir().unwrap(); let wal_dir = dir.path().join("wal"); let kek = [0x42u8; 32]; diff --git a/nodedb-wal/tests/wal_suite/main.rs b/nodedb-wal/tests/wal_suite/main.rs index 57ff99f8e..70dfdb378 100644 --- a/nodedb-wal/tests/wal_suite/main.rs +++ b/nodedb-wal/tests/wal_suite/main.rs @@ -13,3 +13,22 @@ //! declaration in `cases/mod.rs`. mod cases; + +/// Create a temporary directory that also works under `wasm32-wasip1`. +/// +/// `tempfile::tempdir()` aborts there: `std::env::temp_dir()` is +/// `unimplemented!("not supported by WASI yet")` in std +/// (`library/std/src/sys/paths/wasi.rs`), so it fails before it ever reaches the +/// filesystem and no preopen or `TMPDIR` can help. The wasm runner preopens only +/// the working directory (`wasmtime --dir=.`), so on wasm the directory is +/// created under it; native keeps the exact `tempfile::tempdir()` behaviour. +pub(crate) fn tempdir() -> std::io::Result { + #[cfg(target_arch = "wasm32")] + { + tempfile::Builder::new().tempdir_in(".") + } + #[cfg(not(target_arch = "wasm32"))] + { + tempfile::tempdir() + } +} diff --git a/nodedb/src/control/planner/sql_plan_convert/dml/vector_primary.rs b/nodedb/src/control/planner/sql_plan_convert/dml/vector_primary.rs index a79cde8a2..96539addc 100644 --- a/nodedb/src/control/planner/sql_plan_convert/dml/vector_primary.rs +++ b/nodedb/src/control/planner/sql_plan_convert/dml/vector_primary.rs @@ -158,9 +158,9 @@ pub(in super::super) fn convert_vector_primary_insert( let key_column = declared.as_deref().unwrap_or(primary_key); let mut fields = row.payload_fields.clone(); if !is_auto_rowid_pk(primary_key) - && !fields + && fields .get(key_column) - .is_some_and(|v| !matches!(v, SqlValue::Null)) + .is_none_or(|v| matches!(v, SqlValue::Null)) { fields.insert(key_column.to_string(), SqlValue::String(doc_id.clone())); } diff --git a/nodedb/src/control/server/shared/ddl/neutral/database/clone.rs b/nodedb/src/control/server/shared/ddl/neutral/database/clone.rs index ffd340982..2a0fc6532 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/database/clone.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/database/clone.rs @@ -451,10 +451,9 @@ fn clone_chain_depth(state: &SharedState, start_db_id: DatabaseId) -> crate::Res /// Returns `Err` if the system clock is set before the Unix epoch — caller /// must surface the failure rather than silently substituting a sentinel. fn current_wall_ms() -> crate::Result { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) + nodedb_types::clock::since_epoch() .map(|d| d.as_millis() as i64) - .map_err(|e| crate::Error::Internal { - detail: format!("clone_database: system clock predates Unix epoch: {e}"), + .ok_or_else(|| crate::Error::Internal { + detail: "clone_database: system clock predates Unix epoch".to_owned(), }) } diff --git a/scripts/ci/check_wasm_probe.py b/scripts/ci/check_wasm_probe.py new file mode 100644 index 000000000..06f79a74b --- /dev/null +++ b/scripts/ci/check_wasm_probe.py @@ -0,0 +1,427 @@ +#!/usr/bin/env python3 +"""Check that `main` has a scheduled probe and that the wasm job runs tests. + +Both halves of this are invisible in review and easy to undo by accident: + + * a workflow that only fires on `pull_request` never reports on `main`, so a + red `main` waits for the next labelled PR to surface; + * a wasm job that runs `cargo check` never catches a target-specific runtime + panic, and `cargo check` without `--all-targets` never even builds a test + target -- so dev-dependency and test-only wasm breakage stays hidden. + +This gate asserts the shape that closes both, not the prose around it. Run +`--self-test` to prove the assertions still bite: it feeds each check a +configuration built to violate it and fails if the check accepts it. +""" +from __future__ import annotations + +import argparse +import re +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] + +CI_WORKFLOW = Path(".github/workflows/ci.yml") +TEST_WORKFLOW = Path(".github/workflows/test.yml") + +# The crates NodeDB-Lite compiles into its wasm build. Keep in sync with the +# `-p` list in test.yml's `wasm` job; the check below fails on drift either way. +SHARED_CRATES = frozenset({ + "nodedb-types", + "nodedb-codec", + "nodedb-physical", + "nodedb-mem", + "nodedb-crdt", + "nodedb-columnar", + "nodedb-fts", + "nodedb-graph", + "nodedb-spatial", + "nodedb-strict", + "nodedb-sql", + "nodedb-query", + "nodedb-array", + "nodedb-vector", + "nodedb-client", + "nodedb-wal", +}) + +WASM_TARGET = "wasm32-wasip1" +RUNNER_VAR = "CARGO_TARGET_WASM32_WASIP1_RUNNER" + +# The one nodedb-codec test skipped by name in the job: an approximate-recall +# floor that this target misses on float behaviour alone. It is listed here so +# that dropping the skip is a visible change rather than a silent one. The zstd +# *encoder* tests are NOT here -- those are ignored in-source by +# `#[cfg_attr(target_arch = "wasm32", ignore = ...)]`, because there the reason +# is categorical (the target has no encoder) rather than a moving threshold. +EXPECTED_SKIPS = ("top1_recall",) + +# Crates whose wasm test targets are built and executed by the job. All sixteen +# shared crates are in it, and the job's `-p` list must match this set exactly: +# a crate silently dropped from the run loses its coverage, and a crate added +# that cannot actually start on wasm fails the job. +WASM_TESTED_CRATES = frozenset({ + "nodedb-types", + "nodedb-codec", + "nodedb-physical", + "nodedb-mem", + "nodedb-crdt", + "nodedb-columnar", + "nodedb-fts", + "nodedb-graph", + "nodedb-spatial", + "nodedb-strict", + "nodedb-sql", + "nodedb-query", + "nodedb-array", + "nodedb-vector", + "nodedb-client", + "nodedb-wal", +}) + + +class Failure(Exception): + """One gate assertion did not hold.""" + + +def _load_yaml(path: Path) -> dict: + try: + import yaml + except ImportError as exc: # pragma: no cover - environment guard + raise Failure(f"PyYAML is required to parse {path}: {exc}") from exc + try: + return yaml.safe_load(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise Failure(f"{path} is missing") from exc + except yaml.YAMLError as exc: + raise Failure(f"{path} is not valid YAML: {exc}") from exc + + +def _workflow_on(document: dict, path: Path) -> dict: + # PyYAML resolves the bare `on:` key to the boolean True. + trigger = document.get("on", document.get(True)) + if not isinstance(trigger, dict): + raise Failure(f"{path}: `on:` must be a mapping, got {trigger!r}") + return trigger + + +def check_scheduled_main(root: Path) -> None: + """ci.yml must run on a timer as well as on a labelled PR.""" + document = _load_yaml(root / CI_WORKFLOW) + trigger = _workflow_on(document, CI_WORKFLOW) + if "workflow_dispatch" not in trigger: + raise Failure(f"{CI_WORKFLOW}: `workflow_dispatch` trigger was removed") + schedule = trigger.get("schedule") + if not schedule: + raise Failure( + f"{CI_WORKFLOW}: no `schedule` trigger -- nothing reports a red `main`" + ) + crons = [ + entry.get("cron") + for entry in schedule + if isinstance(entry, dict) and entry.get("cron") + ] + if not crons: + raise Failure(f"{CI_WORKFLOW}: `schedule` has no `cron` expression") + + +def _wasm_job(root: Path) -> tuple[dict, dict]: + document = _load_yaml(root / TEST_WORKFLOW) + jobs = document.get("jobs") + if not isinstance(jobs, dict): + raise Failure(f"{TEST_WORKFLOW}: `jobs` must be a mapping") + job = jobs.get("wasm") + if not isinstance(job, dict): + raise Failure(f"{TEST_WORKFLOW}: no `wasm` job") + return document, job + + +def _run_steps(job: dict) -> list[str]: + steps: list[str] = [] + for step in job.get("steps", []) or []: + if not isinstance(step, dict): + continue + # PyYAML resolves the bare `run:` key to the boolean True. + command = step.get("run", step.get(True)) + if isinstance(command, str): + steps.append(command) + return steps + + +def _command_lines(command: str) -> list[str]: + return [ + line.strip() + for line in command.splitlines() + if line.strip() and not line.strip().startswith("#") + ] + + +def _test_step_commands(job: dict) -> list[str]: + """Commands of the steps that actually execute a suite.""" + return [ + command + for command in _run_steps(job) + if any( + re.search(r"\bcargo (nextest run|test)\b", line) + for line in _command_lines(command) + ) + ] + + +def check_wasm_job_runs_tests(root: Path) -> None: + """The wasm job must execute the suites, not merely compile them. + + The `cargo check` shape is rejected only inside the step that is meant to + run tests. A separate compile-only step would be deliberate: it keeps crates + whose test targets cannot start on this target from dropping out of the job + entirely. + """ + _, job = _wasm_job(root) + if not _run_steps(job): + raise Failure(f"{TEST_WORKFLOW}: `wasm` job has no `run` steps") + + test_steps = _test_step_commands(job) + if not test_steps: + raise Failure( + f"{TEST_WORKFLOW}: `wasm` job never runs `cargo test`/`cargo nextest run`" + ) + for command in test_steps: + for line in _command_lines(command): + if "cargo check" in line: + raise Failure( + f"{TEST_WORKFLOW}: the test step of the `wasm` job still runs " + f"`cargo check` ({line!r}) -- a check cannot observe a runtime " + "panic" + ) + + +def check_wasm_target_and_runner(root: Path) -> None: + """The test run must target wasip1 and name a runtime that can execute it.""" + _, job = _wasm_job(root) + commands = "\n".join(_test_step_commands(job)) + if f"--target {WASM_TARGET}" not in commands: + raise Failure(f"{TEST_WORKFLOW}: `wasm` job does not target {WASM_TARGET}") + + env = job.get("env") or {} + runner = env.get(RUNNER_VAR, "") + if not isinstance(runner, str) or "wasmtime" not in runner: + raise Failure( + f"{TEST_WORKFLOW}: {RUNNER_VAR} must name a wasm runtime; got {runner!r}" + ) + if "max-wasm-stack" not in runner: + raise Failure( + f"{TEST_WORKFLOW}: {RUNNER_VAR} must raise the wasm stack " + "(unoptimized test binaries overflow the runtime default)" + ) + + +def check_wasm_job_covers_shared_crates(root: Path) -> None: + """The job's package lists must match what can run, plus a compile fallback. + + Checked in both directions. A crate silently dropped from the test step + loses its coverage; a crate added back that cannot start on this target + would fail the job; and every shared crate must appear in one of the two + steps, so nothing is dropped from the job entirely. + """ + _, job = _wasm_job(root) + test_commands = "\n".join(_test_step_commands(job)) + tested = set(re.findall(r"-p\s+([A-Za-z0-9_-]+)", test_commands)) + + missing = sorted(WASM_TESTED_CRATES - tested) + if missing: + raise Failure( + f"{TEST_WORKFLOW}: `wasm` job does not run tests for " + f"{', '.join(missing)}" + ) + unexpected = sorted(tested - WASM_TESTED_CRATES) + if unexpected: + raise Failure( + f"{TEST_WORKFLOW}: `wasm` job runs tests for {', '.join(unexpected)}, " + "which is not in WASM_TESTED_CRATES -- either its test target cannot " + "start on wasm, or this list needs updating with the evidence" + ) + + all_commands = "\n".join(_run_steps(job)) + every_package = set(re.findall(r"-p\s+([A-Za-z0-9_-]+)", all_commands)) + uncovered = sorted(SHARED_CRATES - every_package) + if uncovered: + raise Failure( + f"{TEST_WORKFLOW}: `wasm` job never mentions {', '.join(uncovered)}; " + "a shared crate that cannot run its tests still has to compile" + ) + + +def check_codec_skips_present(root: Path) -> None: + """The one native-only codec test must stay skipped by name.""" + _, job = _wasm_job(root) + commands = "\n".join(_test_step_commands(job)) + for name in EXPECTED_SKIPS: + if f"--skip {name}" not in commands: + raise Failure( + f"{TEST_WORKFLOW}: `wasm` job no longer skips {name!r}; if that " + "test was made to pass on wasm, drop it from EXPECTED_SKIPS here" + ) + + +CHECKS = ( + ("scheduled-main", check_scheduled_main), + ("wasm-job-runs-tests", check_wasm_job_runs_tests), + ("wasm-target-and-runner", check_wasm_target_and_runner), + ("wasm-job-covers-shared-crates", check_wasm_job_covers_shared_crates), + ("codec-skips-present", check_codec_skips_present), +) + + +def run_checks(root: Path) -> list[str]: + failures: list[str] = [] + for name, check in CHECKS: + try: + check(root) + except Failure as exc: + failures.append(f"[{name}] {exc}") + return failures + + +def self_test(root: Path) -> int: + """Feed every check a configuration that violates exactly it. + + A gate nobody has seen fail is a gate that may already be vacuous: these + mutations prove each assertion still rejects the shape it exists to catch. + """ + import tempfile + + # name -> (file to mutate, mutation, description, checks the mutation is + # expected to trip). The expected set is declared rather than assumed to be + # exactly one: some mutations cascade by design -- downgrading the test run + # to a check also removes the only test step, so the target and coverage + # checks necessarily see it too. Declaring the cascade keeps the self-test + # honest about both an under-reaction and an over-reaction. + mutations = { + "scheduled-main": ( + CI_WORKFLOW, + lambda text: re.sub( + r"^ schedule:\n(?: .*\n)+", "", text, flags=re.MULTILINE + ), + "remove the schedule", + ("scheduled-main",), + ), + "wasm-job-runs-tests": ( + TEST_WORKFLOW, + lambda text: text.replace( + "cargo test --profile ci --target wasm32-wasip1", + "cargo check --profile ci --target wasm32-wasip1", + ), + "downgrade the test run to a check", + ( + "wasm-job-runs-tests", + "wasm-target-and-runner", + "wasm-job-covers-shared-crates", + "codec-skips-present", + ), + ), + "wasm-target-and-runner": ( + TEST_WORKFLOW, + lambda text: text.replace( + "CARGO_TARGET_WASM32_WASIP1_RUNNER: wasmtime", + "CARGO_TARGET_WASM32_WASIP1_RUNNER: echo", + ), + "replace the runtime with a no-op", + ("wasm-target-and-runner",), + ), + "wasm-job-covers-shared-crates": ( + TEST_WORKFLOW, + lambda text: text.replace( + "-p nodedb-crdt -p nodedb-columnar -p nodedb-fts -p nodedb-graph \\\n", + "-p nodedb-crdt -p nodedb-columnar -p nodedb-graph \\\n", + ), + "drop a crate from the package list", + ("wasm-job-covers-shared-crates",), + ), + "codec-skips-present": ( + TEST_WORKFLOW, + lambda text: text.replace("--skip top1_recall", ""), + "un-skip a native-only test", + ("codec-skips-present",), + ), + } + + failures: list[str] = [] + for name, (relative, mutate, description, expected) in mutations.items(): + with tempfile.TemporaryDirectory() as tmp: + mutated_root = Path(tmp) + for workflow in (CI_WORKFLOW, TEST_WORKFLOW): + destination = mutated_root / workflow + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text( + (root / workflow).read_text(encoding="utf-8"), encoding="utf-8" + ) + target = mutated_root / relative + before = target.read_text(encoding="utf-8") + after = mutate(before) + if after == before: + failures.append( + f"[{name}] the self-test mutation changed nothing " + f"({description}) -- it no longer models the violation" + ) + continue + target.write_text(after, encoding="utf-8") + + tripped = { + check_name + for failure in run_checks(mutated_root) + for check_name, _ in CHECKS + if failure.startswith(f"[{check_name}]") + } + if name not in tripped: + failures.append( + f"[{name}] accepted a configuration that {description}" + ) + unexpected = sorted(tripped - set(expected)) + if unexpected: + failures.append( + f"[{name}] mutation {description} also tripped " + f"{unexpected}, which the self-test did not expect" + ) + return failures + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--self-test", + action="store_true", + help="prove each check still rejects the configuration it exists to catch", + ) + parser.add_argument( + "--root", + type=Path, + default=ROOT, + help="repository root (defaults to this script's repository)", + ) + args = parser.parse_args() + root = args.root.resolve() + + if args.self_test: + failures = self_test(root) + if failures: + for failure in failures: + print(f"FAIL {failure}", file=sys.stderr) + return 1 + print(f"self-test: all {len(CHECKS)} checks reject their violation") + return 0 + + failures = run_checks(root) + if failures: + for failure in failures: + print(f"FAIL {failure}", file=sys.stderr) + return 1 + print( + f"wasm probe: ci.yml is scheduled on main; the wasm job executes " + f"{len(WASM_TESTED_CRATES)} shared crates on {WASM_TARGET}" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main())