From 8499b0ea3846ac47b2281ca8bdc27a937b22440d Mon Sep 17 00:00:00 2001 From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com> Date: Wed, 19 Aug 2026 14:12:02 +0700 Subject: [PATCH] Several statements as one unit of work `conn.transaction()` starts a span and hands back a `Transaction` whose `commit()` and `rollback()` end it. The statements in between stay the connection's, because the span belongs to the connection rather than to a second handle on it, and `conn.inTransaction` says whether one is open. `{ readOnly: true }` starts a span that refuses the statement that writes. Nesting is refused by the engine, with the engine's own condition, rather than by a rule kept here that would have to be held in step with it. `await using tx` rolls back. That is the opposite of what the Python client's `with` block does, and the difference is in the language and not in the database: a Python context manager is handed the exception unwinding through it, and a JavaScript disposal is told nothing at all. A disposal that committed would commit half the work of a block that threw, which is the one thing a transaction exists to prevent. So the commit is the word the caller writes, a block that forgets it loses its work loudly, and the alternative was losing half of it quietly. `inTransaction` is an `AtomicBool` beside the connection's lock, written out of the session's own answer by every statement that runs. A tally kept here would have been a second thing to keep in step, and a caller who writes `START TRANSACTION` as a statement would have defeated it. The three places that take the lock now go through one helper, so what an empty slot means, which is a stream holding the connection rather than a closed one, is decided once. The engine pin moves from #338 to current main, which brings in #385: reading a property no column holds is null rather than an error, so nodes match edges and ISO 20.11. The misuse test that asserted the old behaviour moves to the tests for programs that look like misuse and are not, with what it costs written down beside it, which is that a misspelled property is null for every row. --- Cargo.lock | 20 +-- Cargo.toml | 4 +- README.md | 27 +++- binding.cjs | 1 + binding.d.cts | 95 +++++++++++++ etc/zudb.api.md | 16 +++ src/conn.rs | 250 ++++++++++++++++++++++++++-------- src/lib.rs | 1 + src/stream.rs | 9 ++ src/txn.rs | 264 ++++++++++++++++++++++++++++++++++++ test/exports.test.mjs | 1 + test/misuse.test.mjs | 17 ++- test/transaction.test.mjs | 276 ++++++++++++++++++++++++++++++++++++++ test/types/cjs.cts | 28 +++- test/types/esm.mts | 16 +++ types/header.d.ts | 15 +++ zudb.cjs | 1 + zudb.d.cts | 11 ++ zudb.mjs | 1 + 19 files changed, 975 insertions(+), 78 deletions(-) create mode 100644 src/txn.rs create mode 100644 test/transaction.test.mjs diff --git a/Cargo.lock b/Cargo.lock index 219fbba..c9df533 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1277,7 +1277,7 @@ dependencies = [ [[package]] name = "zu" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "zu-common", "zu-encoding", @@ -1293,7 +1293,7 @@ dependencies = [ [[package]] name = "zu-common" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "thiserror", ] @@ -1301,7 +1301,7 @@ dependencies = [ [[package]] name = "zu-encoding" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "ruzstd", "zu-common", @@ -1310,7 +1310,7 @@ dependencies = [ [[package]] name = "zu-exec" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "zu-common", "zu-query", @@ -1320,7 +1320,7 @@ dependencies = [ [[package]] name = "zu-query" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "crossbeam-deque", "zu-common", @@ -1331,7 +1331,7 @@ dependencies = [ [[package]] name = "zu-s3" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "crc32c", "object_store", @@ -1342,7 +1342,7 @@ dependencies = [ [[package]] name = "zu-sqlite" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "rusqlite", "zu-common", @@ -1352,7 +1352,7 @@ dependencies = [ [[package]] name = "zu-storage" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "zu-common", "zu-encoding", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "zu-vector" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "zu-common", ] @@ -1369,7 +1369,7 @@ dependencies = [ [[package]] name = "zu-zu1" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1#92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" +source = "git+https://github.com/tamnd/zu?rev=8aa27d9c9df4087522f4314ea5b1d5850df27f8d#8aa27d9c9df4087522f4314ea5b1d5850df27f8d" dependencies = [ "crc32c", "loom", diff --git a/Cargo.toml b/Cargo.toml index 432f1e0..19e42a5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,8 +18,8 @@ crate-type = ["cdylib"] # with (ADR 0002), so a revision is the honest way to say which one. # A local checkout is used instead with a `paths` override in # `.cargo/config.toml`, which is untracked on purpose. -zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" } -zu-common = { git = "https://github.com/tamnd/zu", rev = "92c9a5e9f1f0d5f4d89bf7321e5710a4fcb861f1" } +zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "8aa27d9c9df4087522f4314ea5b1d5850df27f8d" } +zu-common = { git = "https://github.com/tamnd/zu", rev = "8aa27d9c9df4087522f4314ea5b1d5850df27f8d" } # N-API by way of napi-rs (ADR 0002). `napi9` is the version of N-API # this addon declares it needs, which is what makes one binary work # across Node 24, Node 26, Electron and Bun without a rebuild: the diff --git a/README.md b/README.md index 43c3e75..c7940d8 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ The rows are an array, so iterating them is `for (const row of rows)` and nothin ## What works today -`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Both module formats, typed separately. +`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. Both module formats, typed separately. Build it with `npm run build`, and run the suite with `npm test`. Nothing is published yet, so `npm i zudb` is not a thing you can type at anybody's terminal, but everything it will do is built and installed on every run of the release workflow. @@ -77,6 +77,31 @@ Between the statement and the loop sit two batches, which is the whole of the bu A statement that has to see every row before it can give one, which is `ORDER BY`, `DISTINCT` and the aggregates, runs whole and is handed over in batches afterwards. The loop is the same either way and `summary.streamed` is what tells them apart. +## Several statements as one unit of work + +One statement is atomic on its own. `conn.transaction()` is how two of them stand or fall together: + +```ts +const tx = await conn.transaction(); +await conn.exec(`INSERT (p:Person {id: $id, name: $name})`, { id: 3n, name: "ida" }); +await conn.exec(`INSERT (p:Person {id: $id, name: $name})`, { id: 4n, name: "eve" }); +await tx.commit(); +``` + +The statements are still the connection's, because the span is the connection's and not a second handle to it. `conn.inTransaction` says whether one is open, and it is the session's own answer rather than a tally kept here, so a caller who would rather write `START TRANSACTION`, `COMMIT` and `ROLLBACK` as statements gets the same answer from it. `{ readOnly: true }` starts a span that refuses the statement that writes. Nesting is refused by the engine, with the engine's own condition. + +`await using tx` rolls back. That is the opposite of what the Python client's `with` block does, and the difference is in the language rather than in the database: a Python context manager is handed the exception unwinding through it and can tell a block that ended well from one that failed, and a JavaScript disposal is told nothing at all. A disposal that committed would commit half the work of a block that threw, which is the one thing a transaction exists to prevent. So the commit is the word the caller writes, and leaving the block without writing it undoes the span: + +```ts +{ + await using tx = await conn.transaction(); + await conn.exec(`INSERT (p:Person {id: 5, name: 'zoe'})`); + await tx.commit(); // without this line the insert is undone +} +``` + +A block that ends well and forgets to commit loses its work, which is a loud kind of wrong and shows up the first time the code runs. The alternative was a block that failed and kept half of what it did, which is a quiet kind and shows up in production. Committing or rolling back twice is refused as a `ZuUsageError` rather than ignored, since the statements after the first end belong to no transaction of yours. Leaving the block of a transaction whose connection has already been closed says nothing, because a closed connection took the unwritten span with it and there is nothing left to undo. + ## Asking for numbers instead of bigints `bigIntMode` says how INT64 is spelled on the way out. It goes on one statement, or on a connection for all of them, and a statement on a connection that named one may still name the other: diff --git a/binding.cjs b/binding.cjs index 9446558..3fb4b67 100644 --- a/binding.cjs +++ b/binding.cjs @@ -701,6 +701,7 @@ if (!nativeBinding) { module.exports = nativeBinding module.exports.Connection = nativeBinding.Connection +module.exports.Transaction = nativeBinding.Transaction module.exports.ZuCursor = nativeBinding.ZuCursor module.exports.ZuDate = nativeBinding.ZuDate module.exports.ZuDuration = nativeBinding.ZuDuration diff --git a/binding.d.cts b/binding.d.cts index 2f0e1a8..b2df584 100644 --- a/binding.d.cts +++ b/binding.d.cts @@ -282,6 +282,21 @@ export interface ZuStatementOptions { readonly signal?: AbortSignal } +/** + * What a transaction takes when it starts. + */ +export interface ZuTransactionOptions { + /** + * Starts it `READ ONLY`, which the engine refuses a write inside of + * at the statement that writes rather than at this call. + * + * Worth asking for on a span that only reads, because saying so is + * how a statement that was not meant to write is stopped by the + * database rather than by review. + */ + readonly readOnly?: boolean +} + /** * What a failed call throws. * @@ -333,6 +348,42 @@ export declare class Connection { get readOnly(): boolean /** Whether the connection is still open. */ get open(): boolean + /** + * Whether an explicit transaction is running on this connection. + * + * True inside a `transaction()` and true after a `START + * TRANSACTION` written by hand, because it is asked of the session + * rather than counted here. A statement written on its own runs in + * a transaction of its own and this stays false for it: what it + * answers is whether a span is open, not whether anything is + * atomic. + */ + get inTransaction(): boolean + /** + * Starts a transaction and hands it back. + * + * It starts here rather than at the first statement inside it, so a + * transaction that cannot start says so at the line that asked. A + * connection is inside one transaction at a time and asking for a + * second while one is open is refused by the engine rather than + * nested, because a transaction inside a transaction is a promise + * this database does not make. + * + * ```js + * await using tx = await conn.transaction() + * await conn.exec('INSERT (a:account {uid: 1, balance: 100})') + * await conn.exec('INSERT (b:account {uid: 2, balance: 0})') + * await tx.commit() + * ``` + * + * The `await using` is the rollback nobody remembers to write. It + * undoes the transaction unless the block committed it, which is + * the opposite of what Python's `with` block does here and is the + * only honest reading in JavaScript: a disposal is not told whether + * the scope it is leaving threw, so a disposal that committed would + * commit half of the work of a block that failed. + */ + transaction(options?: ZuTransactionOptions | null): Promise /** * Runs one statement and gives back its rows. * @@ -379,6 +430,50 @@ export declare class Connection { dispose(): Promise } +/** + * A transaction that has been started and not yet ended. + * + * Take one with `Connection.transaction`. It starts when it is taken, + * so a transaction that cannot start says so at the line that asked, + * and the statements that run inside it are the ones written on the + * connection it came from. + */ +export declare class Transaction { + /** + * Whether this transaction was started `READ ONLY`, which the + * engine refuses a write inside of at the statement that writes. + */ + get readOnly(): boolean + /** + * Whether this transaction has already been committed or rolled + * back. + */ + get done(): boolean + /** + * Ends the transaction and keeps what it wrote. + * + * Doing it twice is refused rather than ignored. A second commit is + * a program that has lost track of where its transaction ends, and + * the statements between the two are in neither of them. + */ + commit(): Promise + /** Ends the transaction and throws away what it wrote. */ + rollback(): Promise + /** + * The undo `await using` calls, which is the intended way to scope + * a transaction. + * + * It rolls back, and it does nothing at all when the transaction + * has already ended, which is what makes a committed block and a + * failed one both leave through here without saying anything. + * + * It is also reachable as `Symbol.asyncDispose`, which is what + * `await using` actually looks for and which [`wire_disposal`] puts + * on every transaction as it is made. + */ + dispose(): Promise +} + /** * One statement, read a batch at a time. * diff --git a/etc/zudb.api.md b/etc/zudb.api.md index 3672031..1cb7684 100644 --- a/etc/zudb.api.md +++ b/etc/zudb.api.md @@ -16,10 +16,12 @@ export class Connection { cursor(statement: string, params?: Record | null, options?: ZuStreamOptions | null): ZuCursor dispose(): Promise exec(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise + get inTransaction(): boolean get open(): boolean get path(): string query>(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise> get readOnly(): boolean + transaction(options?: ZuTransactionOptions | null): Promise } // @public @@ -34,6 +36,15 @@ export interface ConnectOptions { // @public export function isZuError(value: unknown): value is ZuError +// @public +export class Transaction { + commit(): Promise + dispose(): Promise + get done(): boolean + get readOnly(): boolean + rollback(): Promise +} + // @public export function version(): string @@ -252,6 +263,11 @@ export class ZuTimestamp { toTemporal(): ZuPlainDateTime | ZuZonedDateTime } +// @public +export interface ZuTransactionOptions { + readonly readOnly?: boolean +} + // @public export type ZuValue = | null diff --git a/src/conn.rs b/src/conn.rs index cd8afe6..7dab81c 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -36,6 +36,7 @@ use crate::cancel::Watch; use crate::error::{aborted, raise, usage}; use crate::stream::{self, Started, ZuCursor}; use crate::temporal; +use crate::txn::StartTask; use crate::value::{Ints, Shape, Spelling, from_js, to_js}; /// What a connection can be opened with. @@ -108,6 +109,18 @@ pub struct Connection { /// to the session and is the same one for the connection's whole /// life, so once is enough. interrupt: Interrupt, + /// Whether an explicit transaction is running on this connection, + /// kept beside the lock for the same reason [`Connection::alive`] + /// is: asking should not queue behind the statement being asked + /// about. + /// + /// Written by every statement that runs, out of the session itself, + /// which is what makes it exact rather than a tally this client + /// keeps. Nothing but a statement can start or end a transaction, so + /// a caller who reads this between two of their own reads the truth, + /// and a caller who reads it from underneath a statement in flight + /// is asking a question that has no answer yet either way. + in_txn: Arc, /// How this connection's statements spell the values they give /// back, unless one of them asks for something else. spelling: Spelling, @@ -218,26 +231,31 @@ impl<'task> ScopedTask<'task> for ConnectTask { interrupt: opened.conn.interrupt(), inner: Arc::new(Mutex::new(Some(opened.conn))), alive: Arc::new(AtomicBool::new(true)), + in_txn: Arc::new(AtomicBool::new(false)), spelling: opened.spelling, path: opened.path, read_only: opened.read_only, } .into_instance(env)?; - wire_disposal(env, &mut instance)?; + wire_disposal(env, &mut instance, "dispose")?; Ok(instance) } } -/// Puts `dispose` on the connection under `Symbol.asyncDispose`, which -/// is the key `await using` looks up and the one thing about this class +/// Puts a method on an instance under `Symbol.asyncDispose`, which is +/// the key `await using` looks up and the one thing about these classes /// that cannot be spelled in the attribute that declares the rest of -/// it: a method's name there is a string, and this key is a symbol. +/// them: a method's name there is a string, and this key is a symbol. /// /// On the instance rather than on the prototype, because reaching the /// prototype from here means calling `Object.getPrototypeOf` through /// three more layers of FFI for a property that is looked up once per -/// connection either way. -fn wire_disposal(env: &Env, instance: &mut ClassInstance<'_, Connection>) -> Result<()> { +/// object either way. +pub(crate) fn wire_disposal( + env: &Env, + instance: &mut ClassInstance<'_, T>, + method: &str, +) -> Result<()> { // `Symbol` is a function, and the well-known symbols hang off it as // properties of that function. let symbols: Function<'_, (), Unknown<'_>> = env.get_global()?.get_named_property("Symbol")?; @@ -247,7 +265,7 @@ fn wire_disposal(env: &Env, instance: &mut ClassInstance<'_, Connection>) -> Res if key.get_type()? != ValueType::Symbol { return Ok(()); } - let dispose: Unknown<'_> = instance.get_named_property("dispose")?; + let dispose: Unknown<'_> = instance.get_named_property(method)?; instance.set_property(key, dispose) } @@ -300,6 +318,68 @@ impl Connection { self.alive.load(Ordering::Acquire) } + /// Whether an explicit transaction is running on this connection. + /// + /// True inside a `transaction()` and true after a `START + /// TRANSACTION` written by hand, because it is asked of the session + /// rather than counted here. A statement written on its own runs in + /// a transaction of its own and this stays false for it: what it + /// answers is whether a span is open, not whether anything is + /// atomic. + #[napi(getter)] + pub fn in_transaction(&self) -> bool { + self.in_txn.load(Ordering::Acquire) + } + + /// Starts a transaction and hands it back. + /// + /// It starts here rather than at the first statement inside it, so a + /// transaction that cannot start says so at the line that asked. A + /// connection is inside one transaction at a time and asking for a + /// second while one is open is refused by the engine rather than + /// nested, because a transaction inside a transaction is a promise + /// this database does not make. + /// + /// ```js + /// await using tx = await conn.transaction() + /// await conn.exec('INSERT (a:account {uid: 1, balance: 100})') + /// await conn.exec('INSERT (b:account {uid: 2, balance: 0})') + /// await tx.commit() + /// ``` + /// + /// The `await using` is the rollback nobody remembers to write. It + /// undoes the transaction unless the block committed it, which is + /// the opposite of what Python's `with` block does here and is the + /// only honest reading in JavaScript: a disposal is not told whether + /// the scope it is leaving threw, so a disposal that committed would + /// commit half of the work of a block that failed. + #[napi( + ts_args_type = "options?: ZuTransactionOptions | null", + ts_return_type = "Promise" + )] + pub fn transaction(&self, options: Option>) -> AsyncTask { + let read_only = match flag(options.as_ref(), "readOnly") { + Ok(read_only) => read_only.unwrap_or(false), + Err(message) => return AsyncTask::new(self.start(false, Some(message))), + }; + let refused = match self.alive.load(Ordering::Acquire) { + true => None, + false => Some(CLOSED.to_string()), + }; + AsyncTask::new(self.start(read_only, refused)) + } + + /// The task that starts one, whether or not it is going to work. + fn start(&self, read_only: bool, refused: Option) -> StartTask { + StartTask::new( + Arc::clone(&self.inner), + Arc::clone(&self.alive), + Arc::clone(&self.in_txn), + read_only, + refused, + ) + } + /// Runs one statement and gives back its rows. /// /// The parameters are named, never positional, because zuQL names @@ -393,6 +473,7 @@ impl Connection { Started { inner: Arc::clone(&self.inner), alive: Arc::clone(&self.alive), + in_txn: Arc::clone(&self.in_txn), statement, params, spelling, @@ -449,6 +530,7 @@ impl Connection { QueryTask { inner: Arc::clone(&self.inner), alive: Arc::clone(&self.alive), + in_txn: Arc::clone(&self.in_txn), statement, params, spelling, @@ -543,10 +625,54 @@ impl From for Failure { } } +/// Takes the connection for the length of one statement, and says why +/// it could not be had when it could not. +/// +/// Every statement that runs on the threadpool goes through here, which +/// is what makes the three things it decides be decided once: what an +/// empty slot means, what a poisoned lock means, and whether the +/// connection is inside a transaction now that the statement has run. +/// That last one is written here rather than by each caller because a +/// caller that forgot would leave `inTransaction` describing some +/// earlier statement. +pub(crate) fn with( + inner: &Mutex>, + alive: &AtomicBool, + in_txn: &AtomicBool, + run: impl FnOnce(&mut zudb::Connection) -> std::result::Result, +) -> std::result::Result { + // A thread that panicked inside the engine left the connection in a + // state nothing here can vouch for, so this says so rather than + // carrying on with it. + let mut held = inner + .lock() + .map_err(|_| Failure::Usage(POISONED.to_string()))?; + // Closed between the call and the thread picking it up, which is a + // race the caller cannot see and this has to check anyway. Or lent + // to a stream that has not finished, which is the same empty slot + // and a different thing to say about it. + let Some(conn) = held.as_mut() else { + return Err(Failure::Usage( + match alive.load(Ordering::Acquire) { + true => STREAMING, + false => CLOSED, + } + .to_string(), + )); + }; + let answered = run(&mut *conn); + in_txn.store(conn.session_mut().in_transaction(), Ordering::Release); + answered +} + /// What a closed connection says, wherever it is noticed. pub(crate) const CLOSED: &str = "the connection is closed, so there is nothing left to run a statement on"; +/// What a connection says when the thread that last held it panicked. +pub(crate) const POISONED: &str = + "the connection was left in an unknown state by a statement that panicked"; + /// What a connection a stream is still reading says to the next /// statement. /// @@ -640,6 +766,31 @@ fn batch_rows(options: Option<&Object<'_>>) -> std::result::Result, } } +/// Reads an option that has to be a boolean, and says what arrived +/// instead. +/// +/// Absent and `null` are the same as unwritten, which is what makes +/// `{ readOnly: wanted }` work for a caller whose `wanted` came out of +/// a config file. Anything else is refused rather than made truthy: a +/// `readOnly: 'false'` that opened a writing transaction would be a +/// string nobody meant read as the opposite of itself. +fn flag(options: Option<&Object<'_>>, name: &str) -> std::result::Result, String> { + let Some(options) = options else { + return Ok(None); + }; + let value: Unknown<'_> = options.get_named_property(name).map_err(|err| err.reason)?; + match value.get_type().map_err(|err| err.reason)? { + ValueType::Undefined | ValueType::Null => Ok(None), + ValueType::Boolean => bool::from_unknown(value) + .map(Some) + .map_err(|err| err.reason), + other => Err(format!( + "{name} is {}, and it is either true or false", + worded(other) + )), + } +} + /// Reads `options.bigIntMode`, which is how this statement spells the /// INT64s it gives back. /// @@ -753,6 +904,9 @@ pub struct QueryTask { /// Whether the connection is still open, which is what tells an /// empty slot that was closed from one a stream is holding. alive: Arc, + /// Where this statement writes whether the connection is inside a + /// transaction now that it has run. + in_txn: Arc, statement: String, params: Vec<(String, Value)>, /// How this statement spells the values it gives back. @@ -773,60 +927,40 @@ impl QueryTask { if let Some(message) = self.refused.take() { return Err(Failure::Usage(message)); } - let mut held = match self.inner.lock() { - Ok(held) => held, - // A thread that panicked inside the engine left the - // connection in a state nothing here can vouch for, so this - // says so rather than carrying on with it. - Err(_) => { - return Err(Failure::Usage( - "the connection was left in an unknown state by a statement that panicked" - .to_string(), - )); + let (statement, params, spelling, watch) = + (&self.statement, &self.params, self.spelling, &self.watch); + with(&self.inner, &self.alive, &self.in_txn, move |conn| { + // From here the connection is this statement's, so this is + // where a signal can start stopping it and where it stops + // being able to. A signal that fired first ends the + // statement without the engine ever seeing it, which is the + // whole point of asking. + if let Some(watch) = watch + && !watch.enter() + { + watch.leave(); + return Err(Failure::Aborted); } - }; - // Closed between the call and the thread picking it up, which is - // a race the caller cannot see and this has to check anyway. Or - // lent to a stream that has not finished, which is the same - // empty slot and a different thing to say about it. - let Some(conn) = held.as_mut() else { - return Err(Failure::Usage( - match self.alive.load(Ordering::Acquire) { - true => STREAMING, - false => CLOSED, + let params: Vec<(&str, Value)> = params + .iter() + .map(|(name, value)| (name.as_str(), value.clone())) + .collect(); + let shape = Shape::of(conn.session_mut().catalog(), spelling); + let result = conn.query_with(statement, ¶ms); + if let Some(watch) = watch { + watch.leave(); + // An interrupt is the engine's answer to somebody having + // asked, and the only somebody here is the caller's + // signal. Reported as an abort rather than as the engine + // condition, because a caller who wrote `catch` around a + // timeout wants their own reason back and not a + // GQLSTATUS. + if watch.asked() && matches!(result, Err(ZuError::Interrupted)) { + return Err(Failure::Aborted); } - .to_string(), - )); - }; - // From here the connection is this statement's, so this is where - // a signal can start stopping it and where it stops being able - // to. A signal that fired first ends the statement without the - // engine ever seeing it, which is the whole point of asking. - if let Some(watch) = &self.watch - && !watch.enter() - { - watch.leave(); - return Err(Failure::Aborted); - } - let params: Vec<(&str, Value)> = self - .params - .iter() - .map(|(name, value)| (name.as_str(), value.clone())) - .collect(); - let shape = Shape::of(conn.session_mut().catalog(), self.spelling); - let result = conn.query_with(&self.statement, ¶ms); - if let Some(watch) = &self.watch { - watch.leave(); - // An interrupt is the engine's answer to somebody having - // asked, and the only somebody here is the caller's signal. - // Reported as an abort rather than as the engine condition, - // because a caller who wrote `catch` around a timeout wants - // their own reason back and not a GQLSTATUS. - if watch.asked() && matches!(result, Err(ZuError::Interrupted)) { - return Err(Failure::Aborted); } - } - Ok((result?, shape)) + Ok((result?, shape)) + }) } /// The exception this rejects the caller's promise with. diff --git a/src/lib.rs b/src/lib.rs index ca1ac23..213337f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -21,6 +21,7 @@ mod conn; mod error; mod stream; mod temporal; +mod txn; mod value; /// The version of the client. diff --git a/src/stream.rs b/src/stream.rs index ebff710..a311d3a 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -268,6 +268,11 @@ impl Drop for Borrowed<'_> { pub struct Started { pub inner: Arc>>, pub alive: Arc, + /// Where this statement writes whether the connection is inside a + /// transaction now that it has run, which a stream does the same way + /// every other statement does even though almost no stream is one of + /// the three words that change it. + pub in_txn: Arc, pub statement: String, pub params: Vec<(String, Value)>, pub spelling: Spelling, @@ -701,6 +706,10 @@ impl Started { None => conn.query_stream(&self.statement, ¶ms, &mut sink), }; + self.in_txn.store( + borrowed.conn().session_mut().in_transaction(), + Ordering::Release, + ); if let Some(guard) = &self.guard { guard.leave(); // An interrupt is the engine's answer to somebody having diff --git a/src/txn.rs b/src/txn.rs new file mode 100644 index 0000000..5d98a03 --- /dev/null +++ b/src/txn.rs @@ -0,0 +1,264 @@ +//! Several statements as one unit of work. +//! +//! A statement written on its own already runs in a transaction of its +//! own, so this is not what makes a write atomic. What it holds is the +//! span: two statements are one unit, and the file keeps the state they +//! started from until one word or the other ends them. +//! +//! ```js +//! await using tx = await conn.transaction() +//! await conn.exec('INSERT (a:account {uid: 1, balance: 100})') +//! await conn.exec('INSERT (b:account {uid: 2, balance: 0})') +//! await tx.commit() +//! ``` +//! +//! The `await using` is the whole reason this is a class rather than a +//! pair of methods, and what it adds is the rollback nobody remembers to +//! write: a transaction that leaves its scope without having been +//! committed is undone, whether it was a `throw`, a `return` out of the +//! middle, or a branch that forgot. +//! +//! That is the opposite of what the Python client's `with` block does, +//! and the difference is in the language rather than in the database. A +//! Python context manager is handed the exception that is unwinding +//! through it, so it can commit when the block ended well and roll back +//! when it did not. A JavaScript disposal is told nothing at all. So a +//! disposal that committed would commit half the work of a block that +//! failed, which is the one thing a transaction exists to prevent, and +//! the commit has to be the word the caller writes. +//! +//! The three statements underneath are `START TRANSACTION`, `COMMIT` and +//! `ROLLBACK`, and a caller who would rather write them can, on this +//! connection, today. What this adds is the undo, and a name for the +//! state, so a program can ask whether it is inside one rather than +//! remembering. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use napi::bindgen_prelude::*; +use napi::{Env, ScopedTask}; +use napi_derive::napi; + +use crate::conn::{Failure, failed, wire_disposal, with}; + +/// A transaction that has been started and not yet ended. +/// +/// Take one with `Connection.transaction`. It starts when it is taken, +/// so a transaction that cannot start says so at the line that asked, +/// and the statements that run inside it are the ones written on the +/// connection it came from. +#[napi] +pub struct Transaction { + /// The same connection the statements inside the transaction run on, + /// held by the same two handles every other statement uses rather + /// than by a reference to the JavaScript object, so that a + /// transaction whose `Connection` was collected still has something + /// to commit. + inner: Arc>>, + alive: Arc, + in_txn: Arc, + read_only: bool, + /// Whether a `COMMIT` or a `ROLLBACK` has already run here. Shared + /// with the tasks, because the word that ends a transaction is a + /// statement and every statement here happens off the event loop. + done: Arc, +} + +#[napi] +impl Transaction { + /// Whether this transaction was started `READ ONLY`, which the + /// engine refuses a write inside of at the statement that writes. + #[napi(getter)] + pub fn read_only(&self) -> bool { + self.read_only + } + + /// Whether this transaction has already been committed or rolled + /// back. + #[napi(getter)] + pub fn done(&self) -> bool { + self.done.load(Ordering::Acquire) + } + + /// Ends the transaction and keeps what it wrote. + /// + /// Doing it twice is refused rather than ignored. A second commit is + /// a program that has lost track of where its transaction ends, and + /// the statements between the two are in neither of them. + #[napi(ts_return_type = "Promise")] + pub fn commit(&self) -> AsyncTask { + self.end(Word::Commit) + } + + /// Ends the transaction and throws away what it wrote. + #[napi(ts_return_type = "Promise")] + pub fn rollback(&self) -> AsyncTask { + self.end(Word::Rollback) + } + + /// The undo `await using` calls, which is the intended way to scope + /// a transaction. + /// + /// It rolls back, and it does nothing at all when the transaction + /// has already ended, which is what makes a committed block and a + /// failed one both leave through here without saying anything. + /// + /// It is also reachable as `Symbol.asyncDispose`, which is what + /// `await using` actually looks for and which [`wire_disposal`] puts + /// on every transaction as it is made. + #[napi(ts_return_type = "Promise")] + pub fn dispose(&self) -> AsyncTask { + self.end(Word::Dispose) + } + + fn end(&self, word: Word) -> AsyncTask { + AsyncTask::new(EndTask { + inner: Arc::clone(&self.inner), + alive: Arc::clone(&self.alive), + in_txn: Arc::clone(&self.in_txn), + done: Arc::clone(&self.done), + word, + }) + } +} + +/// What a transaction is being ended with. +enum Word { + Commit, + Rollback, + /// A rollback that has nothing to say about a transaction which has + /// already ended, because leaving the scope of a committed + /// transaction is the ordinary way one ends. + Dispose, +} + +impl Word { + fn statement(&self) -> &'static str { + match self { + Word::Commit => "COMMIT", + Word::Rollback | Word::Dispose => "ROLLBACK", + } + } +} + +/// Starting one, which is the statement that starts one. +pub struct StartTask { + inner: Arc>>, + alive: Arc, + in_txn: Arc, + read_only: bool, + /// Why this is not going to run, when it is not. + refused: Option, +} + +impl StartTask { + pub(crate) fn new( + inner: Arc>>, + alive: Arc, + in_txn: Arc, + read_only: bool, + refused: Option, + ) -> StartTask { + StartTask { + inner, + alive, + in_txn, + read_only, + refused, + } + } +} + +impl<'task> ScopedTask<'task> for StartTask { + type Output = std::result::Result<(), Failure>; + type JsValue = ClassInstance<'task, Transaction>; + + fn compute(&mut self) -> Result { + if let Some(message) = self.refused.take() { + return Ok(Err(Failure::Usage(message))); + } + // `READ ONLY` is the engine's own spelling and it is enforced at + // the statement that writes rather than here, which is why this + // is two statements and not two code paths. + let statement = match self.read_only { + true => "START TRANSACTION READ ONLY", + false => "START TRANSACTION", + }; + Ok(with(&self.inner, &self.alive, &self.in_txn, |conn| { + conn.query(statement).map(|_| ()).map_err(Failure::from) + })) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + output.map_err(|failure| failed(env, failure, None))?; + let mut instance = Transaction { + inner: Arc::clone(&self.inner), + alive: Arc::clone(&self.alive), + in_txn: Arc::clone(&self.in_txn), + read_only: self.read_only, + done: Arc::new(AtomicBool::new(false)), + } + .into_instance(env)?; + wire_disposal(env, &mut instance, "dispose")?; + Ok(instance) + } +} + +/// Ending one, with whichever of the two words was asked for. +pub struct EndTask { + inner: Arc>>, + alive: Arc, + in_txn: Arc, + done: Arc, + word: Word, +} + +impl<'task> ScopedTask<'task> for EndTask { + type Output = std::result::Result<(), Failure>; + type JsValue = (); + + fn compute(&mut self) -> Result { + // Claimed before the statement runs rather than after it, so + // that two commits issued together cannot both find the + // transaction open and both run. The claim is given back below + // when the statement did not go through. + if self + .done + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return Ok(match self.word { + Word::Dispose => Ok(()), + _ => Err(Failure::Usage(ENDED.to_string())), + }); + } + // A connection closed while a transaction was open took the + // transaction with it, unwritten, so leaving the scope of one + // has nothing left to undo. Only for a disposal: a caller who + // wrote `rollback()` by hand is asking a question and is owed + // the answer that the connection is gone. + if matches!(self.word, Word::Dispose) && !self.alive.load(Ordering::Acquire) { + return Ok(Ok(())); + } + let statement = self.word.statement(); + let answered = with(&self.inner, &self.alive, &self.in_txn, |conn| { + conn.query(statement).map(|_| ()).map_err(Failure::from) + }); + if answered.is_err() { + // Given back, so a commit the engine refused leaves a + // transaction the caller can still roll back rather than one + // that is neither open nor ended. + self.done.store(false, Ordering::Release); + } + Ok(answered) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + output.map_err(|failure| failed(env, failure, None)) + } +} + +/// What a second `commit()` or `rollback()` says. +const ENDED: &str = "this transaction has already ended, and the statements after it \ + belong to no transaction of yours"; diff --git a/test/exports.test.mjs b/test/exports.test.mjs index 9160baf..5d786c3 100644 --- a/test/exports.test.mjs +++ b/test/exports.test.mjs @@ -22,6 +22,7 @@ const SURFACE = [ 'abiVersion', 'isZuError', 'Connection', + 'Transaction', 'ZuStream', 'ZuCursor', 'ZuDate', diff --git a/test/misuse.test.mjs b/test/misuse.test.mjs index 8bd297b..002a406 100644 --- a/test/misuse.test.mjs +++ b/test/misuse.test.mjs @@ -261,12 +261,6 @@ const MISUSES = [ name: 'ZuUsageError', says: ["carries no value for column 'name'", 'every column of a new row has to hold one'], }, - { - what: 'reads a property the table does not have', - run: ({ conn }) => conn.query('MATCH (p:person) RETURN p.nope AS x'), - name: 'ZuUsageError', - says: ["unknown property 'nope'"], - }, ] for (const misuse of MISUSES) { @@ -445,6 +439,17 @@ test('the programs that look like misuse and are not', async (t) => { // is deleted. assert.deepEqual(await conn.query('MATCH (p:nobody) RETURN p.id AS id'), []) + // A property the table does not have reads as null rather than + // failing, which is ISO 20.11 and is what an edge property has always + // answered here. It is the one case in this test that costs + // something: a misspelled property is null for every row instead of a + // refusal, and what catches that back is a declared property list the + // query can be held to rather than a client that guesses. + assert.deepEqual(await conn.query('MATCH (p:person) RETURN p.nope AS x'), [ + { x: null }, + { x: null }, + ]) + // A stream made and never read has not started, so the statement // after it runs rather than being told the connection is busy. This // is the reason a stream starts at its first read. diff --git a/test/transaction.test.mjs b/test/transaction.test.mjs new file mode 100644 index 0000000..65b93b5 --- /dev/null +++ b/test/transaction.test.mjs @@ -0,0 +1,276 @@ +// Several statements as one unit of work. +// +// What is being asserted here is not that a write is atomic, since one +// statement is atomic on its own. It is the span: that two statements +// stand or fall together, that the word which ends them is the caller's, +// and that a span nobody ended is undone rather than kept. +// +// The last of those is the difference from the Python client and it is +// the one worth reading the tests for. A JavaScript disposal is not told +// whether the block it is leaving threw, so `await using` here rolls +// back and the commit is written out. A block that ends well and forgot +// to commit loses its work, which is a loud kind of wrong, and the +// alternative was a block that failed and committed half of it, which is +// a quiet kind. + +import assert from 'node:assert/strict' +import test from 'node:test' + +import { fresh, isZuError, twoPeople } from './helper.mjs' + +const COUNT = 'MATCH (p:person) RETURN count(*) AS n' + +// How many people are in the database, which is what every one of these +// asks after the fact. +async function people(conn) { + const rows = await conn.query(COUNT) + return Number(rows[0].n) +} + +test('two statements committed together are both there', async (t) => { + const { conn } = await twoPeople(t) + + const tx = await conn.transaction() + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + await conn.exec("INSERT (p:person {id: 4, name: 'eve'})") + await tx.commit() + + assert.equal(await people(conn), 4) + assert.equal(tx.done, true) +}) + +test('a rollback leaves the database as it was', async (t) => { + const { conn } = await twoPeople(t) + + const tx = await conn.transaction() + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + // Seen from inside, because a transaction is not a queue: the + // statements have run and the span is what has not ended. + assert.equal(await people(conn), 3) + await tx.rollback() + + assert.equal(await people(conn), 2) + assert.equal(tx.done, true) +}) + +test('await using rolls back a transaction nobody committed', async (t) => { + const { conn } = await twoPeople(t) + + { + await using tx = await conn.transaction() + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + assert.equal(tx.done, false) + } + + assert.equal(await people(conn), 2) +}) + +test('await using keeps what the block committed', async (t) => { + const { conn } = await twoPeople(t) + + { + await using tx = await conn.transaction() + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + await tx.commit() + } + + // The disposal ran on a transaction that had already ended, and had + // nothing to say about it. A second ROLLBACK there would have been an + // error out of a block that did everything right. + assert.equal(await people(conn), 3) +}) + +test('a throw inside the block rolls back and still throws', async (t) => { + const { conn } = await twoPeople(t) + + await assert.rejects( + async () => { + await using tx = await conn.transaction() + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + assert.equal(tx.done, false) + throw new Error('the caller changed their mind') + }, + { message: 'the caller changed their mind' }, + ) + + assert.equal(await people(conn), 2) +}) + +test('the connection says whether it is inside one', async (t) => { + const { conn } = await twoPeople(t) + + assert.equal(conn.inTransaction, false) + const tx = await conn.transaction() + assert.equal(conn.inTransaction, true) + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + // Still true after a statement, which is the thing that would have + // gone wrong if this were a tally kept here rather than the session's + // own answer. + assert.equal(conn.inTransaction, true) + await tx.commit() + assert.equal(conn.inTransaction, false) +}) + +test('a transaction started by hand is one the connection knows about', async (t) => { + const { conn } = await twoPeople(t) + + // The three words are statements, and a caller who would rather write + // them is running the same thing this class runs. + await conn.exec('START TRANSACTION') + assert.equal(conn.inTransaction, true) + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + await conn.exec('ROLLBACK') + + assert.equal(conn.inTransaction, false) + assert.equal(await people(conn), 2) +}) + +test('a statement on its own is not inside a transaction of anybody else', async (t) => { + const { conn } = await twoPeople(t) + + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + + // Its own write was atomic and its own span is over. What + // `inTransaction` answers is whether a span is open, and none is. + assert.equal(conn.inTransaction, false) + assert.equal(await people(conn), 3) +}) + +test('committing twice is refused rather than ignored', async (t) => { + const { conn } = await twoPeople(t) + + const tx = await conn.transaction() + await tx.commit() + + await assert.rejects( + () => tx.commit(), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /already ended/) + // A mistake this client caught, so there is no GQLSTATUS to + // branch on and a caller mapping codes has to tell a missing one + // from an unknown one. + assert.equal(err.code, undefined) + return true + }, + ) +}) + +test('rolling back a committed transaction is refused too', async (t) => { + const { conn } = await twoPeople(t) + + const tx = await conn.transaction() + await tx.commit() + + await assert.rejects(() => tx.rollback(), (err) => isZuError(err, 'ZuUsageError')) +}) + +test('a read only transaction refuses the statement that writes', async (t) => { + const { conn } = await twoPeople(t) + + const tx = await conn.transaction({ readOnly: true }) + assert.equal(tx.readOnly, true) + // Reading is what it is for, and reading works. + assert.equal(await people(conn), 2) + + await assert.rejects( + () => conn.exec("INSERT (p:person {id: 3, name: 'ida'})"), + (err) => { + assert.ok(isZuError(err, 'ZuTransactionError'), `${err.name}: ${err.message}`) + return true + }, + ) + + await tx.rollback() + assert.equal(await people(conn), 2) +}) + +test('a transaction inside a transaction is refused', async (t) => { + const { conn } = await twoPeople(t) + + const tx = await conn.transaction() + await assert.rejects( + () => conn.transaction(), + (err) => { + // The engine's own condition, not this client's, because nesting + // is a thing the database decides and a client that guessed would + // be a second rule to keep in step. + assert.ok(err.code, `${err.name}: ${err.message}`) + return true + }, + ) + await tx.rollback() +}) + +test('a transaction on a closed connection is refused as a rejection', async (t) => { + const { conn } = await fresh(t) + conn.close() + + await assert.rejects( + () => conn.transaction(), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /the connection is closed/) + return true + }, + ) +}) + +test('leaving the block of a transaction whose connection is gone says nothing', async (t) => { + const { conn } = await twoPeople(t) + + // A connection closed underneath an open transaction took the + // transaction with it, unwritten, so there is nothing left to undo + // and the disposal has nothing to report. + { + await using tx = await conn.transaction() + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + assert.equal(tx.done, false) + conn.close() + } + + assert.equal(conn.open, false) +}) + +test('an option that is not a boolean is refused, and says what arrived', async (t) => { + const { conn } = await twoPeople(t) + + await assert.rejects( + () => conn.transaction({ readOnly: 'yes' }), + (err) => { + assert.ok(isZuError(err, 'ZuUsageError'), err.message) + assert.match(err.message, /readOnly is a String/) + return true + }, + ) + + // Refused before anything started, so the connection is exactly where + // it was and the next statement runs. + assert.equal(conn.inTransaction, false) + assert.equal(await people(conn), 2) +}) + +test('an absent option is the same as an unwritten one', async (t) => { + const { conn } = await twoPeople(t) + + const tx = await conn.transaction({ readOnly: undefined }) + assert.equal(tx.readOnly, false) + await conn.exec("INSERT (p:person {id: 3, name: 'ida'})") + await tx.commit() + + assert.equal(await people(conn), 3) +}) + +test('a rolled back transaction leaves the connection ready for the next one', async (t) => { + const { conn } = await twoPeople(t) + + for (const keep of [false, true, false]) { + const tx = await conn.transaction() + await conn.exec("INSERT (p:person {id: 9, name: 'ida'})") + if (keep) await tx.commit() + else await tx.rollback() + } + + assert.equal(await people(conn), 3) + assert.equal(conn.inTransaction, false) +}) diff --git a/test/types/cjs.cts b/test/types/cjs.cts index 572450f..02eba05 100644 --- a/test/types/cjs.cts +++ b/test/types/cjs.cts @@ -2,7 +2,14 @@ // resolution through a different condition to a different file, and so // is worth compiling separately rather than assuming. -import { connect, isZuError, ZuTimestamp, type ZuParam, type ZuStream } from 'zudb' +import { + connect, + isZuError, + ZuTimestamp, + type ZuParam, + type ZuStream, + type ZuTransactionOptions, +} from 'zudb' export async function total(path: string): Promise { // The mode is on the statement here, so the rows it gives back are @@ -44,6 +51,25 @@ export async function moment(path: string, at: bigint): Promise { } } +export async function span(path: string, options: ZuTransactionOptions): Promise { + // The options are an object of their own, so a caller can build one + // and pass it around. A `try` and a `finally` is the other spelling + // of the block below, for a program that cannot write `await using`, + // and the rollback in it is the same word the disposal would run. + const conn = await connect(path) + const tx = await conn.transaction(options) + try { + const rows = await conn.query<{ n: number }>('MATCH (p:person) RETURN count(*) AS n', null, { + bigIntMode: 'number', + }) + await tx.commit() + return rows[0]?.n ?? 0 + } finally { + if (!tx.done) await tx.rollback() + conn.close() + } +} + export async function ids(path: string): Promise { const conn = await connect(path) const stream: ZuStream<{ id: bigint }> = conn.stream('MATCH (p:person) RETURN p.id AS id') diff --git a/test/types/esm.mts b/test/types/esm.mts index f8a1d93..5447a0f 100644 --- a/test/types/esm.mts +++ b/test/types/esm.mts @@ -14,6 +14,7 @@ import { type ZuStream, type ZuSummary, type ZuValue, + type Transaction, } from 'zudb' export async function people(path: string, name: string): Promise { @@ -92,6 +93,21 @@ export async function serialized(path: string, mode: ZuBigIntMode): Promise ({ ...row, id: row.id + 1 }))) } +export async function moved(path: string, from: bigint, to: bigint): Promise { + await using conn = await connect(path) + + // `await using` on a transaction needs the same thing it needs on a + // connection, and the transaction's disposal is declared in the same + // place and for the same reason. + await using tx: Transaction = await conn.transaction() + const open: boolean = conn.inTransaction + await conn.exec('MATCH (p:person) WHERE p.id = $from SET p.id = $to', { from, to }) + await tx.commit() + + // Both are readable after the fact, and both are plain booleans. + return open && tx.done && !tx.readOnly +} + export function retryable(caught: unknown): boolean { // `catch` gives `unknown`, and the guard is what narrows it. Reading // `caught.retryable` without it does not compile. diff --git a/types/header.d.ts b/types/header.d.ts index 40350f0..b432221 100644 --- a/types/header.d.ts +++ b/types/header.d.ts @@ -282,6 +282,21 @@ export interface ZuStatementOptions { readonly signal?: AbortSignal } +/** + * What a transaction takes when it starts. + */ +export interface ZuTransactionOptions { + /** + * Starts it `READ ONLY`, which the engine refuses a write inside of + * at the statement that writes rather than at this call. + * + * Worth asking for on a span that only reads, because saying so is + * how a statement that was not meant to write is stopped by the + * database rather than by review. + */ + readonly readOnly?: boolean +} + /** * What a failed call throws. * diff --git a/zudb.cjs b/zudb.cjs index 235be94..cba3074 100644 --- a/zudb.cjs +++ b/zudb.cjs @@ -163,6 +163,7 @@ module.exports = { abiVersion: binding.abiVersion, isZuError, Connection: binding.Connection, + Transaction: binding.Transaction, ZuStream, // The pull underneath a stream, which `conn.cursor(...)` hands back // and almost nobody should be holding. It is here because it is in diff --git a/zudb.d.cts b/zudb.d.cts index f8a688e..1092625 100644 --- a/zudb.d.cts +++ b/zudb.d.cts @@ -31,6 +31,17 @@ declare module './binding.cjs' { options?: ZuStreamOptions | null, ): ZuStream } + + /** + * The disposal of a transaction, declared here for the same reason + * the connection's is: the key is a symbol and the generator writes + * method names as strings. + * + * It rolls back. Leaving the scope of a transaction that was never + * committed undoes it, which is what makes `await using` the safe + * spelling and the commit the word a caller writes. + */ + interface Transaction extends AsyncDisposable {} } /** diff --git a/zudb.mjs b/zudb.mjs index ac80a24..88e4ace 100644 --- a/zudb.mjs +++ b/zudb.mjs @@ -21,6 +21,7 @@ export const version = zudb.version export const abiVersion = zudb.abiVersion export const isZuError = zudb.isZuError export const Connection = zudb.Connection +export const Transaction = zudb.Transaction export const ZuStream = zudb.ZuStream export const ZuCursor = zudb.ZuCursor export const ZuDate = zudb.ZuDate