diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a769c88b..77b6c565f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,8 +19,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **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). +- **Owned write transactions and browser transactions** (#824). `Minigraf::begin_write_owned()` and `begin_write_owned_with(TxOptions)` return an `OwnedWriteTransaction`: `execute`, `commit` and `rollback` with the borrowed `WriteTransaction`'s results and errors, but it is `Send` and `'static`, keeps the database open, and may begin on one thread and run, commit, roll back or be dropped on another. Dropping it open rolls it back. While it is open the calls a `WriteTransaction` refuses are `INT-001` on the thread that began it (per database, whichever thread uses it meanwhile); other threads' writes, checkpoints and begins wait until it finishes, and their queries read the committed state without waiting. A user function in one of its statements that calls the database gets `INT-001` on whichever thread runs the statement. Every language binding's transaction is built on it. `BrowserDb.beginWrite()` returns a `BrowserWriteTransaction` with a synchronous `execute`, a `commit()` that applies the staged facts as one transaction and resolves after the IndexedDB flush, and a `rollback()`; freeing it open rolls it back. While it is open, `BrowserDb`'s `execute`, `query`, `prepare`, `checkpoint`, `importGraph`, a second `beginWrite` and a prepared query's `execute`/`query` throw or reject with `[INT-001]`; a finished one's `execute` and `commit` are `[API-018]`. No new error codes; no file format change. + ### Changed +- **INT-001 text** (#824): `a WriteTransaction is already in progress on this thread; use tx.execute() instead` now reads `a write transaction is already in progress on this thread; use tx.execute() instead`, since owned and browser transactions raise it too. The code and the calls that raise it for a borrowed transaction are unchanged. + - **INT-015 text** (#870): `{} must be a positive integer` is now `{} must be a non-negative integer`, because `:query-memory-limit` accepts 0. The code and when `:max-derived-facts`/`:max-results` raise it are unchanged. - **INT-020 round-limit text** (#861): `Max iterations (N) exceeded. Possible infinite recursion or cycle in rules.` now reads `Max iterations (N) exceeded. A stratum's rules were still deriving new facts after N rounds; raise the limit with :max-iterations N or OpenOptions::max_iterations.` The code is unchanged. @@ -38,6 +42,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Testing +- **Owned and browser transaction tests** (#824): `tests/owned_tx_test.rs` checks that an owned transaction leaves the database as a borrowed one given the same statements and options (commit, rollback, drop; small and large; file and in-memory, after reopening), the threads rules and the begin refusals; the fault matrix runs its large-transaction and mixed-sync-mode cases through the owned transaction too. `src/browser/tx_tests.rs` (`wasm-bindgen`) checks the browser transaction against the native one over fixed statement sequences, its IndexedDB commit, rollback, drop and failed-flush paths, and the `[INT-001]`/`[API-018]` rules. + - **Acceptance execution gaps** (#837): nonlinear and multistratum recursive scaling, custom transaction chunk/clamp bounds, mandatory Linux RSS assertions, and exact named-test/output verification in the nightly smoke route. The storage diff --git a/CLAUDE.md b/CLAUDE.md index 7ca4f0ace..a551d34ad 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -187,7 +187,7 @@ 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/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`) +6. **`src/db.rs`** — Public API: `Minigraf::open/execute/query/fact_log/prepare/begin_write/begin_write_with/begin_write_owned/begin_write_owned_with/checkpoint/save/verify/rebuild_indexes`, `WriteTransaction`, `OwnedWriteTransaction` (#824: `Send + 'static`, holds a `Minigraf` clone; the write lock is `Mutex` + `Condvar`, `WriteSlot::owned` makes every other writer wait in `Inner::lock_write`; `TxOwner` records its begin thread per database, and `refuse_reentry` gives `INT-001` there as the thread-local flag does for `WriteTransaction`, and each owned statement sets that flag on the thread running it (`RunningHere`); both kinds share `TxState`, and small staging is `Staged`, which `BrowserWriteTransaction` uses too), `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) @@ -197,7 +197,7 @@ cargo run < demos/demo_negation.txt 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 -9. **`src/browser/`** — Browser WASM backend (`browser` feature): `buffer.rs` (`BrowserBufferBackend`, in-memory pages with dirty tracking; free pages a pinned generation may read are withheld from release, `pins::split_released`), `indexeddb.rs` (IndexedDB persistence); `BrowserDb.query` returns a `BrowserCursor` (JSON row batches); `BrowserDb.prepare` returns a `BrowserPreparedQuery` (`execute`/`query` with a bindings object; `withMaxDerivedFacts`/`withQueryMemoryLimit`/`withMaxIterations` return a new one, #872; read by `binds.rs`: plain JS values or `{type, value}` forms, #181; a set as `{type: 'set', value: [..]}` or a JS `Set`, #316); `udfs.rs`: `registerPredicate`/`registerAggregate` (#180; values as `{type, value}` objects, a throw is `QRY-014`; JS values held in a thread-local table, the registry keeps ids). Tests here must be `#[wasm_bindgen_test]` (the module only builds for wasm32); `.github/workflows/wasm.yml` runs them in headless browsers, runs the WASI tests under Wasmtime, and lints both wasm targets. New file-backed integration test files need `#![cfg(not(target_arch = "wasm32"))]` +9. **`src/browser/`** — Browser WASM backend (`browser` feature): `buffer.rs` (`BrowserBufferBackend`, in-memory pages with dirty tracking; free pages a pinned generation may read are withheld from release, `pins::split_released`), `indexeddb.rs` (IndexedDB persistence); `BrowserDb.query` returns a `BrowserCursor` (JSON row batches); `BrowserDb.prepare` returns a `BrowserPreparedQuery` (`execute`/`query` with a bindings object; `withMaxDerivedFacts`/`withQueryMemoryLimit`/`withMaxIterations` return a new one, #872; read by `binds.rs`: plain JS values or `{type, value}` forms, #181; a set as `{type: 'set', value: [..]}` or a JS `Set`, #316); `BrowserDb.beginWrite` returns a `BrowserWriteTransaction` (#824: sync `execute` staging through `db::Staged`, `commit()` a `Promise` resolved after the IndexedDB flush, `rollback()`; `BrowserDbInner::tx_open` makes every other read/write call `INT-001`; finished = `API-018`; tests in `tx_tests.rs`); `udfs.rs`: `registerPredicate`/`registerAggregate` (#180; values as `{type, value}` objects, a throw is `QRY-014`; JS values held in a thread-local table, the registry keeps ids). Tests here must be `#[wasm_bindgen_test]` (the module only builds for wasm32); `.github/workflows/wasm.yml` runs them in headless browsers, runs the WASI tests under Wasmtime, and lints both wasm targets. New file-backed integration test files need `#![cfg(not(target_arch = "wasm32"))]` ### Data Model @@ -250,7 +250,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**2133 tests** (2112 active, 21 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). +**2159 tests** (2138 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 62ad0e7ef..29ffa48f6 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,12 @@ let mut tx = db.begin_write()?; tx.execute(r#"(transact [[:alice :person/age 31]])"#)?; tx.commit()?; +// Owned transaction — Send + 'static, so it can live in an object or move to +// another thread; every language binding's transaction is one +let mut tx = db.begin_write_owned()?; +tx.execute(r#"(transact [[:alice :person/age 32]])"#)?; +std::thread::spawn(move || tx.commit()).join().unwrap()?; + // Time travel — query as of past transaction counter db.execute("(query [:find ?age :as-of 1 :where [:alice :person/age ?age]])")?; @@ -271,7 +277,7 @@ See the [Mobile Integration](https://project-minigraf.github.io/minigraf-docs/la ### For WASM / Browser -Published as [`@minigraf/browser`](https://www.npmjs.com/package/@minigraf/browser) on npm (IndexedDB-backed, `wasm-pack`). WASI build (`wasm32-wasip1`) available as [`@minigraf/wasi`](https://www.npmjs.com/package/@minigraf/wasi) on npm and as a GitHub Releases artifact (Wasmtime / Wasmer). See the [Use Cases](https://project-minigraf.github.io/minigraf-docs/latest/use-cases/). The [playground](https://minigraf-playground.vercel.app/) and the [time travel visualizer](https://project-minigraf.github.io/minigraf-visualizer/) are both built on `@minigraf/browser`. +Published as [`@minigraf/browser`](https://www.npmjs.com/package/@minigraf/browser) on npm (IndexedDB-backed, `wasm-pack`). `BrowserDb.beginWrite()` opens a write transaction (from v3.0.0): `execute` stages and reads back, `await tx.commit()` stores it in IndexedDB, `rollback()` discards it, and the database's other calls throw `[INT-001]` until it ends. The WASI package is the REPL: each command commits on its own, with no explicit transactions. WASI build (`wasm32-wasip1`) available as [`@minigraf/wasi`](https://www.npmjs.com/package/@minigraf/wasi) on npm and as a GitHub Releases artifact (Wasmtime / Wasmer). See the [Use Cases](https://project-minigraf.github.io/minigraf-docs/latest/use-cases/). The [playground](https://minigraf-playground.vercel.app/) and the [time travel visualizer](https://project-minigraf.github.io/minigraf-visualizer/) are both built on `@minigraf/browser`. ### For Python / Node.js / Java / C diff --git a/docs/ERROR_REFERENCE.md b/docs/ERROR_REFERENCE.md index 229f27035..f2bfc255a 100644 --- a/docs/ERROR_REFERENCE.md +++ b/docs/ERROR_REFERENCE.md @@ -241,7 +241,7 @@ with no `CodedError` anywhere in its chain. | API-024 | Bind slot type mismatch | Database API | | API-025 | Query option override not allowed | Database API | | INT-000 | Unclassified internal error | Internal | -| INT-001 | WriteTransaction already in progress on this thread | Internal | +| INT-001 | Write transaction already in progress on this thread | Internal | | INT-002 | Invalid entity (API layer) | Internal | | INT-003 | Invalid value (API layer) | Internal | | INT-004 | Invalid timestamp | Internal | @@ -3557,7 +3557,7 @@ let mut log = src.fact_log(&FactFilter::new())?; **Error text**: `{} is closed` -**Cause**: A language binding object was used after it was closed or finished, for example `append` or `finish` on a log writer after `finish` or `close`. Rust consumes these objects, so the Rust API never returns this code. Bindings cannot consume an object a caller still holds, so they return this error instead. A cursor or fact log reports the end (no more batches) after `close`, not this error. The `{}` names the object. +**Cause**: A language binding object was used after it was closed or finished, for example `append` or `finish` on a log writer after `finish` or `close`, or `execute` or `commit` on a write transaction after it committed (or failed to), rolled back, or its database was closed (#824). A finished transaction's `rollback` does nothing instead. Rust consumes these objects, so the Rust API never returns this code. Bindings cannot consume an object a caller still holds, so they return this error instead. A cursor or fact log reports the end (no more batches) after `close`, not this error. The `{}` names the object. **Resolution**: - Create a new object instead of reusing a finished one. @@ -3704,22 +3704,36 @@ specific code, so this should now be rare. **Scenario**: An I/O error from the underlying filesystem propagates through `?` without ever being wrapped in a `CodedError`. -### INT-001 WriteTransaction already in progress on this thread +### INT-001 Write transaction already in progress on this thread -**Error text**: `a WriteTransaction is already in progress on this thread; use tx.execute() instead` +**Error text**: `a write transaction is already in progress on this thread; use tx.execute() instead` **Cause**: on the same thread that already holds an active `WriteTransaction`, one of these was called: the `Minigraf` methods -`execute()`, `query()`, `fact_log()`, `begin_write()`, `checkpoint()`, -`verify()`, `rebuild_indexes()` or `prepare()`, or a `PreparedQuery`'s -`execute()` or `query()`. Acquiring the write +`execute()`, `query()`, `fact_log()`, `begin_write()`, `begin_write_owned()`, +`checkpoint()`, `verify()`, `rebuild_indexes()` or `prepare()`, or a +`PreparedQuery`'s `execute()` or `query()`. Acquiring the write lock again on that thread would deadlock, so this is detected and rejected instead. (Before v3.0.0, `checkpoint()`, `verify()` and `rebuild_indexes()` did not check and deadlocked, #734.) +The same calls are refused on the thread that began an open +`OwnedWriteTransaction` of that database (#824), whichever thread uses the +transaction meanwhile, and on the thread running one of its statements (a +user function calling back into the database); other threads' writes and begins wait for it +instead, and other databases are not affected. A binding transaction is an +owned one, so the same rule applies to every language binding. In the +browser, every `BrowserDb` call that reads or writes (`execute`, `query`, +`prepare`, `checkpoint`, `importGraph`, a second `beginWrite`, and a +prepared query's `execute`/`query`) is refused while a +`BrowserWriteTransaction` is open, since other asynchronous code could +otherwise mistake the database for the transaction. + **Resolution**: -- Use the existing `WriteTransaction`'s `tx.execute()` for further writes +- Use the existing transaction's `tx.execute()` for further writes instead of calling `db.execute()` again on the same thread. +- End the transaction (`commit()` or `rollback()`) before using the + database directly again. **Scenario**: Code inside a `WriteTransaction` write closure calls `db.execute("(transact ...)")` on the outer `db` handle instead of diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index 24b0e1fb5..858f63de0 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,12 +4,12 @@ | | Default features | With `--features test-hooks` | |---|---|---| -| **Active** | 2,112 | 2,150 | +| **Active** | 2,138 | 2,176 | | **Ignored** | 21 | 30 | -| **Total** | 2,133 | 2,180 | +| **Total** | 2,159 | 2,206 | Counts above are the current Cargo test inventory (active versus ignored), -including #777, #789, #781, #794, #795, #837, #865, #869, #870, #861, #863, #871 and #872 (Linux). Verification results below identify which suites ran. +including #777, #789, #781, #794, #795, #837, #865, #869, #870, #861, #863, #871, #872 and #824 (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, @@ -392,6 +392,15 @@ These are not `#[test]`s; run them with `cargo kani` or `.github/workflows/kani. ## WAL and Recovery +- **Owned write transactions** (#824; `tests/owned_tx_test.rs`, 16 tests; `src/db.rs`, 2 unit tests; `src/fault_matrix.rs`, 7 tests; 1 doctest; full default suite passed, 2,138 passed and 21 ignored, 2026-11-11) + - `OwnedWriteTransaction` is `Send + 'static` (compile-time check). + - `owned_tx_matches_borrowed_tx`: for five statement sequences (writes, retractions, staged reads, rules, `API-011`, parse errors, latest statement wins) × commit/rollback/drop × the database's chunk or a chunk of 2 (large on a file), in memory and on a file, an owned and a borrowed transaction give the same statement outcomes, fact records, rows and `tx_count`, also after reopening. A large owned commit is a checkpoint (no WAL left). + - Threads: begun on one thread and run, committed, rolled back or dropped (large) on another; afterwards every guarded call works on every thread. While open: every guarded call (`execute`, `query`, `prepare`, prepared `execute`/`query`, `fact_log`, `checkpoint`, `verify`, `rebuild_indexes`, both begins) is `INT-001` on the begin thread, also while another thread holds the transaction; another thread's `transact`, rule, `checkpoint`, `verify`, `rebuild_indexes` and both begins wait and then succeed; its `execute` query, cursor, prepared query and fact log read the committed state without waiting; another database is neither guarded nor blocked. + - Begin refusals: `INT-001` inside a borrowed or an owned transaction, `API-014` read-only (file and memory); the handle stays usable. Empty and query-only transactions take no `tx_count`; a failed statement stages nothing and leaves it open; the transaction keeps the database open after its last handle is dropped. + - A user function that writes or queries through the database, from a statement run on a thread that did not begin the transaction, fails the query with `QRY-014` (`INT-001` inside) instead of deadlocking or reading outside the transaction; the transaction then commits. + - Unit tests: a commit refused at commit time (`API-019`) releases the write lock; a poisoned write lock still ends the transaction (`API-001`). + - Fault matrix: the four large write-transaction cases and the three mixed-sync-mode commits run again through an owned transaction. +- **Browser write transactions** (#824; `src/browser/tx_tests.rs`, 12 `#[wasm_bindgen_test]`, not counted above): staging and read-your-own-writes, rules registered at once and kept after a rollback, failed statements; a differential check against the native transaction over five sequences (outcomes, current, `:any-valid-time` with `:db/tx-count`, `:valid-at` and `:as-of` results, `tx_count`); one `tx_count` per commit, facts visible before the flush resolves, commit/rollback/drop each across an IndexedDB reopen; a failed IndexedDB flush rejects the commit as it fails `execute`; a commit refused with `API-019` finishes the transaction; every guarded `BrowserDb` call `[INT-001]` while open, a cursor opened before still reading; finished transactions `[API-018]`; a user function of a transaction's query that calls `beginWrite` or `query` fails the query (`QRY-014`) without a panic. - **Torn WAL tail** (#513) - Opening a WAL writer cuts off a partial last entry (cut in the checksum, the header or the payload), or an entry with a bad checksum and everything after it, then appends where replay reaches. An intact WAL is left byte for byte. - An append that fails part-way makes every later append `WAL-007`. A rejected oversize fact (`WAL-003`, nothing written) does not. @@ -667,7 +676,7 @@ These are not `#[test]`s; run them with `cargo kani` or `.github/workflows/kani. None of these are counted in the summary. The wasm32 targets run in CI (`.github/workflows/wasm.yml`). -- **In a browser** (`wasm-pack test --headless`, Firefox and Chrome): 68 `wasm_bindgen_test` tests (Firefox, 2026-11-10). The final #641 snapshot passed in headless Firefox on the bench; Chrome remains a CI run. +- **In a browser** (`wasm-pack test --headless`, Firefox and Chrome): 80 `wasm_bindgen_test` tests (Firefox, 2026-11-11, after #824). The final #641 snapshot passed in headless Firefox on the bench; Chrome remains a CI run. - Three committed-memory-backend input-order fixtures: scalar Sort omission, streaming counts with Boolean/float representatives and mixed numeric group order, and partition windows preserving float edges and numeric group classes (#641). - Twenty-six for `BrowserDb`, including IndexedDB persistence, the default `allow_overrides` list (#869), the cursor, and the flush returning every dirty page and failing on one it cannot read (#470). - Eleven for `BrowserBufferBackend` (which used `#[test]` and never ran before). diff --git a/docs/superpowers/specs/2026-11-10-824-binding-transactions-design.md b/docs/superpowers/specs/2026-11-10-824-binding-transactions-design.md new file mode 100644 index 000000000..719df0787 --- /dev/null +++ b/docs/superpowers/specs/2026-11-10-824-binding-transactions-design.md @@ -0,0 +1,204 @@ +# Binding write transactions — design (#824) + +**Mission Type:** mixed (code in 9 repositories, plus documentation) +**Base branches:** `v3` in `project-minigraf/minigraf` (core) and `v3` in every binding repository. Nothing targets `main`. +**Approved:** 2026-11-10. + +## 1. User Intent + +> #824 + +Issue #824, "Expose explicit write transactions across all supported language bindings" (milestone v3.0.0): + +> Expose explicit write transactions across all supported Minigraf language bindings, with behavior matching the Rust API: begin a transaction, run multiple reads and writes against its staged state, then commit or roll back. +> +> The immediate consumer is ldbc-benchmarks. FinBench conditional read-write operations need tentative writes, reads that see those writes, and a conditional commit or abort within one operation. Separate auto-committed `execute()` calls cannot preserve that boundary. + +The issue's "Expected contract" and "Completion criteria" sections are part of the intent and are covered by the acceptance criteria below. + +> [Asked how the bindings should hold a core write transaction across calls and threads] Owned transaction in core. + +> [Asked what transaction-control contract the WASI package (the REPL) should have] Document auto-commit only. + +## 2. Goal + +An application in any binding can begin a write transaction, run several `transact`, `retract` and `query` statements in it with queries seeing the staged writes, and then commit everything atomically or roll it back. Abandoned, failed and misused transactions clean up and never leave the write lock held. To make that sound, core gains a transaction that owns what it needs and can move between threads, and `BrowserDb` gains its own transaction. WASI's contract (auto-commit per REPL command, no explicit transactions) is recorded and tested. + +## 3. Terms + +- **Core**: the `minigraf` crate, branch `v3`. At planning time `origin/v3` is `d9755c02`. +- **Binding**: one of `minigraf-python`, `minigraf-java`, `minigraf-android`, `minigraf-swift`, `minigraf-node`, `minigraf-c`, `minigraf-wasm` (clones at `../minigraf-`). **Template** is `minigraf-binding-template`. **Native bindings** are all of these except `minigraf-wasm`. +- **UniFFI bindings**: python, java, android, swift and the template. Each has a Rust crate (`src/lib.rs`) and, except the template, a language wrapper. +- **Borrowed transaction**: core's existing `WriteTransaction<'_>` from `begin_write` / `begin_write_with`. +- **Owned transaction**: the new core transaction of section A. +- **Binding transaction**: the object a binding returns from its begin call. +- **Open**: a transaction that has not finished. **Finished**: after `commit` (successful or failed), after `rollback`, or after the database it belongs to was closed (Node, C). +- **Staged**: written by a statement of an open transaction and not yet committed. +- **Begin thread**: the thread that called the begin function of an open transaction. +- **Guarded calls**: the database calls that core refuses with `INT-001` on a thread with an open borrowed transaction today: `execute`, `query`, `prepare` and prepared `execute`/`query`, `fact_log`, `checkpoint`, `verify`, `rebuild_indexes`, and a second begin (`src/db.rs`, callers of `is_write_tx_active`; `tests/write_tx_guard_test.rs`). +- **Scope form**: a wrapper's language construct that ends a transaction when a block is left (Python `with`, Kotlin/Java `AutoCloseable`, a Swift closure form). + +## 4. Context & Investigation Findings + +### Core (`origin/v3`) + +- `src/db.rs:1183` `begin_write`, `:1217` `begin_write_with(TxOptions)`; `:452` `TxOptions { synchronous, chunk_facts }` (#871). +- `src/db.rs:1897` `WriteTransaction<'a>` holds a `MutexGuard<'a, WriteContext>` and `&'a Inner`: it borrows the database and is not `Send`. +- `src/db.rs:47-59`: the thread-local flag `WRITE_TX_ACTIVE` is set by begin and cleared by `commit`, `rollback` and `Drop` on the thread that runs them. A borrowed transaction cannot safely begin on one thread and end on another. +- `WriteTransaction::execute` (`:1941`): a `transact`/`retract` is staged and returns `QueryResult::Ok`; a `query` reads committed plus staged facts (`:2011`); a rule is registered at once in the shared registry and is not undone by a rollback. No cursor and no prepared query exists inside a transaction. +- `commit(self)` `:2170`, `rollback(self)` `:2298`, `Drop` rolls back `:2342`. A transaction that passes its chunk size on a file database becomes large (#550) and commits as a checkpoint. +- A write from another thread waits on the write lock until the transaction ends. A query from another thread reads the committed state without waiting. +- `docs/superpowers/specs/2026-11-06-862-override-audit-design.md` §6: `TxOptions` reaches the bindings with #824. + +### Browser (core `src/browser/mod.rs`) + +- `BrowserDb` (`:50`) is not built on `Minigraf`. It holds `Rc>` with its own `FactStorage` and page store, takes no options, and has no transaction type. A write is applied and saved synchronously, then flushed to IndexedDB (`apply_write`, `:371`). +- `minigraf-wasm` re-exports core; its `wasi-pkg` is the REPL. `src/repl.rs` has no transaction commands. + +### Bindings (`origin/v3` of each; all pin core `d9755c02`) + +- UniFFI crates: `MiniGrafDb { inner: Arc> }` (python `src/lib.rs:514`); each call clones the `Minigraf` out of that mutex (`db()`, `:651`) so that no binding lock is held while core runs. Objects must be `Send + Sync` and are freed by the host's collector on any thread. The four bindings' `src/lib.rs` are the same file. +- Node (`src/lib.rs:439`): `close()` empties the handle and closes its prepared queries (registry of weak references); calls are synchronous on the JS thread. +- C (`src/lib.rs:88`): per-object `*_last_error`; the database handle keeps a registry of its prepared handles; `include/minigraf.h` is generated by `cbindgen`. +- No binding exposes a transaction or `TxOptions`. Every binding has `consumer-test/` and a `package.yml` workflow. + +### Constraints + +- Core size gate on `v3`: 1,240 KiB. It is not raised by this change. +- Error codes are never removed or recycled. This change is expected to add none: `INT-001` and `API-018` are reused. +- Core PRs to `v3` merge with a merge commit. No PR is merged without the maintainer's explicit confirmation. +- Tests follow `CLAUDE.md` §Testing Conventions (no `{:?}` of `Result`/`Value`/… in assert messages). + +## 5. Acceptance Criteria + +### A. Core PR: owned transaction + +- A1. `Minigraf` has two new public functions that begin an owned transaction, one with the database's options and one taking `TxOptions`. The owned transaction type is `Send` and `'static`, keeps the database open while it exists, and offers `execute`, `commit` and `rollback` with the results and errors of the borrowed transaction's. Dropping it open rolls it back. A compile-time test asserts `Send + 'static`. +- A2. For any sequence of statements, an owned transaction leaves the database in the state a borrowed transaction given the same statements and `TxOptions` leaves it in: after commit, after rollback and after drop; small and large (#550) transactions; file and in-memory databases. Test: `owned_tx_matches_borrowed_tx`, comparing query results and, on a file database, results after reopening. +- A3. An owned transaction may begin on one thread and run `execute`, `commit`, `rollback` or be dropped on another. Afterwards every guarded call succeeds again on every thread, including the begin thread. Tests: `owned_tx_commit_on_other_thread`, `owned_tx_drop_on_other_thread`. +- A4. While an owned transaction is open, on its database: every guarded call from the begin thread fails with `INT-001`; a write or a begin (owned or borrowed) from any other thread waits until the transaction ends and then proceeds; a query from another thread returns the committed state without waiting. A different database is not affected by it. Tests: `owned_tx_guards_begin_thread`, `owned_tx_other_thread_write_waits`, `owned_tx_other_thread_query_reads_committed`, `owned_tx_leaves_other_database_usable`. +- A5. Beginning an owned transaction fails with `INT-001` on a thread that has an open transaction of either kind on that database, with `API-014` on a read-only handle, and leaves the handle usable in both cases. +- A6. The borrowed transaction's public API and behaviour are unchanged: every existing test passes unmodified, including `tests/write_tx_guard_test.rs`, `tests/concurrency_test.rs`, the fault-injection and crash tests, and the lock-poisoning test (`API-001`). `cargo semver-checks` (v3 vs PR base) passes. +- A7. The model-based or fault-matrix test that covers `WriteTransaction` commits today also runs its transaction cases through the owned transaction, or the PR description states which existing suite covers the changed lock and why that is sufficient. +- A8. The size gate passes at 1,240 KiB. If it cannot, the work stops and the measured size is reported; it does not raise the gate. + +### B. Core PR: browser transaction + +- B1. `BrowserDb` has a synchronous begin function returning a transaction object with a synchronous `execute(datalog)`, an asynchronous `commit()` and a synchronous `rollback()`. It takes no options. +- B2. `execute` inside the transaction: a `transact`/`retract` is staged and returns the JSON the browser API uses for core's `QueryResult::Ok`; a `query` returns committed plus staged facts in the JSON shape of `BrowserDb.execute`; a rule is registered at once. A statement that fails stages nothing and leaves the transaction open. +- B3. For any sequence of statements, committing a browser transaction leaves `BrowserDb` returning the same query results as an in-memory core `Minigraf` returns after a transaction with the same statements, including the rejections (`API-011`, invalid valid-time windows) and the latest-statement-wins rule (#477). Test: a `wasm-bindgen` differential test over a fixed list of sequences. +- B4. `commit()` applies the staged facts as one transaction (one `tx_count`; no query sees part of it), and its promise resolves after the IndexedDB flush. After a resolved commit, reopening the database by name returns the committed facts. A failed flush rejects the promise and leaves the state a failed flush of `BrowserDb.execute` leaves today. +- B5. `rollback()`, and dropping or freeing the object while open, discard the staged facts: no query returns them and a reopen does not either. +- B6. While a browser transaction is open, `BrowserDb`'s `execute`, `query`, `prepare`, `checkpoint`, `importGraph`, a second begin, and `execute`/`query` of a prepared query throw or reject with an error whose message starts with `[INT-001]`. A cursor opened before the transaction began keeps returning its rows. After the transaction finishes, all of them work. +- B7. A finished browser transaction: `execute` and `commit` fail with `[API-018]`; `rollback` does nothing. +- B8. `wasm-bindgen` tests in `src/browser/` cover B2–B7, including a persistent (IndexedDB) commit-and-reopen and rollback-and-reopen. + +### C. Core PR: documentation + +- C1. `CHANGELOG.md`, `docs/TEST_COVERAGE.md`, the `CLAUDE.md` test count and module notes, and `README.md` where it describes transactions or the browser API. +- C2. `docs/ERROR_REFERENCE.md` and the `INT-001` registry text describe the owned-transaction case (the code is unchanged), and `API-018` names a finished transaction among its causes. +- C3. This spec is copied to `docs/superpowers/specs/2026-11-10-824-binding-transactions-design.md`. + +### D. Every native binding (6 bindings + template): transaction object + +- D1. The database object has a begin call returning a binding transaction, and a form of it taking transaction options: `synchronous` (the binding's existing spelling of `Full`/`Normal`) and `chunk_facts`, each optional, absent meaning the database's value. A negative `chunk_facts` is rejected as the binding rejects other negative counts. On a read-only database, begin fails with `API-014`. +- D2. The transaction has `execute(datalog)`, returning what the database's `execute` returns for the same core result (a staged write returns the binding's encoding of `QueryResult::Ok`), `commit()` and `rollback()`. It has no cursor and no prepared-query call. +- D3. Read-your-own-writes: after a `transact` and a `retract` in a transaction, a `query` through the transaction returns the staged state; the same query from another thread through the database returns the committed state (Node: checked before begin and after rollback instead). +- D4. `commit()` makes every staged change visible to database queries at once and, on a file database, present after close and reopen. `rollback()` leaves the database as it was, also after close and reopen. +- D5. A statement that fails (parse error, `API-011`, …) reports core's error code, stages nothing, and leaves the transaction open and usable. +- D6. A finished transaction: `execute` and `commit` fail with `API-018`; `rollback` and close/free do nothing and report no error. A `commit` that fails finishes the transaction and rolls it back. +- D7. Abandonment rolls back and releases the write lock: dropping the last reference in the Rust crate (UniFFI, Node), `minigraf_tx_free` on an open transaction (C), and leaving a scope form without a commit, normally or by an exception. After each, a write through the database succeeds from every thread. Tests name each path. +- D8. While a transaction is open, the database object behaves as A4 states: guarded calls on the begin thread fail with `INT-001`; a write or begin from another thread waits and then succeeds; a query from another thread reads the committed state. While a begin waits, queries from other threads through the same database object still run (the binding holds none of its own locks while it waits). +- D9. One transaction object used from two threads at once does not interleave within a statement and cannot corrupt state: each call runs whole, and a call that loses a race with `commit`/`rollback` fails with `API-018`. A transaction begun on one thread may be committed, rolled back or freed on another (A3). +- D10. Closing the database with a transaction open: in Node (`close()`) and C (`minigraf_close`) the transaction is rolled back and finished (`API-018` on later use; a C transaction handle must still be freed, and freeing it after the close is safe). In the UniFFI bindings the transaction keeps the database open, as prepared queries do. +- D11. `chunk_facts` works through the binding: on a file database, a transaction begun with a small `chunk_facts` that stages more facts than that commits, and every fact is returned after reopening. +- D12. A user function registered on the database runs in a query inside a transaction and sees staged values. A user function that makes a guarded call on the database fails that query with `QRY-014`; nothing deadlocks. +- D13. Rules registered inside a transaction remain after a rollback (tested once per binding implementation: UniFFI crate, Node, C). + +### E. Public names + +Names follow each binding's conventions. This table is the contract unless a name collides with an existing one, in which case the PR description states the name used. + +| Where | Begin | Options | Transaction | +|---|---|---|---| +| Core (Rust) | `Minigraf::begin_write_owned()`, `begin_write_owned_with(TxOptions)` | `TxOptions` (existing) | `OwnedWriteTransaction`: `execute`, `commit`, `rollback` | +| Core browser | `BrowserDb.beginWrite()` | none | `BrowserWriteTransaction`: `execute`, `commit` (Promise), `rollback` | +| UniFFI crates (all 5) | `MiniGrafDb.begin_write()`, `begin_write_with(options)` | record `TxOptions { synchronous: SyncMode?, chunk_facts: i64? }` | `MiniGrafWriteTransaction`: `execute`, `commit`, `rollback` | +| Python wrapper | `db.begin_write(synchronous=None, chunk_facts=None)` | keyword arguments | context manager: `with db.begin_write() as tx:` rolls back on exit unless committed | +| Java, Android (Kotlin) | `beginWrite()`, `beginWrite(options)` | `TxOptions` with a builder or named arguments, as the binding's open options do | `AutoCloseable`; `close()` rolls back unless committed | +| Swift | `beginWrite(options:)` | `TxOptions` | plus a closure form that rolls back when the closure throws or returns without committing | +| Node | `db.beginWrite(options?)` | `{ synchronous?: 'full' \| 'normal', chunkFacts?: number }` | `WriteTransaction`: `execute`, `commit`, `rollback`; typed in `index.d.ts` | +| C | `minigraf_begin_write(db, options_json)` (`NULL` = defaults; unknown keys rejected) | JSON `{"synchronous": …, "chunk_facts": …}` | `MiniGrafTx*`: `minigraf_tx_execute`, `minigraf_tx_commit`, `minigraf_tx_rollback`, `minigraf_tx_last_error`, `minigraf_tx_free`; declared in `include/minigraf.h` | + +### F. wasm binding + +- F1. Pins the merged core commit; the generated TypeScript declarations list the begin call and the transaction class. +- F2. `consumer-test/browser/contract.js` runs, against the packed package: a commit with read-your-own-writes, a rollback, and one `[INT-001]` rejection from B6. +- F3. WASI contract: the `wasi-pkg` README states that the package is the REPL, that every command commits on its own, and that it has no explicit transactions. A `wasi-pkg` test pipes a `transact`, a command that fails, and a `query`, and checks that the first command's fact is returned (the failed command did not undo it). `wasi-pkg` builds and its existing tests pass at the new pin. + +### G. Every binding PR (7 bindings + template) + +- G1. `Cargo.toml` pins a commit that is on core `v3` and contains the core PR. No PR is marked ready for merge while it pins a commit that is only on a PR branch. +- G2. The binding's own tests, its `consumer-test`, its `package.yml` and its other CI jobs are green at the new pin. Anything the pin move breaks is fixed in the same PR and listed in the PR description. +- G3. Tests exist at two levels in each native binding: the Rust crate (all of D, including the thread cases D8 and D9 and the reopen cases D4 and D11) and the language wrapper or consumer test through the built package (commit with read-your-own-writes, rollback, scope-form exit by exception, a finished-handle error, commit surviving reopen). +- G4. The README documents: begin, the options, `execute`/`commit`/`rollback`, the scope form with one complete example, what happens on abandonment, the `INT-001` rule for the begin thread and waiting from other threads, database close with an open transaction, that rules are not rolled back, and that an open transaction holds the write lock until it finishes, so it should always be ended explicitly or through the scope form. +- G5. The README or the binding's changelog names the first binding version that has write transactions and the core commit it embeds. +- G6. The PR description references #824 without a closing keyword. + +### H. Documentation outside the binding repositories + +- H1. Docs-site PR (`../minigraf-docs`): the pages that describe the bindings and the browser API (`content/use-cases.md` and whichever page documents transactions) describe binding transactions with one example per binding, the browser difference (no WAL; a commit is durable when its promise resolves), and the WASI contract. `content/comparison.md` is updated if its matrix has a transactions row. New text is inside `` where the surrounding text is shared with v2. Its link check passes. +- H2. No document contradicts another: core docs, the binding READMEs and the docs site agree on names and behaviour. + +### I. Completion + +- I1. Ten PRs exist and are green: core, 7 bindings, template, docs site. +- I2. #824 is closed by hand after the last PR merges, with a comment that lists the ten PRs and, per binding, the pinned core commit and the first transaction-capable version. + +## 6. Edge Cases & Boundaries + +- **Empty transaction.** Commit of a transaction with no staged write succeeds and changes nothing a query can see. +- **Only queries.** A transaction that ran only queries commits and rolls back without error. +- **Large transaction.** A binding transaction that becomes large behaves as core's does; its rollback and abandonment discard its working pages (D7 and D11 apply to it). +- **Failed handle.** After a failed checkpoint (`STG-046`), begin, `execute` and `commit` report what core reports; the binding adds nothing. +- **`synchronous`.** `Normal` cannot be shown to skip a sync from a binding test; it is enough that the option is accepted and the commit succeeds. In-memory databases accept and ignore both options. +- **Collector timing.** A collector may never run. No test may rely on a JVM, Python or JS collector to release a transaction; abandonment is tested through the Rust crate and the scope forms. +- **Waiting begin.** A begin from another thread blocks that thread for as long as the open transaction lasts. This is core's behaviour; the binding documents it and adds no timeout. +- **Transaction moved to another thread.** `INT-001` follows the begin thread, not the thread now using the transaction. A thread that uses a transaction it did not begin and then writes through the database waits on its own transaction. The READMEs warn against it. +- **Node.** There is one JS thread: every guarded call during an open transaction is `INT-001`, never a wait. +- **Browser awaits.** Other asynchronous code may run between a browser transaction's statements. B6 makes its database calls fail loudly instead of reading or writing outside the transaction. +- **Browser commit in flight.** Between `commit()` being called and its promise resolving, the transaction is finished and the facts are visible, as with `BrowserDb.execute` writes today. +- **Double free (C).** `minigraf_tx_free(NULL)` is a no-op, as the other `*_free` functions are. +- **Regressions to avoid.** `execute` and prepared queries outside a transaction behave as before; auto-committed writes keep their result JSON; the C header stays in sync with the exported symbols; Node's `index.d.ts` stays in sync with the addon; core durability behaviour (WAL sync modes, crash recovery, #554 whole-transaction visibility) is unchanged. + +## 7. Key Decisions + +- **Owned transaction in core** (maintainer, 2026-11-10). Core's borrowed transaction cannot be held by a binding object: it borrows the database, is not `Send`, and is tracked by a thread-local flag. The alternative, a worker thread inside each binding, would have meant three separate thread mechanisms and no way to call a Node user function from inside a transaction. The price is a change to the write lock used by every write path shortly before the v3.0.0 cut, which is why A2, A6 and A7 demand equivalence with the borrowed transaction and unchanged existing tests. +- **Guard follows the begin thread, per database.** It keeps Rust's rule (same thread: `INT-001`; other threads: wait) for a transaction that can move, and does not make one database's transaction block another database, which the thread-local flag does for borrowed transactions today. That existing quirk is left alone. +- **WASI: auto-commit only** (maintainer, 2026-11-10). The WASI package is the REPL; no transaction commands are added. +- **Surface equals core's.** `execute`, `commit`, `rollback` only. No cursor, no prepared query inside a transaction, no nested transactions, no savepoints. +- **`TxOptions` comes with the transactions**, as the #862 design §6 planned. The browser gets none: `BrowserDb` takes no options and has no WAL. +- **Finished objects fail with `API-018`**, the code bindings already use for closed objects. +- **Commit is always explicit.** A scope form never commits on normal exit; that keeps one rule in every language and makes a forgotten commit visible in tests instead of silently durable. +- **Browser calls fail during a transaction** rather than reading committed state, because the issue asks that ordinary calls are never mistaken for transaction-local ones. +- **Template included**, as in #180, #181, #316 and #872. + +## 8. Out of Scope + +- REPL transaction commands, and any library API for WASI. +- Cursors, prepared queries, savepoints or nested transactions inside a transaction. +- Options on `BrowserDb`; OPFS (#441). +- Rolling back rule registration. +- Lock timeouts or a non-blocking begin. +- Changing the borrowed transaction's cross-database `INT-001` behaviour. +- The ldbc-benchmarks switch from its JNI bridge to the Java binding. +- Changes to `main` or to v2.x binding branches; releasing or tagging any repository; merging any PR. + +## 9. Possible Directions & Ideas (non-binding) + +- The UniFFI crates' `src/lib.rs` files are the same; doing the template first and carrying the change to the other four may save time. +- A binding PR can be developed against the core PR's head commit and re-pinned to the `v3` merge commit once the maintainer merges core. +- Prior binding rollouts: #872, `docs/superpowers/plans/2026-10-24-181-prepared-ffi.md`, `2026-10-24-180-udf-uniffi.md`. The Node and C prepared-query registries show how `close()` reaches dependent objects. +- `src/db.rs:2011` (`read_snapshot`) shows how staged facts are layered over a snapshot; the browser transaction may be able to use the same snapshot call. +- The borrowed transaction could be expressed through the owned one or left as it is; either is acceptable if A6 holds. +- The laptop has 15 GB of RAM: build one binding at a time. Tests and timings run on the bench machine first. diff --git a/src/browser/mod.rs b/src/browser/mod.rs index d162ae669..07645af60 100644 --- a/src/browser/mod.rs +++ b/src/browser/mod.rs @@ -10,6 +10,9 @@ mod binds; pub mod buffer; /// Async IndexedDB backend for browser WASM persistence. pub mod indexeddb; +/// Tests of `BrowserWriteTransaction` (#824). +#[cfg(all(target_arch = "wasm32", feature = "browser", test))] +mod tx_tests; /// User-defined functions from JavaScript (#180). mod udfs; @@ -39,6 +42,24 @@ struct BrowserDbInner { /// Free pages whose bytes are kept because an open cursor's generation /// may read them (#543); dropped by a later commit once out of force. withheld: Vec, + /// Whether a `BrowserWriteTransaction` is open (#824): every other call + /// that reads or writes is then `INT-001`. + tx_open: bool, +} + +impl BrowserDbInner { + /// `INT-001` while a transaction is open. + fn refuse_in_tx(&self) -> anyhow::Result<()> { + if self.tx_open { + crate::error::bail_coded!(crate::error::ErrorCode::Int001); + } + Ok(()) + } +} + +/// `INT-001` while a transaction of `db` is open. +fn refuse_in_tx(db: &RefCell) -> Result<(), JsValue> { + db.borrow().refuse_in_tx().map_err(to_js_error) } /// Browser-only Minigraf database handle backed by IndexedDB. @@ -73,6 +94,7 @@ impl BrowserDb { pfs, idb: None, withheld: Vec::new(), + tx_open: false, })), }) } @@ -107,6 +129,7 @@ impl BrowserDb { pfs, idb: Some(idb), withheld: Vec::new(), + tx_open: false, })), }) } @@ -120,6 +143,7 @@ impl BrowserDb { /// - Rule: `{"ok": true}` #[wasm_bindgen(js_name = execute)] pub async fn execute(&self, datalog: String) -> Result { + refuse_in_tx(&self.inner)?; let cmd = parse_datalog_command(&datalog).map_err(to_js_error)?; // BrowserDb takes no options: the default override list (#869). if let DatalogCommand::Query(q) = &cmd { @@ -169,6 +193,7 @@ impl BrowserDb { /// `[API-010]` for a query with `$slot` bind slots. #[wasm_bindgen(js_name = query)] pub fn query(&self, datalog: &str) -> Result { + refuse_in_tx(&self.inner)?; let cursor = self.query_inner(datalog).map_err(to_js_error)?; Ok(BrowserCursor { vars: cursor.vars().to_vec(), @@ -190,6 +215,7 @@ impl BrowserDb { /// ``` #[wasm_bindgen(js_name = prepare)] pub fn prepare(&self, datalog: &str) -> Result { + refuse_in_tx(&self.inner)?; let query = self.prepare_inner(datalog).map_err(to_js_error)?; Ok(BrowserPreparedQuery { _db: self.inner.clone(), @@ -203,6 +229,7 @@ impl BrowserDb { /// so `checkpoint()` is only needed after `import_graph()` or explicit bulk ops. /// No-op for in-memory databases. pub async fn checkpoint(&self) -> Result<(), JsValue> { + refuse_in_tx(&self.inner)?; let (flush, idb) = { let mut inner = self.inner.borrow_mut(); inner.pfs.save().map_err(to_js_error)?; @@ -259,6 +286,7 @@ impl BrowserDb { /// queryable and all dirty pages are flushed to IndexedDB. #[wasm_bindgen(js_name = importGraph)] pub async fn import_graph(&self, data: js_sys::Uint8Array) -> Result<(), JsValue> { + refuse_in_tx(&self.inner)?; let bytes = data.to_vec(); if !bytes.len().is_multiple_of(crate::storage::PAGE_SIZE) { return Err(JsValue::from_str( @@ -311,6 +339,42 @@ impl BrowserDb { } Ok(()) } + + /// Begin a write transaction (#824): its `execute` stages writes and + /// reads them back, `commit()` applies them as one transaction and + /// stores it, and `rollback()` discards them. + /// + /// Synchronous. Takes no options. While it is open, every other call on + /// this database that reads or writes (`execute`, `query`, `prepare`, + /// `checkpoint`, `importGraph`, another `beginWrite`, and a prepared + /// query's `execute`/`query`) throws or rejects with `[INT-001]`, so + /// other asynchronous code cannot mistake the database for the + /// transaction. A cursor opened before keeps returning its rows. Always + /// end it, in a `finally` if need be: + /// + /// ```js + /// const tx = db.beginWrite(); + /// try { + /// tx.execute('(transact [[:alice :balance 90] [:bob :balance 110]])'); + /// const rows = JSON.parse(tx.execute('(query [:find ?b :where [:alice :balance ?b]])')); + /// if (rows.results[0][0] < 0) tx.rollback(); else await tx.commit(); + /// } finally { + /// tx.rollback(); // does nothing once committed + /// } + /// ``` + #[wasm_bindgen(js_name = beginWrite)] + pub fn begin_write(&self) -> Result { + // Before the mutable borrow: a user function of a transaction's query + // calling this gets INT-001, not a borrow panic. + refuse_in_tx(&self.inner)?; + let mut inner = self.inner.borrow_mut(); + inner.tx_open = true; + let staged = crate::db::Staged::new(inner.fact_storage.current_tx_count()); + Ok(BrowserWriteTransaction { + db: self.inner.clone(), + staged: Some(staged), + }) + } } impl BrowserDb { @@ -332,6 +396,7 @@ impl BrowserDb { inner.rules.clone(), inner.functions.clone(), crate::query::engine::Limits::default(), + std::sync::Arc::default(), ) } @@ -398,18 +463,7 @@ impl BrowserDb { }) .collect(); - // One batch, so no query sees part of the transaction (#554). - inner - .fact_storage - .load_facts(stamped) - .map_err(to_js_error)?; - - inner.pfs.mark_dirty(); - inner.pfs.save().map_err(to_js_error)?; - - // Collect dirty pages as owned Vec<(u64, Vec)> — no borrows - // escape — and drop the pages this commit freed (#440). - let flush = take_commit_flush(&mut inner)?; + let flush = commit_stamped(&mut inner, stamped)?; let json = if is_retract { format!(r#"{{"retracted":{}}}"#, tx_id) @@ -436,6 +490,167 @@ impl BrowserDb { } } +/// Apply stamped facts as one transaction and save: what an `execute` write +/// and a transaction's commit share. Returns what to send to IndexedDB. +fn commit_stamped( + inner: &mut BrowserDbInner, + stamped: Vec, +) -> Result { + // One batch, so no query sees part of the transaction (#554). + inner + .fact_storage + .load_facts(stamped) + .map_err(to_js_error)?; + + inner.pfs.mark_dirty(); + inner.pfs.save().map_err(to_js_error)?; + + // Collect dirty pages as owned Vec<(u64, Vec)> — no borrows + // escape — and drop the pages this commit freed (#440). + take_commit_flush(inner) +} + +// ── BrowserWriteTransaction ───────────────────────────────────────────────── + +/// A write transaction from `BrowserDb.beginWrite()` (#824). +/// +/// `execute` is synchronous: a `transact`/`retract` is staged and returns +/// `{"ok":true}`; a `query` reads the committed facts plus the staged ones +/// and returns `BrowserDb.execute`'s JSON; a `rule` is registered at once, +/// and stays after a rollback. A statement that fails stages nothing and +/// leaves the transaction open. +/// +/// `commit()` returns a `Promise`: the staged facts are applied as one +/// transaction (one `tx_count`) when it is called, and the promise resolves +/// once they are stored in IndexedDB. `rollback()`, or freeing the object +/// while open, discards them. After either, the transaction is finished: +/// `execute` and `commit` fail with `[API-018]`, and `rollback` does nothing. +#[wasm_bindgen] +pub struct BrowserWriteTransaction { + db: Rc>, + /// `None` once finished. + staged: Option, +} + +#[wasm_bindgen] +impl BrowserWriteTransaction { + /// Run one Datalog command inside the transaction; see the type's docs. + pub fn execute(&mut self, datalog: &str) -> Result { + let Some(staged) = self.staged.as_mut() else { + return Err(finished()); + }; + tx_execute(&self.db, staged, datalog) + .map(query_result_to_json) + .map_err(to_js_error) + } + + /// Apply the staged facts as one transaction and store it. The promise + /// resolves after the IndexedDB flush, and rejects as a failed flush of + /// `BrowserDb.execute` does. The transaction is finished either way: a + /// commit that is refused (for example `[API-019]`, a valid time that + /// ended before the commit time) discards the staged facts. + pub fn commit(&mut self) -> js_sys::Promise { + let Some(staged) = self.staged.take() else { + return js_sys::Promise::reject(&finished()); + }; + let committed = { + let mut inner = self.db.borrow_mut(); + inner.tx_open = false; + tx_commit(&mut inner, staged).map(|flush| { + ( + flush, + inner.idb.as_ref().map(IndexedDbBackend::clone_handle), + ) + }) + }; + match committed { + Err(e) => js_sys::Promise::reject(&e), + Ok((flush, idb)) => wasm_bindgen_futures::future_to_promise(async move { + if let (Some(flush), Some(idb)) = (flush, idb) { + idb.write_pages(flush.puts, flush.deletes).await?; + } + Ok(JsValue::UNDEFINED) + }), + } + } + + /// Discard the staged facts and finish the transaction. Does nothing on a + /// finished one. + pub fn rollback(&mut self) { + if self.staged.take().is_some() { + self.db.borrow_mut().tx_open = false; + } + } +} + +impl Drop for BrowserWriteTransaction { + fn drop(&mut self) { + if self.staged.take().is_some() + && let Ok(mut inner) = self.db.try_borrow_mut() + { + inner.tx_open = false; + } + } +} + +/// `[API-018]`: a call on a finished transaction. +fn finished() -> JsValue { + JsValue::from_str(&crate::MinigrafError::closed("write transaction").to_string()) +} + +/// A transaction's `execute`, as the native transaction's on an in-memory +/// database: stage a write, read the staged state, or register a rule. +fn tx_execute( + db: &RefCell, + staged: &mut crate::db::Staged, + datalog: &str, +) -> anyhow::Result { + use crate::error::{ErrorCode, bail_coded}; + use crate::query::datalog::parser::statement; + + if let Some(kind) = statement::kind_of(datalog) { + let chunk = statement::tx_chunk_facts(statement::DEFAULT_TX_CHUNK_FACTS); + let statement = statement::read_statement(datalog, kind, staged.next_tx_id(), chunk)?; + let facts = statement.facts()?; + crate::graph::storage::check_one_window_per_triple(&facts)?; + staged.stage(facts); + return Ok(QueryResult::Ok); + } + let cmd = parse_datalog_command(datalog)?; + let inner = db.borrow(); + let mut executor = DatalogExecutor::new_with_rules_and_functions( + inner.fact_storage.clone(), + inner.rules.clone(), + inner.functions.clone(), + ); + match cmd { + DatalogCommand::Transact(_) | DatalogCommand::Retract(_) => bail_coded!(ErrorCode::Api002), + DatalogCommand::Query(ref q) => { + crate::overrides::check_overrides(q, crate::Overridable::DEFAULTS)?; + crate::query::datalog::prepared::reject_unbound_slots(q)?; + let snapshot = inner.fact_storage.snapshot()?; + executor.set_snapshot(snapshot.with_staged(staged.read_view())); + executor.execute(cmd) + } + DatalogCommand::Rule(_) => executor.execute(cmd), + } +} + +/// A transaction's commit: `None` when nothing was staged. +fn tx_commit( + inner: &mut BrowserDbInner, + staged: crate::db::Staged, +) -> Result, JsValue> { + let facts = staged.pending_facts; + if facts.is_empty() { + return Ok(None); + } + let (facts, tx_id) = crate::db::Staged::checked_for_commit(facts).map_err(to_js_error)?; + let tx_count = inner.fact_storage.allocate_tx_count(); + let stamped = crate::db::Staged::stamp(facts, tx_id, tx_count); + commit_stamped(inner, stamped).map(Some) +} + // ── BrowserCursor ─────────────────────────────────────────────────────────── /// A cursor from `BrowserDb.query()`: the rows of one query answer, in batches. @@ -512,6 +727,7 @@ impl BrowserPreparedQuery { /// type, `[API-022]` for a missing slot and `[API-024]` for a bind value /// of the wrong kind. pub fn execute(&self, bindings: JsValue) -> Result { + refuse_in_tx(&self._db)?; let binds = binds::read_bindings(&bindings)?; self.query .execute(&as_slice(&binds)) @@ -522,6 +738,7 @@ impl BrowserPreparedQuery { /// Run the query with `bindings` and return a cursor over its rows. The /// answer is fixed when the cursor opens. Throws as `execute` does. pub fn query(&self, bindings: JsValue) -> Result { + refuse_in_tx(&self._db)?; let binds = binds::read_bindings(&bindings)?; let cursor = self .query @@ -1158,7 +1375,7 @@ mod tests { // ── #411 / #440: migrated pages are stored, free pages hold no bytes ── /// Open `name` with its object store emptied, so earlier runs leave nothing. - async fn empty_store(name: &str) -> IndexedDbBackend { + pub(super) async fn empty_store(name: &str) -> IndexedDbBackend { let idb = IndexedDbBackend::open(name).await.expect("idb open"); let keys: Vec = idb .load_all_pages() diff --git a/src/browser/tx_tests.rs b/src/browser/tx_tests.rs new file mode 100644 index 000000000..f145922ef --- /dev/null +++ b/src/browser/tx_tests.rs @@ -0,0 +1,467 @@ +//! `BrowserWriteTransaction` (#824): staging, read-your-own-writes, commit +//! and rollback against IndexedDB, the `[INT-001]` guard on the database +//! while a transaction is open, `[API-018]` after it finishes, and a +//! differential check against the native transaction. + +use super::*; +use wasm_bindgen_futures::JsFuture; +use wasm_bindgen_test::*; + +wasm_bindgen_test_configure!(run_in_browser); + +const NAMES: &str = "(query [:find ?n :where [?e :person/name ?n]])"; +const SETUP: &str = r#"(transact [[:alice :person/name "Alice"] [:alice :person/age 30] + [:bob :person/name "Bob"] [:carol :person/name "Carol"]])"#; + +/// A JSON result's rows, each encoded, sorted. +fn sorted_rows(json: &str) -> Vec { + let v: serde_json::Value = serde_json::from_str(json).unwrap(); + let mut rows: Vec = v["results"] + .as_array() + .unwrap() + .iter() + .map(ToString::to_string) + .collect(); + rows.sort(); + rows +} + +/// The `[CODE]` an error message starts with. +fn code_of(e: &JsValue) -> String { + let s = e.as_string().expect("an error message"); + s.split(']') + .next() + .unwrap_or("") + .trim_start_matches('[') + .to_string() +} + +async fn commit(tx: &mut BrowserWriteTransaction) -> Result { + JsFuture::from(tx.commit()).await +} + +async fn names(db: &BrowserDb) -> Vec { + sorted_rows(&db.execute(NAMES.to_string()).await.expect("query")) +} + +async fn setup() -> BrowserDb { + let db = BrowserDb::open_in_memory().expect("open_in_memory"); + db.execute(SETUP.to_string()).await.expect("setup"); + db +} + +// ─── B2 ────────────────────────────────────────────────────────────────────── + +#[wasm_bindgen_test] +async fn execute_stages_and_reads_own_writes() { + let db = setup().await; + let mut tx = db.begin_write().expect("begin"); + assert_eq!( + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .expect("transact"), + r#"{"ok":true}"# + ); + assert_eq!( + tx.execute(r#"(retract [[:bob :person/name "Bob"]])"#) + .expect("retract"), + r#"{"ok":true}"# + ); + let inside = tx.execute(NAMES).expect("query"); + let v: serde_json::Value = serde_json::from_str(&inside).unwrap(); + assert_eq!(v["variables"], serde_json::json!(["?n"]), "execute's JSON"); + assert_eq!( + sorted_rows(&inside), + vec![r#"["Alice"]"#, r#"["Carol"]"#, r#"["Dave"]"#] + ); + assert_eq!( + tx.execute("(rule [(named ?e ?n) [?e :person/name ?n]])") + .expect("rule"), + r#"{"ok":true}"# + ); + let ruled = tx + .execute("(query [:find ?n :where (named ?e ?n)])") + .expect("rule query"); + assert_eq!(sorted_rows(&ruled).len(), 3, "the rule reads staged facts"); + commit(&mut tx).await.expect("commit"); + assert_eq!( + names(&db).await, + vec![r#"["Alice"]"#, r#"["Carol"]"#, r#"["Dave"]"#] + ); +} + +#[wasm_bindgen_test] +async fn failed_statement_stages_nothing_and_stays_open() { + let db = setup().await; + let mut tx = db.begin_write().expect("begin"); + assert_eq!( + code_of(&tx.execute("(transact [[:a").unwrap_err()), + "PRS-004" + ); + let two_windows = r#"(transact [[:x :p/v 1 {:valid-from "2020-01-01T00:00:00Z"}] + [:x :p/v 1 {:valid-from "2021-01-01T00:00:00Z"}]])"#; + assert_eq!(code_of(&tx.execute(two_windows).unwrap_err()), "API-011"); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .expect("a good statement after the failures"); + commit(&mut tx).await.expect("commit"); + assert_eq!(names(&db).await.len(), 4, "only the good statement"); + let v = db + .execute("(query [:find ?e :where [?e :p/v _]])".to_string()) + .await + .expect("query"); + assert_eq!(sorted_rows(&v).len(), 0, "nothing of the failed one"); +} + +// ─── B3: differential check against the native transaction ────────────────── + +/// Sequences of statements, each run in one transaction and committed: the +/// browser database must end where an in-memory `Minigraf` ends. +const SEQUENCES: &[&[&str]] = &[ + &[], + &[NAMES], + &[ + r#"(transact [[:dave :person/name "Dave"] [:dave :person/age 40]])"#, + NAMES, + r#"(retract [[:bob :person/name "Bob"]])"#, + "(transact [[:alice :person/age 31]])", + "(retract [[:alice :person/age 30]])", + ], + &[ + // Latest statement wins for a triple written twice (#477). + r#"(transact [[:erin :person/name "Erin" {:valid-from "2020-01-01T00:00:00Z"}]])"#, + r#"(transact [[:erin :person/name "Erin" {:valid-from "2021-01-01T00:00:00Z" :valid-to "2030-01-01T00:00:00Z"}]])"#, + // API-011 and an inverted window (API-019): rejected, nothing staged. + r#"(transact [[:x :p/v 1 {:valid-from "2020-01-01T00:00:00Z"}] [:x :p/v 1 {:valid-from "2021-01-01T00:00:00Z"}]])"#, + r#"(transact [[:y :p/v 1 {:valid-from "2021-01-01T00:00:00Z" :valid-to "2020-01-01T00:00:00Z"}]])"#, + "(transact [[:x :p/v", + "(query [:find ?e :where [?e :person/name \"Erin\"]])", + ], + &[ + "(rule [(friendly ?e ?n) [?e :person/name ?n] [?e :person/friend _]])", + r#"(transact [[:frank :person/name "Frank"] [:frank :person/friend :alice] [:gina :person/name "Gina"]])"#, + "(query [:find ?n :where (friendly ?e ?n)])", + r#"(retract [[:carol :person/name "Carol"]])"#, + r#"(transact [[:carol :person/name "Carol"]])"#, + "(query [:find ?n ?a :where [?e :person/name ?n] [?e :person/age ?a]])", + ], +]; + +/// The checks run on both databases after each sequence. +const CHECKS: &[&str] = &[ + "(query [:find ?e ?a ?v :where [?e ?a ?v]])", + "(query [:find ?e ?a ?v ?t :any-valid-time :where [?e ?a ?v] [?e :db/tx-count ?t]])", + "(query [:find ?e ?v :valid-at \"2025-06-01T00:00:00Z\" :where [?e :person/name ?v]])", + "(query [:find ?e ?v :as-of 1 :where [?e :person/name ?v]])", +]; + +/// What a statement returned, comparable across the two databases. +fn outcome(result: Result) -> String { + match result { + Ok(json) if json.contains("\"results\"") => sorted_rows(&json).join(";"), + Ok(json) => json, + Err(code) => code, + } +} + +#[wasm_bindgen_test] +async fn commit_matches_native_transaction() { + for (i, stmts) in SEQUENCES.iter().enumerate() { + let browser = setup().await; + let native = crate::Minigraf::in_memory().unwrap(); + native.execute(SETUP).unwrap(); + + let mut btx = browser.begin_write().expect("begin"); + let mut ntx = native.begin_write().unwrap(); + for stmt in *stmts { + let b = outcome(btx.execute(stmt).map_err(|e| code_of(&e))); + let n = outcome( + ntx.execute(stmt) + .map(query_result_to_json) + .map_err(|e| e.code().to_string()), + ); + assert_eq!(b, n, "sequence {i}: statement outcomes differ"); + } + commit(&mut btx).await.expect("browser commit"); + ntx.commit().unwrap(); + + for check in CHECKS { + let b = browser.execute((*check).to_string()).await.expect("check"); + let n = query_result_to_json(native.execute(check).unwrap()); + assert_eq!( + sorted_rows(&b), + sorted_rows(&n), + "sequence {i}: results differ" + ); + } + let bcount = browser.inner.borrow().fact_storage.current_tx_count(); + assert_eq!(bcount, native.current_tx_count(), "sequence {i}: tx_count"); + } +} + +// ─── B4, B5: commit, rollback and drop against IndexedDB ──────────────────── + +async fn fresh_db(name: &str) -> BrowserDb { + drop(super::tests::empty_store(name).await); + BrowserDb::open(name).await.expect("open") +} + +#[wasm_bindgen_test] +async fn commit_is_one_transaction_and_survives_reopen() { + let name = "minigraf-test-tx-commit"; + let db = fresh_db(name).await; + db.execute(SETUP.to_string()).await.expect("setup"); + let before = db.inner.borrow().fact_storage.current_tx_count(); + let mut tx = db.begin_write().expect("begin"); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + tx.execute(r#"(transact [[:erin :person/name "Erin"]])"#) + .unwrap(); + tx.execute(r#"(retract [[:bob :person/name "Bob"]])"#) + .unwrap(); + let pending = tx.commit(); + // Finished and visible before the flush resolves. + assert_eq!( + names(&db).await, + vec![r#"["Alice"]"#, r#"["Carol"]"#, r#"["Dave"]"#, r#"["Erin"]"#] + ); + JsFuture::from(pending).await.expect("flush"); + assert_eq!( + db.inner.borrow().fact_storage.current_tx_count(), + before + 1, + "one tx_count for the whole transaction" + ); + let counts = db + .execute( + "(query [:find ?t :any-valid-time :where [?e :person/name _] [?e :db/tx-count ?t]])" + .to_string(), + ) + .await + .unwrap(); + assert_eq!( + sorted_rows(&counts), + vec!["[1]", "[1]", "[2]", "[2]"], + "Alice and Carol from the setup, Dave and Erin from the one commit" + ); + drop(db); + let db = BrowserDb::open(name).await.expect("reopen"); + assert_eq!(names(&db).await.len(), 4, "the commit after reopening"); +} + +#[wasm_bindgen_test] +async fn rollback_and_drop_discard_and_survive_reopen() { + let name = "minigraf-test-tx-rollback"; + let db = fresh_db(name).await; + db.execute(SETUP.to_string()).await.expect("setup"); + + let mut tx = db.begin_write().expect("begin"); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + tx.rollback(); + assert_eq!(names(&db).await.len(), 3, "rolled back"); + + let mut tx = db.begin_write().expect("begin after rollback"); + tx.execute(r#"(transact [[:erin :person/name "Erin"]])"#) + .unwrap(); + drop(tx); // as JavaScript's free() does + assert_eq!(names(&db).await.len(), 3, "dropped"); + + drop(db); + let db = BrowserDb::open(name).await.expect("reopen"); + assert_eq!(names(&db).await.len(), 3, "neither after reopening"); +} + +/// A failed IndexedDB flush rejects the commit's promise and leaves what a +/// failed flush of `execute` leaves: the facts in memory, not in the store. +#[wasm_bindgen_test] +async fn failed_flush_rejects_like_execute() { + let name = "minigraf-test-tx-flush"; + let db = fresh_db(name).await; + db.execute(SETUP.to_string()).await.expect("setup"); + db.inner.borrow().idb.as_ref().unwrap().db.close(); + + let failed_execute = db + .execute(r#"(transact [[:dave :person/name "Dave"]])"#.to_string()) + .await; + assert!(failed_execute.is_err(), "execute's flush fails"); + let mut tx = db.begin_write().expect("begin"); + tx.execute(r#"(transact [[:erin :person/name "Erin"]])"#) + .unwrap(); + assert!(commit(&mut tx).await.is_err(), "the commit's flush fails"); + assert_eq!(names(&db).await.len(), 5, "both in memory, as before"); + assert!( + tx.execute(NAMES).is_err(), + "the transaction is finished after a failed flush" + ); + drop(db); + let db = BrowserDb::open(name).await.expect("reopen"); + assert_eq!(names(&db).await.len(), 3, "neither reached the store"); +} + +#[wasm_bindgen_test] +async fn refused_commit_finishes_and_discards() { + let db = setup().await; + let mut tx = db.begin_write().expect("begin"); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + let soon = crate::graph::types::tx_id_now().saturating_add(20); + for f in &mut tx.staged.as_mut().unwrap().pending_facts { + f.valid_to = soon.cast_signed(); + } + while crate::graph::types::tx_id_now() <= soon {} + let err = commit(&mut tx) + .await + .expect_err("window ends before commit"); + assert_eq!(code_of(&err), "API-019"); + assert_eq!(names(&db).await.len(), 3, "nothing committed"); + assert_eq!(code_of(&tx.execute(NAMES).unwrap_err()), "API-018"); +} + +#[wasm_bindgen_test] +async fn empty_and_query_only_transactions() { + let db = setup().await; + let before = db.inner.borrow().fact_storage.current_tx_count(); + let mut tx = db.begin_write().expect("begin"); + commit(&mut tx).await.expect("empty commit"); + let mut tx = db.begin_write().expect("begin"); + tx.execute(NAMES).unwrap(); + commit(&mut tx).await.expect("query-only commit"); + let mut tx = db.begin_write().expect("begin"); + tx.execute(NAMES).unwrap(); + tx.rollback(); + assert_eq!(db.inner.borrow().fact_storage.current_tx_count(), before); +} + +// ─── B6, B7: the guard and finished transactions ──────────────────────────── + +/// Every guarded `BrowserDb` call's error code, or `None` on success. +/// +/// `importGraph` is called only while a transaction is open: it is refused +/// before it reads the blob, and would otherwise replace the database. +async fn guarded( + db: &BrowserDb, + pq: &BrowserPreparedQuery, + open: bool, +) -> Vec<(&'static str, Option)> { + let code = |r: Result<(), JsValue>| r.err().map(|e| code_of(&e)); + let empty = js_sys::Object::new().into(); + let import = if open { + code( + db.import_graph(js_sys::Uint8Array::new_with_length(0)) + .await, + ) + } else { + None + }; + vec![ + ( + "execute", + code(db.execute(NAMES.to_string()).await.map(|_| ())), + ), + ("query", code(db.query(NAMES).map(|_| ()))), + ("prepare", code(db.prepare(NAMES).map(|_| ()))), + ("checkpoint", code(db.checkpoint().await)), + ("importGraph", import), + ("beginWrite", code(db.begin_write().map(|_| ()))), + ("prepared execute", code(pq.execute(empty).map(|_| ()))), + ( + "prepared query", + code(pq.query(js_sys::Object::new().into()).map(|_| ())), + ), + ] +} + +#[wasm_bindgen_test] +async fn database_calls_are_int001_while_open() { + let db = setup().await; + let pq = db.prepare(NAMES).expect("prepare"); + let mut cursor = db.query(NAMES).expect("cursor before begin"); + let mut tx = db.begin_write().expect("begin"); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + for (name, code) in guarded(&db, &pq, true).await { + assert_eq!(code.as_deref(), Some("INT-001"), "{name} while open"); + } + let rows = cursor.next_batch(10).expect("cursor").expect("rows"); + assert_eq!(sorted_rows(&format!(r#"{{"results":{rows}}}"#)).len(), 3); + commit(&mut tx).await.expect("commit"); + + // After the commit everything works. + for (name, code) in guarded(&db, &pq, false).await { + assert_eq!(code, None, "{name} after the commit"); + } + // After a rollback and a drop too. + db.begin_write().expect("begin").rollback(); + drop(db.begin_write().expect("begin")); + assert_eq!(names(&db).await.len(), 4); +} + +#[wasm_bindgen_test] +async fn finished_transaction_is_api018() { + let db = setup().await; + let mut tx = db.begin_write().expect("begin"); + commit(&mut tx).await.expect("commit"); + assert_eq!(code_of(&tx.execute(NAMES).unwrap_err()), "API-018"); + assert_eq!(code_of(&commit(&mut tx).await.unwrap_err()), "API-018"); + tx.rollback(); // does nothing + + let mut tx = db.begin_write().expect("begin"); + tx.rollback(); + tx.rollback(); + assert_eq!(code_of(&tx.execute(NAMES).unwrap_err()), "API-018"); + assert_eq!(code_of(&commit(&mut tx).await.unwrap_err()), "API-018"); + // A finished transaction no longer guards the database. + assert_eq!(names(&db).await.len(), 3); +} + +#[wasm_bindgen_test] +async fn rules_stay_after_rollback() { + let db = setup().await; + let mut tx = db.begin_write().expect("begin"); + tx.execute("(rule [(named ?e ?n) [?e :person/name ?n]])") + .unwrap(); + tx.rollback(); + let r = db + .execute("(query [:find ?n :where (named ?e ?n)])".to_string()) + .await + .expect("the rule is still registered"); + assert_eq!(sorted_rows(&r).len(), 3); +} + +/// A user function of a transaction's query that calls the database (here +/// `beginWrite` and `query`) is refused as `[INT-001]` is, so the query fails with +/// `[QRY-014]`; nothing panics and the transaction stays usable. +#[wasm_bindgen_test] +async fn user_function_calling_the_database_inside_a_transaction() { + let db = setup().await; + let calls = [ + ("begins?", "db.beginWrite(); return true"), + ( + "queries?", + "db.query('(query [:find ?n :where [?e :person/name ?n]])'); return true", + ), + ]; + for (name, body) in calls { + let other = BrowserDb { + inner: db.inner.clone(), + }; + let f = js_sys::Function::new_with_args("db", &format!("return (v) => {{ {body} }}")); + let predicate = f + .call1(&JsValue::NULL, &JsValue::from(other)) + .expect("make the predicate"); + db.register_predicate(name, predicate).expect("register"); + } + let mut tx = db.begin_write().expect("begin"); + for (name, _) in calls { + let q = format!("(query [:find ?n :where [?e :person/name ?n] [({name} ?n)]])"); + let e = tx.execute(&q).expect_err("a call back into the database"); + let msg = e.as_string().expect("a message"); + assert!( + msg.starts_with("[QRY-014]") + && msg.contains("write transaction is already in progress"), + "{name}: {msg}" + ); + } + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .expect("still usable"); + commit(&mut tx).await.expect("commit"); + assert_eq!(names(&db).await.len(), 4); +} diff --git a/src/db.rs b/src/db.rs index 3dc8f3bbf..ebc6a1d86 100644 --- a/src/db.rs +++ b/src/db.rs @@ -59,6 +59,48 @@ pub(crate) fn is_write_tx_active() -> bool { WRITE_TX_ACTIVE.with(|f| f.get()) } +/// The begin thread of a database's open [`OwnedWriteTransaction`] (#824). +/// +/// The borrowed transaction's thread-local flag cannot follow a transaction +/// that moves between threads, so an owned one records the thread that +/// began it here, per database, until it finishes. The calls a borrowed +/// transaction refuses on its thread (`INT-001`) are refused on this one. +/// The mutex is only ever held to read or set the id. +#[derive(Default)] +pub(crate) struct TxOwner { + thread: Mutex>, +} + +impl TxOwner { + /// Whether the current thread began this database's open owned + /// transaction. + fn held_here(&self) -> bool { + let current = std::thread::current().id(); + *self + .thread + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + == Some(current) + } + + fn set(&self, thread: Option) { + *self + .thread + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = thread; + } + + /// Fail with `INT-001` on a thread with an open transaction on this + /// database: a borrowed one (any database, as before #824) or an owned + /// one it began. + pub(crate) fn refuse_reentry(&self) -> Result<()> { + if is_write_tx_active() || self.held_here() { + bail_coded!(ErrorCode::Int001); + } + Ok(()) + } +} + // ─── Page cache sizing ───────────────────────────────────────────────────────── /// Page cache capacity to use for an in-memory (`MemoryBackend`) database. @@ -541,6 +583,27 @@ enum WriteContext { }, } +/// The write lock's state: the context, and whether an owned transaction +/// holds the lock between its calls (#824). While `owned` is set, every +/// writer other than that transaction waits on `Inner::write_free`. +struct WriteSlot { + ctx: WriteContext, + owned: bool, +} + +impl std::ops::Deref for WriteSlot { + type Target = WriteContext; + fn deref(&self) -> &WriteContext { + &self.ctx + } +} + +impl std::ops::DerefMut for WriteSlot { + fn deref_mut(&mut self) -> &mut WriteContext { + &mut self.ctx + } +} + // ─── IntegrityReport ────────────────────────────────────────────────────────── /// The result of [`Minigraf::verify`]. @@ -573,8 +636,13 @@ struct Inner { /// each query reads a snapshot of it (#180). functions: Arc, /// Serialises all writes. Holds `WriteContext` which contains the PFS/WAL - /// for file-backed databases. - write_lock: Mutex, + /// for file-backed databases. Take it with [`Inner::lock_write`], which + /// waits while an owned transaction holds it. + write_lock: Mutex, + /// Signalled when an owned transaction releases the write lock. + write_free: std::sync::Condvar, + /// The begin thread of the open owned transaction, if any. + owner: Arc, /// Configuration options. options: OpenOptions, } @@ -592,7 +660,11 @@ impl Drop for Inner { } if self.options.wal_checkpoint_threshold == usize::MAX { #[cfg(not(target_arch = "wasm32"))] - if let Ok(WriteContext::File { pfs, .. }) = self.write_lock.get_mut() { + if let Ok(WriteSlot { + ctx: WriteContext::File { pfs, .. }, + .. + }) = self.write_lock.get_mut() + { pfs.discard(); } return; @@ -603,6 +675,32 @@ impl Drop for Inner { } } +impl Inner { + /// Take the write lock for a write, a checkpoint or a begin: wait first + /// while an owned transaction holds it (#824). `API-001` if poisoned. + fn lock_write(&self) -> Result> { + let mut slot = self + .write_lock + .lock() + .map_err(|_| err_coded!(ErrorCode::Api001))?; + while slot.owned { + slot = self + .write_free + .wait(slot) + .map_err(|_| err_coded!(ErrorCode::Api001))?; + } + Ok(slot) + } + + /// End an owned transaction's hold on the write lock and wake the + /// writers waiting for it. + fn release_owned(&self, slot: &mut WriteSlot) { + slot.owned = false; + self.owner.set(None); + self.write_free.notify_all(); + } +} + /// Refuse a write, before it takes a `tx_count`, on a handle where an earlier /// write failed: a checkpoint (STG-046) or a WAL append (WAL-007). See #390 /// and #513. @@ -776,7 +874,9 @@ impl Minigraf { fact_storage, rules: Arc::new(RwLock::new(RuleRegistry::new())), functions: Arc::new(Functions::with_builtins()), - write_lock: Mutex::new(ctx), + write_lock: Mutex::new(WriteSlot { ctx, owned: false }), + write_free: std::sync::Condvar::new(), + owner: Arc::default(), options, }), } @@ -893,9 +993,7 @@ impl Minigraf { fn execute_inner(&self, input: &str) -> Result { // Detect same-thread reentrant write (would deadlock on the Mutex). - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } + self.inner.owner.refuse_reentry()?; if let Some(kind) = statement::kind_of(input) { return self.execute_statement(input, kind); @@ -912,11 +1010,7 @@ impl Minigraf { DatalogCommand::Rule(_) => { // Rule registration is a write: it mutates the shared // RuleRegistry, so it runs under write_lock (no WAL needed). - let _ctx = self - .inner - .write_lock - .lock() - .map_err(|_| err_coded!(ErrorCode::Api001))?; + let _ctx = self.inner.lock_write()?; let executor = DatalogExecutor::new_with_rules_and_functions( self.inner.fact_storage.clone(), self.inner.rules.clone(), @@ -946,14 +1040,10 @@ impl Minigraf { )?; self.reject_if_read_only(kind.name())?; { - let mut ctx = self - .inner - .write_lock - .lock() - .map_err(|_| err_coded!(ErrorCode::Api001))?; + let mut ctx = self.inner.lock_write()?; #[cfg(not(target_arch = "wasm32"))] - if statement.is_large() && matches!(*ctx, WriteContext::File { .. }) { + if statement.is_large() && matches!(ctx.ctx, WriteContext::File { .. }) { return self.execute_large(&mut ctx, statement, kind); } @@ -1112,9 +1202,7 @@ impl Minigraf { } fn query_inner(&self, input: &str) -> Result { - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } + self.inner.owner.refuse_reentry()?; let cmd = parse_datalog_command(input)?; match &cmd { DatalogCommand::Query(q) => { @@ -1159,9 +1247,7 @@ impl Minigraf { } fn fact_log_inner(&self, filter: &crate::FactFilter) -> Result { - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } + self.inner.owner.refuse_reentry()?; // No write lock: the snapshot holds whole transactions, and its pin // keeps checkpoints from reusing the pages it reads (#473 spec §4). crate::FactLog::open(&self.inner.fact_storage, filter) @@ -1222,32 +1308,62 @@ impl Minigraf { } fn begin_write_inner(&self, options: TxOptions) -> Result> { - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } + self.inner.owner.refuse_reentry()?; self.reject_if_read_only("begin_write")?; - let guard = self - .inner - .write_lock - .lock() - .map_err(|_| err_coded!(ErrorCode::Api001))?; + let guard = self.inner.lock_write()?; // Set flag only after successfully acquiring the lock set_write_tx_active(true); Ok(WriteTransaction { + state: TxState::new(&self.inner, options), 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(), committed: false, - #[cfg(not(target_arch = "wasm32"))] - large: None, + }) + } + + /// Begin an [`OwnedWriteTransaction`] with the database's options: + /// [`begin_write_owned_with`](Self::begin_write_owned_with) with + /// [`TxOptions::default`]. + /// + /// # Errors + /// + /// As [`begin_write_owned_with`](Self::begin_write_owned_with). + pub fn begin_write_owned(&self) -> Result { + self.begin_write_owned_with(TxOptions::default()) + } + + /// Begin an [`OwnedWriteTransaction`] with `options`, which apply as they + /// do to [`begin_write_with`](Self::begin_write_with) (#824). + /// + /// Takes the write lock, and holds it until the transaction commits, + /// rolls back or is dropped, on whichever thread that happens. While + /// another transaction (of either kind) holds it, this waits. + /// + /// # Errors + /// + /// `INT-001` on a thread with an open `WriteTransaction`, or with an open + /// owned transaction of this database that it began; `API-014` on a + /// read-only handle. A failed call leaves the handle usable. + pub fn begin_write_owned_with( + &self, + options: TxOptions, + ) -> Result { + self.begin_write_owned_inner(options) + .map_err(MinigrafError::from) + } + + fn begin_write_owned_inner(&self, options: TxOptions) -> Result { + self.inner.owner.refuse_reentry()?; + self.reject_if_read_only("begin_write")?; + let mut slot = self.inner.lock_write()?; + slot.owned = true; + self.inner.owner.set(Some(std::thread::current().id())); + let state = TxState::new(&self.inner, options); + drop(slot); + Ok(OwnedWriteTransaction { + db: self.clone(), + state, + finished: false, }) } @@ -1286,15 +1402,9 @@ impl Minigraf { } fn checkpoint_inner(&self) -> Result<()> { - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } + self.inner.owner.refuse_reentry()?; self.reject_if_read_only("checkpoint")?; - let mut ctx = self - .inner - .write_lock - .lock() - .map_err(|_| err_coded!(ErrorCode::Api001))?; + let mut ctx = self.inner.lock_write()?; Self::do_checkpoint(&self.inner.fact_storage, &mut ctx) } @@ -1325,15 +1435,9 @@ impl Minigraf { } fn verify_inner(&self) -> Result { - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } - let ctx = self - .inner - .write_lock - .lock() - .map_err(|_| err_coded!(ErrorCode::Api001))?; - match &*ctx { + self.inner.owner.refuse_reentry()?; + let ctx = self.inner.lock_write()?; + match &ctx.ctx { WriteContext::Memory => Ok(IntegrityReport::default()), #[cfg(not(target_arch = "wasm32"))] WriteContext::File { pfs, .. } | WriteContext::ReadOnly { pfs } => { @@ -1373,16 +1477,10 @@ impl Minigraf { } fn rebuild_indexes_inner(&self) -> Result<()> { - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } + self.inner.owner.refuse_reentry()?; self.reject_if_read_only("rebuild_indexes")?; - let mut ctx = self - .inner - .write_lock - .lock() - .map_err(|_| err_coded!(ErrorCode::Api001))?; - match &mut *ctx { + let mut ctx = self.inner.lock_write()?; + match &mut ctx.ctx { WriteContext::Memory => Ok(()), #[cfg(not(target_arch = "wasm32"))] WriteContext::File { @@ -1421,7 +1519,7 @@ impl Minigraf { /// check when a checkpoint commits. #[cfg(all(test, not(target_arch = "wasm32")))] pub(crate) fn generation(&self) -> u64 { - match &*self.inner.write_lock.lock().unwrap() { + match &self.inner.write_lock.lock().unwrap().ctx { WriteContext::File { pfs, .. } | WriteContext::ReadOnly { pfs } => pfs.generation(), WriteContext::Memory => 0, } @@ -1433,7 +1531,7 @@ impl Minigraf { pub(crate) fn kill(self) { let mut ctx = self.inner.write_lock.lock().unwrap(); if let WriteContext::File { mut pfs, .. } = - std::mem::replace(&mut *ctx, WriteContext::Memory) + std::mem::replace(&mut ctx.ctx, WriteContext::Memory) { pfs.discard(); } @@ -1446,7 +1544,7 @@ impl Minigraf { 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) + std::mem::replace(&mut ctx.ctx, WriteContext::Memory) { if let Some(mut wal) = wal { wal.lose_unsynced().unwrap(); @@ -1459,7 +1557,7 @@ impl Minigraf { /// 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() { + match &self.inner.write_lock.lock().unwrap().ctx { WriteContext::File { wal, .. } => wal.as_ref().map(WalWriter::is_synced), _ => None, } @@ -1553,9 +1651,7 @@ impl Minigraf { ) -> Result { use crate::query::datalog::prepared::prepare_query; - if is_write_tx_active() { - bail_coded!(ErrorCode::Int001); - } + self.inner.owner.refuse_reentry()?; let cmd = parse_datalog_command(query_str)?; @@ -1579,6 +1675,7 @@ impl Minigraf { self.inner.rules.clone(), self.inner.functions.clone(), self.inner.options.limits(), + self.inner.owner.clone(), ) } @@ -1894,23 +1991,29 @@ impl Minigraf { /// ``` /// /// Dropping without committing performs an implicit rollback. +/// +/// It borrows the database and stays on the thread that began it. To hold a +/// transaction in an object that outlives the borrow or moves between +/// threads, use [`OwnedWriteTransaction`]. pub struct WriteTransaction<'a> { - guard: MutexGuard<'a, WriteContext>, + guard: MutexGuard<'a, WriteSlot>, inner: &'a Inner, + state: TxState, + /// Set to `true` after a successful `commit()` to suppress rollback in `Drop`. + committed: bool, +} + +/// What a transaction has staged, shared by [`WriteTransaction`] and +/// [`OwnedWriteTransaction`]. Every method runs under the write lock. +struct TxState { /// 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. - next_pending_tx_count: u64, - /// Synthetic tx_id seed used to keep staged writes monotonic. - next_pending_tx_id: TxId, - /// Set to `true` after a successful `commit()` to suppress rollback in `Drop`. - committed: bool, + /// What the transaction staged while small. + staged: Staged, /// The working state once the transaction is large (#550 spec §4): its /// statements are then in copy-on-write working trees, not staged. #[cfg(not(target_arch = "wasm32"))] @@ -1926,23 +2029,25 @@ enum Current<'s> { Large(statement::Statement<'s>), } -impl<'a> WriteTransaction<'a> { - /// Execute a Datalog command within this transaction. - /// - /// - **Writes** (transact / retract): buffered in-memory; not durable until `commit()`. - /// Returns `Ok(QueryResult::Ok)` immediately (not `Transacted`/`Retracted`). - /// - **Reads** (query): see committed facts in `FactStorage` **plus** all facts - /// buffered in this transaction (read-your-own-writes). - /// - **Rules**: registered immediately into the shared rule registry. - /// - /// # Errors - /// - /// Returns an error if parsing or execution fails. - pub fn execute(&mut self, input: &str) -> Result { - self.execute_inner(input).map_err(MinigrafError::from) +impl TxState { + fn new(inner: &Inner, options: TxOptions) -> Self { + TxState { + sync: options.synchronous.unwrap_or(inner.options.synchronous), + chunk_facts: options + .chunk_facts + .unwrap_or(inner.options.transaction_chunk_facts), + staged: Staged::new(inner.fact_storage.current_tx_count()), + #[cfg(not(target_arch = "wasm32"))] + large: None, + } } - fn execute_inner(&mut self, input: &str) -> Result { + fn execute( + &mut self, + ctx: &mut WriteContext, + inner: &Inner, + input: &str, + ) -> Result { if let Some(kind) = statement::kind_of(input) { #[cfg(not(target_arch = "wasm32"))] if let Some(tx) = &self.large { @@ -1950,32 +2055,31 @@ impl<'a> WriteTransaction<'a> { 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)?; - self.write_statement(chunks)?; + refuse_if_failed(ctx)?; + self.write_statement(ctx, chunks)?; return Ok(QueryResult::Ok); } // A rejected statement stages nothing; the transaction stays // usable. Pass 1 checks windows at the staging time; `commit` // 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 staged_tx_id = self.staged.next_tx_id(); 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 { .. }); + let file = matches!(*ctx, WriteContext::File { .. }); #[cfg(not(target_arch = "wasm32"))] if file && statement.is_large() { - self.become_large(Current::Large(statement))?; + self.become_large(ctx, Current::Large(statement))?; return Ok(QueryResult::Ok); } let facts = statement.facts()?; crate::graph::storage::check_one_window_per_triple(&facts)?; #[cfg(not(target_arch = "wasm32"))] - if file && self.pending_facts.len().saturating_add(facts.len()) > chunk { - self.become_large(Current::Small(facts))?; + if file && self.staged.pending_facts.len().saturating_add(facts.len()) > chunk { + self.become_large(ctx, Current::Small(facts))?; return Ok(QueryResult::Ok); } - self.stage_pending_facts(facts); + self.staged.stage(facts); return Ok(QueryResult::Ok); } let cmd = parse_datalog_command(input)?; @@ -1985,49 +2089,42 @@ impl<'a> WriteTransaction<'a> { Err(err_coded!(ErrorCode::Api002)) } DatalogCommand::Query(ref q) => { - check_overrides(q, &self.inner.options.allow_overrides)?; + check_overrides(q, &inner.options.allow_overrides)?; reject_unbound_slots(q)?; - self.execute_read_command(cmd) + let mut executor = DatalogExecutor::new_with_rules_and_functions( + inner.fact_storage.clone(), + inner.rules.clone(), + inner.functions.clone(), + ); + executor.set_limits(inner.options.limits()); + executor.set_snapshot(self.read_snapshot(ctx, inner)?); + executor.execute(cmd) + } + DatalogCommand::Rule(rule) => { + let mut executor = DatalogExecutor::new_with_rules_and_functions( + inner.fact_storage.clone(), + inner.rules.clone(), + inner.functions.clone(), + ); + executor.set_limits(inner.options.limits()); + executor.execute(DatalogCommand::Rule(rule)) } - DatalogCommand::Rule(rule) => self.execute_rule_command(rule), } } - fn execute_read_command(&self, cmd: DatalogCommand) -> Result { - let mut executor = DatalogExecutor::new_with_rules_and_functions( - self.inner.fact_storage.clone(), - self.inner.rules.clone(), - self.inner.functions.clone(), - ); - executor.set_limits(self.inner.options.limits()); - executor.set_snapshot(self.read_snapshot()?); - executor.execute(cmd) - } - /// The state a query inside the transaction reads (#550 spec §7): for a /// large transaction its working trees, which hold every committed and /// pending fact and its finished statements; otherwise the database with /// the staged facts after its pending ones. - fn read_snapshot(&self) -> Result { - let snapshot = self.inner.fact_storage.snapshot()?; + fn read_snapshot(&self, ctx: &WriteContext, inner: &Inner) -> Result { + let snapshot = inner.fact_storage.snapshot()?; #[cfg(not(target_arch = "wasm32"))] - if let (Some(tx), WriteContext::File { pfs, .. }) = (&self.large, &*self.guard) { + if let (Some(tx), WriteContext::File { pfs, .. }) = (&self.large, ctx) { return Ok(snapshot.with_committed(pfs.large_reader(tx))); } - // A default `valid_from` stays the sentinel until the commit stamps - // it; reads see it as the fact's own staged tx_id. - let staged = self - .pending_facts - .iter() - .cloned() - .map(|mut f| { - if f.asserted && f.valid_from == VALID_FROM_USE_TX_TIME { - f.valid_from = f.tx_id.cast_signed(); - } - f - }) - .collect(); - Ok(snapshot.with_staged(staged)) + #[cfg(target_arch = "wasm32")] + let _ = ctx; + Ok(snapshot.with_staged(self.staged.read_view())) } /// Make the transaction large with `current`, the statement that takes @@ -2035,11 +2132,12 @@ impl<'a> WriteTransaction<'a> { /// 26). The staged statements are checked at the transaction's `tx_id` /// first: a failure rejects `current` and leaves the transaction small. #[cfg(not(target_arch = "wasm32"))] - fn become_large(&mut self, current: Current<'_>) -> Result<()> { + fn become_large(&mut self, ctx: &mut WriteContext, current: Current<'_>) -> Result<()> { let tx_id = crate::graph::types::tx_id_now(); - let keep = last_statement_mask(&self.pending_facts); + let keep = last_statement_mask(&self.staged.pending_facts); let kept = || { - self.pending_facts + self.staged + .pending_facts .iter() .zip(&keep) .filter_map(|(f, k)| k.then_some(f)) @@ -2050,10 +2148,10 @@ impl<'a> WriteTransaction<'a> { for f in kept() { crate::wal::check_fact_size(f)?; } - refuse_if_failed(&self.guard)?; + refuse_if_failed(ctx)?; let WriteContext::File { pfs, wal, db_path, .. - } = &mut *self.guard + } = &mut *ctx else { bail_coded!(ErrorCode::Int019, "a large transaction without a file"); }; @@ -2061,7 +2159,7 @@ impl<'a> WriteTransaction<'a> { // 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) + let mut staged: Vec = std::mem::take(&mut self.staged.pending_facts) .into_iter() .zip(keep) .filter_map(|(f, k)| k.then_some(f)) @@ -2077,8 +2175,8 @@ impl<'a> WriteTransaction<'a> { self.large = Some(tx); written?; match current { - Current::Small(facts) => self.write_statement(std::iter::once(Ok(facts))), - Current::Large(statement) => self.write_statement(statement.chunks()?), + Current::Small(facts) => self.write_statement(ctx, std::iter::once(Ok(facts))), + Current::Large(statement) => self.write_statement(ctx, statement.chunks()?), } } @@ -2086,13 +2184,16 @@ impl<'a> WriteTransaction<'a> { /// a statement base (#550 spec §5): a rejected statement returns the /// working trees to it and leaves the transaction usable. #[cfg(not(target_arch = "wasm32"))] - fn write_statement(&mut self, chunks: impl IntoIterator>) -> Result<()> + fn write_statement( + &mut self, + ctx: &mut WriteContext, + chunks: impl IntoIterator>, + ) -> Result<()> where C: std::ops::DerefMut, { use crate::storage::persistent_facts::ChunkError; - let (WriteContext::File { pfs, .. }, Some(tx)) = (&mut *self.guard, self.large.as_mut()) - else { + let (WriteContext::File { pfs, .. }, Some(tx)) = (ctx, self.large.as_mut()) else { bail_coded!( ErrorCode::Int019, "a large statement without a large transaction" @@ -2114,71 +2215,16 @@ impl<'a> WriteTransaction<'a> { pfs.end_statement(tx, true) } - /// Discard the working state of a large transaction (#550 spec §4.4). - #[cfg(not(target_arch = "wasm32"))] - fn rollback_large(&mut self) { - if let (Some(tx), WriteContext::File { pfs, .. }) = (self.large.take(), &mut *self.guard) { - pfs.rollback_large(tx); - } - } - - fn execute_rule_command( - &self, - rule: crate::query::datalog::types::Rule, - ) -> Result { - let mut executor = DatalogExecutor::new_with_rules_and_functions( - self.inner.fact_storage.clone(), - self.inner.rules.clone(), - self.inner.functions.clone(), - ); - executor.set_limits(self.inner.options.limits()); - 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); - let staged_tx_count = self.next_pending_tx_count; - - self.pending_facts.extend(facts.into_iter().map(|mut fact| { - fact.tx_id = staged_tx_id; - fact.tx_count = staged_tx_count; - fact - })); - - self.next_pending_tx_id = staged_tx_id.saturating_add(1); - self.next_pending_tx_count = self.next_pending_tx_count.saturating_add(1); - } - - /// Commit this transaction atomically. - /// - /// All buffered facts are stamped with a single `tx_count` / `tx_id`, then - /// 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 - /// - /// Returns an error if the WAL write or fact application fails. - pub fn commit(self) -> Result<(), MinigrafError> { - self.commit_inner().map_err(MinigrafError::from) - } - - fn commit_inner(mut self) -> Result<()> { + /// Commit what is staged. On an error the caller rolls back what is + /// left (a large transaction's working state). + fn commit(&mut self, ctx: &mut WriteContext, inner: &Inner) -> Result<()> { #[cfg(not(target_arch = "wasm32"))] if self.large.is_some() { - // A failed handle is refused; the drop rolls the transaction back. - refuse_if_failed(&self.guard)?; + // A failed handle is refused; the caller rolls the transaction back. + refuse_if_failed(ctx)?; // No statement wrote a fact: nothing to commit, no tx_count taken. if !self.large.as_ref().is_some_and(|tx| tx.wrote()) { - self.rollback_large(); + self.rollback(ctx); } if let ( Some(tx), @@ -2188,52 +2234,28 @@ impl<'a> WriteTransaction<'a> { db_path, wal_entry_count, }, - ) = (self.large.take(), &mut *self.guard) + ) = (self.large.take(), ctx) { pfs.commit_large(tx)?; Minigraf::delete_wal(wal, db_path, wal_entry_count)?; } - self.committed = true; - set_write_tx_active(false); return Ok(()); } - let facts_to_commit = std::mem::take(&mut self.pending_facts); + let facts_to_commit = std::mem::take(&mut self.staged.pending_facts); if !facts_to_commit.is_empty() { - // One tx_id cannot tell several versions of a fact apart, so the - // last statement that wrote a triple decides it, as reads inside - // the transaction already see (#477). Each statement passed - // API-011 when staged, so every triple left has one window. - let facts_to_commit = keep_last_statement(facts_to_commit); - // Before allocating: a rejected transaction takes no tx_count. - // A default `valid_from` becomes the commit time, so the windows - // are checked again here (#436). - let tx_id = crate::graph::types::tx_id_now(); - crate::graph::storage::check_valid_windows(&facts_to_commit, tx_id)?; - refuse_if_failed(&self.guard)?; - let tx_count = self.inner.fact_storage.allocate_tx_count(); - - // Stamp facts with tx_id and tx_count - let stamped: Vec = facts_to_commit - .into_iter() - .map(|mut f| { - f.tx_id = tx_id; - f.tx_count = tx_count; - // Fix valid_from if it was left as the sentinel (placeholder for "use tx time") - if f.valid_from == VALID_FROM_USE_TX_TIME && f.asserted { - f.valid_from = tx_id.cast_signed(); - } - f - }) - .collect(); + let (facts_to_commit, tx_id) = Staged::checked_for_commit(facts_to_commit)?; + refuse_if_failed(ctx)?; + let tx_count = inner.fact_storage.allocate_tx_count(); + let stamped = Staged::stamp(facts_to_commit, tx_id, tx_count); // WAL write includes size validation — no separate check needed. // Write WAL entry FIRST — if this fails, no facts have been applied // to shared FactStorage, so the database remains in a clean state. - let should_checkpoint = Self::wal_write_stamped_batch( - &mut self.guard, - &self.inner.options, + let should_checkpoint = WriteTransaction::wal_write_stamped_batch( + ctx, + &inner.options, self.sync, tx_count, &stamped, @@ -2241,15 +2263,70 @@ impl<'a> WriteTransaction<'a> { // WAL succeeded — now apply facts to shared FactStorage, in one // batch so no query sees part of the transaction (#554). - let _ = self.inner.fact_storage.load_facts(stamped)?; + let _ = inner.fact_storage.load_facts(stamped)?; // Trigger auto-checkpoint AFTER facts are in FactStorage so the // checkpoint captures the newly written facts. if should_checkpoint { - Minigraf::do_checkpoint(&self.inner.fact_storage, &mut self.guard)?; + Minigraf::do_checkpoint(&inner.fact_storage, ctx)?; } } + Ok(()) + } + /// Discard everything staged, and a large transaction's working state + /// (#550 spec §4.4). + fn rollback(&mut self, ctx: &mut WriteContext) { + #[cfg(not(target_arch = "wasm32"))] + if let (Some(tx), WriteContext::File { pfs, .. }) = (self.large.take(), ctx) { + pfs.rollback_large(tx); + } + #[cfg(target_arch = "wasm32")] + let _ = ctx; + self.staged.pending_facts.clear(); + } +} + +impl<'a> WriteTransaction<'a> { + /// Execute a Datalog command within this transaction. + /// + /// - **Writes** (transact / retract): buffered in-memory; not durable until `commit()`. + /// Returns `Ok(QueryResult::Ok)` immediately (not `Transacted`/`Retracted`). + /// - **Reads** (query): see committed facts in `FactStorage` **plus** all facts + /// buffered in this transaction (read-your-own-writes). + /// - **Rules**: registered immediately into the shared rule registry. + /// + /// # Errors + /// + /// Returns an error if parsing or execution fails. + pub fn execute(&mut self, input: &str) -> Result { + self.state + .execute(&mut self.guard.ctx, self.inner, input) + .map_err(MinigrafError::from) + } + + /// This transaction's chunk size after the clamp, for tests. + #[cfg(test)] + pub(crate) fn chunk(&self) -> usize { + statement::tx_chunk_facts(self.state.chunk_facts) + } + + /// Commit this transaction atomically. + /// + /// All buffered facts are stamped with a single `tx_count` / `tx_id`, then + /// 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 + /// + /// Returns an error if the WAL write or fact application fails. + pub fn commit(mut self) -> Result<(), MinigrafError> { + // On an error the drop rolls the transaction back. + self.state + .commit(&mut self.guard.ctx, self.inner) + .map_err(MinigrafError::from)?; self.committed = true; set_write_tx_active(false); Ok(()) @@ -2296,14 +2373,104 @@ impl<'a> WriteTransaction<'a> { /// Explicitly roll back the transaction. Also happens implicitly on drop. pub fn rollback(mut self) { - #[cfg(not(target_arch = "wasm32"))] - self.rollback_large(); - self.pending_facts.clear(); + self.state.rollback(&mut self.guard.ctx); self.committed = true; // Suppress rollback in Drop set_write_tx_active(false); } } +/// The facts a small transaction staged (#550 keeps a large one in working +/// trees instead), shared by [`WriteTransaction`], [`OwnedWriteTransaction`] +/// and the browser's transaction (#824), so all of them stage, read and +/// commit alike. +pub(crate) struct Staged { + /// Facts buffered in this transaction (not yet committed to FactStorage). + pub(crate) pending_facts: Vec, + /// Synthetic tx_count assigned to the next staged write batch. + next_pending_tx_count: u64, + /// Synthetic tx_id seed used to keep staged writes monotonic. + next_pending_tx_id: TxId, +} + +impl Staged { + /// Nothing staged, on a store whose last `tx_count` is `current_tx_count`. + pub(crate) fn new(current_tx_count: u64) -> Self { + Staged { + pending_facts: Vec::new(), + next_pending_tx_count: current_tx_count.saturating_add(1), + next_pending_tx_id: crate::graph::types::tx_id_now(), + } + } + + /// The `tx_id` the next staged statement is read at. + pub(crate) fn next_tx_id(&self) -> TxId { + std::cmp::max(crate::graph::types::tx_id_now(), self.next_pending_tx_id) + } + + /// Stage buffered facts with stable synthetic read metadata. + pub(crate) fn stage(&mut self, facts: Vec) { + let staged_tx_id = self.next_tx_id(); + let staged_tx_count = self.next_pending_tx_count; + + self.pending_facts.extend(facts.into_iter().map(|mut fact| { + fact.tx_id = staged_tx_id; + fact.tx_count = staged_tx_count; + fact + })); + + self.next_pending_tx_id = staged_tx_id.saturating_add(1); + self.next_pending_tx_count = self.next_pending_tx_count.saturating_add(1); + } + + /// The staged facts as a query inside the transaction reads them: a + /// default `valid_from` stays the sentinel until the commit stamps it; + /// reads see it as the fact's own staged tx_id. + pub(crate) fn read_view(&self) -> Vec { + self.pending_facts + .iter() + .cloned() + .map(|mut f| { + if f.asserted && f.valid_from == VALID_FROM_USE_TX_TIME { + f.valid_from = f.tx_id.cast_signed(); + } + f + }) + .collect() + } + + /// The staged `facts` a commit writes, and its `tx_id`, or the error that + /// rejects the commit before it takes a `tx_count`. + pub(crate) fn checked_for_commit(facts: Vec) -> Result<(Vec, TxId)> { + // One tx_id cannot tell several versions of a fact apart, so the + // last statement that wrote a triple decides it, as reads inside + // the transaction already see (#477). Each statement passed + // API-011 when staged, so every triple left has one window. + let facts = keep_last_statement(facts); + // Before allocating: a rejected transaction takes no tx_count. + // A default `valid_from` becomes the commit time, so the windows + // are checked again here (#436). + let tx_id = crate::graph::types::tx_id_now(); + crate::graph::storage::check_valid_windows(&facts, tx_id)?; + Ok((facts, tx_id)) + } + + /// Stamp `facts` with the commit's `tx_id` and `tx_count`. + pub(crate) fn stamp(facts: Vec, tx_id: TxId, tx_count: u64) -> Vec { + facts + .into_iter() + .map(|mut f| { + f.tx_id = tx_id; + f.tx_count = tx_count; + // Fix valid_from if it was left as the sentinel (placeholder for "use tx time") + if f.valid_from == VALID_FROM_USE_TX_TIME && f.asserted { + f.valid_from = tx_id.cast_signed(); + } + f + }) + .collect() + } +} + /// Keep, for each `(entity, attribute, value)`, only the staged records of /// the newest statement that wrote it (#477). Staged facts carry their /// statement's synthetic `tx_count`; the order of what is kept is unchanged. @@ -2346,14 +2513,168 @@ impl Drop for WriteTransaction<'_> { // No changes were made to the shared FactStorage during buffering, // so no cleanup is required there. A large transaction's working // pages are discarded (#550 spec §4.4). - #[cfg(not(target_arch = "wasm32"))] - self.rollback_large(); - self.pending_facts.clear(); + self.state.rollback(&mut self.guard.ctx); set_write_tx_active(false); } } } +// ─── OwnedWriteTransaction ──────────────────────────────────────────────────── + +/// An explicit write transaction that owns what it needs (#824): it keeps +/// the database open, is `Send` and `'static`, and may begin on one thread +/// and run statements, commit, roll back or be dropped on another. Language +/// bindings hold it in their transaction objects. Begin one with +/// [`Minigraf::begin_write_owned`] or [`Minigraf::begin_write_owned_with`]. +/// +/// It stages, reads and commits exactly as a [`WriteTransaction`] given the +/// same statements and options does, and holds the database's write lock +/// until it finishes: a write, a checkpoint or a begin from another thread +/// waits until then, and a query from another thread reads the committed +/// state without waiting. On the thread that began it, every call on the +/// database that a `WriteTransaction` refuses is refused (`INT-001`) while +/// it is open, whichever thread uses the transaction meanwhile, and on the +/// thread running one of its statements (so a user function that calls back +/// into the database gets `INT-001`). A thread that did not begin it and, +/// between its statements, writes through the database waits on its own +/// transaction. +/// +/// ``` +/// use minigraf::Minigraf; +/// +/// let db = Minigraf::in_memory().unwrap(); +/// let mut tx = db.begin_write_owned().unwrap(); +/// tx.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap(); +/// let tx = std::thread::spawn(move || { +/// tx.execute(r#"(transact [[:alice :person/age 30]])"#).unwrap(); +/// tx +/// }) +/// .join() +/// .unwrap(); +/// tx.commit().unwrap(); +/// ``` +/// +/// Dropping it without committing rolls it back. +pub struct OwnedWriteTransaction { + db: Minigraf, + state: TxState, + /// Set once the transaction committed, failed to commit or rolled back: + /// the drop then has nothing to release. + finished: bool, +} + +impl std::fmt::Debug for OwnedWriteTransaction { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OwnedWriteTransaction") + .field("finished", &self.finished) + .finish_non_exhaustive() + } +} + +impl OwnedWriteTransaction { + /// Execute a Datalog command within this transaction, as + /// [`WriteTransaction::execute`] does: a `transact`/`retract` is staged + /// and returns `QueryResult::Ok`, a query reads the committed state plus + /// what is staged, and a rule is registered at once. + /// + /// # Errors + /// + /// As [`WriteTransaction::execute`]; a failed statement stages nothing and + /// leaves the transaction open. `API-001` if the write lock is poisoned. + pub fn execute(&mut self, input: &str) -> Result { + // The write lock is this transaction's (`WriteSlot::owned`): no other + // caller holds the mutex for longer than it takes to see that. + let inner = &self.db.inner; + let mut slot = inner + .write_lock + .lock() + .map_err(|_| MinigrafError::from(err_coded!(ErrorCode::Api001)))?; + // While the statement runs, this thread holds the write mutex: a user + // function it calls must get INT-001 from the database, as on the + // begin thread, not lock the mutex again or read outside the + // transaction. + let _running = RunningHere::enter(); + self.state + .execute(&mut slot.ctx, inner, input) + .map_err(MinigrafError::from) + } + + /// Commit this transaction atomically, as [`WriteTransaction::commit`] + /// does, and release the write lock. A commit that fails rolls the + /// transaction back. + /// + /// # Errors + /// + /// As [`WriteTransaction::commit`]; `API-001` if the write lock is + /// poisoned. + pub fn commit(mut self) -> Result<(), MinigrafError> { + let result = match self.db.inner.write_lock.lock() { + Ok(mut slot) => { + let committed = self.state.commit(&mut slot.ctx, &self.db.inner); + if committed.is_err() { + self.state.rollback(&mut slot.ctx); + } + self.db.inner.release_owned(&mut slot); + committed + } + Err(poisoned) => { + let mut slot = poisoned.into_inner(); + self.state.rollback(&mut slot.ctx); + self.db.inner.release_owned(&mut slot); + Err(err_coded!(ErrorCode::Api001)) + } + }; + self.finished = true; + result.map_err(MinigrafError::from) + } + + /// Roll the transaction back and release the write lock. Also happens + /// when it is dropped open. + pub fn rollback(mut self) { + self.finish_rollback(); + } + + fn finish_rollback(&mut self) { + if self.finished { + return; + } + // A poisoned lock is still released: the transaction ends either way. + let mut slot = self + .db + .inner + .write_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + self.state.rollback(&mut slot.ctx); + self.db.inner.release_owned(&mut slot); + self.finished = true; + } +} + +/// Marks the current thread as running a transaction statement (the +/// thread-local flag `refuse_reentry` reads) until dropped, then restores it. +struct RunningHere(bool); + +impl RunningHere { + fn enter() -> Self { + let was = is_write_tx_active(); + set_write_tx_active(true); + RunningHere(was) + } +} + +impl Drop for RunningHere { + fn drop(&mut self) { + set_write_tx_active(self.0); + } +} + +impl Drop for OwnedWriteTransaction { + fn drop(&mut self) { + self.finish_rollback(); + } +} + // ─── Unit tests ─────────────────────────────────────────────────────────────── #[cfg(all(test, not(target_arch = "wasm32")))] @@ -2496,7 +2817,27 @@ mod tests { tx.execute(r#"(transact [[:alice :person/age 30]])"#) .unwrap(); let soon = crate::graph::types::tx_id_now().saturating_add(20); - for f in &mut tx.pending_facts { + for f in &mut tx.state.staged.pending_facts { + f.valid_to = soon.cast_signed(); + } + while crate::graph::types::tx_id_now() <= soon { + std::thread::yield_now(); + } + let err = tx.commit().expect_err("window ends before the commit time"); + assert_eq!(err.code(), "API-019"); + assert_eq!(db.current_tx_count(), 0, "no tx_count taken"); + } + + /// A failed commit of an owned transaction finishes it, rolls it back + /// and releases the write lock (#824). + #[test] + fn owned_commit_failure_releases_the_lock() { + let db = Minigraf::in_memory().unwrap(); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:alice :person/age 30]])"#) + .unwrap(); + let soon = crate::graph::types::tx_id_now().saturating_add(20); + for f in &mut tx.state.staged.pending_facts { f.valid_to = soon.cast_signed(); } while crate::graph::types::tx_id_now() <= soon { @@ -2505,6 +2846,34 @@ mod tests { let err = tx.commit().expect_err("window ends before the commit time"); assert_eq!(err.code(), "API-019"); assert_eq!(db.current_tx_count(), 0, "no tx_count taken"); + db.execute(r#"(transact [[:bob :person/age 40]])"#) + .expect("the write lock is free"); + assert_eq!(db.current_tx_count(), 1); + } + + /// A panic inside an owned transaction's statement poisons the write + /// lock: the transaction still finishes and later calls see API-001, as + /// after any poisoning (#824). + #[test] + fn owned_tx_finishes_on_a_poisoned_lock() { + let db = Minigraf::in_memory().unwrap(); + let tx = db.begin_write_owned().unwrap(); + let poison = db.clone(); + let _ = std::thread::spawn(move || { + let _guard = poison.inner.write_lock.lock().unwrap(); + panic!("poison the write lock"); + }) + .join(); + assert_eq!(tx.commit().expect_err("poisoned").code(), "API-001"); + let slot = db + .inner + .write_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + assert!(!slot.owned, "the commit released the write lock"); + drop(slot); + let tx = db.begin_write_owned(); + assert_eq!(tx.expect_err("poisoned").code(), "API-001"); } // ── read-your-own-writes ────────────────────────────────────────────────── diff --git a/src/error.rs b/src/error.rs index 37eb3ed52..d9162df63 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1671,7 +1671,7 @@ pub(crate) const REGISTRY: &[(ErrorCode, &str, &str, ErrorCategory)] = &[ ( ErrorCode::Int001, "INT-001", - "a WriteTransaction is already in progress on this thread; use tx.execute() instead", + "a write transaction is already in progress on this thread; use tx.execute() instead", ErrorCategory::Internal, ), ( diff --git a/src/fault_matrix.rs b/src/fault_matrix.rs index 061797a25..c00b22481 100644 --- a/src/fault_matrix.rs +++ b/src/fault_matrix.rs @@ -20,6 +20,10 @@ //! `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. +//! +//! The transaction cases run again through an `OwnedWriteTransaction` +//! (#824), which shares the borrowed transaction's staging and commit but +//! holds the write lock differently: [`with_owned`]. use crate::db::{Minigraf, OpenOptions, SyncMode, TxOptions}; use crate::fact_log::FactFilter; @@ -116,9 +120,57 @@ const X_TX_AFTER: &str = r#"(transact [[:x1 :p/x 2 {:valid-from "2020-01-01T00:00:00Z"}] [:x10 :p/x 10] [:x9 :p/x 9]])"#; const X_TX_FACTS: usize = 12; +thread_local! { + /// Run the transaction cases through `begin_write_owned_with` (#824). + static OWNED: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + +/// Run `f` with the transaction cases using an `OwnedWriteTransaction`. +fn with_owned(f: impl FnOnce()) { + struct Reset; + impl Drop for Reset { + fn drop(&mut self) { + OWNED.with(|o| o.set(false)); + } + } + OWNED.with(|o| o.set(true)); + let _reset = Reset; + f(); +} + +/// A transaction of either kind, as [`with_owned`] chose. +enum AnyTx<'a> { + Borrowed(crate::db::WriteTransaction<'a>), + Owned(crate::db::OwnedWriteTransaction), +} + +impl AnyTx<'_> { + fn begin(db: &Minigraf, options: TxOptions) -> Result, crate::MinigrafError> { + if OWNED.with(std::cell::Cell::get) { + db.begin_write_owned_with(options).map(AnyTx::Owned) + } else { + db.begin_write_with(options).map(AnyTx::Borrowed) + } + } + + fn execute(&mut self, stmt: &str) -> Result { + match self { + AnyTx::Borrowed(tx) => tx.execute(stmt), + AnyTx::Owned(tx) => tx.execute(stmt), + } + } + + fn commit(self) -> Result<(), crate::MinigrafError> { + match self { + AnyTx::Borrowed(tx) => tx.commit(), + AnyTx::Owned(tx) => tx.commit(), + } + } +} + /// Run the `WriteTransaction` of `Op::Large { tx: true, .. }`. fn large_write_tx(db: &Minigraf) -> Result<(), crate::MinigrafError> { - let mut tx = db.begin_write()?; + let mut tx = AnyTx::begin(db, TxOptions::default())?; tx.execute(X_TX_BEFORE)?; tx.execute(X_LARGE)?; match tx.execute(X_TX_FAILS) { @@ -448,6 +500,50 @@ fn faults_during_a_large_write_transaction_on_a_new_file_with_a_wal() { }); } +#[test] +fn faults_during_a_large_owned_write_transaction() { + with_owned(|| { + run_all(Op::Large { + one_slot: false, + wal: true, + tx: true, + }); + }); +} + +#[test] +fn faults_during_a_large_owned_write_transaction_that_creates_the_wal() { + with_owned(|| { + run_all(Op::Large { + one_slot: false, + wal: false, + tx: true, + }); + }); +} + +#[test] +fn faults_during_a_large_owned_write_transaction_on_a_new_file() { + with_owned(|| { + run_all(Op::Large { + one_slot: true, + wal: false, + tx: true, + }); + }); +} + +#[test] +fn faults_during_a_large_owned_write_transaction_on_a_new_file_with_a_wal() { + with_owned(|| { + run_all(Op::Large { + one_slot: true, + wal: true, + tx: true, + }); + }); +} + /// The disk fills part-way through a checkpoint: every write from then on /// fails. The checkpoint fails, the handle refuses writes with STG-046, and /// after space is freed a reopen holds every committed fact. @@ -609,7 +705,7 @@ impl History { } fn commit_with(db: &Minigraf, mode: SyncMode, stmt: &str) -> Result<(), crate::MinigrafError> { - let mut tx = db.begin_write_with(TxOptions::new().synchronous(mode))?; + let mut tx = AnyTx::begin(db, TxOptions::new().synchronous(mode))?; tx.execute(stmt)?; tx.commit() } @@ -768,3 +864,18 @@ fn faults_during_a_normal_commit_on_a_full_database() { fn faults_during_a_full_commit_on_a_normal_database() { run_all_mixed(Mix::FullOnNormal); } + +#[test] +fn faults_during_an_owned_full_commit_after_a_normal_one() { + with_owned(|| run_all_mixed(Mix::FullAfterNormal)); +} + +#[test] +fn faults_during_an_owned_normal_commit_on_a_full_database() { + with_owned(|| run_all_mixed(Mix::NormalOnFull)); +} + +#[test] +fn faults_during_an_owned_full_commit_on_a_normal_database() { + with_owned(|| run_all_mixed(Mix::FullOnNormal)); +} diff --git a/src/lib.rs b/src/lib.rs index 347a78571..d8712ece9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -108,7 +108,10 @@ pub mod browser; #[cfg(not(target_arch = "wasm32"))] pub use db::OpenOptionsWithPath; -pub use db::{IntegrityReport, Minigraf, OpenOptions, SyncMode, TxOptions, WriteTransaction}; +pub use db::{ + IntegrityReport, Minigraf, OpenOptions, OwnedWriteTransaction, SyncMode, TxOptions, + WriteTransaction, +}; pub use error::{ErrorCategory, MinigrafError}; pub use overrides::Overridable; pub use query::datalog::functions::UdfError; diff --git a/src/query/datalog/prepared.rs b/src/query/datalog/prepared.rs index 4e52f24c8..9c5352b4c 100644 --- a/src/query/datalog/prepared.rs +++ b/src/query/datalog/prepared.rs @@ -1,5 +1,5 @@ use crate::cursor::Cursor; -use crate::error::{ErrorCode, MinigrafError, bail_coded, err_coded}; +use crate::error::{ErrorCode, MinigrafError, bail_coded}; use crate::graph::FactStorage; use crate::graph::storage::Snapshot; use crate::graph::types::Value; @@ -89,6 +89,8 @@ pub struct PreparedQuery { /// The database's `max_iterations`, `max_derived_facts` and /// `query_memory_limit`, or the setters' values (#870, #861). limits: crate::query::engine::Limits, + /// The database's owned-transaction begin thread, for `INT-001` (#824). + owner: Arc, } impl std::fmt::Debug for PreparedQuery { @@ -173,9 +175,7 @@ impl PreparedQuery { /// - Missing bind value for a slot present in the query. /// - Type mismatch (e.g. `Val` supplied for an `:as-of` slot). pub fn execute(&self, bindings: &[(&str, BindValue)]) -> Result { - if crate::db::is_write_tx_active() { - return Err(err_coded!(ErrorCode::Int001).into()); - } + self.owner.refuse_reentry()?; self.fact_storage .snapshot() .and_then(|snapshot| self.execute_inner(bindings, snapshot)) @@ -190,9 +190,7 @@ impl PreparedQuery { /// # Errors /// The same as [`PreparedQuery::execute`]. pub fn query(&self, bindings: &[(&str, BindValue)]) -> Result { - if crate::db::is_write_tx_active() { - return Err(err_coded!(ErrorCode::Int001).into()); - } + self.owner.refuse_reentry()?; let run = || { let snapshot = self.fact_storage.snapshot()?; let (executor, command) = self.bind(bindings, snapshot.clone())?; @@ -247,6 +245,7 @@ pub(crate) fn prepare_query( rules: Arc>, functions: Arc, limits: crate::query::engine::Limits, + owner: Arc, ) -> Result { validate_no_attribute_slots(&query)?; let slot_names = collect_slot_names(&query); @@ -257,6 +256,7 @@ pub(crate) fn prepare_query( rules, functions, limits, + owner, }) } diff --git a/tests/owned_tx_test.rs b/tests/owned_tx_test.rs new file mode 100644 index 000000000..dcc473e89 --- /dev/null +++ b/tests/owned_tx_test.rs @@ -0,0 +1,700 @@ +// File-backed (Minigraf::open, tempfile): native only, like the other file tests. +#![cfg(not(target_arch = "wasm32"))] +//! `OwnedWriteTransaction` (#824): a write transaction that owns its +//! database handle, so a language binding can hold it in an object that is +//! used, committed or dropped on any thread. It must leave the database as +//! the borrowed `WriteTransaction` does, follow its begin thread for +//! `INT-001`, and make other threads' writers wait. + +use minigraf::{ + FactFilter, Minigraf, OpenOptions, OwnedWriteTransaction, QueryResult, TxOptions, + WriteTransaction, +}; +use std::path::Path; +use std::sync::mpsc; +use std::thread; +use std::time::Duration; + +/// How long a call that should wait is given to return anyway, and how long +/// a call that should not wait may take before the test calls it stuck. +const WAITS: Duration = Duration::from_millis(300); +const STUCK: Duration = Duration::from_secs(30); + +/// A compile-time check: the owned transaction can be held in a `'static` +/// object and moved between threads (#824 A1). +#[test] +fn owned_tx_is_send_and_static() { + fn assert_send_static() {} + assert_send_static::(); +} + +// ─── Equivalence with the borrowed transaction (A2) ────────────────────────── + +/// One fact version, comparable across databases: a default valid-from +/// (the commit time) is shown as `None`, because two databases commit at +/// different times. +type Rec = (String, String, String, u64, Option, i64, bool); + +fn records(db: &Minigraf) -> Vec { + let mut v: Vec = db + .fact_log(&FactFilter::new()) + .unwrap() + .map(|r| { + let r = r.unwrap(); + let from = (r.valid_from != r.tx_id.cast_signed()).then_some(r.valid_from); + ( + r.entity.to_string(), + r.attribute, + format!("{:?}", r.value), + r.tx_count, + from, + r.valid_to, + r.asserted, + ) + }) + .collect(); + v.sort(); + v +} + +/// The rows of a query over every current fact, sorted. +fn rows(db: &Minigraf) -> Vec { + let QueryResult::QueryResults { results, .. } = db + .execute("(query [:find ?e ?a ?v :where [?e ?a ?v]])") + .unwrap() + else { + panic!("not a query result"); + }; + let mut v: Vec = results.iter().map(|r| format!("{r:?}")).collect(); + v.sort(); + v +} + +/// What a statement inside a transaction returned: its rows or error code. +fn outcome(result: Result) -> String { + match result { + Ok(QueryResult::QueryResults { results, .. }) => { + let mut v: Vec = results.iter().map(|r| format!("{r:?}")).collect(); + v.sort(); + v.join(";") + } + Ok(QueryResult::Ok) => "ok".to_string(), + Ok(_) => "other".to_string(), + Err(e) => e.code().to_string(), + } +} + +const SETUP: &str = r#"(transact [[:alice :person/name "Alice"] [:alice :person/age 30] + [:bob :person/name "Bob"] [:carol :person/name "Carol"]])"#; + +/// Statement sequences: writes, retractions, reads of staged state, rules, +/// rejections (API-011, parse errors, API-002 via `parse`), several windows +/// of a triple across statements (latest statement wins, #477), and enough +/// facts to pass a chunk of 2. +const SEQUENCES: &[&[&str]] = &[ + &[], + &["(query [:find ?n :where [?e :person/name ?n]])"], + &[ + r#"(transact [[:dave :person/name "Dave"] [:dave :person/age 40]])"#, + "(query [:find ?n :where [?e :person/name ?n]])", + r#"(retract [[:bob :person/name "Bob"]])"#, + "(query [:find ?n :where [?e :person/name ?n]])", + "(transact [[:alice :person/age 31]])", + ], + &[ + r#"(transact [[:erin :person/name "Erin" {:valid-from "2020-01-01T00:00:00Z"}]])"#, + r#"(transact [[:erin :person/name "Erin" {:valid-from "2021-01-01T00:00:00Z" :valid-to "2030-01-01T00:00:00Z"}]])"#, + r#"(transact [[:x :p/v 1 {:valid-from "2020-01-01T00:00:00Z"}] [:x :p/v 1 {:valid-from "2021-01-01T00:00:00Z"}]])"#, + "(transact [[:x :p/v", + "(query [:find ?e :where [?e :person/name \"Erin\"]])", + ], + &[ + "(rule [(named ?e ?n) [?e :person/name ?n]])", + r#"(transact [[:frank :person/name "Frank"] [:frank :person/friend :alice] [:gina :person/name "Gina"]])"#, + "(query [:find ?n :where (named ?e ?n)])", + r#"(retract [[:carol :person/name "Carol"] [:alice :person/age 30]])"#, + r#"(transact [[:hal :person/name "Hal"] [:hal :person/age 50] [:ida :person/name "Ida"]])"#, + "(query [:find ?n ?a :where [?e :person/name ?n] [?e :person/age ?a]])", + ], +]; + +#[derive(Clone, Copy, Debug)] +enum End { + Commit, + Rollback, + Drop, +} + +/// Run `stmts` in a borrowed transaction, return each statement's outcome. +fn run_borrowed(db: &Minigraf, opts: TxOptions, stmts: &[&str], end: End) -> Vec { + let mut tx: WriteTransaction<'_> = db.begin_write_with(opts).unwrap(); + let out = stmts.iter().map(|s| outcome(tx.execute(s))).collect(); + match end { + End::Commit => tx.commit().unwrap(), + End::Rollback => tx.rollback(), + End::Drop => drop(tx), + } + out +} + +fn run_owned(db: &Minigraf, opts: TxOptions, stmts: &[&str], end: End) -> Vec { + let mut tx = db.begin_write_owned_with(opts).unwrap(); + let out = stmts.iter().map(|s| outcome(tx.execute(s))).collect(); + match end { + End::Commit => tx.commit().unwrap(), + End::Rollback => tx.rollback(), + End::Drop => drop(tx), + } + out +} + +fn open_file(path: &Path) -> Minigraf { + OpenOptions::new().path(path).open().unwrap() +} + +#[test] +fn owned_tx_matches_borrowed_tx() { + for (i, stmts) in SEQUENCES.iter().enumerate() { + for end in [End::Commit, End::Rollback, End::Drop] { + // `None`: the database's chunk; `Some(2)`: large on a file. + for chunk in [None, Some(2)] { + let opts = || match chunk { + Some(n) => TxOptions::new().chunk_facts(n), + None => TxOptions::new(), + }; + let case = format!("sequence {i}, {end:?}, chunk {chunk:?}"); + + // In memory. + let a = Minigraf::in_memory().unwrap(); + let b = Minigraf::in_memory().unwrap(); + a.execute(SETUP).unwrap(); + b.execute(SETUP).unwrap(); + let out_a = run_borrowed(&a, opts(), stmts, end); + let out_b = run_owned(&b, opts(), stmts, end); + assert!( + out_a == out_b, + "statement outcomes differ in memory: {case}" + ); + assert!( + records(&a) == records(&b), + "records differ in memory: {case}" + ); + assert!(rows(&a) == rows(&b), "rows differ in memory: {case}"); + assert_eq!(a.current_tx_count(), b.current_tx_count(), "{case}"); + + // On a file, before and after reopening. + let dir = tempfile::tempdir().unwrap(); + let (pa, pb) = (dir.path().join("a.graph"), dir.path().join("b.graph")); + let (out_a, out_b, rec_a, rec_b) = { + let a = open_file(&pa); + let b = open_file(&pb); + a.execute(SETUP).unwrap(); + b.execute(SETUP).unwrap(); + let out_a = run_borrowed(&a, opts(), stmts, end); + let out_b = run_owned(&b, opts(), stmts, end); + assert!(rows(&a) == rows(&b), "rows differ on a file: {case}"); + (out_a, out_b, records(&a), records(&b)) + }; + assert!( + out_a == out_b, + "statement outcomes differ on a file: {case}" + ); + assert!(rec_a == rec_b, "records differ on a file: {case}"); + let (a, b) = (open_file(&pa), open_file(&pb)); + assert!( + records(&a) == records(&b), + "records differ after reopen: {case}" + ); + assert!(rows(&a) == rows(&b), "rows differ after reopen: {case}"); + } + } + } +} + +/// The sequences above that pass a chunk of 2 do become large: a query of +/// the working trees sees the staged facts, and the commit is a checkpoint +/// (no WAL left). +#[test] +fn owned_tx_large_commit_is_a_checkpoint() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("g.graph"); + let db = open_file(&path); + db.execute(SETUP).unwrap(); + let mut tx = db + .begin_write_owned_with(TxOptions::new().chunk_facts(2)) + .unwrap(); + tx.execute(r#"(transact [[:a :p/n "A"] [:b :p/n "B"] [:c :p/n "C"] [:d :p/n "D"]])"#) + .unwrap(); + let QueryResult::QueryResults { results, .. } = tx + .execute("(query [:find ?n :where [?e :p/n ?n]])") + .unwrap() + else { + panic!("not a query result"); + }; + assert_eq!(results.len(), 4, "the staged facts are read"); + tx.commit().unwrap(); + let mut wal = path.as_os_str().to_owned(); + wal.push(".wal"); + assert!( + !Path::new(&wal).exists(), + "a large commit is a checkpoint and deletes the WAL" + ); + drop(db); + let db = open_file(&path); + assert_eq!(rows(&db).len(), 8, "every fact after reopening"); +} + +// ─── Threads (A3, A4) ──────────────────────────────────────────────────────── + +/// Every guarded call, by name, with its error code or `None` on success. +fn guarded_calls(db: &Minigraf) -> Vec<(&'static str, Option<&'static str>)> { + let code = |r: Result<(), minigraf::MinigrafError>| r.err().map(|e| e.code()); + let prepared = || db.prepare("(query [:find ?n :where [?e :person/name ?n]])"); + vec![ + ( + "execute transact", + code( + db.execute(r#"(transact [[:zed :person/name "Zed"]])"#) + .map(|_| ()), + ), + ), + ( + "execute query", + code( + db.execute("(query [:find ?e :where [?e :person/name _]])") + .map(|_| ()), + ), + ), + ( + "query", + code( + db.query("(query [:find ?e :where [?e :person/name _]])") + .map(|_| ()), + ), + ), + ("prepare", code(prepared().map(|_| ()))), + ( + "fact_log", + code(db.fact_log(&FactFilter::new()).map(|_| ())), + ), + ("checkpoint", code(db.checkpoint())), + ("verify", code(db.verify().map(|_| ()))), + ("rebuild_indexes", code(db.rebuild_indexes())), + ("begin_write", code(db.begin_write().map(|_| ()))), + ( + "begin_write_owned", + code(db.begin_write_owned().map(|_| ())), + ), + ] +} + +/// Prepared `execute`/`query` of a query prepared before the transaction. +fn prepared_calls(pq: &minigraf::PreparedQuery) -> Vec<(&'static str, Option<&'static str>)> { + vec![ + ("prepared execute", pq.execute(&[]).err().map(|e| e.code())), + ("prepared query", pq.query(&[]).err().map(|e| e.code())), + ] +} + +fn assert_all(calls: &[(&'static str, Option<&'static str>)], want: Option<&str>, when: &str) { + for (name, code) in calls { + assert_eq!(*code, want, "{name} {when}"); + } +} + +fn file_db(dir: &tempfile::TempDir) -> Minigraf { + let db = open_file(&dir.path().join("g.graph")); + db.execute(SETUP).unwrap(); + db +} + +#[test] +fn owned_tx_guards_begin_thread() { + let dir = tempfile::tempdir().unwrap(); + let db = file_db(&dir); + let pq = db + .prepare("(query [:find ?n :where [?e :person/name ?n]])") + .unwrap(); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + assert_all(&guarded_calls(&db), Some("INT-001"), "on the begin thread"); + assert_all(&prepared_calls(&pq), Some("INT-001"), "on the begin thread"); + + // Moved to another thread, it still guards the begin thread. + let (back, rx) = mpsc::channel(); + let handle = thread::spawn(move || { + tx.execute(r#"(transact [[:erin :person/name "Erin"]])"#) + .unwrap(); + rx.recv().unwrap(); + tx.commit().unwrap(); + }); + assert_all( + &guarded_calls(&db), + Some("INT-001"), + "on the begin thread while another thread holds the transaction", + ); + back.send(()).unwrap(); + handle.join().unwrap(); + + // Finished on the other thread: every call works here again. + assert_all(&prepared_calls(&pq), None, "after the commit"); + assert_all(&guarded_calls(&db), None, "after the commit"); + assert_eq!(rows(&db).len(), 7, "both staged facts and Zed committed"); +} + +#[test] +fn owned_tx_commit_on_other_thread() { + let dir = tempfile::tempdir().unwrap(); + let db = file_db(&dir); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + let db2 = db.clone(); + thread::spawn(move || { + tx.execute(r#"(retract [[:bob :person/name "Bob"]])"#) + .unwrap(); + tx.commit().unwrap(); + // The committing thread did not begin it: nothing is guarded here. + assert_all(&guarded_calls(&db2), None, "on the committing thread"); + }) + .join() + .unwrap(); + assert_all( + &guarded_calls(&db), + None, + "on the begin thread after commit", + ); + let names = db + .execute("(query [:find ?n :where [?e :person/name ?n]])") + .unwrap(); + let QueryResult::QueryResults { results, .. } = names else { + panic!("not a query result"); + }; + // Alice, Carol, Dave and Zed (from the first guarded_calls). + assert_eq!(results.len(), 4, "Dave committed, Bob retracted"); +} + +#[test] +fn owned_tx_rollback_on_other_thread() { + let dir = tempfile::tempdir().unwrap(); + let db = file_db(&dir); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + thread::spawn(move || tx.rollback()).join().unwrap(); + let before = rows(&db); + assert_eq!(before.len(), 4, "the staged fact is gone"); + assert_all( + &guarded_calls(&db), + None, + "on the begin thread after rollback", + ); +} + +#[test] +fn owned_tx_drop_on_other_thread() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("g.graph"); + { + let db = file_db(&dir); + let mut tx = db + .begin_write_owned_with(TxOptions::new().chunk_facts(2)) + .unwrap(); + // Large: its working pages are discarded by the drop. + tx.execute(r#"(transact [[:a :p/n "A"] [:b :p/n "B"] [:c :p/n "C"]])"#) + .unwrap(); + thread::spawn(move || drop(tx)).join().unwrap(); + assert_eq!(rows(&db).len(), 4, "nothing of the dropped transaction"); + assert_all(&guarded_calls(&db), None, "on the begin thread after drop"); + let db2 = db.clone(); + thread::spawn(move || { + db2.execute(r#"(transact [[:y :p/n "Y"]])"#).unwrap(); + }) + .join() + .unwrap(); + } + let db = open_file(&path); + let names: Vec = rows(&db); + // SETUP's 4 facts, Zed (guarded_calls) and Y. + assert_eq!(names.len(), 6, "after reopening: no dropped fact"); +} + +/// The transaction keeps the database open after every other handle is +/// dropped, and the close-time checkpoint runs after it finishes. +#[test] +fn owned_tx_keeps_database_open() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("g.graph"); + let tx = { + let db = file_db(&dir); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + tx + }; + tx.commit().unwrap(); + let db = open_file(&path); + assert_eq!( + rows(&db).len(), + 5, + "the commit after the last handle dropped" + ); +} + +/// Run `f` on a new thread; `Some` of its result if it returns within `t`. +fn within( + t: Duration, + f: impl FnOnce() -> T + Send + 'static, +) -> (Option, mpsc::Receiver) { + let (send, recv) = mpsc::channel(); + thread::spawn(move || { + let _ = send.send(f()); + }); + match recv.recv_timeout(t) { + Ok(v) => (Some(v), recv), + Err(_) => (None, recv), + } +} + +#[test] +fn owned_tx_other_thread_write_waits() { + type Call = fn(&Minigraf) -> Result<(), minigraf::MinigrafError>; + let calls: &[(&str, Call)] = &[ + ("execute transact", |db| { + db.execute(r#"(transact [[:w :person/name "W"]])"#) + .map(|_| ()) + }), + ("execute rule", |db| { + db.execute("(rule [(named2 ?e ?n) [?e :person/name ?n]])") + .map(|_| ()) + }), + ("checkpoint", Minigraf::checkpoint), + ("verify", |db| db.verify().map(|_| ())), + ("rebuild_indexes", Minigraf::rebuild_indexes), + ("begin_write", |db| db.begin_write()?.commit()), + ("begin_write_owned", |db| db.begin_write_owned()?.commit()), + ]; + for (name, call) in calls { + let dir = tempfile::tempdir().unwrap(); + let db = file_db(&dir); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + let db2 = db.clone(); + let call = *call; + let (early, rest) = within(WAITS, move || call(&db2).map_err(|e| e.code())); + assert!(early.is_none(), "{name} from another thread did not wait"); + tx.commit().unwrap(); + let result = rest + .recv_timeout(STUCK) + .expect("the waiting call returns after the commit"); + assert_eq!(result, Ok(()), "{name} after the commit"); + } +} + +#[test] +fn owned_tx_other_thread_query_reads_committed() { + let dir = tempfile::tempdir().unwrap(); + let db = file_db(&dir); + let pq = db + .prepare("(query [:find ?n :where [?e :person/name ?n]])") + .unwrap(); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + tx.execute(r#"(retract [[:bob :person/name "Bob"]])"#) + .unwrap(); + let staged = outcome(tx.execute("(query [:find ?n :where [?e :person/name ?n]])")); + let db2 = db.clone(); + let (got, _) = within(STUCK, move || { + let names = outcome(db2.execute("(query [:find ?n :where [?e :person/name ?n]])")); + let cursor_rows = db2 + .query("(query [:find ?n :where [?e :person/name ?n]])") + .unwrap() + .count(); + let prepared = outcome(pq.execute(&[])); + let log = db2.fact_log(&FactFilter::new()).unwrap().count(); + (names, cursor_rows, prepared, log) + }); + let (names, cursor_rows, prepared, log) = got.expect("queries from another thread do not wait"); + assert_eq!( + names, + r#"[String("Alice")];[String("Bob")];[String("Carol")]"# + ); + assert_eq!(cursor_rows, 3, "a cursor reads the committed state"); + assert_eq!( + prepared, names, + "a prepared query reads the committed state" + ); + assert_eq!(log, 4, "the fact log holds the committed records"); + assert_eq!( + staged, r#"[String("Alice")];[String("Carol")];[String("Dave")]"#, + "the transaction reads its staged state" + ); + tx.commit().unwrap(); +} + +#[test] +fn owned_tx_leaves_other_database_usable() { + let dir = tempfile::tempdir().unwrap(); + let db = file_db(&dir); + let other = Minigraf::in_memory().unwrap(); + let pq = other + .prepare("(query [:find ?n :where [?e :person/name ?n]])") + .unwrap(); + let tx = db.begin_write_owned().unwrap(); + // On the begin thread and on another, the other database is not guarded + // and does not wait. + assert_all(&guarded_calls(&other), None, "on another database"); + assert_all(&prepared_calls(&pq), None, "on another database"); + let (got, _) = within(STUCK, move || { + other + .execute(r#"(transact [[:q :person/name "Q"]])"#) + .map(|_| ()) + .map_err(|e| e.code()) + }); + assert_eq!(got, Some(Ok(())), "another database's write does not wait"); + tx.rollback(); +} + +// ─── Begin refusals (A5) ───────────────────────────────────────────────────── + +#[test] +fn owned_begin_refused_inside_a_transaction() { + let dir = tempfile::tempdir().unwrap(); + let db = file_db(&dir); + { + let tx = db.begin_write().unwrap(); + let err = db.begin_write_owned().expect_err("inside a borrowed one"); + assert_eq!(err.code(), "INT-001"); + tx.commit().unwrap(); + } + { + let tx = db.begin_write_owned().unwrap(); + let err = db.begin_write_owned().expect_err("inside an owned one"); + assert_eq!(err.code(), "INT-001"); + let err = db + .begin_write_owned_with(TxOptions::new().chunk_facts(5)) + .expect_err("inside an owned one, with options"); + assert_eq!(err.code(), "INT-001"); + tx.commit().unwrap(); + } + // The handle stays usable. + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + tx.commit().unwrap(); + assert_eq!(rows(&db).len(), 5); +} + +#[test] +fn owned_begin_refused_on_read_only() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("g.graph"); + drop(file_db(&dir)); + let db = OpenOptions::new() + .read_only(true) + .path(&path) + .open() + .unwrap(); + let err = db.begin_write_owned().expect_err("read-only"); + assert_eq!(err.code(), "API-014"); + let err = db + .begin_write_owned_with(TxOptions::new()) + .expect_err("read-only"); + assert_eq!(err.code(), "API-014"); + assert_eq!(rows(&db).len(), 4, "the handle still reads"); + let mem = OpenOptions::new().read_only(true).open_memory().unwrap(); + assert_eq!( + mem.begin_write_owned().expect_err("read-only").code(), + "API-014" + ); +} + +// ─── Edge cases ────────────────────────────────────────────────────────────── + +#[test] +fn owned_tx_empty_and_query_only() { + let db = Minigraf::in_memory().unwrap(); + db.execute(SETUP).unwrap(); + let before = db.current_tx_count(); + db.begin_write_owned().unwrap().commit().unwrap(); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute("(query [:find ?n :where [?e :person/name ?n]])") + .unwrap(); + tx.commit().unwrap(); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute("(query [:find ?n :where [?e :person/name ?n]])") + .unwrap(); + tx.rollback(); + assert_eq!(db.current_tx_count(), before, "no transaction taken"); +} + +/// A statement that fails stages nothing and leaves the transaction open. +#[test] +fn owned_tx_failed_statement_stays_open() { + let db = Minigraf::in_memory().unwrap(); + let mut tx = db.begin_write_owned().unwrap(); + assert_eq!( + tx.execute("(transact [[:a").expect_err("parse").code(), + "PRS-004" + ); + let err = tx + .execute(r#"(transact [[:x :p/v 1 {:valid-from "2020-01-01T00:00:00Z"}] [:x :p/v 1 {:valid-from "2021-01-01T00:00:00Z"}]])"#) + .expect_err("two windows of a triple"); + assert_eq!(err.code(), "API-011"); + tx.execute(r#"(transact [[:a :p/n "A"]])"#).unwrap(); + tx.commit().unwrap(); + assert_eq!(rows(&db).len(), 1, "only the good statement"); +} + +// ─── User functions off the begin thread ──────────────────────────────────── + +/// A user function that calls back into the database, from a statement of an +/// owned transaction run on a thread that did not begin it, gets INT-001 +/// (QRY-014 for the query) instead of locking the write mutex its own thread +/// holds or reading outside the transaction. The transaction stays usable +/// and the lock is released when it ends. +#[test] +fn owned_tx_user_function_off_the_begin_thread() { + let db = Minigraf::in_memory().unwrap(); + db.execute(SETUP).unwrap(); + let w = db.clone(); + db.try_register_predicate("writes?", move |_| { + w.execute(r#"(transact [[:z :person/name "Z"]])"#) + .map(|_| true) + .map_err(|e| minigraf::UdfError::new(e.code())) + }) + .unwrap(); + let r = db.clone(); + db.try_register_predicate("reads?", move |_| { + r.execute(NAMES_Q) + .map(|_| true) + .map_err(|e| minigraf::UdfError::new(e.code())) + }) + .unwrap(); + let mut tx = db.begin_write_owned().unwrap(); + tx.execute(r#"(transact [[:dave :person/name "Dave"]])"#) + .unwrap(); + let (got, _) = within(STUCK, move || { + let mut errors = Vec::new(); + for f in ["writes?", "reads?"] { + let q = format!("(query [:find ?n :where [?e :person/name ?n] [({f} ?n)]])"); + let e = tx.execute(&q).expect_err("a call back into the database"); + errors.push((e.code(), e.to_string().contains("INT-001"))); + } + // Still usable: a statement and the commit. + tx.execute(r#"(transact [[:erin :person/name "Erin"]])"#) + .unwrap(); + tx.commit().unwrap(); + errors + }); + let errors = got.expect("no deadlock"); + assert_eq!(errors, vec![("QRY-014", true), ("QRY-014", true)]); + // The flag is the thread's own: after the statements, nothing is guarded. + assert_all(&guarded_calls(&db), None, "after the commit"); + assert_eq!(rows(&db).len(), 7, "Dave, Erin and Zed with the setup"); +} + +const NAMES_Q: &str = "(query [:find ?n :where [?e :person/name ?n]])";