Expose sqlite3_interrupt() as db.interrupt() — closes #406 (#407)

Author: ospfranco

README.md

@@ -25,6 +25,7 @@ Some of the big supported features:
 - Custom tokenizers
 - Load runtime extensions
 - JSONB support
+- Native query interruption via `db.interrupt()`
 
 It also contains a simple [Key-Value store](https://op-engineering.github.io/op-sqlite/docs/key_value_storage) you can use without adding one more dependency to your app.
 

cpp/DBHostObject.cpp

@@ -294,6 +294,12 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
 
   function_map["close"] = HFN(this) {
     invalidated = true;
+    // Abort pending native SQLite work before waiting on the thread pool.
+#if !defined(OP_SQLITE_USE_LIBSQL) && !defined(OP_SQLITE_USE_TURSO)
+    if (db != nullptr) {
+      sqlite3_interrupt(db);
+    }
+#endif
     // Drain any in-flight async queries before closing the db handle.
     // Without this, a queued/running execute() on the thread pool may
     // dereference the freed sqlite3* pointer → heap corruption / SIGABRT.
@@ -309,12 +315,39 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) {
     return {};
   });
 
+  function_map["interrupt"] = HFN(this) {
+    if (invalidated) {
+      throw std::runtime_error("[op-sqlite][interrupt] database is closed");
+    }
+
+#ifdef OP_SQLITE_USE_LIBSQL
+    throw std::runtime_error("[op-sqlite][interrupt] sqlite3_interrupt is not "
+                             "supported with libsql");
+#elif defined(OP_SQLITE_USE_TURSO)
+    throw std::runtime_error("[op-sqlite][interrupt] sqlite3_interrupt is not "
+                             "supported with Turso");
+#else
+    if (db == nullptr) {
+      throw std::runtime_error("[op-sqlite][interrupt] database is null");
+    }
+
+    sqlite3_interrupt(db);
+    return {};
+#endif
+  });
+
   function_map["delete"] = HFN(this) {
     if (count != 0) {
       throw std::runtime_error("[op-sqlite] Delete no longer takes arguments");
     }
 
     invalidated = true;
+    // Abort pending native SQLite work before waiting on the thread pool.
+#if !defined(OP_SQLITE_USE_LIBSQL) && !defined(OP_SQLITE_USE_TURSO)
+    if (db != nullptr) {
+      sqlite3_interrupt(db);
+    }
+#endif
     // Drain any in-flight async queries before closing/removing the db handle.
     // Without this, queued/running work may dereference a freed sqlite handle.
     thread_pool->waitFinished();

docs/docs/api.md

@@ -120,6 +120,28 @@ On web, `execute()` runs the full SQL string passed to it.
 On native, `execute()` currently runs only the first prepared statement.
 If you need identical behavior across platforms, avoid multi-statement SQL strings.
 
+## Interrupting a Query
+
+On native, `interrupt()` aborts any pending database operation on this connection. It is safe to call from a thread different from the one running the operation. The interrupted query returns `SQLITE_INTERRUPT`; any in-flight transaction is rolled back. This calls SQLite's native [`sqlite3_interrupt()`](https://sqlite.org/c3ref/interrupt.html).
+
+`interrupt()` is not available when op-sqlite is built with the `libsql` or `turso` backend.
+
+```tsx
+const query = db.execute(longRunningQuery);
+
+setTimeout(() => {
+  db.interrupt();
+}, 100);
+
+try {
+  await query;
+} catch (error) {
+  // SQLITE_INTERRUPT
+}
+```
+
+On web, `interrupt()` is not supported.
+
 ### Execute with Host Objects
 
 It’s possible to return HostObjects when using a query. The benefit is that HostObjects are only created in C++ and only when you try to access a value inside of them a C++ value → JS value conversion happens. This means creation is fast, property access is slow. The use case is clear if you are returning **massive** amount of objects but only displaying/accessing a few of them at the time.

example/src/tests/queries.ts

@@ -14,7 +14,7 @@ import {
 	expect,
 	it,
 } from "@op-engineering/op-test";
-import { chance } from "./utils";
+import { chance, sleep } from "./utils";
 
 // import pkg from '../../package.json'
 
@@ -107,6 +107,91 @@ describe("Queries tests", () => {
 		}
 	});
 
