diff --git a/CHANGELOG.md b/CHANGELOG.md index e831114ac..45b30ca21 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **`:query-memory-limit`, limit bind slots and `PreparedQuery` limit setters** (#870, part of #862). `:query-memory-limit N` replaces `OpenOptions::query_memory_limit` for one query, up or down; `0` is allowed and lets only a query that holds no state run (QRY-012 otherwise). It is gated by the new `Overridable::QueryMemoryLimit`, which is **not** in `Overridable::DEFAULTS`, so a database refuses it with API-025 unless the embedder lists it; `BrowserDb` always refuses it. `:max-derived-facts`, `:max-results` and `:query-memory-limit` take a `$slot` filled with `BindValue::Val(Value::Integer(n))` (INT-014/INT-015 out of range as for a literal, API-024 for another type, API-022 when missing). `PreparedQuery::with_max_derived_facts` and `with_query_memory_limit` replace the database values for one statement; they are not gated, and an allowed keyword still overrides them. See [docs/CONFIGURATION.md](docs/CONFIGURATION.md). The v3 binary-size gate rises from 1,234 to 1,240 KiB (+672 B measured here; the rest is room for #861). +- **Per-transaction WAL sync mode and chunk size: `TxOptions` and `Minigraf::begin_write_with`** (#871, part of #862). `TxOptions` (`#[non_exhaustive]`; fields `synchronous: Option` and `chunk_facts: Option`, builders `new`, `synchronous`, `chunk_facts`) sets either option for one `WriteTransaction`; `None` means the database's value, and `begin_write()` is `begin_write_with(TxOptions::default())`. `synchronous` decides whether that transaction's WAL entry is synced, whatever `OpenOptions::synchronous` is, so a bulk load can commit with `Normal` on a `Full` database; a `Full` commit syncs the whole WAL, so earlier `Normal` entries become durable with it. It has no effect on a transaction that becomes large (a checkpoint, durable in both modes; its WAL guard header is always synced, a deviation from the #862 design recorded there) or on an in-memory database. `chunk_facts` replaces `OpenOptions::transaction_chunk_facts` for that transaction, with the same clamp (1..=1,000,000). The options are not stored in the file, not gated by `allow_overrides`, and apply to that transaction only. `begin_write_with` fails as `begin_write` does (`INT-001`, `API-014`). The WAL writer no longer keeps a sync mode: each append is given its transaction's. See [docs/DURABILITY.md](docs/DURABILITY.md) for what survives a crash when modes are mixed. + - **Configurable recursion round limit** (#861, part of #862). `OpenOptions::max_iterations` sets the most rounds one stratum of a rule query may run (default 1,000,000, see Breaking changes); `:max-iterations N` or `$slot` replaces it for one query, up or down (INT-012/013/014/015 as for `:max-derived-facts`), gated by the new `Overridable::MaxIterations`, which **is** in `Overridable::DEFAULTS`; `PreparedQuery::with_max_iterations` replaces it for one statement, ungated, and an allowed keyword still overrides it. Rounds are still counted per stratum. See [docs/CONFIGURATION.md](docs/CONFIGURATION.md). ### Changed diff --git a/CLAUDE.md b/CLAUDE.md index c15a2dc1d..551386011 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -187,13 +187,13 @@ cargo run < demos/demo_negation.txt 5. **`src/repl.rs`** — Interactive REPL; TTY-aware (suppresses prompts/banner for piped input) -6. **`src/db.rs`** — Public API: `Minigraf::open/execute/query/fact_log/prepare/begin_write/checkpoint/save/verify/rebuild_indexes`, `WriteTransaction`, `IntegrityReport`, `OpenOptions::page_cache_size`, `OpenOptions::read_only` (#429: shared lock, nothing written, writes are `API-014`), `OpenOptions::query_memory_limit` (#549: `QRY-012`, off by default), `OpenOptions::transaction_chunk_facts` (#550), `OpenOptions::max_iterations` (#861: rounds per stratum, default 1,000,000), `OpenOptions::allow_overrides` (#869: which query keywords may override settings, `API-025`; see `docs/CONFIGURATION.md`) +6. **`src/db.rs`** — Public API: `Minigraf::open/execute/query/fact_log/prepare/begin_write/begin_write_with/checkpoint/save/verify/rebuild_indexes`, `WriteTransaction`, `TxOptions` (#871: one transaction's `synchronous` and `chunk_facts`; `None` = the database value; the WAL writer keeps no mode, each append gets its transaction's; no effect on a large transaction, whose WAL guard is always synced), `IntegrityReport`, `OpenOptions::page_cache_size`, `OpenOptions::read_only` (#429: shared lock, nothing written, writes are `API-014`), `OpenOptions::query_memory_limit` (#549: `QRY-012`, off by default), `OpenOptions::transaction_chunk_facts` (#550), `OpenOptions::max_iterations` (#861: rounds per stratum, default 1,000,000), `OpenOptions::allow_overrides` (#869: which query keywords may override settings, `API-025`; see `docs/CONFIGURATION.md`) - `src/overrides.rs`: `Overridable` (`DEFAULTS`, `keyword`; `QueryMemoryLimit` not in `DEFAULTS`, #870; `MaxIterations` in it, #861) and `check_overrides`, called after parsing at every query entry point (`BrowserDb` uses `DEFAULTS`) - `src/cursor.rs`: `Cursor` / `Batch` — owned, `Send` cursor returned by `Minigraf::query` and `PreparedQuery::query` (#432); pulls engine queries one internal batch at a time (`close`/drop stops the work; a read error after some rows comes on the next call), other queries computed at open; holds a `Snapshot` (one state of the store, its committed generation pinned, #543) until dropped - `src/fact_log.rs`: `FactLog` / `FactFilter` / `FactRecord` / `FactOrder` — every fact version streamed by `Minigraf::fact_log` (#430): tx order (multi-pass, bounded window) or storage order, key-level filters; an open log pins the committed generation (`storage::pins`), so checkpoints run but do not reuse its pages (`API-013` deprecated, #543) - `src/log_writer.rs`: `LogWriter` — builds a new file from `FactRecord`s with their tx and valid-time kept (#431): tx-ordered appends with holes (`API-015`/`API-016`), batched copy-on-write checkpoints at tx boundaries, no WAL, built at `.partial` and renamed by `finish` (`STG-043` if the target exists) -7. **`src/wal.rs`** — Fact-level sidecar WAL, CRC32-protected entries, crash recovery. Like `storage::verify` and `storage::dir_sync`, not compiled for wasm32 (no file backend there) +7. **`src/wal.rs`** — Fact-level sidecar WAL, CRC32-protected entries, crash recovery; `append_entry` takes the `SyncMode` of each append (#871), and in tests the writer tracks its synced length (lost-sync fault, `Minigraf::power_loss`). Like `storage::verify` and `storage::dir_sync`, not compiled for wasm32 (no file backend there) 8. **`src/error.rs`** — Structured error codes: `MinigrafError`, `ErrorCategory`, `ErrorCode` registry (PRS/QRY/STG/WAL/API/INT codes); must match `docs/ERROR_REFERENCE.md`. `MinigrafError::invalid_argument` (API-017) and `MinigrafError::closed` (API-018) exist for language bindings @@ -250,7 +250,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**2108 tests** (2087 active, 21 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). +**2132 tests** (2111 active, 21 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). See `docs/TEST_COVERAGE.md` for the full per-file breakdown. The #760 deterministic storage suite adds three native unit tests in a separate `--cfg minigraf_shuttle` build, outside the default count. Run only `graph::storage::deterministic::` with that cfg; ordinary unit tests do not enter its scheduler. `deterministic-schedules.yml` runs relevant v3 PRs on Linux/macOS/Windows plus Linux MSRV, and larger nightly/manual searches with replay artifacts. The identical main workflow activates the v3 schedule after both PRs merge. See `docs/testing/deterministic-schedules.md` for scope, controls and findings. diff --git a/README.md b/README.md index ba62f3edc..62ad0e7ef 100644 --- a/README.md +++ b/README.md @@ -316,7 +316,7 @@ Point lookups, time travel on one entity, and joins and rule queries that a cons File-backed databases limit string values to 4,068 bytes (`WAL-003`). In-memory databases have no limit. -**Durability tuning:** every write is `fsync`'d immediately by default (`SyncMode::Full`). Bulk loaders/migrations that can safely re-run from a checkpoint watermark on failure can trade that for throughput with `OpenOptions::new().synchronous(SyncMode::Normal)` — `checkpoint()` still fsyncs unconditionally in both modes. See the [Performance Tuning](https://project-minigraf.github.io/minigraf-docs/latest/performance-tuning/#configuration-knobs) for the full tradeoff and the write-batching pattern that pairs with it. +**Durability tuning:** every write is `fsync`'d immediately by default (`SyncMode::Full`). Bulk loaders/migrations that can safely re-run from a checkpoint watermark on failure can trade that for throughput with `OpenOptions::new().synchronous(SyncMode::Normal)`, or for one transaction with `db.begin_write_with(TxOptions::new().synchronous(SyncMode::Normal))` on an otherwise `Full` database — `checkpoint()` still fsyncs unconditionally in both modes, and a later `Full` commit also makes earlier `Normal` ones durable. See the [Performance Tuning](https://project-minigraf.github.io/minigraf-docs/latest/performance-tuning/#configuration-knobs) for the full tradeoff and the write-batching pattern that pairs with it. **Configuration:** [docs/CONFIGURATION.md](docs/CONFIGURATION.md) lists every `OpenOptions` setting, which ones query text can override, and how `OpenOptions::allow_overrides` lets the embedder refuse those keywords (`API-025`, #869). diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 0217413eb..2485bf9ee 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -21,8 +21,8 @@ code are not gated. | `max_results` (deprecated) | no limit of its own | query text: `:max-results N` or `$slot` | Alone, replaces the database value; with `:max-derived-facts`, the lower applies | | `query_memory_limit` | no limit | query text: `:query-memory-limit N` or `$slot` (not in `DEFAULTS`); code: `PreparedQuery::with_query_memory_limit` | Replaces the database value for that query, up or down; a keyword overrides the setter | | `max_iterations` | 1,000,000 rounds per stratum | query text: `:max-iterations N` or `$slot`; code: `PreparedQuery::with_max_iterations` | Replaces the database value for that query, up or down; a keyword overrides the setter | -| `transaction_chunk_facts` | 65,536 | — (per-transaction `TxOptions` planned, #871) | — | -| `synchronous` | `SyncMode::Full` | — (per-transaction `TxOptions` planned, #871) | — | +| `transaction_chunk_facts` | 65,536 | code: `TxOptions::chunk_facts` with `Minigraf::begin_write_with` | Replaces the database value for that `WriteTransaction` only, up or down, with the same clamp (1..=1,000,000); later transactions and `execute` use the database value | +| `synchronous` | `SyncMode::Full` | code: `TxOptions::synchronous` with `Minigraf::begin_write_with` | Decides whether that `WriteTransaction`'s WAL entry is synced, either way; later transactions and `execute` use the database value. A `Full` commit also makes earlier `Normal` ones durable. No effect on a large transaction (always durable) or an in-memory database. See [DURABILITY.md](DURABILITY.md) | | `wal_checkpoint_threshold` | 1000 | open only | Database-wide checkpoint policy | | `page_cache_size` | 256 pages | open only | Shared cache | | `allow_unlocked` | `false` | open only | Decided when the file is opened | diff --git a/docs/DURABILITY.md b/docs/DURABILITY.md index 607307f08..5410fcb78 100644 --- a/docs/DURABILITY.md +++ b/docs/DURABILITY.md @@ -43,6 +43,17 @@ A crash during a checkpoint is no different: the file keeps the previous checkpo Batching is the other lever: `begin_write()`, several `execute()` calls and one `commit()` write one WAL entry with one sync, in either mode. +### Per-transaction mode + +`Minigraf::begin_write_with(TxOptions::new().synchronous(mode))` chooses the mode of one `WriteTransaction`'s commit, whatever the database's mode (#871); `begin_write()` and `execute()` use the database's. The mode is not kept: the next transaction uses the database's again. So one WAL can hold entries committed both ways: + +- A **process crash** keeps every committed transaction, in either mode. +- A **power loss** keeps a prefix of the commit order, never a transaction in part and never a later one without an earlier one. The prefix includes every transaction up to the last `Full` commit, the last checkpoint, or whichever is later: a `Full` commit syncs the whole WAL file, so the `Normal` entries before it become durable with it. Entries committed with `Normal` after the last `Full` commit may be lost. +- A `Normal` commit never weakens a transaction committed before it. +- A **large transaction** (more facts than `transaction_chunk_facts`) commits as a checkpoint, with no WAL entry, so it is durable in both modes; the header of the WAL it creates before its first page write is always synced. The per-transaction mode has no effect on it. + +For example, a database opened `Full` can run a bulk load as `Normal` transactions and end it with one `Full` transaction (or a `checkpoint()`), which makes the whole load durable at once. + **What durability assumes.** A sync is only as good as the storage under it. Minigraf assumes that when `fdatasync`/`fsync` (`FlushFileBuffers` on Windows) returns, the data is on stable storage. Disks or controllers with a volatile write cache that ignore flushes, virtual machines with an "unsafe" disk cache mode, and some network or FUSE filesystems break that assumption, and a power loss there can lose committed transactions or tear a page. A torn or rotted page is detected by its checksum on the next read (`STG-029`); it is not silently read. **Browser (`BrowserDb`).** There is no WAL. Each `transact` and `retract` is a checkpoint whose pages, meta page and freed page keys are written in one IndexedDB `readwrite` transaction, and the returned promise resolves after that transaction completes. Closing the tab before then loses that write, whole. How durable a completed IndexedDB transaction is on a power loss is up to the browser. diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index c6433b995..7c2aa481c 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,12 +4,12 @@ | | Default features | With `--features test-hooks` | |---|---|---| -| **Active** | 2,087 | 2,125 | +| **Active** | 2,111 | 2,149 | | **Ignored** | 21 | 30 | -| **Total** | 2,108 | 2,155 | +| **Total** | 2,132 | 2,179 | Counts above are the current Cargo test inventory (active versus ignored), -including #777, #789, #781, #794, #795, #837, #865, #869, #870, #861 and #863 (Linux). Verification results below identify which suites ran. +including #777, #789, #781, #794, #795, #837, #865, #869, #870, #861, #863 and #871 (Linux). Verification results below identify which suites ran. **#863 nightly candidates**: four engine unit tests check the `test-hooks` path events and memory peaks (both builds). With `test-hooks` only, @@ -219,6 +219,7 @@ These tests compare the database against an independent answer rather than a sin - Kills a child process running a seeded workload (multi-valued batches, batched retracts, five checkpoint modes) at a random point. - The reopened file must hold exactly the transactions the child logged as committed, or those plus the one in flight. - Checked through full-scan, AEVT, AVET, EAVT and `:as-of` queries and `verify`; again after a second reopen; and after a checkpoint plus reopen. + - A `PerTxSync` variant (#871, every build) commits every transaction through `begin_write_with` with a sync mode chosen per transaction from the seed (the first `Normal`, the second `Full`) on a database opened `Full`. - With `test-hooks`, a `Large` variant makes most transactions large, runs every third one as a `WriteTransaction` of three statements, and rolls back a large statement after writing pages (#550). - **Golden-file corpus** (`tests/golden_corpus_test.rs`, #391 *(v2.x)*) - `tests/golden/` holds v7 files written by minigraf 2.0.3: one checkpoint, several checkpoints, indexes rebuilt on open, a crash-left WAL, an already-checkpointed WAL, same-transaction multi-values. @@ -396,6 +397,14 @@ These are not `#[test]`s; run them with `cargo kani` or `.github/workflows/kani. - An append that fails part-way makes every later append `WAL-007`. A rejected oversize fact (`WAL-003`, nothing written) does not. - Through `Minigraf`: the failed transaction is not applied, later `transact` and `commit` are `WAL-007`, and a `checkpoint()` makes writes work again. - In `tests/wal_test.rs`: a commit made after reopening a crash-left WAL with each kind of torn tail survives a second crash. +- **Per-transaction options** (full default and `test-hooks` suites passed on the alternate bench machine with Rust 1.89, 2026-11-10; `src/tx_options_tests.rs`, 17 tests; `src/wal.rs`, 2 tests; `src/fault_matrix.rs`, 3 tests; 2 doctests; `src/db.rs`, 1 test rewritten; #871) + - `TxOptions` builders and fields; `begin_write_with` fails with `INT-001` and `API-014` as `begin_write` does and leaves the handle usable. + - A small commit syncs the WAL exactly when its effective mode is `Full`, for the database modes × `None`/`Full`/`Normal`, whether the commit creates the WAL or appends to it. The mode never stays with the WAL writer: implicit transactions and `begin_write()` follow the database mode after a commit, a rollback or a drop, including after a `Normal` transaction that became large (creating the WAL guard) and was rolled back. The stored-mode test of #302 now checks that a `Normal` database's commits leave the WAL unsynced at both writer call sites. + - Test models: a lost sync loses every WAL byte since the last good sync, including whole `Normal` entries; `Minigraf::power_loss` ends a handle with the WAL cut back to its last sync. + - On `Full` and `Normal` databases: `Normal` then kill keeps A; `Normal` then power loss keeps a prefix (A lost), with a later commit surviving a kill; `Normal`, `Full`, power loss keeps both; `Full`, `Normal`, power loss keeps A. Every reopen is compared record by record through `fact_log`. + - A large `Normal` transaction survives a power loss whole with `verify` clean; a `Normal` commit at the checkpoint threshold still checkpoints; options on an in-memory database change nothing. + - `chunk_facts`: `Some(2)` makes three facts large (one statement or staged ones), `Some(1000)` keeps them small on a database with chunk 2, `0` acts as 1 and `usize::MAX` as 1,000,000; it applies to that transaction only; the thread-local test hook still wins. The facts committed by the same statements are the same with any options. + - Fault matrix: every fault at every write and sync of a `Full` commit after a `Normal` one, a `Normal` commit on a `Full` database (it makes no sync, so the sync faults never fire there), and a `Full` commit on a `Normal` database. The reopened file must be a prefix of the commit order from the last sync or checkpoint, never part of a transaction; only a lost sync may lose a `Normal` entry; `verify` is clean and a commit after recovery survives a kill. - **WAL entries of any size** (`src/wal.rs`, 3 tests, one replacing the 1,000,000-fact cap test; #686) - An entry of 1,000,001 facts and the entries after it replay, and reopening the writer cuts none of them off. - A fact count the bytes after it cannot hold (one more than a quarter of them, or `u64::MAX`) ends replay at that entry, and reopening the writer cuts the WAL off there. diff --git a/docs/superpowers/specs/2026-11-06-862-override-audit-design.md b/docs/superpowers/specs/2026-11-06-862-override-audit-design.md index e7dca5303..74d4d25ca 100644 --- a/docs/superpowers/specs/2026-11-06-862-override-audit-design.md +++ b/docs/superpowers/specs/2026-11-06-862-override-audit-design.md @@ -111,6 +111,7 @@ impl Minigraf { pub fn begin_write_with(&self, opts: TxOptions) -> Result usize { /// /// Independent of [`Minigraf::checkpoint`], which always fsyncs the main /// `.graph` file regardless of this setting — see [`OpenOptions::synchronous`]. +/// The database's mode is set at open; one `WriteTransaction` can choose its +/// own with [`TxOptions::synchronous`] and [`Minigraf::begin_write_with`]. +/// A `Full` commit syncs the whole WAL, so it also makes earlier `Normal` +/// commits durable. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SyncMode { /// `fdatasync` after every WAL write. Default; matches Minigraf's @@ -175,6 +179,9 @@ pub struct OpenOptions { /// setting. /// /// No effect on in-memory databases, which have no WAL. + /// + /// One `WriteTransaction` can use the other mode: see + /// [`TxOptions::synchronous`] and [`Minigraf::begin_write_with`]. pub synchronous: SyncMode, /// Open the database read-only. Defaults to `false`. /// @@ -221,6 +228,9 @@ pub struct OpenOptions { /// Kept between 1 and 1,000,000 (the most facts one WAL entry holds). In-memory /// databases never write chunks; a statement is still read a chunk at a /// time. + /// + /// One `WriteTransaction` can use another size: see + /// [`TxOptions::chunk_facts`] and [`Minigraf::begin_write_with`]. pub transaction_chunk_facts: usize, /// The query keywords that may override this database's settings (#862). /// Defaults to [`Overridable::DEFAULTS`], every keyword that could @@ -411,6 +421,82 @@ impl OpenOptions { } } +// ─── TxOptions ─────────────────────────────────────────────────────────────── + +/// Options for one explicit write transaction, passed to +/// [`Minigraf::begin_write_with`] (#871). +/// +/// Each field is `None` by default, which means the database's value from +/// [`OpenOptions`]. A value applies to that transaction only: later +/// transactions, and `transact`/`retract` through [`Minigraf::execute`], use +/// the database's values again. Nothing about it is stored in the file. +/// +/// `#[non_exhaustive]`: construct via [`TxOptions::new`]/[`TxOptions::default`] +/// plus the builder methods. +/// +/// ```no_run +/// # #[cfg(not(target_arch = "wasm32"))] { +/// use minigraf::{OpenOptions, SyncMode, TxOptions}; +/// +/// let db = OpenOptions::new().path("bulk.graph").open().unwrap(); +/// // A bulk load that can be re-run: no WAL sync at its commit. +/// let mut tx = db +/// .begin_write_with(TxOptions::new().synchronous(SyncMode::Normal)) +/// .unwrap(); +/// tx.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap(); +/// tx.commit().unwrap(); +/// # } +/// ``` +#[derive(Debug, Clone, Default, PartialEq, Eq)] +#[non_exhaustive] +pub struct TxOptions { + /// The WAL durability mode of this transaction's commit, or `None` for + /// [`OpenOptions::synchronous`]. + /// + /// It decides only whether the WAL entry this transaction's commit + /// appends is synced: `Full` syncs it, `Normal` does not, whatever the + /// database's mode. A `Full` commit syncs the whole WAL, so it also makes + /// every entry committed before it durable, including entries committed + /// with `Normal`; a `Normal` commit never weakens one committed before it. + /// See [`SyncMode`] for what survives a crash. + /// + /// No effect on a transaction that becomes large (see + /// [`OpenOptions::transaction_chunk_facts`]): it commits as a + /// checkpoint, which is durable in either mode, and its WAL guard is + /// always synced. No effect on an in-memory database, which has no WAL. + pub synchronous: Option, + /// The chunk size of this transaction, or `None` for + /// [`OpenOptions::transaction_chunk_facts`]. + /// + /// Every `transact`/`retract` of the transaction is read a chunk of this + /// many facts at a time, and on a file database the transaction becomes + /// large once its facts pass it. Kept between 1 and 1,000,000, as the + /// database value is. + pub chunk_facts: Option, +} + +impl TxOptions { + /// Every option `None`: the database's values. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Set this transaction's WAL durability mode. See [`TxOptions::synchronous`]. + #[must_use] + pub fn synchronous(mut self, mode: SyncMode) -> Self { + self.synchronous = Some(mode); + self + } + + /// Set this transaction's chunk size. See [`TxOptions::chunk_facts`]. + #[must_use] + pub fn chunk_facts(mut self, facts: usize) -> Self { + self.chunk_facts = Some(facts); + self + } +} + /// `OpenOptions` combined with a file path, ready to open. #[cfg(not(target_arch = "wasm32"))] pub struct OpenOptionsWithPath { @@ -669,11 +755,7 @@ impl Minigraf { // `g + 1` behind this WAL, and without it a file with one valid // meta slot would be refused (#550 spec §4.5). pfs.force_dirty(); - Some(WalWriter::open_or_create_at( - &wal_path, - opts.synchronous, - pfs.generation(), - )?) + Some(WalWriter::open_or_create_at(&wal_path, pfs.generation())?) } else { None }; @@ -903,6 +985,7 @@ impl Minigraf { let should_checkpoint = WriteTransaction::wal_write_stamped_batch( &mut ctx, &self.inner.options, + self.inner.options.synchronous, tx_count, &stamped, )?; @@ -951,12 +1034,7 @@ impl Minigraf { }; // The WAL guard: a crash before the commit point finds a WAL whose // base is this generation, never an orphan `g + 1` page (§4.5). - Self::open_wal( - wal, - db_path, - self.inner.options.synchronous, - pfs.generation(), - )?; + Self::open_wal(wal, db_path, pfs.generation())?; let mut tx = pfs.begin_large(crate::graph::types::tx_id_now())?; for chunk in chunks { let written = match chunk { @@ -983,12 +1061,11 @@ impl Minigraf { fn open_wal<'w>( wal: &'w mut Option, db_path: &Path, - sync: SyncMode, generation: u64, ) -> Result<&'w mut WalWriter> { if wal.is_none() { let wal_path = Self::wal_path_for(db_path); - *wal = Some(WalWriter::open_or_create_at(&wal_path, sync, generation)?); + *wal = Some(WalWriter::open_or_create_at(&wal_path, generation)?); } wal.as_mut().ok_or_else(|| err_coded!(ErrorCode::Api009)) } @@ -1092,18 +1169,59 @@ impl Minigraf { // ── Explicit transaction ────────────────────────────────────────────────── - /// Begin an explicit write transaction. + /// Begin an explicit write transaction with the database's options: + /// [`begin_write_with`](Self::begin_write_with) with + /// [`TxOptions::default`]. To choose the WAL durability mode or the chunk + /// size of one transaction, use `begin_write_with`. /// /// Acquires the write lock; held until `commit()`, `rollback()`, or drop. /// /// # Errors /// - /// Returns an error if a `WriteTransaction` is already active on **this thread**. + /// Returns an error if a `WriteTransaction` is already active on **this + /// thread** (`INT-001`), or the handle is read-only (`API-014`). pub fn begin_write(&self) -> Result, MinigrafError> { - self.begin_write_inner().map_err(MinigrafError::from) + self.begin_write_with(TxOptions::default()) } - fn begin_write_inner(&self) -> Result> { + /// Begin an explicit write transaction with `options` (#871): its WAL + /// durability mode ([`TxOptions::synchronous`]) and chunk size + /// ([`TxOptions::chunk_facts`]). A `None` option is the database's value + /// from [`OpenOptions`]. The options apply to this transaction only. + /// + /// Acquires the write lock; held until `commit()`, `rollback()`, or drop. + /// + /// ``` + /// use minigraf::{Minigraf, SyncMode, TxOptions}; + /// + /// // On a file database this commit is not synced, and the transaction + /// // is written a chunk of 10,000 facts at a time once it passes that + /// // many; in memory (as here, so the example runs everywhere) the + /// // options change nothing. + /// let db = Minigraf::in_memory().unwrap(); + /// let opts = TxOptions::new() + /// .synchronous(SyncMode::Normal) + /// .chunk_facts(10_000); + /// let mut tx = db.begin_write_with(opts.clone()).unwrap(); + /// tx.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap(); + /// tx.commit().unwrap(); + /// // Later transactions use the database's options again. + /// db.execute(r#"(transact [[:bob :person/name "Bob"]])"#).unwrap(); + /// ``` + /// + /// # Errors + /// + /// Fails as [`begin_write`](Self::begin_write) does: `INT-001` while a + /// `WriteTransaction` is active on **this thread**, `API-014` on a + /// read-only handle. A failed call leaves the handle usable. + pub fn begin_write_with( + &self, + options: TxOptions, + ) -> Result, MinigrafError> { + self.begin_write_inner(options).map_err(MinigrafError::from) + } + + fn begin_write_inner(&self, options: TxOptions) -> Result> { if is_write_tx_active() { bail_coded!(ErrorCode::Int001); } @@ -1118,6 +1236,12 @@ impl Minigraf { Ok(WriteTransaction { guard, inner: &self.inner, + sync: options + .synchronous + .unwrap_or(self.inner.options.synchronous), + chunk_facts: options + .chunk_facts + .unwrap_or(self.inner.options.transaction_chunk_facts), pending_facts: Vec::new(), next_pending_tx_count: self.inner.fact_storage.current_tx_count().saturating_add(1), next_pending_tx_id: crate::graph::types::tx_id_now(), @@ -1315,6 +1439,32 @@ impl Minigraf { } } + /// End the handle as a power loss between operations would (#871): as + /// [`kill`](Self::kill), and the WAL is cut back to its length at its + /// last successful sync, losing every entry committed since without one. + #[cfg(all(test, not(target_arch = "wasm32")))] + pub(crate) fn power_loss(self) { + let mut ctx = self.inner.write_lock.lock().unwrap(); + if let WriteContext::File { mut pfs, wal, .. } = + std::mem::replace(&mut *ctx, WriteContext::Memory) + { + if let Some(mut wal) = wal { + wal.lose_unsynced().unwrap(); + } + pfs.discard(); + } + } + + /// Whether the WAL is open and every byte written to it is synced, for + /// tests that check whether a commit synced it. + #[cfg(all(test, not(target_arch = "wasm32")))] + pub(crate) fn wal_synced(&self) -> Option { + match &*self.inner.write_lock.lock().unwrap() { + WriteContext::File { wal, .. } => wal.as_ref().map(WalWriter::is_synced), + _ => None, + } + } + /// Internal checkpoint logic (operates on an already-held write-lock guard). fn do_checkpoint(_fact_storage: &FactStorage, ctx: &mut WriteContext) -> Result<()> { match ctx { @@ -1747,6 +1897,12 @@ impl Minigraf { pub struct WriteTransaction<'a> { guard: MutexGuard<'a, WriteContext>, inner: &'a Inner, + /// This transaction's WAL durability mode (`TxOptions::synchronous`, + /// else the database's). + sync: SyncMode, + /// This transaction's configured chunk size, before the clamp + /// (`TxOptions::chunk_facts`, else the database's). + chunk_facts: usize, /// Facts buffered in this transaction (not yet committed to FactStorage). pending_facts: Vec, /// Synthetic tx_count assigned to the next staged write batch. @@ -1791,7 +1947,7 @@ impl<'a> WriteTransaction<'a> { #[cfg(not(target_arch = "wasm32"))] if let Some(tx) = &self.large { // Every fact is stamped with the transaction's tx_id. - let chunk = statement::tx_chunk_facts(self.inner.options.transaction_chunk_facts); + let chunk = statement::tx_chunk_facts(self.chunk_facts); let statement = statement::read_statement(input, kind, tx.tx_id(), chunk)?; let chunks = statement.chunks()?; refuse_if_failed(&self.guard)?; @@ -1803,7 +1959,7 @@ impl<'a> WriteTransaction<'a> { // checks them again at the commit time. let staged_tx_id = std::cmp::max(crate::graph::types::tx_id_now(), self.next_pending_tx_id); - let chunk = statement::tx_chunk_facts(self.inner.options.transaction_chunk_facts); + let chunk = statement::tx_chunk_facts(self.chunk_facts); let statement = statement::read_statement(input, kind, staged_tx_id, chunk)?; #[cfg(not(target_arch = "wasm32"))] let file = matches!(*self.guard, WriteContext::File { .. }); @@ -1901,13 +2057,9 @@ impl<'a> WriteTransaction<'a> { else { bail_coded!(ErrorCode::Int019, "a large transaction without a file"); }; - // The WAL guard (§4.5), before any page is written. - Minigraf::open_wal( - wal, - db_path, - self.inner.options.synchronous, - pfs.generation(), - )?; + // The WAL guard (§4.5), before any page is written. Its header is + // synced whatever the transaction's mode (#871). + Minigraf::open_wal(wal, db_path, pfs.generation())?; let mut tx = pfs.begin_large(tx_id)?; let mut staged: Vec = std::mem::take(&mut self.pending_facts) .into_iter() @@ -1983,6 +2135,12 @@ impl<'a> WriteTransaction<'a> { executor.execute(DatalogCommand::Rule(rule)) } + /// This transaction's chunk size after the clamp, for tests. + #[cfg(test)] + pub(crate) fn chunk(&self) -> usize { + statement::tx_chunk_facts(self.chunk_facts) + } + /// Stage buffered facts with stable synthetic read metadata. fn stage_pending_facts(&mut self, facts: Vec) { let staged_tx_id = std::cmp::max(crate::graph::types::tx_id_now(), self.next_pending_tx_id); @@ -2001,8 +2159,9 @@ impl<'a> WriteTransaction<'a> { /// Commit this transaction atomically. /// /// All buffered facts are stamped with a single `tx_count` / `tx_id`, then - /// the WAL entry is written and fsynced **before** any fact is applied to the - /// shared `FactStorage`. This guarantees that if the WAL write fails the + /// the WAL entry is written (and synced, unless the transaction's mode is + /// `SyncMode::Normal`: [`TxOptions::synchronous`]) **before** any fact is + /// applied to the shared `FactStorage`. This guarantees that if the WAL write fails the /// database is left completely unchanged (clean rollback — no cleanup needed). /// /// # Errors @@ -2075,6 +2234,7 @@ impl<'a> WriteTransaction<'a> { let should_checkpoint = Self::wal_write_stamped_batch( &mut self.guard, &self.inner.options, + self.sync, tx_count, &stamped, )?; @@ -2095,7 +2255,8 @@ impl<'a> WriteTransaction<'a> { Ok(()) } - /// Write a pre-stamped batch of facts to the WAL. + /// Write a pre-stamped batch of facts to the WAL, synced if `sync` is + /// `SyncMode::Full`. /// /// Accepts an already-computed `tx_count` and `facts` slice. /// Called while the write lock is held. @@ -2107,6 +2268,7 @@ impl<'a> WriteTransaction<'a> { fn wal_write_stamped_batch( ctx: &mut WriteContext, opts: &OpenOptions, + sync: SyncMode, tx_count: u64, facts: &[Fact], ) -> Result { @@ -2120,9 +2282,8 @@ impl<'a> WriteTransaction<'a> { wal_entry_count, } => { // Lazily open the WAL writer if not already open. - let wal_writer = - Minigraf::open_wal(wal, db_path, opts.synchronous, pfs.generation())?; - wal_writer.append_entry(tx_count, facts)?; + let wal_writer = Minigraf::open_wal(wal, db_path, pfs.generation())?; + wal_writer.append_entry(tx_count, facts, sync)?; pfs.mark_dirty(); *wal_entry_count = wal_entry_count.saturating_add(1); @@ -2830,19 +2991,12 @@ mod tests { db.execute(r#"(transact [[:alice :name "Alice"]])"#) .unwrap(); - let ctx = db.inner.write_lock.lock().unwrap(); - match &*ctx { - WriteContext::File { wal, .. } => { - let wal_writer = wal.as_ref().expect("WAL should be open after a write"); - assert_eq!( - wal_writer.sync_mode(), - SyncMode::Normal, - "lazy-reopen call site should honor opts.synchronous" - ); - } - _ => panic!("expected read-write file-backed WriteContext"), - } - drop(ctx); + // The writer keeps no mode (#871): the append was not synced. + assert_eq!( + db.wal_synced(), + Some(false), + "lazy-reopen call site should honor opts.synchronous" + ); // Handle drops here without checkpointing: wal_checkpoint_threshold == // usize::MAX suppresses both auto-checkpoint and the close-time // checkpoint in `Inner::drop`, so the .wal file survives on disk. @@ -2867,20 +3021,17 @@ mod tests { .open() .unwrap(); - let ctx = db.inner.write_lock.lock().unwrap(); - match &*ctx { - WriteContext::File { wal, .. } => { - let wal_writer = wal - .as_ref() - .expect("WAL should be opened eagerly since the .wal file pre-existed"); - assert_eq!( - wal_writer.sync_mode(), - SyncMode::Normal, - "initial-open call site should honor opts.synchronous" - ); - } - _ => panic!("expected read-write file-backed WriteContext"), - } + assert_eq!( + db.wal_synced(), + Some(true), + "WAL should be opened eagerly since the .wal file pre-existed" + ); + db.execute(r#"(transact [[:bob :name "Bob"]])"#).unwrap(); + assert_eq!( + db.wal_synced(), + Some(false), + "initial-open call site should honor opts.synchronous" + ); } // ── failed commit leaves database unchanged ─────────────────────────────── diff --git a/src/fault_matrix.rs b/src/fault_matrix.rs index 30e8eb29d..061797a25 100644 --- a/src/fault_matrix.rs +++ b/src/fault_matrix.rs @@ -15,8 +15,13 @@ //! - commits once more, kills, reopens, and finds that commit too. //! //! `k` runs from 0 until the operation finishes without reaching the fault. +//! +//! The mixed-mode operations ([`Mix`], #871) commit `X` through +//! `begin_write_with` with a WAL sync mode other than the database's, or +//! after entries committed without a sync. A lost sync may then lose those +//! entries, so their reopen check is the prefix guarantee instead. -use crate::db::{Minigraf, OpenOptions}; +use crate::db::{Minigraf, OpenOptions, SyncMode, TxOptions}; use crate::fact_log::FactFilter; use crate::storage::fault::{self, Fault, Site}; use std::path::Path; @@ -157,6 +162,12 @@ fn bulk(prefix: &str, n: usize) -> String { /// `TransactNewWal`, two entries pending in the WAL. fn build(path: &Path, op: Op) -> Minigraf { let db = open(path, op); + fill(&db, op); + db +} + +/// The commits of [`build`], on `db`. +fn fill(db: &Minigraf, op: Op) { if let Op::Large { one_slot: true, wal, @@ -168,7 +179,7 @@ fn build(path: &Path, op: Op) -> Minigraf { db.execute(r#"(transact [[:e1 :p/tag "c"] [:e3 :p/ref :e1]])"#) .unwrap(); } - return db; + return; } db.execute(&bulk("a", 150)).unwrap(); db.execute(r#"(transact [[:e1 :p/tag "a"] [:e1 :p/tag "b"] [:e2 :p/n 1]])"#) @@ -183,7 +194,6 @@ fn build(path: &Path, op: Op) -> Minigraf { db.execute(r#"(transact [[:e1 :p/tag "c"] [:e3 :p/ref :e1]])"#) .unwrap(); } - db } /// The records `after` has beyond `before`, if `before` is all in `after`. @@ -546,3 +556,215 @@ fn no_commit_after_a_failed_meta_fsync() { reopen_and_check(&path, Op::Checkpoint, &live, false); } } + +// ─── Mixed WAL sync modes (#871) ───────────────────────────────────────────── + +/// A mixed-mode operation: `X` committed through `begin_write_with`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Mix { + /// (i) On a `Full` database whose WAL ends in an entry committed with + /// `Normal`, `X` commits with `Full`. + FullAfterNormal, + /// (ii) On a `Full` database whose WAL holds synced entries, `X` commits + /// with `Normal`. + NormalOnFull, + /// (iii) On a `Normal` database whose WAL holds unsynced entries, `X` + /// commits with `Full`. + FullOnNormal, +} + +impl Mix { + fn db_mode(self) -> SyncMode { + match self { + Mix::FullAfterNormal | Mix::NormalOnFull => SyncMode::Full, + Mix::FullOnNormal => SyncMode::Normal, + } + } + + fn x_mode(self) -> SyncMode { + match self { + Mix::FullAfterNormal | Mix::FullOnNormal => SyncMode::Full, + Mix::NormalOnFull => SyncMode::Normal, + } + } +} + +/// The states a file may come back in: the records after each commit, in +/// commit order, and the first one every reopen must reach. +struct History { + states: Vec>, + /// The last state at a successful WAL sync or a checkpoint: what even a + /// lost sync keeps. + durable: usize, +} + +impl History { + fn push(&mut self, db: &Minigraf) { + self.states.push(snapshot(db)); + // No WAL: the last commit was a checkpoint. + if db.wal_synced().is_none_or(|synced| synced) { + self.durable = self.states.len() - 1; + } + } +} + +fn commit_with(db: &Minigraf, mode: SyncMode, stmt: &str) -> Result<(), crate::MinigrafError> { + let mut tx = db.begin_write_with(TxOptions::new().synchronous(mode))?; + tx.execute(stmt)?; + tx.commit() +} + +/// Run `mix` with `f` at every call index. Returns how many indexes hit. +fn run_mixed(mix: Mix, f: Fault) -> u64 { + let op = Op::Transact; + let mut hits = 0; + for k in 0u64.. { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("f.graph"); + let open_db = || { + OpenOptions::new() + .synchronous(mix.db_mode()) + .wal_checkpoint_threshold(1000) + .path(&path) + .open() + .expect("open") + }; + let db = open_db(); + // Two checkpoints, then two entries in the database mode. + fill(&db, Op::TransactNewWal); + let mut history = History { + states: Vec::new(), + durable: 0, + }; + history.push(&db); + db.execute(&bulk("c", 20)).unwrap(); + history.push(&db); + db.execute(r#"(transact [[:e1 :p/tag "c"] [:e3 :p/ref :e1]])"#) + .unwrap(); + history.push(&db); + if mix == Mix::FullAfterNormal { + commit_with(&db, SyncMode::Normal, r#"(transact [[:n1 :p/n "normal"]])"#).unwrap(); + history.push(&db); + } + assert_eq!( + db.wal_synced(), + Some(mix == Mix::NormalOnFull), + "the WAL before X" + ); + let before = snapshot(&db); + + fault::arm(f, k); + let result = commit_with(&db, mix.x_mode(), X); + let tripped = fault::disarm(); + + let live = snapshot(&db); + let applied = extra(&before, &live).expect("the handle lost a fact"); + assert!( + applied.is_empty() || is_whole_x(&applied, op), + "the handle applied part of a transaction" + ); + if result.is_ok() { + assert!( + is_whole_x(&applied, op), + "a committed transaction is missing" + ); + history.push(&db); + } else { + assert!(applied.is_empty(), "a failed commit was applied"); + } + + // A later write succeeds, or is refused as failed. + let live = match db.execute(r#"(transact [[:y :p/y "later"]])"#) { + Ok(_) => { + history.push(&db); + snapshot(&db) + } + Err(e) => { + let c = code(e); + assert!(c == "WAL-007" || c == "STG-046", "later write: {c}"); + assert!(result.is_err(), "a write refused after a success"); + live + } + }; + db.kill(); + + // Reopen: a prefix of the commit order, from the durable state on. + let db = open_db(); + let reopened = snapshot(&db); + let at = history.states.iter().rposition(|s| *s == reopened); + // A failed X's entry may still have reached the file whole: then + // every transaction before it is there too. + let failed_x_on_disk = + result.is_err() && extra(&before, &reopened).is_some_and(|more| is_whole_x(&more, op)); + assert!( + at.is_some_and(|at| at >= history.durable) || failed_x_on_disk, + "the reopened file is not a prefix of the commit order from its last sync" + ); + if f == Fault::SyncLost && tripped == Some(Site::WalSync) { + // X's lost sync takes every unsynced entry before it too: the + // file is back at the last synced state, below the one before X. + let before_x = history.states.len() - 1; + assert!( + at == Some(history.durable) && history.durable < before_x, + "a lost sync kept an unsynced entry" + ); + } + if f != Fault::SyncLost { + // Only a lost sync may lose an entry committed without one. + assert!( + extra(&live, &reopened).is_some(), + "a fact the handle held was lost without a lost sync" + ); + } + assert!( + db.verify().unwrap().problems.is_empty(), + "verify after reopen" + ); + + // One more commit after recovery must survive a kill. + db.execute(r#"(transact [[:z :p/z "after"]])"#).unwrap(); + let with_z = snapshot(&db); + db.kill(); + let db = open_db(); + assert!(snapshot(&db) == with_z, "a commit after recovery was lost"); + assert!(db.verify().unwrap().problems.is_empty(), "verify after Z"); + db.kill(); + + if tripped.is_none() { + assert!(result.is_ok(), "failed without a fault"); + break; + } + hits += 1; + } + hits +} + +fn run_all_mixed(mix: Mix) { + for f in FAULTS { + let hits = run_mixed(mix, f); + // A `Normal` commit to an existing WAL makes no sync for a sync + // fault to hit; every other fault fires. + let syncs = mix.x_mode() == SyncMode::Full; + let sync_fault = matches!(f, Fault::SyncEio | Fault::SyncLost); + if sync_fault && !syncs { + assert_eq!(hits, 0, "a Normal commit synced the WAL"); + } else { + assert!(hits > 0, "the fault never fired"); + } + } +} + +#[test] +fn faults_during_a_full_commit_after_a_normal_one() { + run_all_mixed(Mix::FullAfterNormal); +} + +#[test] +fn faults_during_a_normal_commit_on_a_full_database() { + run_all_mixed(Mix::NormalOnFull); +} + +#[test] +fn faults_during_a_full_commit_on_a_normal_database() { + run_all_mixed(Mix::FullOnNormal); +} diff --git a/src/lib.rs b/src/lib.rs index 65acb83df..347a78571 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -96,6 +96,8 @@ pub(crate) mod query; pub mod repl; pub(crate) mod storage; pub(crate) mod temporal; +#[cfg(all(test, not(target_arch = "wasm32")))] +mod tx_options_tests; /// The sidecar WAL of file-backed databases; wasm32 targets have no file backend. #[cfg(not(target_arch = "wasm32"))] pub(crate) mod wal; @@ -106,7 +108,7 @@ pub mod browser; #[cfg(not(target_arch = "wasm32"))] pub use db::OpenOptionsWithPath; -pub use db::{IntegrityReport, Minigraf, OpenOptions, SyncMode, WriteTransaction}; +pub use db::{IntegrityReport, Minigraf, OpenOptions, SyncMode, TxOptions, WriteTransaction}; pub use error::{ErrorCategory, MinigrafError}; pub use overrides::Overridable; pub use query::datalog::functions::UdfError; diff --git a/src/storage/persistent_facts/large_tx.rs b/src/storage/persistent_facts/large_tx.rs index 8a912a120..a7b47905b 100644 --- a/src/storage/persistent_facts/large_tx.rs +++ b/src/storage/persistent_facts/large_tx.rs @@ -1247,9 +1247,7 @@ mod tests { db.kill(); // A WAL every entry of which is checkpointed (none here): only its // header, as a large transaction's guard leaves it. - let _ = - crate::wal::WalWriter::open_or_create_at(&wal_of(&path), crate::db::SyncMode::Full, 2) - .unwrap(); + let _ = crate::wal::WalWriter::open_or_create_at(&wal_of(&path), 2).unwrap(); let db = Minigraf::open(&path).unwrap(); db.checkpoint().unwrap(); assert_eq!(db.generation(), 3); diff --git a/src/tx_options_tests.rs b/src/tx_options_tests.rs new file mode 100644 index 000000000..bd6178198 --- /dev/null +++ b/src/tx_options_tests.rs @@ -0,0 +1,629 @@ +//! `TxOptions` and `Minigraf::begin_write_with` (#871): each option applies +//! to one transaction only, and what survives a process kill or a power loss +//! when WAL sync modes are mixed per transaction on one WAL file. +//! +//! A power loss is [`Minigraf::power_loss`]: the WAL is cut back to its +//! length at its last successful sync, then the handle ends as a kill would. + +use crate::db::{Minigraf, OpenOptions, SyncMode, TxOptions}; +use crate::fact_log::FactFilter; +use std::path::{Path, PathBuf}; + +/// One fact version, comparable across handles. +type Rec = (String, String, String, u64, u64, i64, i64, bool); + +fn records(db: &Minigraf) -> Vec { + let mut v: Vec = db + .fact_log(&FactFilter::new()) + .unwrap() + .map(|r| { + let r = r.unwrap(); + ( + r.entity.to_string(), + r.attribute, + format!("{:?}", r.value), + r.tx_count, + r.tx_id, + r.valid_from, + r.valid_to, + r.asserted, + ) + }) + .collect(); + v.sort(); + v +} + +fn wal_of(path: &Path) -> PathBuf { + let mut p = path.as_os_str().to_owned(); + p.push(".wal"); + PathBuf::from(p) +} + +/// A file database in `mode` that never checkpoints by itself. +fn open(path: &Path, mode: SyncMode) -> Minigraf { + OpenOptions::new() + .synchronous(mode) + .wal_checkpoint_threshold(usize::MAX) + .path(path) + .open() + .unwrap() +} + +/// Commit `stmt` in one transaction begun with `opts`. +fn commit(db: &Minigraf, opts: TxOptions, stmt: &str) { + let mut tx = db.begin_write_with(opts).unwrap(); + tx.execute(stmt).unwrap(); + tx.commit().unwrap(); +} + +fn tx(mode: Option) -> TxOptions { + match mode { + Some(m) => TxOptions::new().synchronous(m), + None => TxOptions::new(), + } +} + +const MODES: [SyncMode; 2] = [SyncMode::Full, SyncMode::Normal]; + +// ─── API ───────────────────────────────────────────────────────────────────── + +#[test] +fn tx_options_builders_set_their_fields() { + let opts = TxOptions::new(); + assert_eq!(opts, TxOptions::default()); + assert_eq!(opts.synchronous, None); + assert_eq!(opts.chunk_facts, None); + let opts = opts.synchronous(SyncMode::Normal).chunk_facts(7); + assert_eq!(opts.synchronous, Some(SyncMode::Normal)); + assert_eq!(opts.chunk_facts, Some(7)); + assert_eq!(opts.clone(), opts); + assert!(format!("{opts:?}").contains("Normal")); + let root: crate::TxOptions = opts; + assert_eq!(root.chunk_facts, Some(7)); +} + +#[test] +fn begin_write_with_fails_as_begin_write_does() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + { + let _tx = db.begin_write().unwrap(); + let err = db + .begin_write_with(TxOptions::new().synchronous(SyncMode::Normal)) + .err() + .expect("a second transaction on this thread"); + assert_eq!(err.code(), "INT-001"); + } + // The failed call left the handle usable. + commit( + &db, + TxOptions::new().synchronous(SyncMode::Normal), + "(transact [[:a :p/n 1]])", + ); + drop(db); + + let ro = OpenOptions::new() + .read_only(true) + .path(&path) + .open() + .unwrap(); + let err = ro + .begin_write_with(TxOptions::new().chunk_facts(2)) + .err() + .expect("a read-only handle"); + assert_eq!(err.code(), "API-014"); + let err = ro.begin_write().err().expect("a read-only handle"); + assert_eq!(err.code(), "API-014"); + assert_eq!(records(&ro).len(), 1, "the read-only handle still reads"); +} + +// ─── synchronous ───────────────────────────────────────────────────────────── + +/// AC 4: a small commit syncs the WAL exactly when its effective mode is +/// `Full`, in all four combinations, whether the commit creates the WAL or +/// appends to one. +#[test] +fn commit_syncs_by_the_transaction_mode() { + for db_mode in MODES { + for tx_mode in [None, Some(SyncMode::Full), Some(SyncMode::Normal)] { + let effective = tx_mode.unwrap_or(db_mode); + for wal_exists in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, db_mode); + if wal_exists { + db.execute("(transact [[:a :p/n 0]])").unwrap(); + assert!(wal_of(&path).exists()); + } else { + assert_eq!(db.wal_synced(), None, "no WAL yet"); + } + commit(&db, tx(tx_mode), "(transact [[:a :p/n 1]])"); + assert_eq!( + db.wal_synced(), + Some(effective == SyncMode::Full), + "the commit synced the WAL by its effective mode" + ); + } + } + } +} + +/// AC 5: the mode of a transaction does not stay with the WAL writer, also +/// when the transaction's own commit created the WAL file. +#[test] +fn transaction_mode_applies_to_that_transaction_only() { + for creates_wal in [true, false] { + for db_mode in MODES { + let other = match db_mode { + SyncMode::Full => SyncMode::Normal, + SyncMode::Normal => SyncMode::Full, + }; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, db_mode); + if !creates_wal { + commit( + &db, + TxOptions::new().synchronous(SyncMode::Full), + "(transact [[:a :p/n 0]])", + ); + } + commit( + &db, + TxOptions::new().synchronous(other), + "(transact [[:a :p/n 1]])", + ); + assert_eq!(db.wal_synced(), Some(other == SyncMode::Full)); + + // An implicit transaction follows the database mode again. + db.execute("(transact [[:a :p/n 2]])").unwrap(); + assert_eq!(db.wal_synced(), Some(db_mode == SyncMode::Full)); + + // So does `begin_write()`. + commit( + &db, + TxOptions::new().synchronous(other), + "(transact [[:a :p/n 3]])", + ); + let mut t = db.begin_write().unwrap(); + t.execute("(transact [[:a :p/n 4]])").unwrap(); + t.commit().unwrap(); + assert_eq!(db.wal_synced(), Some(db_mode == SyncMode::Full)); + + // A rolled-back or dropped transaction leaves nothing behind. + commit( + &db, + TxOptions::new().synchronous(other), + "(transact [[:a :p/n 5]])", + ); + let mut t = db + .begin_write_with(TxOptions::new().synchronous(other)) + .unwrap(); + t.execute("(transact [[:a :p/n 6]])").unwrap(); + t.rollback(); + drop( + db.begin_write_with(TxOptions::new().synchronous(other)) + .unwrap(), + ); + db.execute("(transact [[:a :p/n 7]])").unwrap(); + assert_eq!(db.wal_synced(), Some(db_mode == SyncMode::Full)); + } + } +} + +/// AC 5: a `Normal` transaction on a `Full` database that became large +/// (creating the WAL guard) and was rolled back; the next implicit +/// transaction syncs. +#[test] +fn rolled_back_large_normal_transaction_leaves_no_mode_behind() { + for drop_it in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + let mut t = db + .begin_write_with( + TxOptions::new() + .synchronous(SyncMode::Normal) + .chunk_facts(2), + ) + .unwrap(); + t.execute("(transact [[:a :p/n 1] [:b :p/n 2] [:c :p/n 3]])") + .unwrap(); + assert!(wal_of(&path).exists(), "the WAL guard exists"); + if drop_it { + drop(t); + } else { + t.rollback(); + } + assert_eq!(db.wal_synced(), Some(true), "the guard header is synced"); + db.execute("(transact [[:d :p/n 4]])").unwrap(); + assert_eq!(db.wal_synced(), Some(true), "the implicit commit synced"); + db.power_loss(); + let db = open(&path, SyncMode::Full); + assert_eq!(records(&db).len(), 1, "only the implicit commit"); + } +} + +/// AC 7: a large transaction begun with `Normal` commits as a checkpoint +/// and survives a power loss whole. +#[test] +fn large_normal_transaction_survives_a_power_loss() { + for db_mode in MODES { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, db_mode); + db.execute("(transact [[:base :p/n 0]])").unwrap(); + db.checkpoint().unwrap(); + let mut t = db + .begin_write_with( + TxOptions::new() + .synchronous(SyncMode::Normal) + .chunk_facts(2), + ) + .unwrap(); + t.execute("(transact [[:a :p/n 1]])").unwrap(); + t.execute("(transact [[:b :p/n 2] [:c :p/n 3] [:d :p/n 4]])") + .unwrap(); + t.execute("(transact [[:e :p/n 5]])").unwrap(); + t.commit().unwrap(); + assert!(!wal_of(&path).exists(), "a large commit is a checkpoint"); + let live = records(&db); + assert_eq!(live.len(), 6); + db.power_loss(); + + let db = open(&path, db_mode); + assert!(records(&db) == live, "the whole large transaction survives"); + assert!(db.verify().unwrap().problems.is_empty(), "verify"); + } +} + +/// AC 8: the options are accepted on an in-memory database, and change +/// nothing there. +#[test] +fn options_on_an_in_memory_database() { + let stmt = "(transact [[:a :p/n 1] [:b :p/n 2] [:c :p/n 3]])"; + let plain = Minigraf::in_memory().unwrap(); + commit(&plain, TxOptions::new(), stmt); + for opts in [ + TxOptions::new().synchronous(SyncMode::Normal), + TxOptions::new().synchronous(SyncMode::Full), + TxOptions::new().chunk_facts(1), + TxOptions::new().chunk_facts(0), + TxOptions::new() + .chunk_facts(usize::MAX) + .synchronous(SyncMode::Normal), + ] { + let db = Minigraf::in_memory().unwrap(); + commit(&db, opts, stmt); + assert_eq!(db.wal_synced(), None, "no WAL"); + assert!( + without_tx(&records(&db)) == without_tx(&records(&plain)), + "the same facts as without options" + ); + } +} + +/// AC 8: a `Normal` commit that brings the WAL to the threshold still +/// checkpoints. +#[test] +fn normal_commit_still_auto_checkpoints() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = OpenOptions::new() + .wal_checkpoint_threshold(2) + .path(&path) + .open() + .unwrap(); + let normal = TxOptions::new().synchronous(SyncMode::Normal); + commit(&db, normal.clone(), "(transact [[:a :p/n 1]])"); + assert!(wal_of(&path).exists(), "one entry pending"); + commit(&db, normal, "(transact [[:a :p/n 2]])"); + assert!(!wal_of(&path).exists(), "the threshold checkpointed"); + let live = records(&db); + db.power_loss(); + let db = Minigraf::open(&path).unwrap(); + assert!(records(&db) == live, "the checkpoint is durable"); +} + +// ─── Mixed modes, crash after each (AC 10) ─────────────────────────────────── + +/// How a handle ends. +#[derive(Clone, Copy, PartialEq)] +enum End { + Kill, + PowerLoss, +} + +/// On a database in `db_mode` holding a checkpointed fact and one synced +/// WAL entry, commit one transaction per mode in `modes` through +/// `begin_write_with`, end the handle with `end`, and reopen. Returns the +/// records after each commit (index 0: before the first) and the reopened +/// handle. +fn mixed( + db_mode: SyncMode, + modes: &[SyncMode], + end: End, + path: &Path, +) -> (Vec>, Minigraf) { + let db = open(path, db_mode); + db.execute("(transact [[:base :p/n 0]])").unwrap(); + db.checkpoint().unwrap(); + commit( + &db, + TxOptions::new().synchronous(SyncMode::Full), + "(transact [[:synced :p/n 0]])", + ); + let mut after = vec![records(&db)]; + for (i, mode) in modes.iter().enumerate() { + commit( + &db, + TxOptions::new().synchronous(*mode), + &format!("(transact [[:t{i} :p/n {i}] [:t{i} :p/name \"tx {i}\"]])"), + ); + after.push(records(&db)); + } + match end { + End::Kill => db.kill(), + End::PowerLoss => db.power_loss(), + } + (after, open(path, db_mode)) +} + +/// After recovery, one more commit survives a kill and a reopen. +fn commit_after_recovery(db: Minigraf, path: &Path, db_mode: SyncMode) { + assert!( + db.verify().unwrap().problems.is_empty(), + "verify after reopen" + ); + db.execute(r#"(transact [[:z :p/z "after"]])"#).unwrap(); + let with_z = records(&db); + db.kill(); + let db = open(path, db_mode); + assert!(records(&db) == with_z, "a commit after recovery was lost"); + assert!(db.verify().unwrap().problems.is_empty(), "verify after Z"); +} + +#[test] +fn normal_then_kill_keeps_it() { + for db_mode in MODES { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let (after, db) = mixed(db_mode, &[SyncMode::Normal], End::Kill, &path); + assert!(records(&db) == after[1], "A survives a process kill"); + commit_after_recovery(db, &path, db_mode); + } +} + +#[test] +fn normal_then_power_loss_keeps_a_prefix() { + for db_mode in MODES { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let (after, db) = mixed(db_mode, &[SyncMode::Normal], End::PowerLoss, &path); + let got = records(&db); + assert!(after.contains(&got), "a prefix of the commit order"); + assert!(got == after[0], "the unsynced A is lost"); + commit_after_recovery(db, &path, db_mode); + } +} + +#[test] +fn normal_then_full_then_power_loss_keeps_both() { + for db_mode in MODES { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let (after, db) = mixed( + db_mode, + &[SyncMode::Normal, SyncMode::Full], + End::PowerLoss, + &path, + ); + assert!(records(&db) == after[2], "B's sync made A durable too"); + commit_after_recovery(db, &path, db_mode); + } +} + +#[test] +fn full_then_normal_then_power_loss_keeps_the_full_one() { + for db_mode in MODES { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let (after, db) = mixed( + db_mode, + &[SyncMode::Full, SyncMode::Normal], + End::PowerLoss, + &path, + ); + let got = records(&db); + // B may be whole or absent; the model loses it. + assert!( + got == after[1] || got == after[2], + "A, and B whole or absent" + ); + assert!(got == after[1], "the unsynced B is lost in the model"); + commit_after_recovery(db, &path, db_mode); + } +} + +// ─── chunk_facts ───────────────────────────────────────────────────────────── + +/// Whether the transaction committed as a large one: its commit is a +/// checkpoint, which deletes the WAL (a small commit leaves an entry in it). +fn committed_large(path: &Path) -> bool { + !wal_of(path).exists() +} + +/// Records without their transaction stamps: what a commit wrote, +/// whatever its `tx_id`/`tx_count` (a default `valid_from` is the `tx_id`). +fn without_tx(recs: &[Rec]) -> Vec<(String, String, String, bool, i64, bool)> { + let mut v: Vec<_> = recs + .iter() + .map(|r| { + ( + r.0.clone(), + r.1.clone(), + r.2.clone(), + r.5 == r.4.cast_signed(), + r.6, + r.7, + ) + }) + .collect(); + v.sort(); + v +} + +const THREE: &str = "(transact [[:a :p/n 1] [:b :p/n 2] [:c :p/n 3]])"; + +#[test] +fn chunk_facts_replaces_the_database_chunk_size() { + // Default database chunk, transaction chunk 2: three facts are large. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + commit(&db, TxOptions::new().chunk_facts(2), THREE); + assert!(committed_large(&path), "three facts pass a chunk of 2"); + assert_eq!(records(&db).len(), 3); + + // Staged statements that pass it together make it large too. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + let mut t = db + .begin_write_with(TxOptions::new().chunk_facts(2)) + .unwrap(); + t.execute("(transact [[:a :p/n 1] [:b :p/n 2]])").unwrap(); + t.execute("(transact [[:c :p/n 3]])").unwrap(); + t.commit().unwrap(); + assert!(committed_large(&path), "staged facts pass a chunk of 2"); + + // Database chunk 2, transaction chunk 1000: three facts are small. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = OpenOptions::new() + .transaction_chunk_facts(2) + .wal_checkpoint_threshold(usize::MAX) + .path(&path) + .open() + .unwrap(); + commit(&db, TxOptions::new().chunk_facts(1000), THREE); + assert!(!committed_large(&path), "a WAL entry"); + assert_eq!(db.wal_synced(), Some(true)); + assert_eq!(records(&db).len(), 3); +} + +#[test] +fn chunk_facts_is_clamped() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + { + let t = db + .begin_write_with(TxOptions::new().chunk_facts(0)) + .unwrap(); + assert_eq!(t.chunk(), 1); + } + { + let t = db + .begin_write_with(TxOptions::new().chunk_facts(usize::MAX)) + .unwrap(); + assert_eq!( + t.chunk(), + crate::query::datalog::parser::statement::MAX_TX_CHUNK_FACTS + ); + } + // A chunk of 0 behaves as 1: two facts are large. + commit( + &db, + TxOptions::new().chunk_facts(0), + "(transact [[:a :p/n 1] [:b :p/n 2]])", + ); + assert!(committed_large(&path), "two facts pass a chunk of 1"); +} + +#[test] +fn chunk_facts_applies_to_that_transaction_only() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + commit(&db, TxOptions::new().chunk_facts(2), THREE); + assert!(committed_large(&path)); + { + let t = db.begin_write().unwrap(); + assert_eq!(t.chunk(), 65_536, "the database chunk size again"); + } + commit( + &db, + TxOptions::new(), + "(transact [[:d :p/n 1] [:e :p/n 2] [:f :p/n 3]])", + ); + assert!(!committed_large(&path), "the next transaction is small"); + db.execute("(transact [[:g :p/n 1] [:h :p/n 2] [:i :p/n 3]])") + .unwrap(); + assert!(!committed_large(&path), "an implicit transaction is small"); + assert_eq!(records(&db).len(), 9); +} + +/// The thread-local test hook keeps precedence over `TxOptions::chunk_facts`, +/// as it has over the database value. +#[test] +fn test_hook_beats_chunk_facts() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + let _chunk = crate::query::datalog::parser::statement::set_tx_chunk_facts(Some(2)); + { + let t = db + .begin_write_with(TxOptions::new().chunk_facts(1000)) + .unwrap(); + assert_eq!(t.chunk(), 2); + } + commit(&db, TxOptions::new().chunk_facts(1000), THREE); + assert!(committed_large(&path), "the hook's chunk of 2 applies"); +} + +/// AC 15: committed data does not depend on the options. +#[test] +fn committed_data_does_not_depend_on_the_options() { + let statements = [ + r#"(transact [[:a :p/n 1] [:a :p/n 2] [:b :p/name "bee"] [:c :p/ref :a]])"#, + r#"(transact [[:d :p/n 4 {:valid-from "2020-01-01T00:00:00Z"}] [:a :p/n 1]])"#, + "(retract [[:a :p/n 2]])", + r#"(transact [[:e :p/long "lllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllll"]])"#, + ]; + let run = |opts: TxOptions| { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.graph"); + let db = open(&path, SyncMode::Full); + db.execute("(transact [[:a :p/n 2] [:z :p/n 0]])").unwrap(); + db.checkpoint().unwrap(); + let mut t = db.begin_write_with(opts).unwrap(); + for s in statements { + t.execute(s).unwrap(); + } + t.commit().unwrap(); + let live = without_tx(&records(&db)); + db.kill(); + let db = open(&path, SyncMode::Full); + assert!( + without_tx(&records(&db)) == live, + "the reopened file agrees" + ); + live + }; + let expected = run(TxOptions::new()); + for opts in [ + TxOptions::new().synchronous(SyncMode::Normal), + TxOptions::new().synchronous(SyncMode::Full), + TxOptions::new().chunk_facts(1), + TxOptions::new().chunk_facts(3), + TxOptions::new() + .chunk_facts(2) + .synchronous(SyncMode::Normal), + TxOptions::new().chunk_facts(1_000_000), + ] { + assert!(run(opts) == expected, "the same committed data"); + } +} diff --git a/src/wal.rs b/src/wal.rs index 2cc3249c4..bd0916bb3 100644 --- a/src/wal.rs +++ b/src/wal.rs @@ -211,22 +211,29 @@ pub struct WalEntry { /// Appends committed transaction entries to the WAL sidecar file. /// /// Created by `Minigraf::open()` for file-backed databases. -/// Not used for in-memory databases. +/// Not used for in-memory databases. The writer keeps no [`SyncMode`]: each +/// append is given its own, so one transaction's mode never reaches the next +/// (#871). pub struct WalWriter { file: File, - sync_mode: SyncMode, /// An append failed, so the file may end in a partial entry. Later /// appends would land after it, where replay never reaches (#513). failed: bool, - /// The file's length after the last successful append. + /// The file's length after the last successful append (test models only). + #[cfg(test)] end: u64, + /// The file's length at its last successful sync: what a lost sync or a + /// power loss leaves of it, in the test models. An append without a sync + /// (`SyncMode::Normal`) does not move it. + #[cfg(test)] + synced: u64, } impl WalWriter { /// [`WalWriter::open_or_create_at`] with base generation 1, for tests. #[cfg(test)] - pub fn open_or_create(path: &Path, sync_mode: SyncMode) -> Result { - Self::open_or_create_at(path, sync_mode, 1) + pub fn open_or_create(path: &Path) -> Result { + Self::open_or_create_at(path, 1) } /// Open an existing WAL or create a new one. @@ -234,11 +241,7 @@ impl WalWriter { /// If creating, writes the WAL header with `base_generation`, the generation /// of the meta page active now. If opening, validates the header (keeping its /// base generation) and seeks to the end for appending. - pub fn open_or_create_at( - path: &Path, - sync_mode: SyncMode, - base_generation: u64, - ) -> Result { + pub fn open_or_create_at(path: &Path, base_generation: u64) -> Result { // Try atomic create-new first (no TOCTOU window) match OpenOptions::new() .read(true) @@ -251,12 +254,15 @@ impl WalWriter { // The header is durable; make the WAL's directory entry // durable too, or a power loss can lose the whole file (#389). sync_parent_dir(path)?; + #[cfg_attr(not(test), allow(unused_variables))] let end = file.seek(SeekFrom::End(0))?; return Ok(WalWriter { file, - sync_mode, failed: false, + #[cfg(test)] end, + #[cfg(test)] + synced: end, }); } Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {} @@ -293,52 +299,79 @@ impl WalWriter { sync_parent_dir(path)?; } } + #[cfg_attr(not(test), allow(unused_variables))] let end = file.seek(SeekFrom::End(0))?; Ok(WalWriter { file, - sync_mode, failed: false, + #[cfg(test)] end, + // The test models take what an earlier handle left as on disk. + #[cfg(test)] + synced: end, }) } - /// The `SyncMode` this writer was opened with. Used only in tests, to - /// verify that `SyncMode` reaches `WalWriter` from every call site. + /// Whether every byte written to the file has been synced since. Tests + /// use it to see whether a commit synced the WAL. #[cfg(test)] - pub(crate) fn sync_mode(&self) -> SyncMode { - self.sync_mode + pub(crate) fn is_synced(&self) -> bool { + self.synced == self.end + } + + /// Cut the file back to its length at its last successful sync, as a + /// power loss would leave it (a test model; #871). + #[cfg(test)] + pub(crate) fn lose_unsynced(&mut self) -> Result<()> { + self.file.set_len(self.synced)?; + self.end = self.synced; + Ok(()) } /// Serialize `facts` as a WAL entry and append it to the file. /// /// The entry is written atomically from the caller's perspective: /// a partial write produces a bad CRC32, which the reader discards. - /// Then flushes to disk, unless `sync_mode` is `SyncMode::Normal` — see [`SyncMode`]. + /// Then flushes to disk, unless `sync` is `SyncMode::Normal` — see [`SyncMode`]. + /// A flush covers every earlier entry too, including ones appended + /// without one. /// /// After a failed write or sync, every later append fails with WAL-007: /// the file may end in a partial entry, and replay stops there (#513). A /// new writer (after a checkpoint deletes the WAL, or on reopen) starts /// clean. Whether the failed entry itself is durable is unknown: it is /// replayed on reopen if it reached the disk whole, and lost otherwise. - pub fn append_entry(&mut self, tx_count: u64, facts: &[Fact]) -> Result<()> { + pub fn append_entry(&mut self, tx_count: u64, facts: &[Fact], sync: SyncMode) -> Result<()> { if self.failed { bail_coded!(ErrorCode::Wal007); } let entry_bytes = serialize_entry(tx_count, facts)?; - let written = self.write_entry(&entry_bytes); + let written = self.write_entry(&entry_bytes, sync); if written.is_err() { self.failed = true; } written } - fn write_entry(&mut self, entry_bytes: &[u8]) -> Result<()> { + fn write_entry(&mut self, entry_bytes: &[u8], sync: SyncMode) -> Result<()> { wal_write(&mut self.file, entry_bytes)?; - match self.sync_mode { - SyncMode::Full => wal_sync(&mut self.file, false, self.end)?, - SyncMode::Normal => {} + #[cfg(test)] + { + self.end = self.end.saturating_add(entry_bytes.len() as u64); + } + if sync == SyncMode::Full { + // A lost sync loses every write since the last good one, this + // entry and any appended without a sync before it. + #[cfg(test)] + let synced_len = self.synced; + #[cfg(not(test))] + let synced_len = 0; + wal_sync(&mut self.file, false, synced_len)?; + #[cfg(test)] + { + self.synced = self.end; + } } - self.end = self.end.saturating_add(entry_bytes.len() as u64); Ok(()) } @@ -556,7 +589,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - let _writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let _writer = WalWriter::open_or_create(&path).unwrap(); let mut reader = WalReader::open(&path).unwrap(); let entries = reader.read_entries().unwrap(); @@ -571,8 +604,10 @@ mod tests { let alice = Uuid::new_v4(); let fact = make_fact(alice, ":name", Value::String("Alice".to_string()), 1); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); - writer.append_entry(1, std::slice::from_ref(&fact)).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); + writer + .append_entry(1, std::slice::from_ref(&fact), SyncMode::Full) + .unwrap(); let mut reader = WalReader::open(&path).unwrap(); let entries = reader.read_entries().unwrap(); @@ -594,8 +629,10 @@ mod tests { // Normal mode skips the per-write flush; the entry must still be // durable-within-process (write_all()'d) and correctly replayable. - let mut writer = WalWriter::open_or_create(&path, SyncMode::Normal).unwrap(); - writer.append_entry(1, std::slice::from_ref(&fact)).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); + writer + .append_entry(1, std::slice::from_ref(&fact), SyncMode::Normal) + .unwrap(); let mut reader = WalReader::open(&path).unwrap(); let entries = reader.read_entries().unwrap(); @@ -616,8 +653,8 @@ mod tests { make_fact(alice, ":age", Value::Integer(30), 1), ]; - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); - writer.append_entry(1, &facts).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); + writer.append_entry(1, &facts, SyncMode::Full).unwrap(); let mut reader = WalReader::open(&path).unwrap(); let entries = reader.read_entries().unwrap(); @@ -639,7 +676,7 @@ mod tests { let alice = Uuid::new_v4(); let bob = Uuid::new_v4(); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); writer .append_entry( 1, @@ -649,12 +686,14 @@ mod tests { Value::String("Alice".to_string()), 1, )], + SyncMode::Full, ) .unwrap(); writer .append_entry( 2, &[make_fact(bob, ":name", Value::String("Bob".to_string()), 2)], + SyncMode::Full, ) .unwrap(); @@ -674,7 +713,7 @@ mod tests { let bob = Uuid::new_v4(); // First open: create WAL and write entry with tx_count=1 - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); writer .append_entry( 1, @@ -684,16 +723,18 @@ mod tests { Value::String("Alice".to_string()), 1, )], + SyncMode::Full, ) .unwrap(); drop(writer); // Second open: exercises the fallback branch (file already exists) - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); writer .append_entry( 2, &[make_fact(bob, ":name", Value::String("Bob".to_string()), 2)], + SyncMode::Full, ) .unwrap(); drop(writer); @@ -750,10 +791,12 @@ mod tests { std::fs::write(&path, b"MW").unwrap(); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); let alice = Uuid::new_v4(); let fact = make_fact(alice, ":name", Value::String("Alice".to_string()), 1); - writer.append_entry(1, std::slice::from_ref(&fact)).unwrap(); + writer + .append_entry(1, std::slice::from_ref(&fact), SyncMode::Full) + .unwrap(); drop(writer); let mut reader = WalReader::open(&path).unwrap(); @@ -784,8 +827,8 @@ mod tests { let fact = make_fact(alice, ":name", Value::String("Alice".to_string()), 1); // Write a valid entry - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); - writer.append_entry(1, &[fact]).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); + writer.append_entry(1, &[fact], SyncMode::Full).unwrap(); drop(writer); // Append garbage bytes after the valid entry to simulate a partial second write @@ -804,10 +847,10 @@ mod tests { /// The bytes of a WAL holding entries `1..=n`. fn wal_with_entries(path: &Path, n: u64) -> Vec { - let mut writer = WalWriter::open_or_create(path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(path).unwrap(); for tx in 1..=n { let fact = make_fact(Uuid::new_v4(), ":name", Value::Integer(tx as i64), tx); - writer.append_entry(tx, &[fact]).unwrap(); + writer.append_entry(tx, &[fact], SyncMode::Full).unwrap(); } drop(writer); std::fs::read(path).unwrap() @@ -831,14 +874,14 @@ mod tests { let second = two.len() - one_len; for cut in [1, 4, 12, 20, second - 1] { std::fs::write(&path, &two[..one_len + cut]).unwrap(); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); assert_eq!( std::fs::metadata(&path).unwrap().len(), one_len as u64, "the partial entry is cut off on open" ); let fact = make_fact(Uuid::new_v4(), ":name", Value::Integer(7), 7); - writer.append_entry(7, &[fact]).unwrap(); + writer.append_entry(7, &[fact], SyncMode::Full).unwrap(); drop(writer); assert_eq!(tx_counts(&path), [1, 7], "the new entry is reachable"); } @@ -856,7 +899,7 @@ mod tests { // Damage entry 2; entry 3 after it is unreachable and goes too. bytes[one_len] ^= 0xFF; std::fs::write(&path, &bytes).unwrap(); - let writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let writer = WalWriter::open_or_create(&path).unwrap(); drop(writer); assert_eq!(std::fs::metadata(&path).unwrap().len(), one_len as u64); assert_eq!(tx_counts(&path), [1]); @@ -867,7 +910,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); let bytes = wal_with_entries(&path, 3); - let writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let writer = WalWriter::open_or_create(&path).unwrap(); drop(writer); assert!(std::fs::read(&path).unwrap() == bytes, "nothing to cut"); } @@ -876,17 +919,17 @@ mod tests { fn test_wal_failed_append_refuses_later_appends_with_wal_007() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); let fact = |tx: u64| make_fact(Uuid::new_v4(), ":name", Value::Integer(1), tx); - writer.append_entry(1, &[fact(1)]).unwrap(); + writer.append_entry(1, &[fact(1)], SyncMode::Full).unwrap(); fault::arm(fault::Fault::Torn(10), 0); writer - .append_entry(2, &[fact(2)]) + .append_entry(2, &[fact(2)], SyncMode::Full) .expect_err("the torn append fails"); assert_eq!(fault::disarm(), Some(Site::WalWrite)); let err: crate::error::MinigrafError = writer - .append_entry(3, &[fact(3)]) + .append_entry(3, &[fact(3)], SyncMode::Full) .expect_err("later appends are refused") .into(); assert_eq!(err.code(), "WAL-007"); @@ -894,8 +937,8 @@ mod tests { assert_eq!(tx_counts(&path), [1], "nothing after the torn entry"); // A new writer cuts off the partial entry and appends normally. - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); - writer.append_entry(4, &[fact(4)]).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); + writer.append_entry(4, &[fact(4)], SyncMode::Full).unwrap(); drop(writer); assert_eq!(tx_counts(&path), [1, 4]); } @@ -930,7 +973,7 @@ mod tests { fn test_wal_rejected_entry_does_not_poison_the_writer() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); let huge = make_fact( Uuid::new_v4(), ":name", @@ -938,10 +981,10 @@ mod tests { 1, ); writer - .append_entry(1, &[huge]) + .append_entry(1, &[huge], SyncMode::Full) .expect_err("WAL-003, before anything is written"); let fact = make_fact(Uuid::new_v4(), ":name", Value::Integer(1), 2); - writer.append_entry(2, &[fact]).unwrap(); + writer.append_entry(2, &[fact], SyncMode::Full).unwrap(); drop(writer); assert_eq!(tx_counts(&path), [2]); } @@ -950,7 +993,7 @@ mod tests { fn test_wal_delete_file() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + WalWriter::open_or_create(&path).unwrap(); assert!(path.exists()); WalWriter::delete_file(&path).unwrap(); assert!(!path.exists()); @@ -964,7 +1007,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); let entity = Uuid::new_v4(); let fact = Fact::new( @@ -975,7 +1018,7 @@ mod tests { ); // Write an entry with one fact - writer.append_entry(1, &[fact]).unwrap(); + writer.append_entry(1, &[fact], SyncMode::Full).unwrap(); writer.file.sync_all().unwrap(); // Overwrite the first fact's length, after the entry's checksum, @@ -1016,7 +1059,7 @@ mod tests { } // Reopening the writer cuts the WAL off at that entry. - drop(WalWriter::open_or_create(&path, SyncMode::Full).unwrap()); + drop(WalWriter::open_or_create(&path).unwrap()); assert_eq!(std::fs::metadata(&path).unwrap().len(), one_len as u64); assert_eq!(tx_counts(&path), [1]); } @@ -1052,19 +1095,21 @@ mod tests { const N: usize = 1_000_001; let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); let fact = make_fact(Uuid::new_v4(), ":n", Value::Integer(1), 1); - writer.append_entry(1, &vec![fact; N]).unwrap(); + writer + .append_entry(1, &vec![fact; N], SyncMode::Full) + .unwrap(); let fact = make_fact(Uuid::new_v4(), ":n", Value::Integer(2), 2); - writer.append_entry(2, &[fact]).unwrap(); + writer.append_entry(2, &[fact], SyncMode::Full).unwrap(); drop(writer); let len = std::fs::metadata(&path).unwrap().len(); // Reopening cuts off nothing, and an append lands after both. - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); assert_eq!(std::fs::metadata(&path).unwrap().len(), len); let fact = make_fact(Uuid::new_v4(), ":n", Value::Integer(3), 3); - writer.append_entry(3, &[fact]).unwrap(); + writer.append_entry(3, &[fact], SyncMode::Full).unwrap(); drop(writer); let entries = WalReader::open(&path).unwrap().read_entries().unwrap(); @@ -1077,7 +1122,7 @@ mod tests { fn test_wal_writer_open_fails_in_a_missing_directory() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("missing").join("test.wal"); - assert!(WalWriter::open_or_create(&path, SyncMode::Full).is_err()); + assert!(WalWriter::open_or_create(&path).is_err()); } #[test] @@ -1089,7 +1134,7 @@ mod tests { std::fs::write(&path, torn).unwrap(); fault::arm(fault::Fault::Eio, 0); - assert!(WalWriter::open_or_create(&path, SyncMode::Full).is_err()); + assert!(WalWriter::open_or_create(&path).is_err()); assert_eq!(fault::disarm(), Some(Site::WalTruncate)); assert!(std::fs::read(&path).unwrap() == torn, "nothing was cut"); assert_eq!(tx_counts(&path), [1]); @@ -1148,13 +1193,13 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("oversized.wal"); - let mut writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&path).unwrap(); let entity = Uuid::new_v4(); let huge_string = "a".repeat(MAX_VALUE_BYTES + 1); let fact = Fact::new(entity, ":test".to_string(), Value::String(huge_string), 1); - let result = writer.append_entry(1, &[fact]); + let result = writer.append_entry(1, &[fact], SyncMode::Full); let err = result.expect_err("oversized fact should be rejected"); let minigraf_err: crate::error::MinigrafError = err.into(); assert_eq!(minigraf_err.code(), "WAL-003"); @@ -1168,8 +1213,7 @@ mod tests { use uuid::Uuid; let dir = tempfile::tempdir().unwrap(); - let mut writer = - WalWriter::open_or_create(&dir.path().join("ident.wal"), SyncMode::Full).unwrap(); + let mut writer = WalWriter::open_or_create(&dir.path().join("ident.wal")).unwrap(); let e = Uuid::new_v4(); let at = |n: usize| format!(":{}", "a".repeat(n - 1)); let ok = Fact::new( @@ -1178,11 +1222,11 @@ mod tests { Value::Keyword(at(MAX_IDENT_BYTES)), 1, ); - writer.append_entry(1, &[ok]).unwrap(); + writer.append_entry(1, &[ok], SyncMode::Full).unwrap(); let long_attr = Fact::new(e, at(MAX_IDENT_BYTES + 1), Value::Integer(1), 2); let long_kw = Fact::new(e, ":a".into(), Value::Keyword(at(MAX_IDENT_BYTES + 1)), 2); for f in [long_attr, long_kw] { - let err = writer.append_entry(2, &[f]).unwrap_err(); + let err = writer.append_entry(2, &[f], SyncMode::Full).unwrap_err(); assert_eq!(crate::error::MinigrafError::from(err).code(), "WAL-003"); } } @@ -1213,7 +1257,7 @@ mod tests { let path = dir.path().join("test.wal"); take_sync_log(); - let _writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let _writer = WalWriter::open_or_create(&path).unwrap(); let log = take_sync_log(); assert_eq!( @@ -1237,10 +1281,10 @@ mod tests { use crate::storage::dir_sync::take_sync_log; let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - drop(WalWriter::open_or_create(&path, SyncMode::Full).unwrap()); + drop(WalWriter::open_or_create(&path).unwrap()); take_sync_log(); - let _writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let _writer = WalWriter::open_or_create(&path).unwrap(); assert!( take_sync_log().is_empty(), @@ -1258,7 +1302,7 @@ mod tests { std::fs::write(&path, b"MW").unwrap(); take_sync_log(); - let _writer = WalWriter::open_or_create(&path, SyncMode::Full).unwrap(); + let _writer = WalWriter::open_or_create(&path).unwrap(); let log = take_sync_log(); assert_eq!( @@ -1277,7 +1321,7 @@ mod tests { use crate::storage::dir_sync::take_sync_log; let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("test.wal"); - drop(WalWriter::open_or_create(&path, SyncMode::Full).unwrap()); + drop(WalWriter::open_or_create(&path).unwrap()); take_sync_log(); WalWriter::delete_file(&path).unwrap(); @@ -1320,10 +1364,10 @@ mod tests { fn test_wal_v2_header_records_base_generation() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("base.wal"); - drop(WalWriter::open_or_create_at(&path, SyncMode::Full, 42).unwrap()); + drop(WalWriter::open_or_create_at(&path, 42).unwrap()); assert_eq!(WalReader::open(&path).unwrap().base_generation(), Some(42)); // Reopening for append keeps the original base generation. - drop(WalWriter::open_or_create_at(&path, SyncMode::Full, 99).unwrap()); + drop(WalWriter::open_or_create_at(&path, 99).unwrap()); assert_eq!(WalReader::open(&path).unwrap().base_generation(), Some(42)); } @@ -1337,8 +1381,8 @@ mod tests { std::fs::write(&path, header).unwrap(); assert_eq!(WalReader::open(&path).unwrap().base_generation(), Some(1)); // A v2.x WAL can still be appended to. - let mut w = WalWriter::open_or_create_at(&path, SyncMode::Full, 5).unwrap(); - w.append_entry(1, &[]).unwrap(); + let mut w = WalWriter::open_or_create_at(&path, 5).unwrap(); + w.append_entry(1, &[], SyncMode::Full).unwrap(); drop(w); let mut r = WalReader::open(&path).unwrap(); assert_eq!(r.read_entries().unwrap().len(), 1); @@ -1351,4 +1395,55 @@ mod tests { std::fs::write(&path, [0u8; 5]).unwrap(); assert_eq!(WalReader::open(&path).unwrap().base_generation(), None); } + + /// #871: a lost sync loses every write since the WAL's last good sync, + /// so a `Normal` entry before a `Full` one whose sync is lost goes too. + #[test] + fn test_lost_sync_loses_earlier_normal_entries() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("lost.wal"); + let fact = make_fact(Uuid::new_v4(), ":n", Value::Integer(1), 1); + let mut writer = WalWriter::open_or_create(&path).unwrap(); + writer + .append_entry(1, std::slice::from_ref(&fact), SyncMode::Full) + .unwrap(); + writer + .append_entry(2, std::slice::from_ref(&fact), SyncMode::Normal) + .unwrap(); + assert!(!writer.is_synced(), "a Normal append does not sync"); + fault::arm(fault::Fault::SyncLost, 0); + assert!( + writer + .append_entry(3, std::slice::from_ref(&fact), SyncMode::Full) + .is_err(), + "the lost sync fails the append" + ); + assert_eq!(fault::disarm(), Some(Site::WalSync)); + drop(writer); + assert_eq!(tx_counts(&path), vec![1], "only the synced entry is left"); + } + + /// #871: a `Full` append syncs every entry before it; the power-loss + /// model cuts only what was written after the last sync. + #[test] + fn test_full_append_syncs_earlier_normal_entries() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("full.wal"); + let fact = make_fact(Uuid::new_v4(), ":n", Value::Integer(1), 1); + let mut writer = WalWriter::open_or_create(&path).unwrap(); + assert!(writer.is_synced(), "the header is synced"); + writer + .append_entry(1, std::slice::from_ref(&fact), SyncMode::Normal) + .unwrap(); + writer + .append_entry(2, std::slice::from_ref(&fact), SyncMode::Full) + .unwrap(); + assert!(writer.is_synced(), "a Full append syncs the whole file"); + writer + .append_entry(3, std::slice::from_ref(&fact), SyncMode::Normal) + .unwrap(); + writer.lose_unsynced().unwrap(); + drop(writer); + assert_eq!(tx_counts(&path), vec![1, 2]); + } } diff --git a/tests/crash_kill_test.rs b/tests/crash_kill_test.rs index 0919dace4..33eaf6e29 100644 --- a/tests/crash_kill_test.rs +++ b/tests/crash_kill_test.rs @@ -21,7 +21,11 @@ //! This also covers the original #308 report (a kill mid-checkpoint left a //! header checksum mismatch): every round must reopen. //! -//! With the `test-hooks` feature a sixth variant makes most transactions +//! A sixth variant commits every transaction through `begin_write_with` +//! with a WAL sync mode chosen per transaction from the seed, on a database +//! opened `SyncMode::Full` (#871): `Normal` and `Full` entries share one WAL. +//! +//! With the `test-hooks` feature a seventh variant makes most transactions //! large (#550): they write pages and commit as a checkpoint, with no WAL //! entry, and every fourth one is followed by a large statement that fails //! after writing pages and rolls back. Every third runs as a @@ -38,7 +42,7 @@ //! Design: docs/superpowers/specs/2026-10-08-crash-kill-data-check-design.md #![cfg(not(target_arch = "wasm32"))] -use minigraf::db::{Minigraf, OpenOptions, SyncMode}; +use minigraf::db::{Minigraf, OpenOptions, SyncMode, TxOptions}; use minigraf::{QueryResult, Value}; use std::collections::{BTreeSet, HashMap}; use std::io::Write; @@ -181,6 +185,10 @@ enum Variant { AutoCheckpoint, /// As `CheckpointEvery3`, with `SyncMode::Normal` (no WAL fsync). NormalSync, + /// Every transaction through `begin_write_with`, its sync mode chosen + /// per transaction ([`tx_sync_mode`]) on a `Full` database (#871); + /// `checkpoint()` after every fifth. + PerTxSync, /// Statements of more than [`LARGE_CHUNK`] facts are large (#550), a /// failing large statement follows every fourth transaction, every third /// is a `WriteTransaction` ([`in_write_tx`]); `checkpoint()` after every @@ -195,10 +203,26 @@ const VARIANTS: &[Variant] = &[ Variant::WalOnly, Variant::AutoCheckpoint, Variant::NormalSync, + Variant::PerTxSync, #[cfg(feature = "test-hooks")] Variant::Large, ]; +/// The `PerTxSync` variant's sync mode for transaction `k`, from the seed: +/// the first is `Normal` and the second `Full`, so both occur. +fn tx_sync_mode(rng: &mut Rng, k: u64) -> SyncMode { + let normal = match k { + 1 => true, + 2 => false, + _ => rng.below(2) == 0, + }; + if normal { + SyncMode::Normal + } else { + SyncMode::Full + } +} + /// The chunk size of the `Large` variant's child. #[cfg(feature = "test-hooks")] const LARGE_CHUNK: usize = 2; @@ -235,6 +259,7 @@ impl Variant { Variant::AutoCheckpoint => opts.wal_checkpoint_threshold(2), Variant::NormalSync => opts.synchronous(SyncMode::Normal), Variant::CheckpointEach | Variant::CheckpointEvery3 => opts, + Variant::PerTxSync => opts.synchronous(SyncMode::Full), #[cfg(feature = "test-hooks")] Variant::Large => opts, } @@ -245,6 +270,7 @@ impl Variant { Variant::CheckpointEach => true, Variant::CheckpointEvery3 | Variant::NormalSync => k % 3 == 0, Variant::WalOnly | Variant::AutoCheckpoint => false, + Variant::PerTxSync => k % 5 == 0, #[cfg(feature = "test-hooks")] Variant::Large => k % 5 == 0, } @@ -501,17 +527,27 @@ fn crash_kill_child_entrypoint() { #[cfg(feature = "test-hooks")] let _chunk = large.then(|| minigraf::__tx_chunk_facts(Some(LARGE_CHUNK))); let mut rng = Rng::new(seed); + // A separate stream, so the workload (and so the model) is unchanged. + let mut sync_rng = Rng::new(seed.rotate_left(29)); + let per_tx = matches!(variant, Variant::PerTxSync); let mut live = BTreeSet::new(); for k in 1.. { let (stmt, next) = next_tx(&mut rng, &live); - #[cfg(feature = "test-hooks")] - if large && k % 3 == 0 { - in_write_tx(&db, &stmt); + if per_tx { + let opts = TxOptions::new().synchronous(tx_sync_mode(&mut sync_rng, k)); + let mut tx = db.begin_write_with(opts).expect("child begin_write_with"); + tx.execute(&stmt).expect("child statement"); + tx.commit().expect("child commit"); } else { + #[cfg(feature = "test-hooks")] + if large && k % 3 == 0 { + in_write_tx(&db, &stmt); + } else { + db.execute(&stmt).expect("child transaction"); + } + #[cfg(not(feature = "test-hooks"))] db.execute(&stmt).expect("child transaction"); } - #[cfg(not(feature = "test-hooks"))] - db.execute(&stmt).expect("child transaction"); log.write_all(format!("{k}\n").as_bytes()) .expect("child logs transaction"); live = next;