diff --git a/README.md b/README.md index fd769f7..2332034 100644 --- a/README.md +++ b/README.md @@ -35,10 +35,22 @@ The rows are an array, so iterating them is `for (const row of rows)` and nothin ## What works today -`connect`, `query`, `exec`, `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`. Read-only connections, memory and thread limits. The full error surface above. +`connect`, `query`, `exec`, `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`. Read-only connections, memory and thread limits. An `AbortSignal` on any statement. The full error surface above. 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. +## Stopping a statement + +Every statement takes a third argument, and what is in it today is a signal: + +```ts +const rows = await conn.query(statement, params, { signal: AbortSignal.timeout(50) }); +``` + +It is the signal JavaScript already has, so a timeout written like the one above, the signal a framework hands a request handler, and an `AbortSignal.any([...])` composed out of both all work here without anything being adapted. When it fires, the engine's interrupt is raised, the executor notices it at a boundary it was already stopping at, and the statement ends inside a vector of rows rather than at the end of the scan. The connection is left exactly as it was, so the statement after a stopped one runs normally. + +What the promise rejects with is the signal's own reason, which is what `fetch` does: `AbortSignal.timeout(50)` rejects with the runtime's `TimeoutError`, `controller.abort(new RequestGone())` rejects with the `RequestGone` you made, and a bare `controller.abort()` rejects with the runtime's `AbortError`. A signal that has already fired stops the statement before the engine sees it at all. A signal that never fires costs one listener, taken off again when the statement ends, whether it answered, failed or was stopped. + ## Installing, once there is something to install `npm i zudb`, and that is the whole of it. The install downloads one file, runs nothing, and needs no compiler: the root package carries the loader and no binary, each platform has its own package holding exactly one addon, and npm picks the one for the machine out of `optionalDependencies` by its `os`, `cpu` and `libc`. There is no `postinstall`, no `node-gyp`, no `node-pre-gyp` and no fetch from anywhere but the registry, which is what makes the package installable behind a proxy, inside a locked-down CI image, and on a machine with no toolchain on it. @@ -60,7 +72,7 @@ Anything outside that table has no binary and no source build to fall back on, s ## Still to come -`AsyncIterable` and Web Streams over a result, and `AbortSignal` wired to the engine's interrupt. `bigIntMode`. `toTemporal()` and `{ temporal: true }`, for the runtimes where Temporal is unflagged: it reached Stage 4 in March 2026 and is unflagged in Node 26, but Node 24 is still the active LTS and Safari is still behind a flag, which is why the stable types are the four classes above. Dual ESM and CJS, with types first in every export condition. Bun and Deno in CI, and the WASM build for the browser. +`AsyncIterable` and Web Streams over a result. `bigIntMode`. `toTemporal()` and `{ temporal: true }`, for the runtimes where Temporal is unflagged: it reached Stage 4 in March 2026 and is unflagged in Node 26, but Node 24 is still the active LTS and Safari is still behind a flag, which is why the stable types are the four classes above. Dual ESM and CJS, with types first in every export condition. Bun and Deno in CI, and the WASM build for the browser. ## Runtimes diff --git a/bench/query.mjs b/bench/query.mjs index 22a99eb..60ae8e9 100644 --- a/bench/query.mjs +++ b/bench/query.mjs @@ -58,6 +58,10 @@ async function time(run) { return best } +// A signal nobody ever fires, which is the shape almost every signal +// passed to a database has. +const idle = new AbortController() + const cases = [ { name: 'scan, two columns', @@ -84,6 +88,15 @@ const cases = [ per: 'statement', run: () => conn.query('MATCH (p:person) RETURN count(*) AS n'), }, + { + // What watching a signal costs a statement nobody stops, which is + // every statement in a server that passes the request's signal down. + // A listener added and taken off again, once per statement and not + // once per row, so the number to read this against is the one above. + name: 'aggregate, one signal in', + per: 'statement', + run: () => conn.query('MATCH (p:person) RETURN count(*) AS n', null, { signal: idle.signal }), + }, { name: 'aggregate, one parameter in', per: 'statement', diff --git a/index.d.ts b/index.d.ts index b551ad9..0c7fca5 100644 --- a/index.d.ts +++ b/index.d.ts @@ -88,6 +88,29 @@ export interface ZuRows> extends Array { readonly notices: ZuNotice[] } +/** + * What a statement takes beside its parameters. + * + * An object rather than a bare signal, because the options that follow + * it belong in the same place and a third argument that changes meaning + * is one nobody can read at a call site. + */ +export interface ZuStatementOptions { + /** + * Stops the statement when it fires, through the same interrupt a + * shell answers `Ctrl-C` with: the executor notices at the boundary it + * was already stopping at, the statement ends, and the connection is + * exactly as it was. The promise rejects with whatever the signal + * gives as its reason, which is what `fetch` does, so + * `AbortSignal.timeout(50)` rejects with a `TimeoutError` and + * `controller.abort(new MyError())` rejects with `MyError`. + * + * A signal that has already fired stops the statement before the + * engine sees it at all. + */ + readonly signal?: AbortSignal +} + /** * What a failed call throws. * @@ -147,7 +170,7 @@ export declare class Connection { * statement does not use is an error from the engine rather than a * value quietly ignored. */ - query>(statement: string, params?: Record | null): Promise> + query>(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise> /** * Runs one statement for its effect and gives back nothing. * @@ -155,7 +178,7 @@ export declare class Connection { * which is what a schema statement or a write wants: a result nobody * reads still costs a row object per row on the way out. */ - exec(statement: string, params?: Record | null): Promise + exec(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise /** * Closes the connection and releases the database. * diff --git a/src/cancel.rs b/src/cancel.rs new file mode 100644 index 0000000..1e27088 --- /dev/null +++ b/src/cancel.rs @@ -0,0 +1,182 @@ +//! An `AbortSignal`, wired to the statement it is meant to stop. +//! +//! Cancellation in JavaScript has one shape and every library uses it, +//! so a statement takes an `AbortSignal` rather than a handle of its +//! own: the timeout a caller already has from `AbortSignal.timeout`, the +//! signal a web framework hands a request handler, and the one a caller +//! composes with `AbortSignal.any` all work here without anything being +//! adapted. What arrives on the other side is the engine's interrupt, +//! which is one word the executor reads at a boundary it was already +//! stopping at, so a statement that is asked to stop does so within a +//! vector of rows and leaves the connection exactly as it was. +//! +//! Two things make this harder than adding a listener. The interrupt +//! belongs to the connection rather than to the statement, so a stop +//! raised a moment too late would end the next statement instead of the +//! one it was meant for. And a signal can fire before the statement +//! reaches the thread it runs on, which is a stop nobody is there to +//! hear. Both are answered the same way: the listener and the statement +//! share two words and set them in the opposite order, so whichever of +//! them is second sees what the other did. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +use napi::bindgen_prelude::*; +use napi::{Env, ValueType}; +use zudb::Interrupt; + +/// The two words a listener and a statement share. +/// +/// `asked` is set by the listener and read by the statement, `running` +/// the other way about. Both are sequentially consistent, which is what +/// makes the pair work: the listener sets `asked` before it reads +/// `running`, the statement sets `running` before it reads `asked`, and +/// an ordering where neither sees the other is one this ordering +/// forbids. +#[derive(Debug)] +struct Shared { + asked: AtomicBool, + running: AtomicBool, + interrupt: Interrupt, +} + +impl Shared { + /// The listener's half: say that somebody asked, then stop the + /// statement if one is running. + fn ask(&self) { + self.asked.store(true, Ordering::SeqCst); + if self.running.load(Ordering::SeqCst) { + self.interrupt.stop(); + } + } +} + +/// A signal watching one statement. +/// +/// It is made on the thread that owns the runtime, because that is the +/// only thread that may add a listener, and it is taken apart there too. +pub struct Watch { + shared: Arc, + /// Kept so the reason can be read when the statement gives up, and + /// so the listener can be taken off again. + signal: ObjectRef, + listener: FunctionRef<(), ()>, +} + +impl Watch { + /// Watches `signal`, and stops `interrupt` when it fires. + /// + /// A signal that has already fired is still watched rather than + /// refused here, so that one place decides what an aborted statement + /// does and the answer does not depend on how early the caller was. + pub fn new(env: &Env, signal: Object<'_>, interrupt: Interrupt) -> Result { + let shared = Arc::new(Shared { + asked: AtomicBool::new(signal.get_named_property::("aborted")?), + running: AtomicBool::new(false), + interrupt, + }); + + let held = Arc::clone(&shared); + let listener: Function<'_, (), ()> = + env.create_function_from_closure("abort", move |_| { + held.ask(); + Ok(()) + })?; + + let mut once = Object::new(env)?; + once.set("once", true)?; + // `FnArgs` rather than the tuple on its own, because a bare tuple + // is one argument to napi and a JavaScript function called with + // one argument that happens to be three values is a function + // called wrongly. The failure that teaches this is + // `addEventListener` complaining that its arguments were not + // specified while three of them sit in the call. + let add: Function<'_, FnArgs<(Unknown<'_>, Unknown<'_>, Unknown<'_>)>, Unknown<'_>> = + signal.get_named_property("addEventListener")?; + add.apply( + signal, + ( + env.create_string("abort")?.into_unknown(env)?, + listener.into_unknown(env)?, + once.into_unknown(env)?, + ) + .into(), + )?; + + Ok(Watch { + shared, + signal: signal.create_ref()?, + listener: listener.create_ref()?, + }) + } + + /// The statement is about to run. `false` means it must not: the + /// signal fired first. + pub fn enter(&self) -> bool { + self.shared.running.store(true, Ordering::SeqCst); + !self.shared.asked.load(Ordering::SeqCst) + } + + /// The statement is done with the connection. + /// + /// The flag goes down first, so a signal firing now raises nothing, + /// and then the interrupt is put back down, so a stop that landed in + /// the moment between the statement finishing and this call cannot + /// end whatever runs next on the same connection. + pub fn leave(&self) { + self.shared.running.store(false, Ordering::SeqCst); + self.shared.interrupt.clear(); + } + + /// Whether the signal fired at all, which is what tells an interrupt + /// the caller asked for apart from one they did not. + pub fn asked(&self) -> bool { + self.shared.asked.load(Ordering::SeqCst) + } + + /// What the signal gives as its reason, if it gives one. + /// + /// Every runtime sets a reason on a signal that has fired, an + /// `AbortError` when the caller named nothing of their own, so this + /// is almost always something. It is an option rather than a promise + /// of one because a rejection with `undefined` in it is worse than + /// the plain sentence this client would write instead. + pub fn reason<'env>(&self, env: &'env Env) -> Option> { + let signal = self.signal.get_value(env).ok()?; + let reason: Unknown<'env> = signal.get_named_property("reason").ok()?; + match reason.get_type() { + Ok(ValueType::Undefined) | Ok(ValueType::Null) | Err(_) => None, + Ok(_) => Some(reason), + } + } + + /// Takes the listener off the signal and releases both references. + /// + /// A signal outlives the statement it stopped, often by the length + /// of a whole request, and a listener left on one is a statement's + /// worth of memory that never goes away. This is why the statement + /// keeps the function it added rather than adding a fresh closure + /// and hoping `once` covers it: `once` only fires for a signal that + /// aborts, and most of them never do. + pub fn release(self, env: &Env) -> Result<()> { + let signal = self.signal.get_value(env)?; + let listener = self.listener.borrow_back(env)?; + let remove: Function<'_, FnArgs<(Unknown<'_>, Unknown<'_>)>, Unknown<'_>> = + signal.get_named_property("removeEventListener")?; + remove.apply( + signal, + ( + env.create_string("abort")?.into_unknown(env)?, + listener.into_unknown(env)?, + ) + .into(), + )?; + + // The function's reference releases itself when it drops, on + // this thread or through the environment's own collector if it + // ever drops on another. The object's does not, and says so on + // stderr when it is forgotten, so it is released here. + self.signal.unref(env) + } +} diff --git a/src/conn.rs b/src/conn.rs index eced97b..85dc061 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -23,9 +23,10 @@ use napi::bindgen_prelude::*; use napi::{Env, ScopedTask, ValueType}; use napi_derive::napi; use zudb::query::{QueryResult, Value}; -use zudb::{Config, Database, ZuError}; +use zudb::{Config, Database, Interrupt, ZuError}; -use crate::error::{raise, usage}; +use crate::cancel::Watch; +use crate::error::{aborted, raise, usage}; use crate::value::{Names, from_js, to_js}; /// What a connection can be opened with. @@ -65,6 +66,15 @@ pub struct Connection { /// rather than inside it, because asking a connection whether it is /// closed should not queue behind a ten second statement. alive: Arc, + /// The word a statement running on this connection reads at every + /// boundary, taken once when the connection was opened. + /// + /// Kept here rather than asked of the connection when a statement + /// wants one, because asking means taking the lock and the caller + /// asking is on the thread that must never wait: the handle belongs + /// to the session and is the same one for the connection's whole + /// life, so once is enough. + interrupt: Interrupt, path: String, read_only: bool, } @@ -110,6 +120,7 @@ impl<'task> ScopedTask<'task> for ConnectTask { fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { let opened = output.map_err(|err| raise(env, err))?; let mut instance = Connection { + interrupt: opened.conn.interrupt(), inner: Arc::new(Mutex::new(Some(opened.conn))), alive: Arc::new(AtomicBool::new(true)), path: opened.path, @@ -199,7 +210,7 @@ impl Connection { /// value quietly ignored. #[napi( ts_generic_types = "Row = Record", - ts_args_type = "statement: string, params?: Record | null", + ts_args_type = "statement: string, params?: Record | null, options?: ZuStatementOptions | null", ts_return_type = "Promise>" )] pub fn query( @@ -207,8 +218,9 @@ impl Connection { env: &Env, statement: String, params: Option>, + options: Option>, ) -> AsyncTask { - AsyncTask::new(self.task(env, statement, params)) + AsyncTask::new(self.task(env, statement, params, options)) } /// Runs one statement for its effect and gives back nothing. @@ -217,7 +229,7 @@ impl Connection { /// which is what a schema statement or a write wants: a result nobody /// reads still costs a row object per row on the way out. #[napi( - ts_args_type = "statement: string, params?: Record | null", + ts_args_type = "statement: string, params?: Record | null, options?: ZuStatementOptions | null", ts_return_type = "Promise" )] pub fn exec( @@ -225,8 +237,9 @@ impl Connection { env: &Env, statement: String, params: Option>, + options: Option>, ) -> AsyncTask { - AsyncTask::new(ExecTask(self.task(env, statement, params))) + AsyncTask::new(ExecTask(self.task(env, statement, params, options))) } /// The task one statement runs as, whether or not it is going to @@ -237,23 +250,32 @@ impl Connection { /// who wrote `await` or `.catch` has somewhere to catch it. A native /// method that throws for a closed connection and rejects for a /// failed statement is a method every caller has to wrap twice. - fn task(&self, env: &Env, statement: String, params: Option>) -> QueryTask { + fn task( + &self, + env: &Env, + statement: String, + params: Option>, + options: Option>, + ) -> QueryTask { // The parameters are read here rather than on the threadpool // thread, because reading a JavaScript value is something only - // the thread that owns the runtime may do. + // the thread that owns the runtime may do. So is adding the + // listener the signal is watched through. let bound = if self.alive.load(Ordering::Acquire) { bind(env, params) + .and_then(|params| Ok((params, watch(env, options, self.interrupt.clone())?))) } else { Err(CLOSED.to_string()) }; - let (params, refused) = match bound { - Ok(params) => (params, None), - Err(message) => (Vec::new(), Some(message)), + let (params, watch, refused) = match bound { + Ok((params, watch)) => (params, watch, None), + Err(message) => (Vec::new(), None, Some(message)), }; QueryTask { inner: Arc::clone(&self.inner), statement, params, + watch, refused, } } @@ -303,6 +325,8 @@ pub enum Failure { Engine(ZuError), /// This client refused the call before the engine saw it. Usage(String), + /// The caller's signal fired, before the statement or during it. + Aborted, } impl From for Failure { @@ -311,19 +335,51 @@ impl From for Failure { } } -impl Failure { - /// The exception this rejects the caller's promise with. - fn raise(self, env: &Env) -> Error { - match self { - Failure::Engine(err) => raise(env, err), - Failure::Usage(message) => usage(env, message), +/// What a closed connection says, wherever it is noticed. +const CLOSED: &str = "the connection is closed, so there is nothing left to run a statement on"; + +/// What an abort says when the signal that fired named no reason of its +/// own, which is a signal built by hand rather than by a runtime. +const ABORTED: &str = "the statement was stopped by the signal it was given"; + +/// Reads `options.signal` and starts watching it. +/// +/// Absent options and an absent signal are the same thing and are the +/// common case, so both cost one property read and no listener. A +/// `signal` that is not an `AbortSignal` is refused here rather than +/// where the listener fails to be added, because the caller's mistake is +/// the value they passed. +fn watch( + env: &Env, + options: Option>, + interrupt: Interrupt, +) -> std::result::Result, String> { + let Some(options) = options else { + return Ok(None); + }; + let signal: Unknown<'_> = options + .get_named_property("signal") + .map_err(|err| err.reason)?; + match signal.get_type().map_err(|err| err.reason)? { + ValueType::Undefined | ValueType::Null => Ok(None), + ValueType::Object => { + let signal = signal.coerce_to_object().map_err(|err| err.reason)?; + if signal + .get_named_property::>("aborted") + .and_then(|aborted| aborted.get_type()) + .map_err(|err| err.reason)? + != ValueType::Boolean + { + return Err("signal is an object that is not an AbortSignal".to_string()); + } + Watch::new(env, signal, interrupt) + .map(Some) + .map_err(|err| err.reason) } + other => Err(format!("signal is a {other}, which is not an AbortSignal")), } } -/// What a closed connection says, wherever it is noticed. -const CLOSED: &str = "the connection is closed, so there is nothing left to run a statement on"; - /// Reads the parameter object into the values the engine binds. /// /// Every failure comes back as the message to refuse the call with, @@ -352,6 +408,8 @@ pub struct QueryTask { inner: Arc>>, statement: String, params: Vec<(String, Value)>, + /// The signal watching this statement, when the caller gave one. + watch: Option, /// Why this statement is not going to run, when it is not. refused: Option, } @@ -383,14 +441,62 @@ impl QueryTask { let Some(conn) = held.as_mut() else { return Err(Failure::Usage(CLOSED.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 names = Names::of(conn.session_mut().catalog()); - let result = conn.query_with(&self.statement, ¶ms)?; - Ok((result, names)) + 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?, names)) + } + + /// The exception this rejects the caller's promise with. + /// + /// An abort rejects with the signal's own reason, which is what + /// `fetch` does: a caller who wrote `AbortSignal.timeout(50)` gets + /// back the `TimeoutError` that signal carries, and one who wrote + /// `controller.abort(new MyError())` gets their own object rather + /// than a description of it. + fn failed(&self, env: &Env, failure: Failure) -> Error { + match failure { + Failure::Engine(err) => raise(env, err), + Failure::Usage(message) => usage(env, message), + Failure::Aborted => self + .watch + .as_ref() + .and_then(|watch| watch.reason(env)) + .map_or_else(|| aborted(env, ABORTED), Error::from), + } + } + + /// Takes the listener back off the signal, whatever happened. + fn release(&mut self, env: &Env) -> Result<()> { + match self.watch.take() { + Some(watch) => watch.release(env), + None => Ok(()), + } } } @@ -403,9 +509,13 @@ impl<'task> ScopedTask<'task> for QueryTask { } fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { - let (result, names) = output.map_err(|failure| failure.raise(env))?; + let (result, names) = output.map_err(|failure| self.failed(env, failure))?; rows(env, &result, &names) } + + fn finally(mut self, env: Env) -> Result<()> { + self.release(&env) + } } /// The rows, as an array of objects keyed by column name. @@ -481,7 +591,11 @@ impl<'task> ScopedTask<'task> for ExecTask { } fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { - output.map_err(|failure| failure.raise(env)) + output.map_err(|failure| self.0.failed(env, failure)) + } + + fn finally(mut self, env: Env) -> Result<()> { + self.0.release(&env) } } diff --git a/src/error.rs b/src/error.rs index 49bd126..611f3fe 100644 --- a/src/error.rs +++ b/src/error.rs @@ -128,6 +128,26 @@ fn usage_object<'env>(env: &'env Env, message: &str) -> Result> { Ok(object) } +/// A statement stopped by a signal that named no reason of its own. +/// +/// Named `AbortError` rather than anything of this client's, because +/// that is the name every runtime gives the reason it makes for a signal +/// nobody gave one to, and a caller testing `err.name === 'AbortError'` +/// is testing the one thing that is true of both. +pub fn aborted(env: &Env, message: &str) -> Error { + match aborted_object(env, message).and_then(|object| object.into_unknown(env)) { + Ok(value) => Error::from(value), + Err(broken) => broken, + } +} + +fn aborted_object<'env>(env: &'env Env, message: &str) -> Result> { + let mut object = blank(env, message)?; + object.set("name", "AbortError")?; + object.set("retryable", false)?; + Ok(object) +} + /// An `Error` with `message` and nothing else on it yet. /// /// napi writes the status it was handed into `code`, and `code` here is diff --git a/src/lib.rs b/src/lib.rs index 6cd690c..accf897 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,7 @@ //! ABI has nothing to say about and a JavaScript program cannot do //! without. +mod cancel; mod conn; mod error; mod value; diff --git a/test/abort.test.mjs b/test/abort.test.mjs new file mode 100644 index 0000000..1bb58f3 --- /dev/null +++ b/test/abort.test.mjs @@ -0,0 +1,180 @@ +// Stopping a statement with the signal every JavaScript program already +// has. + +import assert from 'node:assert/strict' +import { getEventListeners } from 'node:events' +import test from 'node:test' + +import { fresh, isZuError, twoPeople } from './helper.mjs' + +// A statement that is still running a moment after it was asked for, +// which is the only kind there is anything to interrupt. Every triple of +// people, which is a cross product rather than anything a person would +// write, because the point is to spend time in the executor and not to +// mean something. +const HEAVY = 'MATCH (a:person), (b:person), (c:person) RETURN count(a.name + b.name) AS n' + +// Enough of them that the statement above takes hundreds of milliseconds +// even where the engine is fastest, since a statement that finishes in +// the time a timer takes to fire is one no test can catch in the middle. +// It is cubed, so this is a bigger number than it looks. +const PEOPLE = 450 + +async function crowd(t) { + const made = await fresh(t) + await made.conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + const rest = Array.from( + { length: PEOPLE - 1 }, + (_, ix) => `(p${ix}:person {id: ${ix + 2}, name: 'p${ix}'})`, + ).join(', ') + await made.conn.exec(`INSERT ${rest}`) + return made +} + +// How long the heavy statement takes when nobody stops it, measured here +// rather than written down, because a debug build on a busy laptop and a +// release build on a quiet machine are two orders of magnitude apart and +// a number chosen for one of them is a test that fails on the other. +// What the tests then assert is a ratio: asked to stop a tenth of the way +// through, a statement comes back long before it would have finished. +async function pace(conn) { + const started = process.hrtime.bigint() + await conn.query(HEAVY) + return Number(process.hrtime.bigint() - started) / 1e6 +} + +function ms(started) { + return Number(process.hrtime.bigint() - started) / 1e6 +} + +// A tenth of the way through, in whole milliseconds, since that is what +// a timer takes. +function early(plain) { + return Math.max(5, Math.round(plain / 10)) +} + +test('a signal that has already fired stops the statement before the engine sees it', async (t) => { + const { conn } = await twoPeople(t) + + const err = await conn + .query('MATCH (p:person) RETURN p.name AS name', null, { signal: AbortSignal.abort() }) + .then(() => null, (caught) => caught) + + assert.ok(err instanceof Error) + assert.equal(err.name, 'AbortError') +}) + +test('the promise rejects with the reason the signal was given', async (t) => { + const { conn } = await twoPeople(t) + const mine = new Error('the request went away') + const controller = new AbortController() + controller.abort(mine) + + const err = await conn + .query('MATCH (p:person) RETURN p.name AS name', null, { signal: controller.signal }) + .then(() => null, (caught) => caught) + + // The caller's own object, not a description of it, which is what + // makes an existing `catch` work unchanged. + assert.equal(err, mine) +}) + +test('a signal that fires during a statement stops it, and the connection carries on', async (t) => { + const { conn } = await crowd(t) + const plain = await pace(conn) + const controller = new AbortController() + + const started = process.hrtime.bigint() + const running = conn.query(HEAVY, null, { signal: controller.signal }) + setTimeout(() => controller.abort(), early(plain)) + const err = await running.then(() => null, (caught) => caught) + const took = ms(started) + + assert.equal(err?.name, 'AbortError') + assert.ok(took < plain / 2, `took ${took}ms of the ${plain}ms it takes to run, so it ran on`) + + // The interrupt belongs to the connection rather than to the statement, + // so the statement after an abort is the one that would suffer if it + // were left standing. + const rows = await conn.query('MATCH (p:person) RETURN count(*) AS n') + assert.equal(rows[0].n, BigInt(PEOPLE)) +}) + +test('a timeout is a signal like any other', async (t) => { + const { conn } = await crowd(t) + const plain = await pace(conn) + + const started = process.hrtime.bigint() + const err = await conn + .query(HEAVY, null, { signal: AbortSignal.timeout(early(plain)) }) + .then(() => null, (caught) => caught) + const took = ms(started) + + // What `AbortSignal.timeout` puts on the signal, arriving unchanged, + // which is the whole of what a statement timeout has to be in this + // client. + assert.equal(err?.name, 'TimeoutError') + assert.ok(took < plain / 2, `took ${took}ms of the ${plain}ms it takes to run, so it ran on`) + + const rows = await conn.query('MATCH (p:person) RETURN count(*) AS n') + assert.equal(rows[0].n, BigInt(PEOPLE)) +}) + +test('exec takes a signal on the same terms', async (t) => { + const { conn } = await crowd(t) + const plain = await pace(conn) + + const started = process.hrtime.bigint() + const err = await conn + .exec(HEAVY, null, { signal: AbortSignal.timeout(early(plain)) }) + .then(() => null, (caught) => caught) + + assert.equal(err?.name, 'TimeoutError') + assert.ok(ms(started) < plain / 2) +}) + +test('a signal that never fires is left as it was found', async (t) => { + const { conn } = await twoPeople(t) + const controller = new AbortController() + + // A request handler's signal outlives the statements run under it, + // often by a whole request, so a listener left behind is a leak that + // grows with the traffic rather than with the code. + for (let round = 0; round < 8; round += 1) { + await conn.query('MATCH (p:person) RETURN p.name AS name', null, { signal: controller.signal }) + await conn.exec('MATCH (p:person) RETURN p.name AS name', null, { signal: controller.signal }) + } + + assert.deepEqual(getEventListeners(controller.signal, 'abort'), []) +}) + +test('a statement that fails on its own is not mistaken for one that was stopped', async (t) => { + const { conn } = await twoPeople(t) + const controller = new AbortController() + + const err = await conn + .query('MATCH (p:person) RETURN p.nope +', null, { signal: controller.signal }) + .then(() => null, (caught) => caught) + + assert.ok(isZuError(err, 'ZuSyntaxError')) + assert.deepEqual(getEventListeners(controller.signal, 'abort'), []) +}) + +test('options without a signal, and a signal that is not one, are told apart', async (t) => { + const { conn } = await twoPeople(t) + const statement = 'MATCH (p:person) RETURN p.name AS name' + + // The ways of saying nothing, all of which run the statement. + for (const options of [null, undefined, {}, { signal: null }, { signal: undefined }]) { + const rows = await conn.query(statement, null, options) + assert.equal(rows.length, 2) + } + + for (const signal of [42, 'later', {}, new Date()]) { + const err = await conn + .query(statement, null, { signal }) + .then(() => null, (caught) => caught) + assert.ok(isZuError(err, 'ZuUsageError'), `a ${typeof signal} signal was accepted`) + assert.match(err.message, /AbortSignal/) + } +}) diff --git a/types/header.d.ts b/types/header.d.ts index 4d8b19e..c9ca687 100644 --- a/types/header.d.ts +++ b/types/header.d.ts @@ -88,6 +88,29 @@ export interface ZuRows> extends Array { readonly notices: ZuNotice[] } +/** + * What a statement takes beside its parameters. + * + * An object rather than a bare signal, because the options that follow + * it belong in the same place and a third argument that changes meaning + * is one nobody can read at a call site. + */ +export interface ZuStatementOptions { + /** + * Stops the statement when it fires, through the same interrupt a + * shell answers `Ctrl-C` with: the executor notices at the boundary it + * was already stopping at, the statement ends, and the connection is + * exactly as it was. The promise rejects with whatever the signal + * gives as its reason, which is what `fetch` does, so + * `AbortSignal.timeout(50)` rejects with a `TimeoutError` and + * `controller.abort(new MyError())` rejects with `MyError`. + * + * A signal that has already fired stops the statement before the + * engine sees it at all. + */ + readonly signal?: AbortSignal +} + /** * What a failed call throws. *