+	it("interrupt is safe to call with no in-flight query", () => {
+		if (isLibsql() || isTurso()) {
+			return;
+		}
+
+		let threw = false;
+		try {
+			db.interrupt();
+		} catch (_e) {
+			threw = true;
+		}
+
+		expect(threw).toEqual(false);
+	});
+
+	it("interrupt aborts an in-flight query and rolls back the transaction", async () => {
+		if (isLibsql() || isTurso()) {
+			return;
+		}
+
+		await db.execute("DROP TABLE IF EXISTS InterruptTest;");
+		await db.execute("CREATE TABLE InterruptTest (n INTEGER);");
+
+		const longQuery = `
+			WITH RECURSIVE seq(n) AS (
+				SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 100000000
+			)
+			INSERT INTO InterruptTest SELECT n FROM seq;
+		`;
+
+		const queryPromise = db.execute(longQuery);
+
+		await sleep(50);
+		db.interrupt();
+
+		let interrupted = false;
+		try {
+			await queryPromise;
+		} catch (e: any) {
+			interrupted = /interrupt|interrupted|abort|code 9|SQLITE_INTERRUPT/i.test(
+				String(e?.message ?? e),
+			);
+		}
+
+		expect(interrupted).toEqual(true);
+
+		const count = await db.execute("SELECT COUNT(*) AS n FROM InterruptTest;");
+		expect(count.rows[0]!.n).toEqual(0);
+	});
+
+	it("close interrupts an in-flight query before teardown", async () => {
+		if (isLibsql() || isTurso()) {
+			return;
+		}
+
+		await db.execute("DROP TABLE IF EXISTS CloseInterruptTest;");
+		await db.execute("CREATE TABLE CloseInterruptTest (n INTEGER);");
+
+		const longQuery = `
+			WITH RECURSIVE seq(n) AS (
+				SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 100000000
+			)
+			INSERT INTO CloseInterruptTest SELECT n FROM seq;
+		`;
+
+		const queryPromise = db.execute(longQuery);
+
+		await sleep(50);
+		const startedAt = Date.now();
+		db.close();
+		const elapsedMs = Date.now() - startedAt;
+
+		await queryPromise.catch(() => undefined);
+		expect(elapsedMs < 2000).toEqual(true);
+
+		const cleanupDb = open({
+			name: "queries.sqlite",
+			encryptionKey: "test",
+		});
+		cleanupDb.delete();
+
+		// @ts-expect-error Prevent afterEach from deleting a closed handle.
+		db = null;
+	});
+
 	it("executeSync", () => {
 		const res = db.executeSync("SELECT 1");
 		expect(res.rowsAffected).toEqual(0);

src/functions.ts

@@ -85,6 +85,7 @@ function enhanceDB(db: _InternalDB, options: DBParams): DB {
 		setReservedBytes: db.setReservedBytes,
 		getReservedBytes: db.getReservedBytes,
 		close: db.close,
+		interrupt: db.interrupt,
 		closeAsync: async () => {
 			db.close();
 		},

src/functions.web.ts

@@ -188,6 +188,7 @@ function enhanceWebDb(
 		closeAsync: async () => {
 			await db.closeAsync?.();
 		},
+		interrupt: unsupported("interrupt"),
 		delete: unsupported("delete"),
 		attach: unsupported("attach"),
 		detach: unsupported("detach"),
@@ -352,6 +353,9 @@ async function createWebDb(params: {
 				dbId,
 			});
 		},
+		interrupt: () => {
+			throwSyncApiError("interrupt");
+		},
 		delete: () => {
 			throwSyncApiError("delete");
 		},

src/types.ts

@@ -102,6 +102,7 @@ export type PreparedStatement = {
 export type _InternalDB = {
 	close: () => void;
 	closeAsync?: () => Promise<void>;
+	interrupt: () => void;
 	delete: () => void;
 	attach: (params: {
 		secondaryDbFileName: string;
@@ -153,6 +154,14 @@ export type _InternalDB = {
 export type DB = {
 	close: () => void;
 	closeAsync: () => Promise<void>;
+	/**
+	 * Aborts any pending database operation on this connection.
+	 *
+	 * Calls SQLite's native sqlite3_interrupt(). Safe to call from a thread
+	 * different from the one running the operation. An interrupted operation
+	 * returns SQLITE_INTERRUPT and any in-flight transaction is rolled back.
+	 */
+	interrupt: () => void;
 	delete: () => void;
 	attach: (params: {
 		secondaryDbFileName: string;