Add query timeout option to interrupt long-running queries

A single background task with a min-heap manages all query
deadlines efficiently. When a query starts, a TimeoutGuard is
acquired; if the deadline expires before the guard is dropped,
the connection is interrupted via sqlite3_interrupt().

Author: penberg

compat.js

@@ -28,6 +28,26 @@ function convertError(err) {
   return err;
 }
 
+function isQueryOptions(value) {
+  return value != null
+    && typeof value === "object"
+    && !Array.isArray(value)
+    && Object.prototype.hasOwnProperty.call(value, "queryTimeout");
+}
+
+function splitBindParameters(bindParameters) {
+  if (bindParameters.length === 0) {
+    return { params: undefined, queryOptions: undefined };
+  }
+  if (bindParameters.length > 1 && isQueryOptions(bindParameters[bindParameters.length - 1])) {
+    return {
+      params: bindParameters.length === 2 ? bindParameters[0] : bindParameters.slice(0, -1),
+      queryOptions: bindParameters[bindParameters.length - 1],
+    };
+  }
+  return { params: bindParameters.length === 1 ? bindParameters[0] : bindParameters, queryOptions: undefined };
+}
+
 /**
  * Database represents a connection that can prepare and execute SQL statements.
  */
@@ -176,9 +196,9 @@ class Database {
    *
    * @param {string} sql - The SQL statement string to execute.
    */
-  exec(sql) {
+  exec(sql, queryOptions) {
     try {
-      databaseExecSync(this.db, sql);
+      databaseExecSync(this.db, sql, queryOptions);
     } catch (err) {
       throw convertError(err);
     }
@@ -263,7 +283,8 @@ class Statement {
    */
   run(...bindParameters) {
     try {
-      return statementRunSync(this.stmt, ...bindParameters);
+      const { params, queryOptions } = splitBindParameters(bindParameters);
+      return statementRunSync(this.stmt, params, queryOptions);
     } catch (err) {
       throw convertError(err);
     }
@@ -276,7 +297,8 @@ class Statement {
    */
   get(...bindParameters) {
     try {
-      return statementGetSync(this.stmt, ...bindParameters);
+      const { params, queryOptions } = splitBindParameters(bindParameters);
+      return statementGetSync(this.stmt, params, queryOptions);
     } catch (err) {
       throw convertError(err);
     }
@@ -289,7 +311,8 @@ class Statement {
    */
   iterate(...bindParameters) {
     try {
-      const it = statementIterateSync(this.stmt, ...bindParameters);
+      const { params, queryOptions } = splitBindParameters(bindParameters);
+      const it = statementIterateSync(this.stmt, params, queryOptions);
       return {
         next: () => iteratorNextSync(it),
         [Symbol.iterator]() {

docs/api.md

@@ -22,6 +22,7 @@ You can use the `options` parameter to specify various options. Options supporte
 - `syncPeriod`: synchronize the database periodically every `syncPeriod` seconds.
 - `authToken`: authentication token for the provider URL (optional).
 - `timeout`: number of milliseconds to wait on locked database before returning `SQLITE_BUSY` error
+- `queryTimeout`: maximum number of milliseconds a query is allowed to run before being interrupted with `SQLITE_INTERRUPT` error
 
 The function returns a `Database` object.
 
@@ -97,13 +98,14 @@ const stmt = db.prepare("SELECT * FROM users");
 
 Loads a SQLite3 extension
 
-### exec(sql) ⇒ this
+### exec(sql[, queryOptions]) ⇒ this
 
 Executes a SQL statement.
 
 | Param  | Type                | Description                          |
 | ------ | ------------------- | ------------------------------------ |
 | sql    | <code>string</code> | The SQL statement string to execute. |
+| queryOptions | <code>object</code> | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). |
 
 ### interrupt() ⇒ this
 
@@ -119,39 +121,43 @@ Closes the database connection.
 
 ## Methods
 
-### run([...bindParameters]) ⇒ object
+### run([...bindParameters][, queryOptions]) ⇒ object
 
 Executes the SQL statement and returns an info object.
 
 | Param          | Type                          | Description                                      |
 | -------------- | ----------------------------- | ------------------------------------------------ |
 | bindParameters | <code>array of objects</code> | The bind parameters for executing the statement. |
+| queryOptions   | <code>object</code>           | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). |
 
 The returned info object contains two properties: `changes` that describes the number of modified rows and `info.lastInsertRowid` that represents the `rowid` of the last inserted row.
 
-### get([...bindParameters]) ⇒ row
+### get([...bindParameters][, queryOptions]) ⇒ row
 
 Executes the SQL statement and returns the first row.
 
 | Param          | Type                          | Description                                      |
 | -------------- | ----------------------------- | ------------------------------------------------ |
 | bindParameters | <code>array of objects</code> | The bind parameters for executing the statement. |
+| queryOptions   | <code>object</code>           | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). |
 
-### all([...bindParameters]) ⇒ array of rows
+### all([...bindParameters][, queryOptions]) ⇒ array of rows
 
 Executes the SQL statement and returns an array of the resulting rows.
 
 | Param          | Type                          | Description                                      |
 | -------------- | ----------------------------- | ------------------------------------------------ |
 | bindParameters | <code>array of objects</code> | The bind parameters for executing the statement. |
+| queryOptions   | <code>object</code>           | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). |
 
-### iterate([...bindParameters]) ⇒ iterator
+### iterate([...bindParameters][, queryOptions]) ⇒ iterator
 
 Executes the SQL statement and returns an iterator to the resulting rows.
 
 | Param          | Type                          | Description                                      |
 | -------------- | ----------------------------- | ------------------------------------------------ |
 | bindParameters | <code>array of objects</code> | The bind parameters for executing the statement. |
+| queryOptions   | <code>object</code>           | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). |
 
 ### pluck([toggleState]) ⇒ this
 

index.d.ts

@@ -13,6 +13,11 @@ export interface Options {
   encryptionCipher?: string
   encryptionKey?: string
   remoteEncryptionKey?: string
+  queryTimeout?: number
+}
+/** Per-query execution options. */
+export interface QueryOptions {
+  queryTimeout?: number
 }
 export declare function connect(path: string, opts?: Options | undefined | null): Promise<Database>
 /** Result of a database sync operation. */
@@ -27,12 +32,12 @@ export declare function databasePrepareSync(db: Database, sql: string): Statemen
 /** Syncs the database in blocking mode. */
 export declare function databaseSyncSync(db: Database): SyncResult
 /** Executes SQL in blocking mode. */
-export declare function databaseExecSync(db: Database, sql: string): void
+export declare function databaseExecSync(db: Database, sql: string, queryOptions?: QueryOptions | undefined | null): void
 /** Gets first row from statement in blocking mode. */
-export declare function statementGetSync(stmt: Statement, params?: unknown | undefined | null): unknown
+export declare function statementGetSync(stmt: Statement, params?: unknown | undefined | null, queryOptions?: QueryOptions | undefined | null): unknown
 /** Runs a statement in blocking mode. */
-export declare function statementRunSync(stmt: Statement, params?: unknown | undefined | null): RunResult
-export declare function statementIterateSync(stmt: Statement, params?: unknown | undefined | null): RowsIterator
+export declare function statementRunSync(stmt: Statement, params?: unknown | undefined | null, queryOptions?: QueryOptions | undefined | null): RunResult
+export declare function statementIterateSync(stmt: Statement, params?: unknown | undefined | null, queryOptions?: QueryOptions | undefined | null): RowsIterator
 /** SQLite `run()` result object */
 export interface RunResult {
   changes: number
@@ -116,7 +121,7 @@ export declare class Database {
    * * `env` - The environment.
    * * `sql` - The SQL statement to execute.
    */
-  exec(sql: string): Promise<void>
+  exec(sql: string, queryOptions?: QueryOptions | undefined | null): Promise<void>
   /**
    * Syncs the database.
    *
@@ -153,7 +158,7 @@ export declare class Statement {
    *
    * * `params` - The parameters to bind to the statement.
    */
-  run(params?: unknown | undefined | null): RunResult
+  run(params?: unknown | undefined | null, queryOptions?: QueryOptions | undefined | null): RunResult
   /**
    * Executes a SQL statement and returns the first row.
    *
@@ -162,7 +167,7 @@ export declare class Statement {
    * * `env` - The environment.
    * * `params` - The parameters to bind to the statement.
    */
-  get(params?: unknown | undefined | null): object
+  get(params?: unknown | undefined | null, queryOptions?: QueryOptions | undefined | null): object
   /**
    * Create an iterator over the rows of a statement.
    *
@@ -171,7 +176,7 @@ export declare class Statement {
    * * `env` - The environment.
    * * `params` - The parameters to bind to the statement.
    */
-  iterate(params?: unknown | undefined | null): object
+  iterate(params?: unknown | undefined | null, queryOptions?: QueryOptions | undefined | null): object
   raw(raw?: boolean | undefined | null): this
   pluck(pluck?: boolean | undefined | null): this
   timing(timing?: boolean | undefined | null): this

integration-tests/tests/async.test.js

@@ -398,6 +398,56 @@ test.serial("Timeout option", async (t) => {
   fs.unlinkSync(path);
 });
 
+test.serial("Query timeout option interrupts long-running query", async (t) => {
+  const queryTimeout = 100;
+  const [db, errorType] = await connect(":memory:", { queryTimeout });
+  const stmt = await db.prepare(
+    "WITH RECURSIVE infinite_loop(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM infinite_loop) SELECT * FROM infinite_loop;"
+  );
+
+  await t.throwsAsync(async () => {
+    await stmt.all();
+  }, {
+    instanceOf: errorType,
+    message: "interrupted",
+    code: "SQLITE_INTERRUPT",
+  });
+
+  db.close();
+});
+
+test.serial("Query timeout option allows short-running query", async (t) => {
+  const [db] = await connect(":memory:", { queryTimeout: 100 });
+  const stmt = await db.prepare("SELECT 1 AS value");
+  t.deepEqual(await stmt.get(), { value: 1 });
+  db.close();
+});
+
+test.serial("Per-query timeout option interrupts long-running Statement.all()", async (t) => {
+  const [db, errorType] = await connect(":memory:");
+  const stmt = await db.prepare(
+    "WITH RECURSIVE infinite_loop(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM infinite_loop) SELECT * FROM infinite_loop;"
+  );
+
+  await t.throwsAsync(async () => {
+    await stmt.all(undefined, { queryTimeout: 100 });
+  }, {
+    instanceOf: errorType,
+    message: "interrupted",
+    code: "SQLITE_INTERRUPT",
+  });
+
+  db.close();
+});
+
+test.serial("Per-query timeout option is accepted by Database.exec()", async (t) => {
+  const [db] = await connect(":memory:");
+  await db.exec("SELECT 1", { queryTimeout: 100 });
+  t.pass();
+
+  db.close();
+});
+
 test.serial("Concurrent writes over same connection", async (t) => {
   const db = t.context.db;
   await db.exec(`

integration-tests/tests/sync.test.js

@@ -457,6 +457,75 @@ test.serial("Timeout option", async (t) => {
   fs.unlinkSync(path);
 });
 
+test.serial("Query timeout option interrupts long-running query", async (t) => {
+  if (t.context.provider === "sqlite") {
+    t.assert(true);
+    return;
+  }
+
+  const [db, errorType] = await connect(":memory:", { queryTimeout: 100 });
+  const stmt = db.prepare(
+    "WITH RECURSIVE infinite_loop(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM infinite_loop) SELECT * FROM infinite_loop;"
+  );
+
+  t.throws(() => {
+    stmt.all();
+  }, {
+    instanceOf: errorType,
+    message: "interrupted",
+    code: "SQLITE_INTERRUPT",
+  });
+
+  db.close();
+});
+
+test.serial("Query timeout option allows short-running query", async (t) => {
+  if (t.context.provider === "sqlite") {
+    t.assert(true);
+    return;
+  }
+
+  const [db] = await connect(":memory:", { queryTimeout: 100 });
+  const stmt = db.prepare("SELECT 1 AS value");
+  t.deepEqual(stmt.get(), { value: 1 });
+  db.close();
+});
+
+test.serial("Per-query timeout option interrupts long-running Statement.all()", async (t) => {
+  if (t.context.provider === "sqlite") {
+    t.assert(true);
+    return;
+  }
+
+  const [db, errorType] = await connect(":memory:");
+  const stmt = db.prepare(
+    "WITH RECURSIVE infinite_loop(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM infinite_loop) SELECT * FROM infinite_loop;"
+  );
+
+  t.throws(() => {
+    stmt.all(undefined, { queryTimeout: 100 });
+  }, {
+    instanceOf: errorType,
+    message: "interrupted",
+    code: "SQLITE_INTERRUPT",
+  });
+
+  db.close();
+});
+
+test.serial("Per-query timeout option is accepted by Database.exec()", async (t) => {
+  if (t.context.provider === "sqlite") {
+    t.assert(true);
+    return;
+  }
+
+  const [db] = await connect(":memory:");
+  db.exec("SELECT 1", { queryTimeout: 100 });
+  t.pass();
+
+  db.close();
+});
+
 test.serial("Statement.reader [SELECT is true]", async (t) => {
   const db = t.context.db;
   const stmt = db.prepare("SELECT * FROM users WHERE id = ?");