From ec861d1fc964386809604f6e976c8757cad24c32 Mon Sep 17 00:00:00 2001 From: EJ Campbell Date: Mon, 21 Sep 2026 13:57:32 -0700 Subject: [PATCH] Reset a statement after run() the way better-sqlite3 does `run()` reset a statement only before stepping it, and returned as soon as the step returned. SQLite leaves a statement in progress in two cases that this missed: * The step failed with `SQLITE_BUSY`. SQLite suspends the statement so the caller can step it again, and it keeps counting as an active writer. Every `COMMIT` on the connection then fails with `SQLITE_BUSY: cannot commit transaction - SQL statements in progress` until the statement is reused or garbage collected, although the lock is free. Through `@libsql/client`, one write batch that fails busy makes the following batches on the same client fail too. * The step returned a row. libsql's `run()` treats `SQLITE_ROW` as success, so `INSERT ... RETURNING` or `PRAGMA journal_mode=WAL` run through `run()` stopped after the first row. In autocommit mode the write is only committed when the statement halts, so the row stayed invisible to other connections, the write lock stayed held, the connection's own next `COMMIT` failed with the same error, and `changes` was reported as 0 because SQLite records it at halt. better-sqlite3 steps and then unconditionally resets in `run()`, and reads `changes` and `lastInsertRowid` after the reset. Do the same in both the sync and the promise API. In the promise API the reset runs inside the tokio future, on the thread that stepped the statement, before the result is handed back to the JavaScript thread. Tests, in both `sync.test.js` and `async.test.js`, keep the statement referenced so that garbage collection cannot hide the bug: * `BEGIN IMMEDIATE` fails with `SQLITE_BUSY` against a second connection holding the write lock, and a write transaction on the same connection must then commit. * `INSERT ... RETURNING` via `run()` reports `changes` of 1, is visible to and does not block a second connection, and the same connection can then run a transaction. * `SELECT` via `run()` does not keep a read transaction open in rollback-journal mode. * `PRAGMA journal_mode=WAL` via `run()` does not block the next `COMMIT`. All of them pass on better-sqlite3 as well. Before the fix, the busy and RETURNING cases fail on libsql. Co-authored-by: Pekka Enberg --- integration-tests/tests/async.test.js | 76 +++++++++++++++++++++++++++ integration-tests/tests/sync.test.js | 75 ++++++++++++++++++++++++++ src/lib.rs | 26 ++++++++- 3 files changed, 175 insertions(+), 2 deletions(-) diff --git a/integration-tests/tests/async.test.js b/integration-tests/tests/async.test.js index f6536a7..f09d608 100644 --- a/integration-tests/tests/async.test.js +++ b/integration-tests/tests/async.test.js @@ -987,6 +987,82 @@ test.serial("Database.batch() rejects non-array argument", async (t) => { await t.throwsAsync(() => db.batch("SELECT 1"), { instanceOf: TypeError }); }); +test.serial("A statement that failed with SQLITE_BUSY does not block the next COMMIT", async (t) => { + const path = genDatabaseFilename(); + const [holder] = await connect(path); + await holder.exec("PRAGMA journal_mode=WAL"); + await holder.exec("CREATE TABLE t(x)"); + const [db] = await connect(path, { timeout: 50 }); + await holder.exec("BEGIN IMMEDIATE"); + // Keep the failed statement referenced so that garbage collection cannot finalize it. + const begin = await db.prepare("BEGIN IMMEDIATE"); + await t.throwsAsync(() => begin.run(), { code: "SQLITE_BUSY" }); + await holder.exec("ROLLBACK"); + await (await db.prepare("BEGIN IMMEDIATE")).run(); + await (await db.prepare("INSERT INTO t VALUES (1)")).run(); + await (await db.prepare("COMMIT")).run(); + const row = await (await db.prepare("SELECT count(*) AS n FROM t")).get(); + t.is(row.n, 1); + db.close(); + holder.close(); + for (const suffix of ["", "-wal", "-shm"]) fs.rmSync(path + suffix, { force: true }); +}); + +test.serial("Statement.run() on INSERT ... RETURNING commits the write", async (t) => { + const path = genDatabaseFilename(); + const [db] = await connect(path); + await db.exec("CREATE TABLE t(x)"); + // Keep the statement referenced so that garbage collection cannot finalize it. + const insert = await db.prepare("INSERT INTO t VALUES (?) RETURNING x"); + const info = await insert.run(42); + t.is(info.changes, 1); + t.is(info.lastInsertRowid, 1); + // The write is visible to, and does not block, another connection. + const [other] = await connect(path, { timeout: 100 }); + t.is((await (await other.prepare("SELECT count(*) AS n FROM t")).get()).n, 1); + await (await other.prepare("INSERT INTO t VALUES (2)")).run(); + // The same connection can run a transaction afterwards. + await (await db.prepare("BEGIN IMMEDIATE")).run(); + await (await db.prepare("INSERT INTO t VALUES (3)")).run(); + await (await db.prepare("COMMIT")).run(); + t.is((await (await db.prepare("SELECT count(*) AS n FROM t")).get()).n, 3); + other.close(); + db.close(); + for (const suffix of ["", "-journal", "-wal", "-shm"]) fs.rmSync(path + suffix, { force: true }); +}); + +test.serial("Statement.run() on a SELECT does not keep a read transaction open", async (t) => { + const path = genDatabaseFilename(); + const [db] = await connect(path); + await db.exec("CREATE TABLE t(x)"); + await db.exec("INSERT INTO t VALUES (1)"); + // Keep the statement referenced so that garbage collection cannot finalize it. + const select = await db.prepare("SELECT x FROM t"); + t.is((await select.run()).changes, 0); + // In rollback-journal mode an open read transaction would block this write. + const [other] = await connect(path, { timeout: 100 }); + await (await other.prepare("INSERT INTO t VALUES (2)")).run(); + t.is((await (await db.prepare("SELECT count(*) AS n FROM t")).get()).n, 2); + other.close(); + db.close(); + for (const suffix of ["", "-journal"]) fs.rmSync(path + suffix, { force: true }); +}); + +test.serial("Statement.run() on a PRAGMA returning a row does not block the next COMMIT", async (t) => { + const path = genDatabaseFilename(); + const [db] = await connect(path); + // Keep the statement referenced so that garbage collection cannot finalize it. + const pragma = await db.prepare("PRAGMA journal_mode=WAL"); + await pragma.run(); + await (await db.prepare("CREATE TABLE t(x)")).run(); + await (await db.prepare("BEGIN IMMEDIATE")).run(); + await (await db.prepare("INSERT INTO t VALUES (1)")).run(); + await (await db.prepare("COMMIT")).run(); + t.is((await (await db.prepare("SELECT count(*) AS n FROM t")).get()).n, 1); + db.close(); + for (const suffix of ["", "-wal", "-shm"]) fs.rmSync(path + suffix, { force: true }); +}); + const connect = async (path_opt, options = {}) => { const path = path_opt ?? "hello.db"; const provider = process.env.PROVIDER; diff --git a/integration-tests/tests/sync.test.js b/integration-tests/tests/sync.test.js index 7bcb05c..5e7f7a9 100644 --- a/integration-tests/tests/sync.test.js +++ b/integration-tests/tests/sync.test.js @@ -801,6 +801,81 @@ test.serial("Database.batch() rejects non-array argument", async (t) => { t.throws(() => db.batch("SELECT 1"), { instanceOf: TypeError }); }); +test.serial("A statement that failed with SQLITE_BUSY does not block the next COMMIT", async (t) => { + const path = genDatabaseFilename(); + const [holder] = await connect(path); + holder.exec("PRAGMA journal_mode=WAL"); + holder.exec("CREATE TABLE t(x)"); + const [db, errorType] = await connect(path, { timeout: 50 }); + holder.exec("BEGIN IMMEDIATE"); + // Keep the failed statement referenced so that garbage collection cannot finalize it. + const begin = db.prepare("BEGIN IMMEDIATE"); + t.throws(() => begin.run(), { instanceOf: errorType, code: "SQLITE_BUSY" }); + holder.exec("ROLLBACK"); + db.prepare("BEGIN IMMEDIATE").run(); + db.prepare("INSERT INTO t VALUES (1)").run(); + db.prepare("COMMIT").run(); + t.is(db.prepare("SELECT count(*) AS n FROM t").get().n, 1); + db.close(); + holder.close(); + for (const suffix of ["", "-wal", "-shm"]) fs.rmSync(path + suffix, { force: true }); +}); + +test.serial("Statement.run() on INSERT ... RETURNING commits the write", async (t) => { + const path = genDatabaseFilename(); + const [db] = await connect(path); + db.exec("CREATE TABLE t(x)"); + // Keep the statement referenced so that garbage collection cannot finalize it. + const insert = db.prepare("INSERT INTO t VALUES (?) RETURNING x"); + const info = insert.run(42); + t.is(info.changes, 1); + t.is(info.lastInsertRowid, 1); + // The write is visible to, and does not block, another connection. + const [other] = await connect(path, { timeout: 100 }); + t.is(other.prepare("SELECT count(*) AS n FROM t").get().n, 1); + other.prepare("INSERT INTO t VALUES (2)").run(); + // The same connection can run a transaction afterwards. + db.prepare("BEGIN IMMEDIATE").run(); + db.prepare("INSERT INTO t VALUES (3)").run(); + db.prepare("COMMIT").run(); + t.is(db.prepare("SELECT count(*) AS n FROM t").get().n, 3); + other.close(); + db.close(); + for (const suffix of ["", "-journal", "-wal", "-shm"]) fs.rmSync(path + suffix, { force: true }); +}); + +test.serial("Statement.run() on a SELECT does not keep a read transaction open", async (t) => { + const path = genDatabaseFilename(); + const [db] = await connect(path); + db.exec("CREATE TABLE t(x)"); + db.exec("INSERT INTO t VALUES (1)"); + // Keep the statement referenced so that garbage collection cannot finalize it. + const select = db.prepare("SELECT x FROM t"); + t.is(select.run().changes, 0); + // In rollback-journal mode an open read transaction would block this write. + const [other] = await connect(path, { timeout: 100 }); + other.prepare("INSERT INTO t VALUES (2)").run(); + t.is(db.prepare("SELECT count(*) AS n FROM t").get().n, 2); + other.close(); + db.close(); + for (const suffix of ["", "-journal"]) fs.rmSync(path + suffix, { force: true }); +}); + +test.serial("Statement.run() on a PRAGMA returning a row does not block the next COMMIT", async (t) => { + const path = genDatabaseFilename(); + const [db] = await connect(path); + // Keep the statement referenced so that garbage collection cannot finalize it. + const pragma = db.prepare("PRAGMA journal_mode=WAL"); + pragma.run(); + db.prepare("CREATE TABLE t(x)").run(); + db.prepare("BEGIN IMMEDIATE").run(); + db.prepare("INSERT INTO t VALUES (1)").run(); + db.prepare("COMMIT").run(); + t.is(db.prepare("SELECT count(*) AS n FROM t").get().n, 1); + db.close(); + for (const suffix of ["", "-wal", "-shm"]) fs.rmSync(path + suffix, { force: true }); +}); + const connect = async (path_opt, options = {}) => { const path = path_opt ?? "hello.db"; const provider = process.env.PROVIDER; diff --git a/src/lib.rs b/src/lib.rs index d4c0270..e6eb4a7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1066,7 +1066,18 @@ impl Statement { let future = async move { let _timeout_guard = register_timeout(&stmt, query_timeout); - stmt.run(params).await.map_err(Error::from)?; + let result = stmt.run(params).await; + // Reset the statement whether or not the step succeeded, as + // better-sqlite3 does. SQLite leaves a statement in progress after + // a step that returned a row (INSERT ... RETURNING, PRAGMA + // journal_mode=...) or that failed with SQLITE_BUSY. Until it is + // reset, an autocommit write stays uncommitted, its locks stay + // held, and every COMMIT on the connection fails with "SQL + // statements in progress". SQLite only records changes() when the + // statement halts, which the reset forces, so the counters below + // are read after it. + stmt.reset(); + result.map_err(Error::from)?; let changes = if conn.total_changes() == total_changes_before { 0 } else { @@ -1386,7 +1397,18 @@ pub fn statement_run_sync( let total_changes_before = conn.total_changes(); let start = std::time::Instant::now(); - inner_stmt.run(params).await.map_err(Error::from)?; + let result = inner_stmt.run(params).await; + // Reset the statement whether or not the step succeeded, as + // better-sqlite3 does. SQLite leaves a statement in progress after + // a step that returned a row (INSERT ... RETURNING, PRAGMA + // journal_mode=...) or that failed with SQLITE_BUSY. Until it is + // reset, an autocommit write stays uncommitted, its locks stay + // held, and every COMMIT on the connection fails with "SQL + // statements in progress". SQLite only records changes() when the + // statement halts, which the reset forces, so the counters below + // are read after it. + inner_stmt.reset(); + result.map_err(Error::from)?; let changes = if conn.total_changes() == total_changes_before { 0 } else {