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