diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2523b9f..476b4f1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -91,6 +91,133 @@ jobs: - run: npm run test:temporal if: matrix.node == 24 + # The same suite again, over an addon built with AddressSanitizer. + # + # There is no `unsafe` in this crate, which is the reason to run this + # rather than the reason not to: what a binding gets wrong is not + # arithmetic on a raw pointer but a handle used after the scope that + # owned it closed, a buffer read on a thread the runtime had already + # taken back, an engine allocation freed on one side of the boundary + # and touched from the other. None of those are `unsafe` blocks here + # and every one of them is a use-after-free somewhere, which is the + # thing this tool exists to find. + # + # ASan's runtime has to be loaded before anything it instruments, and + # the addon is opened by `require` long after node has started, so it + # is preloaded rather than linked: `-Zexternal-clangrt` tells rustc + # not to bundle its own copy, and LD_PRELOAD supplies clang's. The + # path is asked for rather than written down, because it moves with + # the LLVM version the image ships. Leak detection is off here and + # the job below is where it lives, since ASan and LSan report the + # same run twice and a report that arrives in two jobs is a report + # nobody can attribute. + sanitizer: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-node@v7 + with: + node-version: 24 + - uses: Swatinem/rust-cache@v2 + # The instrumented build is nightly, because `-Zsanitizer` is, + # and it is the only thing here that is: what ships is built by + # the pinned compiler in every other job. + - run: rustup toolchain install nightly --profile minimal + - run: sudo apt-get update && sudo apt-get install -y libclang-rt-18-dev + - run: npm ci + - name: The addon, instrumented + env: + RUSTUP_TOOLCHAIN: nightly + RUSTFLAGS: -Zsanitizer=address -Zexternal-clangrt + CC: clang + CXX: clang++ + run: npm run build:debug + - name: The suite, watched + run: | + runtime=$(clang -print-file-name=libclang_rt.asan-$(uname -m).so) + test -f "$runtime" + LD_PRELOAD="$runtime" ASAN_OPTIONS=detect_leaks=0:abort_on_error=1 \ + npm test + + # The third tool over the same suite, and the one that watches what + # the sanitizer cannot. ASan instruments the source it compiles, so + # it sees nothing node itself does with the memory it hands the + # addon; Valgrind instruments the instructions that run, so both + # sides of the boundary are watched and so is every prebuilt thing + # either of them links. It also reports a read of memory nobody + # wrote, which ASan does not look for at all. + # + # Definite leaks only, and for the reason a Rust process always gives: + # one-time allocations held at exit are still reachable and not lost, + # and counting those would fire this gate on every run and teach + # everyone to ignore it. --error-exitcode is what makes a report a + # failure rather than something in a log nobody opens. + # + # tools/valgrind.supp is three lines of rule and says what it leaves + # out. It exists because which node this runs on decides what leaks: a + # node linked against the system OpenSSL needs nothing suppressed, and + # the one the hosted images install has OpenSSL inside it and holds + # twenty four bytes of compression table forever. The rule names the + # binary that allocated rather than the leak that was reported, which + # is what stops it growing a line every bad week. + # + # --jitless turns V8's compilers off, and it is here so that the + # uninitialised-value check can stay on. Maglev branches on memory + # nobody wrote while it compiles, on a thread of its own, and reports + # it from inside the node binary a dozen frames from anything this + # package wrote. Suppressing that would mean suppressing a whole class + # of error inside node, and that class is the one worth having: an + # addon that hands a half-filled buffer back is exactly what this tool + # sees and ASan does not. The flag is --jitless and not --no-opt + # because --no-opt left Maglev on and the job stayed red: the tiers + # have their own switches now and naming them one by one is a list + # that goes stale the next time V8 grows one. --jitless is the flag + # that cannot be partially true. The suite runs it in ten seconds + # against a job that is already fifty times slower than the real + # thing, so it costs nothing it was measuring. + leaks: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-node@v7 + with: + node-version: 24 + - uses: Swatinem/rust-cache@v2 + - run: sudo apt-get update && sudo apt-get install -y valgrind + - run: npm ci + # An ordinary build. Valgrind wants the instructions the addon + # actually ships, and an instrumented one would be a different + # program with a different allocator underneath it. + - run: npm run build:debug + # The gate is validated the only way a gate can be: a deliberate + # leak has to fail it, and it has to be a leak of the shape being + # looked for, out of a shared object node loaded rather than out + # of node. Preloaded rather than required, because a real napi + # module is a lot of code to write for four thousand bytes and + # nothing about the check needs node to have called it. + - name: A leak the job is meant to catch, caught + run: | + cat > /tmp/leak.c <<'EOF' + #include + __attribute__((constructor)) static void leak(void) { + void *lost = malloc(4096); + (void)lost; + } + EOF + cc -shared -fPIC -o /tmp/leak.so /tmp/leak.c + set +e + LD_PRELOAD=/tmp/leak.so valgrind --error-exitcode=1 --leak-check=full \ + --show-leak-kinds=definite --errors-for-leak-kinds=definite \ + --suppressions=tools/valgrind.supp -q \ + node --jitless -e '' + test $? -eq 1 || { echo "the leak gate did not fire on a leak"; exit 1; } + - name: The suite, counted + run: | + valgrind --error-exitcode=1 --leak-check=full \ + --show-leak-kinds=definite --errors-for-leak-kinds=definite \ + --suppressions=tools/valgrind.supp -q \ + node --jitless --test "test/*.test.mjs" + # Bun and Deno run the same suite over the same binary, because the # binary is the same one: N-API is the ABI all three implement, and a # runtime this package claims and nothing runs on is a runtime this @@ -112,7 +239,13 @@ jobs: - run: npm run build:debug # `bun test` and not `bun run`, because the tests are written # against `node:test` and Bun's shim for it refuses to register a - # test outside its own runner. + # test outside its own runner. The shim brings Bun's own five + # second per-test timeout with it, which node does not have and + # two of these tests do not fit inside: they build a table of + # sixty thousand people so that a stream has something to be + # faster than, and the build alone is most of five seconds on a + # shared runner. `--timeout` puts the limit where it catches a + # hang rather than a fixture. - run: npm run test:bun deno: diff --git a/README.md b/README.md index 7a4e447..e3d2461 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,8 @@ The rows are an array, so iterating them is `for (const row of rows)` and nothin - **Nothing blocks the event loop.** Every native call runs on libuv's threadpool and hands back a promise before the statement has started. There is no synchronous variant, and the ones that arrive later will say in their own documentation that they belong in scripts, not servers. - **`await using` is the intended scoping.** A connection is `Symbol.asyncDispose`, and `close()` stays public for callers who cannot use the syntax. - **A failure is an ordinary `Error`.** Every `catch`, logger and rejection handler already knows what to do with one. What makes it a zu error is the fields, and none of them has to be parsed back out of the message: `code` is the GQLSTATUS and picks the branch, `condition` is the standard's own words for it, `line` and `column` and `excerpt` underline the token, and `retryable` decides whether a retry loop goes round again. A mistake this client caught before the engine saw it carries no `code` and is named `ZuUsageError`, so a caller mapping codes to branches can tell a missing code from one it does not recognize. `isZuError(caught)` is the exported guard for the `catch` clause, where the value is `unknown` and could be anything at all, and in TypeScript it narrows to the full shape. -- **A refusal is a rejection.** A closed connection and a parameter of a type nothing can bind are refused inside the promise rather than thrown out of the call, so one `await` catches everything one statement can do. +- **A refusal is a rejection.** A closed connection, a statement that is not a string and a parameter of a type nothing can bind are all refused inside the promise rather than thrown out of the call, so one `await` catches everything one statement can do and no caller has to wrap the same call twice. That holds for the arguments too: passing a number where a statement goes is a `ZuUsageError` the promise rejects with, not a `TypeError` off the stack. +- **Parameters are named, and nothing about them is guessed.** An object keyed by the names the statement uses, without the `$`. An array is refused rather than bound by position, because zu has no positional parameters and binding one by index would run the statement with none of the values the caller passed and say nothing about it. A value that contains itself is refused too, at a nesting depth no real value reaches. ## What works today @@ -70,6 +71,8 @@ Three ways to read it, all the same statement read once. `for await` over the st Ending early is the case worth knowing about, because it is the reason streaming is different from `query`. A `break`, a `throw`, a `return()` on the iterator, a `cancel()`, or leaving the block of an `await using` all stop the statement and wait for it to let go of the connection, so the next statement on that connection runs rather than queueing behind a scan nobody is reading. The rows already read stand, and `summary.stopped` says the reader stopped it. The statement itself does not start until the first read, so a stream made and never read is not a scan holding anything. +A connection runs one statement at a time, and a stream that has started is that statement until it ends. So a `query` on the same connection while a stream is half-read is refused rather than queued: what it would be queueing behind is the loop that is waiting for it, and a program that stops is worse than a program that is told to read the stream out, cancel it, or open a second connection. Two scans that should overlap want two connections, which is one line and no lock. + Between the statement and the loop sit two batches, which is the whole of the buffering: a reader slower than the scan stops the scan rather than filling memory behind it. `{ batchRows: 512 }` sets what a batch may hold, which is what to name when the rows are going somewhere with a size of its own. On 50k rows here a stream costs about 460ns a row against 370ns for `query`, reading a batch at a time costs about 320ns, and reading the first batch and stopping costs 1.1ms against 18.6ms for the whole scan, which is what the whole thing is for. 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. diff --git a/package.json b/package.json index 13dc067..051aff6 100644 --- a/package.json +++ b/package.json @@ -63,7 +63,7 @@ "build:debug": "napi build --platform --js binding.cjs --dts binding.d.cts", "test": "node --test \"test/*.test.mjs\"", "test:temporal": "node --harmony-temporal --test \"test/*.test.mjs\"", - "test:bun": "bun test test/", + "test:bun": "bun test --timeout 60000 test/", "test:deno": "deno test --no-check --allow-read --allow-write --allow-env --allow-ffi \"test/*.test.mjs\"", "check:types": "tsc --noEmit --project test/types/tsconfig.json", "check:package": "attw --pack .", diff --git a/src/conn.rs b/src/conn.rs index 97c6e14..cd8afe6 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -14,6 +14,13 @@ //! holds the mutex for as long as it runs, and two statements that //! should overlap want two connections. It is there so that a program //! which shares one by accident waits rather than corrupts. +//! +//! Waits, except behind a stream. A stream ends when its reader says +//! so, so a statement that queued behind a half-read one would be +//! waiting for the caller who is waiting for it, and a program that +//! stops is worse than a program that is told no. A stream takes the +//! connection out of the slot instead of locking it, and the next +//! statement finds the slot empty and says [`STREAMING`]. use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; @@ -113,15 +120,24 @@ pub struct Connection { /// Creates one when the path holds nothing, which is what a first /// program expects and what every embedded database does. A read-only /// connection never creates anything. -#[napi(ts_return_type = "Promise")] -pub fn connect(env: &Env, path: String, options: Option) -> AsyncTask { +#[napi( + ts_args_type = "path: string, options?: ConnectOptions | undefined | null", + ts_return_type = "Promise" +)] +pub fn connect( + env: &Env, + path: Unknown<'_>, + options: Option, +) -> AsyncTask { // Whether this runtime has `Temporal` is a question only the thread // that owns the runtime may ask, so it is asked here and carried to // the thread that opens the database, where the answer decides // whether there is anything to open. let has_temporal = temporal::present(env).unwrap_or(false); + let path = text(&path, "path"); AsyncTask::new(ConnectTask { - path, + path: path.as_deref().unwrap_or_default().to_string(), + refused: path.err(), options, has_temporal, }) @@ -129,6 +145,8 @@ pub fn connect(env: &Env, path: String, options: Option) -> Asyn pub struct ConnectTask { path: String, + /// What this client refused the call with, before any of it ran. + refused: Option, options: Option, has_temporal: bool, } @@ -138,6 +156,12 @@ impl<'task> ScopedTask<'task> for ConnectTask { type JsValue = ClassInstance<'task, Connection>; fn compute(&mut self) -> Result { + // Before anything else, because a path that is not a path names + // no file to open and a message about the one it made instead + // would be a message about this client's own confusion. + if let Some(message) = self.refused.take() { + return Ok(Err(Failure::Usage(message))); + } let read_only = self .options .as_ref() @@ -290,8 +314,8 @@ impl Connection { pub fn query( &self, env: &Env, - statement: String, - params: Option>, + statement: Unknown<'_>, + params: Option>, options: Option>, ) -> AsyncTask { AsyncTask::new(self.task(env, statement, params, options)) @@ -309,8 +333,8 @@ impl Connection { pub fn exec( &self, env: &Env, - statement: String, - params: Option>, + statement: Unknown<'_>, + params: Option>, options: Option>, ) -> AsyncTask { AsyncTask::new(ExecTask(self.task(env, statement, params, options))) @@ -329,8 +353,8 @@ impl Connection { pub fn cursor( &self, env: &Env, - statement: String, - params: Option>, + statement: Unknown<'_>, + params: Option>, options: Option>, ) -> ZuCursor { // Read here rather than on the statement's thread, because @@ -338,9 +362,11 @@ impl Connection { // owns the runtime may do. So is adding the listener the signal // is watched through. let bound = if self.alive.load(Ordering::Acquire) { - self.spell(options.as_ref()).and_then(|spelling| { + text(&statement, "statement").and_then(|statement| { + let spelling = self.spell(options.as_ref())?; let batch_rows = batch_rows(options.as_ref())?; Ok(( + statement, bind(env, params)?, spelling, batch_rows, @@ -350,11 +376,18 @@ impl Connection { } else { Err(CLOSED.to_string()) }; - let (params, spelling, batch_rows, watch, refused) = match bound { - Ok((params, spelling, batch_rows, watch)) => { - (params, spelling, batch_rows, watch, None) + let (statement, params, spelling, batch_rows, watch, refused) = match bound { + Ok((statement, params, spelling, batch_rows, watch)) => { + (statement, params, spelling, batch_rows, watch, None) } - Err(message) => (Vec::new(), self.spelling, None, None, Some(message)), + Err(message) => ( + String::new(), + Vec::new(), + self.spelling, + None, + None, + Some(message), + ), }; stream::open( Started { @@ -382,8 +415,8 @@ impl Connection { fn task( &self, env: &Env, - statement: String, - params: Option>, + statement: Unknown<'_>, + params: Option>, options: Option>, ) -> QueryTask { // The parameters are read here rather than on the threadpool @@ -391,8 +424,10 @@ impl Connection { // 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) { - self.spell(options.as_ref()).and_then(|spelling| { + text(&statement, "statement").and_then(|statement| { + let spelling = self.spell(options.as_ref())?; Ok(( + statement, bind(env, params)?, spelling, watch(env, options, self.interrupt.clone())?, @@ -401,12 +436,19 @@ impl Connection { } else { Err(CLOSED.to_string()) }; - let (params, spelling, watch, refused) = match bound { - Ok((params, spelling, watch)) => (params, spelling, watch, None), - Err(message) => (Vec::new(), self.spelling, None, Some(message)), + let (statement, params, spelling, watch, refused) = match bound { + Ok((statement, params, spelling, watch)) => (statement, params, spelling, watch, None), + Err(message) => ( + String::new(), + Vec::new(), + self.spelling, + None, + Some(message), + ), }; QueryTask { inner: Arc::clone(&self.inner), + alive: Arc::clone(&self.alive), statement, params, spelling, @@ -505,6 +547,18 @@ impl From for Failure { pub(crate) const CLOSED: &str = "the connection is closed, so there is nothing left to run a statement on"; +/// What a connection a stream is still reading says to the next +/// statement. +/// +/// A connection runs one statement at a time, and a stream is a +/// statement that ends when its reader says so. So a second statement +/// issued while a stream is outstanding cannot be made to wait: the +/// thing it would wait for is the caller, who is waiting for it. That is +/// a program that stops, which is the worst answer a database can give, +/// and this is the sentence that replaces it. +pub(crate) const STREAMING: &str = "a stream on this connection has not finished, and a connection runs one statement at a time: \ + read the stream to the end, cancel it, or open a second connection"; + /// 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"; @@ -610,6 +664,41 @@ fn int_mode(options: Option<&Object<'_>>, connection: Ints) -> std::result::Resu } } +/// Reads an argument that has to be a string, and says what arrived +/// instead. +/// +/// napi refuses the wrong type on its own, and what it refuses with is +/// `Failed to convert JavaScript value \`Number 42 \` into rust type +/// \`String\``, thrown out of the call with a `code` of `StringExpected` +/// rather than handed to the promise the caller is awaiting. Both +/// halves of that are wrong for this client: the message describes this +/// crate's insides to somebody who mistyped a variable, and a throw +/// from a method whose every other failure is a rejection is a method +/// callers have to wrap twice. So the argument arrives unread and this +/// is what reads it. +fn text(value: &Unknown<'_>, what: &str) -> std::result::Result { + match value.get_type().map_err(|err| err.reason)? { + ValueType::String => String::from_unknown(*value).map_err(|err| err.reason), + other => Err(format!( + "the {what} is {}, and a {what} is a string", + worded(other) + )), + } +} + +/// What arrived, in the words a person writing JavaScript uses for it. +/// +/// `a Undefined` is what the type's own name gives, and the two values +/// that need this are exactly the two a mistake produces most: a +/// variable that was never set and a lookup that found nothing. +fn worded(kind: ValueType) -> String { + match kind { + ValueType::Undefined => "undefined".to_string(), + ValueType::Null => "null".to_string(), + other => format!("a {other}"), + } +} + /// Reads the parameter object into the values the engine binds. /// /// Every failure comes back as the message to refuse the call with, @@ -618,11 +707,36 @@ fn int_mode(options: Option<&Object<'_>>, connection: Ints) -> std::result::Resu /// would rather hear it as a rejection than as a throw. pub(crate) fn bind( env: &Env, - params: Option>, + params: Option>, ) -> std::result::Result, String> { let Some(params) = params else { return Ok(Vec::new()); }; + // An argument of the wrong shape is refused rather than read for + // whatever keys it happens to have. An array read as an object + // binds its parameters as `0` and `1`, a string binds one per + // character, and both of those are a statement that runs with none + // of the values the caller passed and answers nothing. + match params.get_type().map_err(|err| err.reason)? { + ValueType::Undefined | ValueType::Null => return Ok(Vec::new()), + ValueType::Object => {} + other => { + return Err(format!( + "the parameters are {}, and parameters are an object keyed by the names the \ + statement uses, without the $", + worded(other) + )); + } + } + let params = Object::from_unknown(params).map_err(|err| err.reason)?; + if params.is_array().map_err(|err| err.reason)? { + return Err( + "the parameters are an array, and zu names its parameters rather than \ + numbering them: pass an object keyed by the names the statement uses, without \ + the $" + .to_string(), + ); + } let mut bound = Vec::new(); for name in Object::keys(¶ms).map_err(|err| err.reason)? { let value: Unknown<'_> = params @@ -636,6 +750,9 @@ pub(crate) fn bind( pub struct QueryTask { inner: Arc>>, + /// Whether the connection is still open, which is what tells an + /// empty slot that was closed from one a stream is holding. + alive: Arc, statement: String, params: Vec<(String, Value)>, /// How this statement spells the values it gives back. @@ -669,9 +786,17 @@ impl QueryTask { } }; // Closed between the call and the thread picking it up, which is - // a race the caller cannot see and this has to check anyway. + // 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(CLOSED.to_string())); + return Err(Failure::Usage( + match self.alive.load(Ordering::Acquire) { + true => STREAMING, + false => 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 diff --git a/src/stream.rs b/src/stream.rs index 376f014..ebff710 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -34,7 +34,7 @@ use zudb::query::Value; use zudb::{Batch, Flow, Streamed, ZuError}; use crate::cancel::{Guard, Watch}; -use crate::conn::{CLOSED, Failure, beside, failed, notices}; +use crate::conn::{CLOSED, Failure, STREAMING, beside, failed, notices}; use crate::value::{Shape, Spelling, to_js}; /// How many batches may sit between the statement and the reader. @@ -185,6 +185,79 @@ impl Pipe { } } +/// The connection, out of the slot it lives in for as long as the +/// statement runs. +/// +/// Taken out rather than locked, and this is the difference between a +/// stream and every other statement. A statement that holds the lock is +/// a statement anything else on that connection waits behind, and +/// waiting is right when the thing being waited for finishes on its +/// own. A stream finishes when its reader says so, and its reader is +/// the thread that would be doing the waiting, so a second statement +/// that queued behind a half-read stream would be a program that stops +/// and never starts again. An empty slot can be asked about instead: +/// [`crate::conn::STREAMING`] is what the next statement is told, and +/// it is told at once. +/// +/// Put back by the drop, on every path out of the statement, because a +/// connection that stayed out of its slot would be one this client had +/// closed on the caller's behalf. +struct Borrowed<'held> { + inner: &'held Arc>>, + alive: &'held AtomicBool, + conn: Option, +} + +impl<'held> Borrowed<'held> { + fn take( + inner: &'held Arc>>, + alive: &'held AtomicBool, + ) -> std::result::Result { + let mut held = inner + .lock() + .map_err(|_| Failure::Usage(POISONED.to_string()))?; + // Both readings of an empty slot are checked under the one lock + // that decides them, so a stream and a close that arrive + // together cannot both find the connection theirs. + let Some(conn) = held.take() else { + return Err(Failure::Usage( + match alive.load(Ordering::Acquire) { + true => STREAMING, + false => CLOSED, + } + .to_string(), + )); + }; + Ok(Borrowed { + inner, + alive, + conn: Some(conn), + }) + } + + fn conn(&mut self) -> &mut zudb::Connection { + self.conn + .as_mut() + .expect("the connection is taken out only by the drop") + } +} + +impl Drop for Borrowed<'_> { + fn drop(&mut self) { + let conn = self.conn.take(); + if let Ok(mut held) = self.inner.lock() { + // A close that arrived while the statement was running + // found the slot empty and left `alive` false. Putting the + // connection back there would be opening again what the + // caller closed, so it is dropped here instead, which is + // what the close itself would have done. + if self.alive.load(Ordering::Acquire) { + *held = conn; + } + } + } +} + /// Everything the statement needs, held until the first read. /// /// A stream starts when it is first read rather than when it is asked @@ -591,13 +664,8 @@ impl Started { /// Takes the connection, runs the statement, and hands every batch /// over. fn stream(&self, live: &Live) -> std::result::Result { - let mut held = self - .inner - .lock() - .map_err(|_| Failure::Usage(POISONED.to_string()))?; - let Some(conn) = held.as_mut() else { - return Err(Failure::Usage(CLOSED.to_string())); - }; + let mut borrowed = Borrowed::take(&self.inner, &self.alive)?; + let conn = borrowed.conn(); // From here the connection is this statement's, so this is // where a signal can start stopping it. A signal that fired // first ends the statement without the engine ever seeing it. diff --git a/src/value.rs b/src/value.rs index ed96bdb..efc1cdc 100644 --- a/src/value.rs +++ b/src/value.rs @@ -731,6 +731,28 @@ fn as_class(env: &Env, value: Temporal) -> Result> { /// apart from a boundary failure and turns it into the rejection the /// caller is already awaiting. pub fn from_js(env: &Env, name: &str, value: Unknown<'_>) -> Result { + nested(env, name, value, 0) +} + +/// How deep a parameter may nest before this stops reading it. +/// +/// A list of lists of records is a value somebody meant to send, and a +/// value that contains itself is a call that would otherwise walk until +/// the stack ran out and take the process with it. There is no depth +/// between the two that anybody writes on purpose, so the limit is set +/// where a real value never reaches and a cycle always does. +const DEEP: usize = 64; + +fn nested(env: &Env, name: &str, value: Unknown<'_>, depth: usize) -> Result { + if depth > DEEP { + return Err(Error::new( + Status::InvalidArg, + format!( + "parameter {name} nests deeper than {DEEP}, which is what a value that contains \ + itself looks like" + ), + )); + } match value.get_type()? { ValueType::Null | ValueType::Undefined => Ok(Value::Null), ValueType::Boolean => Ok(Value::Bool(bool::from_unknown(value)?)), @@ -761,7 +783,7 @@ pub fn from_js(env: &Env, name: &str, value: Unknown<'_>) -> Result { Ok(Value::Float(n)) } } - ValueType::Object => from_object(env, name, value), + ValueType::Object => from_object(env, name, value, depth), other => Err(Error::new( Status::InvalidArg, format!("parameter {name} is a {other}, which is not a value a statement can hold"), @@ -769,7 +791,7 @@ pub fn from_js(env: &Env, name: &str, value: Unknown<'_>) -> Result { } } -fn from_object(env: &Env, name: &str, value: Unknown<'_>) -> Result { +fn from_object(env: &Env, name: &str, value: Unknown<'_>, depth: usize) -> Result { if let Some(temporal) = temporal_from(env, &value)? { return Ok(Value::Temporal(temporal)); } @@ -779,7 +801,7 @@ fn from_object(env: &Env, name: &str, value: Unknown<'_>) -> Result { let mut items = Vec::with_capacity(len as usize); for ix in 0..len { let item: Unknown<'_> = object.get_element(ix)?; - items.push(from_js(env, name, item)?); + items.push(nested(env, name, item, depth + 1)?); } return Ok(Value::List(items)); } @@ -796,7 +818,7 @@ fn from_object(env: &Env, name: &str, value: Unknown<'_>) -> Result { let mut fields = Vec::new(); for key in Object::keys(&object)? { let field: Unknown<'_> = object.get_named_property(key.as_str())?; - fields.push((key, from_js(env, name, field)?)); + fields.push((key, nested(env, name, field, depth + 1)?)); } Ok(Value::record(fields)) } diff --git a/test/misuse.test.mjs b/test/misuse.test.mjs new file mode 100644 index 0000000..8bd297b --- /dev/null +++ b/test/misuse.test.mjs @@ -0,0 +1,486 @@ +// Deliberately wrong programs, and what each of them is told. +// +// DX3 asks for a misuse suite in every client: no crash, no hang, no +// leak, and a clear error for every program that is wrong on purpose. +// Clear is the hard word, so it is spelled out here as four things a +// message has to do. It names the thing the caller named, being the +// file they opened, the parameter they passed, the option they spelled. +// It says what was expected instead, wherever there is something to +// say. It is the engine's own sentence rather than a syscall's, since +// "failed to fill whole buffer" is a true statement about a read that +// tells nobody which file was not a database. And it never describes +// this crate's insides, because "Failed to convert JavaScript value +// `Number 42 ` into rust type `String`" is a message about napi to +// somebody who mistyped a variable. +// +// Clear also means the right class and the right shape. The class is +// `err.name`, and a caller branches on it: a mistake this client caught +// is a `ZuUsageError` with no `code`, a condition the engine raised is +// the class of its GQLSTATUS. The shape is that every one of these is a +// rejection and none of them is a throw, which is the last test in the +// first half: a method whose failures arrive two different ways is a +// method every caller has to wrap twice. +// +// No crash is the suite running at all. No hang is the pair of tests +// about a connection with a stream half-read on it, which is the one +// place where waiting would be forever. No leak is checked from outside +// the call that would cause one, three ways: every case is followed by +// a read on the connection it was aimed at, the failing connects are +// repeated past the descriptor limit a process starts with, and the +// descriptors themselves are counted where the operating system will +// say. +// +// The last test is the half of a misuse suite that is usually missing. +// The programs that look wrong and are not, each of which is a decision +// somebody would otherwise reverse by accident. + +import assert from 'node:assert/strict' +import { mkdtemp, readdir, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import test from 'node:test' + +import { connect } from 'zudb' + +import { fresh, isZuError, twoPeople } from './helper.mjs' + +const READ = 'MATCH (p:person) RETURN p.id AS id' + +// A file that is not a database, written where a caller would have a +// file that is not a database. +async function junk(dir, name, contents) { + const path = join(dir, name) + await writeFile(path, contents) + return path +} + +// A database of its own, written, closed and opened read-only. Written +// and closed first, because asking for read-only is asking for one that +// is already there. +async function readOnly(t, dir) { + const path = join(dir, 'reader.zu1') + const writer = await connect(path) + await writer.exec("INSERT (p:person {id: 1, name: 'ada'})") + writer.close() + const conn = await connect(path, { readOnly: true }) + t.after(() => conn.close()) + return conn +} + +// A connection of its own, closed. Its own, because the case that uses +// it would otherwise close the connection every other case is checked +// against afterwards. +async function closed(dir) { + const conn = await connect(join(dir, 'closed.zu1')) + conn.close() + return conn +} + +// A stream on the given connection with one batch read out of it, which +// is the state that makes that connection busy: the statement has +// started, it has not ended, and what ends it is a reader. +// +// It takes a database with enough rows in it to still be scanning, since +// a stream that already reached the last row has already given the +// connection back and is not half-read at all. +async function halfRead(conn) { + const stream = conn.stream(READ, null, { batchRows: 1 }) + await stream.batches().next() + return stream +} + +// A database with `count` people in it. One statement per row is a +// write per row, so the rows past the first go in batches. The first is +// written on its own, because that is the insert that declares the +// table. +async function people(t, count) { + const made = await fresh(t) + await made.conn.exec("INSERT (p:person {id: 1, name: 'p1'})") + for (let start = 2; start <= count; start += 500) { + const parts = [] + for (let id = start; id < Math.min(start + 500, count + 1); id++) { + parts.push(`(p${id}:person {id: ${id}, name: 'p${id}'})`) + } + await made.conn.exec(`INSERT ${parts.join(', ')}`) + } + return made +} + +// Enough rows that a stream is still scanning after one batch of one +// row has been read, which is what makes the connection busy. +const CROWD = 2000 + +// Every case gets a connection with two people in it, unless it asks +// for a crowd, and a directory to make a mess in. `run` returning +// normally fails the test: every program in this table is wrong. A +// stream a case pushed onto `streams` is stopped before the connection +// is asked whether it still works. +const MISUSES = [ + { + what: 'connects to a file too small to be a database', + run: ({ dir }) => junk(dir, 'small.zu1', 'not a database at all').then(connect), + name: 'ZuConnectionError', + says: ['small.zu1', '21 bytes', 'too short to be a zu1 database'], + }, + { + what: 'connects to a file the right size and the wrong kind', + run: ({ dir }) => junk(dir, 'big.zu1', 'x'.repeat(40960)).then(connect), + name: 'ZuConnectionError', + says: ['big.zu1', 'not a zu1 file'], + }, + { + what: 'connects read-only to a database that is not there', + // Read-only, because `connect` on a path with nothing at it makes a + // database and that is the documented answer. Asking for read-only + // is asking for one that already exists. + run: ({ dir }) => connect(join(dir, 'nowhere.zu1'), { readOnly: true }), + name: 'ZuConnectionError', + says: ['nowhere.zu1'], + }, + { + what: 'passes a path that is not a string', + run: () => connect(42), + name: 'ZuUsageError', + says: ['the path is a Number', 'a path is a string'], + }, + { + what: 'writes through a connection it opened read-only', + run: async ({ dir, t }) => + (await readOnly(t, dir)).exec("INSERT (p:person {id: 3, name: 'zoe'})"), + name: 'ZuUsageError', + says: ['reader.zu1', 'read-only'], + }, + { + what: 'runs text that will not parse', + run: ({ conn }) => conn.query('MATCH (p:person) RETRUN p.id'), + name: 'ZuSyntaxError', + code: '42001', + says: ['42001', 'line 1, column 18', "found 'RETRUN'"], + }, + { + what: 'leaves out a parameter the statement reads', + run: ({ conn }) => conn.query('MATCH (p:person) WHERE p.id = $id RETURN p.id AS id'), + name: 'ZuSyntaxError', + code: '42002', + says: ['42002', 'missing parameter $id'], + }, + { + what: 'passes a statement that is not a string', + run: ({ conn }) => conn.query(42), + name: 'ZuUsageError', + says: ['the statement is a Number', 'a statement is a string'], + }, + { + what: 'calls query with no arguments at all', + run: ({ conn }) => conn.query(), + name: 'ZuUsageError', + says: ['the statement is undefined', 'a statement is a string'], + }, + { + what: 'passes a parameter of a type zu has no value for', + run: ({ conn }) => conn.query('RETURN $x AS x', { x: () => 1 }), + name: 'ZuUsageError', + says: ['parameter x', 'Function', 'not a value a statement can hold'], + }, + { + what: 'passes a parameter that contains itself', + run: ({ conn }) => { + const knot = {} + knot.self = knot + return conn.query('RETURN $x AS x', { x: knot }) + }, + name: 'ZuUsageError', + says: ['parameter x', 'nests deeper than 64', 'contains itself'], + }, + { + what: 'passes the parameters as an array', + run: ({ conn }) => conn.query('MATCH (p:person) WHERE p.id = $id RETURN p.id AS id', [1]), + name: 'ZuUsageError', + says: ['the parameters are an array', 'names its parameters', 'without the $'], + }, + { + what: 'passes the parameters as a string', + run: ({ conn }) => conn.query(READ, 'id=1'), + name: 'ZuUsageError', + says: ['the parameters are a String', 'without the $'], + }, + { + what: 'names a bigIntMode nobody can spell', + run: ({ conn }) => conn.query(READ, null, { bigIntMode: 'bigInt' }), + name: 'ZuUsageError', + says: ['bigIntMode is "bigInt"', '"bigint" and "number"'], + }, + { + what: 'passes something that is not an AbortSignal as the signal', + run: ({ conn }) => conn.query(READ, null, { signal: 'stop' }), + name: 'ZuUsageError', + says: ['signal is a String', 'not an AbortSignal'], + }, + { + what: 'asks a stream for batches of no rows', + run: ({ conn }) => conn.stream(READ, null, { batchRows: 0 }).batches().next(), + name: 'ZuUsageError', + says: ['batchRows is 0', 'one at the least'], + }, + { + what: 'runs a statement on a connection it closed', + run: ({ dir }) => closed(dir).then((gone) => gone.query(READ)), + name: 'ZuUsageError', + says: ['the connection is closed', 'nothing left to run a statement on'], + }, + { + what: 'runs a statement while a stream on the same connection is half-read', + people: CROWD, + run: async ({ conn, streams }) => { + streams.push(await halfRead(conn)) + return conn.query(READ) + }, + name: 'ZuUsageError', + says: ['a stream on this connection has not finished', 'open a second connection'], + }, + { + what: 'opens a second stream while the first is half-read', + people: CROWD, + run: async ({ conn, streams }) => { + streams.push(await halfRead(conn)) + return conn.stream(READ).batches().next() + }, + name: 'ZuUsageError', + says: ['a stream on this connection has not finished', 'cancel it'], + }, + { + what: 'divides by zero', + run: ({ conn }) => conn.exec("INSERT (p:person {id: 1 / 0, name: 'zoe'})"), + name: 'ZuDataError', + code: '22012', + says: ['22012', 'division by zero'], + }, + { + what: 'writes a row with a column missing', + run: ({ conn }) => conn.exec('INSERT (p:person {id: 3})'), + 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) { + test(`a program that ${misuse.what} is told what is wrong`, async (t) => { + const count = misuse.people ?? 2 + const { conn, dir } = misuse.people ? await people(t, count) : await twoPeople(t) + const streams = [] + + await assert.rejects(() => misuse.run({ conn, dir, t, streams }), (err) => { + assert.ok(isZuError(err, misuse.name), `expected ${misuse.name}, got ${err.name}: ${err}`) + for (const phrase of misuse.says) { + assert.ok(err.message.includes(phrase), `the message is missing '${phrase}': ${err.message}`) + } + // The engine's sentence, not the read that noticed, and not this + // crate's insides either. + assert.doesNotMatch(err.message, /failed to fill whole buffer|rust type|napi/i) + // A mistake this client caught carries no GQLSTATUS, since the + // engine never saw the statement and there is no condition to + // report, and a caller mapping codes to branches has to be able + // to tell that from a code it does not recognize. What the engine + // did raise carries the code the table names, and nothing + // anywhere carries the status napi would have written. + assert.equal(err.code, misuse.code) + assert.equal('code' in err, misuse.code !== undefined) + return true + }) + + // Every stream the case left running is stopped, since a connection + // with a scan on it is busy by design and the question below is + // whether the failure took anything with it. + for (const stream of streams) await stream.cancel() + + // And the connection it was aimed at is still a connection: the + // failure took nothing with it. + assert.equal(conn.open, true) + assert.equal((await conn.query(READ)).length, count) + }) +} + +test('every mistake this client catches is a rejection and never a throw', async (t) => { + const { conn } = await twoPeople(t) + + // The calls a wrong program makes on a connection, each of which + // fails before the engine has seen anything. A native method that + // throws for some of these and rejects for the rest is a method every + // caller has to write two handlers for, and the one they leave out is + // the one that takes the process down. + const calls = [ + () => conn.query(42), + () => conn.query(), + () => conn.exec(null), + () => conn.query(READ, [1]), + () => conn.query(READ, null, { bigIntMode: 'nope' }), + () => conn.query(READ, null, { signal: 'stop' }), + () => conn.query('RETURN $x AS x', { x: Symbol('nope') }), + () => connect(42), + ] + + for (const call of calls) { + let returned + assert.doesNotThrow(() => { + returned = call() + }, `${call} threw where it should have rejected`) + assert.equal(typeof returned.then, 'function', `${call} gave back something that is not a promise`) + await assert.rejects(() => returned, (err) => isZuError(err, 'ZuUsageError')) + } +}) + +test('five hundred failed connections leave nothing open', async (t) => { + const dir = await mkdtemp(join(tmpdir(), 'zu-node-')) + t.after(async () => { + const { rm } = await import('node:fs/promises') + await rm(dir, { recursive: true, force: true }) + }) + const missing = join(dir, 'nowhere.zu1') + const small = await junk(dir, 'small.zu1', 'not a database at all') + + const before = await descriptors() + for (let n = 0; n < 500; n++) { + await assert.rejects(() => connect(missing, { readOnly: true }), { name: 'ZuConnectionError' }) + await assert.rejects(() => connect(small), { name: 'ZuConnectionError' }) + } + + // A descriptor per failure would have run out long ago, and a + // database made now is one that can be written and read. + const conn = await connect(join(dir, 'after.zu1')) + t.after(() => conn.close()) + await conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + assert.equal((await conn.query(READ)).length, 1) + + // And where the operating system will say how many are open, a + // thousand failures cost a handful rather than a thousand. + const after = await descriptors() + if (before !== null && after !== null) { + assert.ok(after - before < 20, `${after - before} descriptors were left open by 1000 failures`) + } +}) + +test('a thousand connections opened and closed leave nothing behind', async (t) => { + const { path } = await twoPeople(t) + + const before = await descriptors() + for (let n = 0; n < 1000; n++) { + const conn = await connect(path) + await conn.dispose() + } + const after = await descriptors() + + if (before !== null && after !== null) { + assert.ok(after - before < 20, `${after - before} descriptors were left open by 1000 connections`) + } +}) + +test('a hundred streams stopped halfway leave the connection as it was', async (t) => { + const { conn } = await twoPeople(t) + + for (let n = 0; n < 100; n++) { + for await (const row of conn.stream(READ, null, { batchRows: 1 })) { + assert.equal(typeof row.id, 'bigint') + break + } + } + + // Each of those left a statement to be stopped and a thread to be + // joined, and the connection they all ran on still runs statements. + assert.equal((await conn.query(READ)).length, 2) +}) + +test('a statement that failed wrote nothing and left the connection alone', async (t) => { + const { conn } = await twoPeople(t) + + await assert.rejects(() => conn.exec("INSERT (p:person {id: 1 / 0, name: 'zoe'})"), { + name: 'ZuDataError', + }) + assert.equal((await conn.query(READ)).length, 2) + + // And the statement after the failed one is an ordinary statement: + // the write that failed took no lock and left no half-written row. + await conn.exec("INSERT (p:person {id: 3, name: 'iris'})") + assert.equal((await conn.query(READ)).length, 3) +}) + +test('a connection closed under a stream ends it rather than hanging on it', async (t) => { + const { conn } = await twoPeople(t) + + const stream = conn.stream(READ, null, { batchRows: 1 }) + const batches = stream.batches() + await batches.next() + // Closing does not wait for a reader that may never come back, and + // reading does not wait for a connection that has gone. Both of those + // waiting for each other is the one failure worse than an error. + conn.close() + + let read = 1 + for (;;) { + const step = await batches.next() + if (step.done) break + read += 1 + assert.ok(read < 10, 'the stream never ended') + } + assert.equal(conn.open, false) +}) + +test('the programs that look like misuse and are not', async (t) => { + const { conn, path } = await twoPeople(t) + + // A parameter the statement does not read is not an error. A caller + // that passes one object to several statements is doing something + // reasonable, and refusing it would make that object the union of + // what every statement wants. + assert.equal((await conn.query(READ, { unread: 1 })).length, 2) + + // A label nothing carries matches nothing. A pattern with no answer + // is the ordinary answer to a question about a graph, and the other + // reading gives a query that fails on the day the last row of a label + // is deleted. + assert.deepEqual(await conn.query('MATCH (p:nobody) RETURN p.id AS id'), []) + + // 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. + conn.stream(READ) + assert.equal((await conn.query(READ)).length, 2) + + // Reading a cursor that was cancelled is the end of the stream rather + // than a failure, and cancelling one twice is cancelling it once. A + // caller cleaning up in a `finally` is allowed to ask twice. + const cursor = conn.cursor(READ) + await cursor.cancel() + await cursor.cancel() + assert.equal(await cursor.next(), null) + + // Closing twice is not an error either, and neither is disposing of + // something already closed, which is what an explicit `close` inside + // an `await using` block leaves behind. + const twice = await connect(path) + twice.close() + twice.close() + await twice.dispose() + assert.equal(twice.open, false) +}) + +// How many files this process has open, or `null` where the system will +// not say. Linux and macOS both publish it as a directory, and Windows +// publishes nothing, which is why every use of this is guarded rather +// than skipped: a descriptor leak is not platform-specific and the two +// platforms that can see one are enough to find it. +async function descriptors() { + for (const where of ['/proc/self/fd', '/dev/fd']) { + try { + return (await readdir(where)).length + } catch { + continue + } + } + return null +} diff --git a/tools/valgrind.supp b/tools/valgrind.supp new file mode 100644 index 0000000..a411575 --- /dev/null +++ b/tools/valgrind.supp @@ -0,0 +1,74 @@ +# What the leak job does not count, and why. +# +# Node leaks a little on purpose, and which little depends on how the +# binary was built. A node linked against the system OpenSSL leaks +# nothing this job can see, because the library is somebody else's file +# and the loader never unloads it; the node that GitHub's runners +# install has OpenSSL statically inside it, so its one-time table of +# built-in compression methods is twenty four bytes allocated at the +# first use of the default library context and never freed, and it lands +# in this report as a definite leak inside the node binary. That is a +# real allocation and it is not this addon's, and no version of this +# addon can free it. +# +# So the rule is one rule, and it is narrow on purpose. A leak is +# ignored only when the frame that allocated it is inside the node +# binary itself. Memory this addon allocates has zudb.*.node in that +# frame instead and is reported, which is what the job is for. The usual +# shape of a suppression file, a list that grows a line every time +# somebody wants the build green, would suppress the thing being looked +# for on the first bad week; this one cannot, because it names the +# allocator rather than the symptom. +# +# The gap it leaves, said out loud: a napi handle this addon leaks was +# allocated by node on its behalf, so it lands under this rule. That +# leak is the one the suite catches from the other side, where it opens +# and closes a thousand connections and counts the process descriptors +# before and after. + +{ + node allocating for itself, malloc + Memcheck:Leak + match-leak-kinds: definite,possible + fun:malloc + obj:*/node +} + +{ + node allocating for itself, realloc + Memcheck:Leak + match-leak-kinds: definite,possible + fun:realloc + obj:*/node +} + +{ + node allocating for itself, calloc + Memcheck:Leak + match-leak-kinds: definite,possible + fun:calloc + obj:*/node +} + +# Node is C++ and most of what it keeps comes through here rather than +# through malloc: `node -e ''` alone holds on to the script object it +# compiled. This addon is Rust and allocates through the C entries +# above, so naming the C++ ones costs it nothing. The names are the +# mangled ones because that is what valgrind matches against, whatever +# it prints: _Znwm is operator new and _Znam is operator new[]. + +{ + node allocating for itself, new + Memcheck:Leak + match-leak-kinds: definite,possible + fun:_Znw* + obj:*/node +} + +{ + node allocating for itself, new[] + Memcheck:Leak + match-leak-kinds: definite,possible + fun:_Zna* + obj:*/node +}