diff --git a/Cargo.lock b/Cargo.lock index f6e7d2b..75de74c 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "thiserror", ] @@ -1301,7 +1301,7 @@ dependencies = [ [[package]] name = "zu-encoding" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" 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=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "zu-common", ] @@ -1369,7 +1369,7 @@ dependencies = [ [[package]] name = "zu-zu1" version = "0.0.1" -source = "git+https://github.com/tamnd/zu?rev=95c7c9909f3a2624515d27eb436da52936016960#95c7c9909f3a2624515d27eb436da52936016960" +source = "git+https://github.com/tamnd/zu?rev=130f67db924bcd0f766ee814b0da2edae32150d4#130f67db924bcd0f766ee814b0da2edae32150d4" dependencies = [ "crc32c", "loom", diff --git a/Cargo.toml b/Cargo.toml index c1d3ec5..f8081f2 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 = "95c7c9909f3a2624515d27eb436da52936016960" } -zu-common = { git = "https://github.com/tamnd/zu", rev = "95c7c9909f3a2624515d27eb436da52936016960" } +zudb = { package = "zu", git = "https://github.com/tamnd/zu", rev = "130f67db924bcd0f766ee814b0da2edae32150d4" } +zu-common = { git = "https://github.com/tamnd/zu", rev = "130f67db924bcd0f766ee814b0da2edae32150d4" } # 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 c90840d..5417c59 100644 --- a/README.md +++ b/README.md @@ -53,9 +53,29 @@ for (const { name } of await conn.query(`MATCH (p:Person) RETURN p.name AS name` It is the whole engine and not a reduced one: writes, transactions, the appender, registered frames and streams, all of it, on bytes that are not a file. `conn.memory` says which kind you have, since `path` cannot quite answer it on a filesystem that allows a colon in a name. Nothing survives the last connection, which is the point: a test, a script, or five minutes with the language costs no cleanup and leaves no `social.zu1` in a directory somebody has to notice later. +## A second connection, made from the first + +`duplicate()` is another connection to the same database, made from a connection rather than from a path. It is how a pool is written. + +```ts +import { connect } from "zudb"; + +await using conn = await connect("social.zu1"); +await using other = await conn.duplicate(); + +const rows = await other.query(`MATCH (p:Person) RETURN p.name AS name`); +console.log(rows.length); +``` + +It forks off the database the connection already holds rather than opening the file again, so it costs a schema load and no path lookup, and it works on a database in memory, where there is no path to open a second time. That was the gap worth closing: a pool that seeds itself and lets the first connection go had no way to a second one at all. + +The two are connections in every sense rather than two names for one. Each has its own prepared statements, its own caches and its own transaction, so a task taking one from a pool is not in whatever transaction the last borrower left open, and closing one does not close the other. What they share is the write side: they queue behind each other to write and each sees what the other has committed. Two of them also read at once, where two statements on one connection queue, which is the other reason to reach for this. + +The switches come across, including `bigIntMode` and `temporal`, because a pool handing out connections that answered differently from the one it was seeded with would be a trap. Other clients spell this call `cursor()`, after the way every embedded database has spelled it for thirty years. That name is taken here by `conn.cursor()`, which is a cursor over the rows of one statement and a different thing entirely, so this one says what it does. + ## 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, databases in memory, 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. An appender, for loading rows a batch at a time, and `load` for building a whole database out of columns and an edge list. Registered frames, so an Arrow table or an object of typed arrays is something a statement can match on without the rows being copied. `columnar`, for a result read down its columns as the buffers themselves rather than across its rows as objects. Prepared statements, compiled at the line that asked and run as often as wanted, and `explain` and `profile`, as a tree a program walks and as the listing a person reads. Both module formats, typed separately. +`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. `duplicate`, for a second connection made from the first. 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, databases in memory, 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. An appender, for loading rows a batch at a time, and `load` for building a whole database out of columns and an edge list. Registered frames, so an Arrow table or an object of typed arrays is something a statement can match on without the rows being copied. `columnar`, for a result read down its columns as the buffers themselves rather than across its rows as objects. Prepared statements, compiled at the line that asked and run as often as wanted, and `explain` and `profile`, as a tree a program walks and as the listing a person reads. 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. diff --git a/binding.d.cts b/binding.d.cts index 0ebaae3..fb9cba8 100644 --- a/binding.d.cts +++ b/binding.d.cts @@ -1017,6 +1017,41 @@ export declare class Connection { * because profiling it would apply the write. */ profile(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise + /** + * Another connection to the same database, made from this one. + * + * This is how a pool is written. `connect()` opens the file again + * and looks the database up by path; this forks off the one this + * connection already holds, which costs a schema load and no + * lookup, and works on a database in memory, where there is no + * path to open a second time. + * + * ```js + * await using other = await conn.duplicate() + * const rows = await other.query('MATCH (p:person) RETURN p.name AS name') + * ``` + * + * The two are connections in every sense rather than two names + * for one. Each has its own prepared statements, its own caches + * and its own transaction, so a task taking one from a pool is not + * in whatever transaction the last borrower left open, and closing + * one does not close the other. What they share is the write side: + * they queue behind each other to write and each sees what the + * other has committed, which is what two connections to one file + * have always done. + * + * Other clients call this `cursor()`, after the way every + * embedded database has spelled it for thirty years. That name is + * taken here by [`Connection::cursor`], which is a cursor over the + * rows of one statement and a different thing entirely, so this + * one says what it does. + * + * The switches this connection was opened with come across, + * including how it spells the values it gives back, because a pool + * handing out connections that answered differently from the one it + * was seeded with would be a trap nobody would look for. + */ + duplicate(): Promise /** * Runs one statement and gives back a cursor over its rows. * diff --git a/etc/zudb.api.md b/etc/zudb.api.md index 33d71a0..457d549 100644 --- a/etc/zudb.api.md +++ b/etc/zudb.api.md @@ -31,6 +31,7 @@ export class Connection { columnar(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise cursor(statement: string, params?: Record | null, options?: ZuStreamOptions | null): ZuCursor dispose(): Promise + duplicate(): Promise exec(statement: string, params?: Record | null, options?: ZuStatementOptions | null): Promise explain(statement: string): Promise get inTransaction(): boolean diff --git a/src/conn.rs b/src/conn.rs index 09d1f79..8bb085a 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -365,6 +365,57 @@ fn memory(config: Config) -> std::result::Result { }) } +/// Forking a second connection off the database one already holds. +/// +/// It takes the connection's lock like any statement, because the fork +/// reads the schema through the write side, and it is a task like any +/// statement for the same reason: a schema load on the runtime's +/// thread is the loop stopped for the length of one. +pub struct DuplicateTask { + inner: Arc>>, + alive: Arc, + in_txn: Arc, + spelling: Spelling, + path: String, + read_only: bool, + memory: bool, + /// Why this is not going to run, when it is not. + refused: Option, +} + +impl<'task> ScopedTask<'task> for DuplicateTask { + type Output = std::result::Result; + type JsValue = ClassInstance<'task, Connection>; + + fn compute(&mut self) -> Result { + if let Some(message) = self.refused.take() { + return Ok(Err(Failure::Usage(message))); + } + Ok(with(&self.inner, &self.alive, &self.in_txn, |conn| { + conn.duplicate().map_err(Failure::from) + })) + } + + fn resolve(&mut self, env: &'task Env, output: Self::Output) -> Result { + let made = output.map_err(|failure| failed(env, failure, None))?; + let mut instance = Connection { + interrupt: made.interrupt(), + inner: Arc::new(Mutex::new(Some(made))), + alive: Arc::new(AtomicBool::new(true)), + // Its own, and false: a fork is outside whatever + // transaction the connection it came from is in. + in_txn: Arc::new(AtomicBool::new(false)), + spelling: self.spelling, + path: self.path.clone(), + read_only: self.read_only, + memory: self.memory, + } + .into_instance(env)?; + wire_disposal(env, &mut instance, "dispose")?; + Ok(instance) + } +} + /// Opens or creates, then connects. /// /// A read-only open of a path that holds nothing fails as an open @@ -797,6 +848,56 @@ impl Connection { )) } + /// Another connection to the same database, made from this one. + /// + /// This is how a pool is written. `connect()` opens the file again + /// and looks the database up by path; this forks off the one this + /// connection already holds, which costs a schema load and no + /// lookup, and works on a database in memory, where there is no + /// path to open a second time. + /// + /// ```js + /// await using other = await conn.duplicate() + /// const rows = await other.query('MATCH (p:person) RETURN p.name AS name') + /// ``` + /// + /// The two are connections in every sense rather than two names + /// for one. Each has its own prepared statements, its own caches + /// and its own transaction, so a task taking one from a pool is not + /// in whatever transaction the last borrower left open, and closing + /// one does not close the other. What they share is the write side: + /// they queue behind each other to write and each sees what the + /// other has committed, which is what two connections to one file + /// have always done. + /// + /// Other clients call this `cursor()`, after the way every + /// embedded database has spelled it for thirty years. That name is + /// taken here by [`Connection::cursor`], which is a cursor over the + /// rows of one statement and a different thing entirely, so this + /// one says what it does. + /// + /// The switches this connection was opened with come across, + /// including how it spells the values it gives back, because a pool + /// handing out connections that answered differently from the one it + /// was seeded with would be a trap nobody would look for. + #[napi(ts_return_type = "Promise")] + pub fn duplicate(&self) -> AsyncTask { + let refused = match self.alive.load(Ordering::Acquire) { + true => None, + false => Some(CLOSED.to_string()), + }; + AsyncTask::new(DuplicateTask { + inner: Arc::clone(&self.inner), + alive: Arc::clone(&self.alive), + in_txn: Arc::clone(&self.in_txn), + spelling: self.spelling, + path: self.path.clone(), + read_only: self.read_only, + memory: self.memory, + refused, + }) + } + /// Runs one statement and gives back a cursor over its rows. /// /// The pull underneath `stream`, which is what a program uses. The diff --git a/test/connect.test.mjs b/test/connect.test.mjs index 122b960..b8bc814 100644 --- a/test/connect.test.mjs +++ b/test/connect.test.mjs @@ -184,3 +184,104 @@ test('a database in memory cannot be opened read-only', async (t) => { return true }) }) + +test('a duplicate is another connection to the same database', async (t) => { + // On a database in memory because that is the one with no path to + // reopen: nothing but the shared write side could carry the rows. + const conn = await connect() + t.after(() => conn.close()) + await conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + + const other = await conn.duplicate() + t.after(() => other.close()) + assert.deepEqual([...(await other.query('MATCH (p:person) RETURN p.name AS name'))], [{ name: 'ada' }]) + await other.exec("INSERT (p:person {id: 2, name: 'zoe'})") + assert.equal((await conn.query('MATCH (p:person) RETURN p.name AS name')).length, 2) +}) + +test('a duplicate says what the connection it came from says', async (t) => { + const { path } = await fresh(t) + const conn = await connect(path, { readOnly: true }) + t.after(() => conn.close()) + const other = await conn.duplicate() + t.after(() => other.close()) + + assert.equal(other.path, conn.path) + assert.equal(other.readOnly, true) + assert.equal(other.memory, false) + assert.equal(other.open, true) +}) + +test('a duplicate outlives the connection it was made from', async (t) => { + // It is a connection and not a view of one, which is what lets a pool + // hand its seed back. + const conn = await connect() + await conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + const other = await conn.duplicate() + t.after(() => other.close()) + + conn.close() + assert.deepEqual([...(await other.query('MATCH (p:person) RETURN p.name AS name'))], [{ name: 'ada' }]) +}) + +test('a duplicate has a transaction of its own', async (t) => { + const conn = await connect() + t.after(() => conn.close()) + await conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + const other = await conn.duplicate() + t.after(() => other.close()) + + const work = await conn.transaction() + assert.equal(conn.inTransaction, true) + assert.equal(other.inTransaction, false) + await work.rollback() +}) + +test('a duplicate spells values the way the connection it came from does', async (t) => { + const conn = await connect({ bigIntMode: 'number' }) + t.after(() => conn.close()) + await conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + const other = await conn.duplicate() + t.after(() => other.close()) + + assert.deepEqual([...(await other.query('MATCH (p:person) RETURN p.id AS id'))], [{ id: 1 }]) +}) + +test('two connections run at once where one would queue', async (t) => { + const conn = await connect() + t.after(() => conn.close()) + await conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + const other = await conn.duplicate() + t.after(() => other.close()) + + const both = await Promise.all([ + conn.query('MATCH (p:person) RETURN p.name AS name'), + other.query('MATCH (p:person) RETURN p.name AS name'), + ]) + assert.deepEqual(both.map((rows) => [...rows]), [[{ name: 'ada' }], [{ name: 'ada' }]]) +}) + +test('a closed connection duplicates nothing', async () => { + const conn = await connect() + conn.close() + + await assert.rejects(() => conn.duplicate(), (err) => { + assert.equal(err.name, 'ZuUsageError') + assert.match(err.message, /closed/) + return true + }) +}) + +test('a duplicate closes with `await using` like the one it came from', async (t) => { + const conn = await connect() + t.after(() => conn.close()) + await conn.exec("INSERT (p:person {id: 1, name: 'ada'})") + + let held + { + await using other = await conn.duplicate() + held = other + assert.equal((await other.query('MATCH (p:person) RETURN p.name AS name')).length, 1) + } + assert.equal(held.open, false) +}) diff --git a/test/readme.test.mjs b/test/readme.test.mjs index 62d9adc..620bddd 100644 --- a/test/readme.test.mjs +++ b/test/readme.test.mjs @@ -81,7 +81,7 @@ async function installed(t, program) { } test('the README prints programs and fragments and knows which is which', async () => { - assert.equal((await programs()).length, 3, "the README's whole programs") + assert.equal((await programs()).length, 4, "the README's whole programs") assert.ok((await blocks('ts')).length > (await programs()).length, 'and its fragments') }) @@ -115,6 +115,19 @@ test('the program with no path leaves the directory empty', { skip: !NODE }, asy assert.deepEqual(await readdir(dir), ['main.ts', 'node_modules', 'package.json'].sort()) }) +test('the duplicate program runs as printed', { skip: !NODE }, async (t) => { + const program = (await programs()).find((block) => block.includes('conn.duplicate()')) + assert.ok(program, 'the block that makes a second connection') + + const { dir, file } = await installed(t, program) + const { stdout } = await run(process.execPath, [file], { cwd: dir }) + + // The database it makes is empty, so the count is the honest zero + // rather than a number the page had to arrange for. + assert.equal(stdout.trim(), '0') + await stat(join(dir, 'social.zu1')) +}) + test('every whole program in the README runs', { skip: !NODE }, async (t) => { for (const program of await programs()) { const { dir, file } = await installed(t, program)