Fix race condition in query timeouts (#218)

The previous fix (0a3d2bd) eagerly released TimeoutGuard when an
iterator finished, but a race remained: the background task could peek
the heap, see an active entry, and call conn.interrupt() while the query
was finishing on another thread. Because SQLite's interrupt flag is
connection-scoped, a late-arriving signal would pend on the connection
and fire on the next operation — typically a prepare() or EXPLAIN issued
right after a timed-out stmt.get(), surfacing as a spurious
SQLITE_INTERRUPT.

Fix:

- Make QueryTimeoutManager per-connection (holds Weak<Connection>) and
track a single `current_operation_id`. The background thread only
interrupts when the expired entry's id matches the currently registered
operation; older entries are drained without side effects.
- Serialize operations on a connection with an execution_lock held
across prepare(), execute_batch(), and Statement.{get,all,run}. This
gives "current operation" a well-defined meaning and closes the race
between guard release and interrupt dispatch.
- Defense-in-depth: if prepare() returns SQLITE_INTERRUPT, probe the
connection with SELECT 1 (up to three times) to consume any signal that
still slipped through before retrying.
- Swap tokio Notify/task + per-entry AtomicBool for std::thread +
Condvar. The manager is single-consumer and no longer needs an async
runtime.

Tests:

- Statement.get() with both defaultQueryTimeout and per-query
queryTimeout options (async + sync).
- Regression test: 20 iterations of timing out stmt.get(..., {
queryTimeout: 50 }) and immediately running EXPLAIN QUERY PLAN on the
same connection — each EXPLAIN must succeed.

Author: penberg

integration-tests/tests/async.test.js

@@ -416,6 +416,29 @@ test.serial("Query timeout option interrupts long-running query", async (t) => {
   db.close();
 });
 
+test.serial("Query timeout option interrupts long-running Statement.get()", async (t) => {
+  const [db, errorType] = await connect(":memory:", { defaultQueryTimeout: 100 });
+  const stmt = await db.prepare(`
+    WITH RECURSIVE numbers(value) AS (
+      SELECT 1
+      UNION ALL
+      SELECT value + 1 FROM numbers WHERE value < 1000000000
+    )
+    SELECT sum(value) FROM numbers;
+  `);
+
+  const error = await t.throwsAsync(async () => {
+    await stmt.get();
+  });
+  t.truthy(error);
+  t.true(error instanceof Error);
+  t.true(error instanceof errorType);
+  t.true(error.message.toLowerCase().includes("interrupt"));
+  t.is(error.code, "SQLITE_INTERRUPT");
+
+  db.close();
+});
+
 test.serial("Query timeout option allows short-running query", async (t) => {
   const [db] = await connect(":memory:", { defaultQueryTimeout: 100 });
   const stmt = await db.prepare("SELECT 1 AS value");
@@ -464,6 +487,57 @@ test.serial("Per-query timeout option interrupts long-running Statement.all()",
   db.close();
 });
 
+test.serial("Per-query timeout option interrupts long-running Statement.get()", async (t) => {
+  const [db, errorType] = await connect(":memory:");
+  const stmt = await db.prepare(`
+    WITH RECURSIVE numbers(value) AS (
+      SELECT 1
+      UNION ALL
+      SELECT value + 1 FROM numbers WHERE value < 1000000000
+    )
+    SELECT sum(value) FROM numbers;
+  `);
+
+  const error = await t.throwsAsync(async () => {
+    await stmt.get(undefined, { queryTimeout: 100 });
+  });
+  t.truthy(error);
+  t.true(error instanceof Error);
+  t.true(error instanceof errorType);
+  t.true(error.message.toLowerCase().includes("interrupt"));
+  t.is(error.code, "SQLITE_INTERRUPT");
+
+  db.close();
+});
+
+test.serial("Timeout on Statement.get() does not leak into later prepare()/EXPLAIN", async (t) => {
+  t.timeout(30_000);
+  const [db, errorType] = await connect(":memory:");
+  const longRunningStmt = await db.prepare(`
+    WITH RECURSIVE numbers(value) AS (
+      SELECT 1
+      UNION ALL
+      SELECT value + 1 FROM numbers WHERE value < 1000000000
+    )
+    SELECT sum(value) FROM numbers;
+  `);
+
+  for (let i = 0; i < 20; i++) {
+    const error = await t.throwsAsync(async () => {
+      await longRunningStmt.get(undefined, { queryTimeout: 50 });
+    });
+    t.truthy(error);
+    t.true(error instanceof errorType);
+    t.true(error.message.toLowerCase().includes("interrupt"));
+
+    const explainStmt = await db.prepare("EXPLAIN QUERY PLAN SELECT 1");
+    const explainRows = await explainStmt.all();
+    t.true(explainRows.length > 0);
+  }
+
+  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 });

integration-tests/tests/sync.test.js

@@ -479,6 +479,34 @@ test.serial("Query timeout option interrupts long-running query", async (t) => {
   db.close();
 });
 
+test.serial("Query timeout option interrupts long-running Statement.get()", async (t) => {
+  if (t.context.provider === "sqlite") {
+    t.assert(true);
+    return;
+  }
+
+  const [db, errorType] = await connect(":memory:", { defaultQueryTimeout: 100 });
+  const stmt = db.prepare(`
+    WITH RECURSIVE numbers(value) AS (
+      SELECT 1
+      UNION ALL
+      SELECT value + 1 FROM numbers WHERE value < 1000000000
+    )
+    SELECT sum(value) FROM numbers;
+  `);
+
+  const error = t.throws(() => {
+    stmt.get();
+  });
+  t.truthy(error);
+  t.true(error instanceof Error);
+  t.true(error instanceof errorType);
+  t.true(error.message.toLowerCase().includes("interrupt"));
+  t.is(error.code, "SQLITE_INTERRUPT");
+
+  db.close();
+});
+
 test.serial("Query timeout option allows short-running query", async (t) => {
   if (t.context.provider === "sqlite") {
     t.assert(true);
@@ -513,6 +541,67 @@ test.serial("Per-query timeout option interrupts long-running Statement.all()",
   db.close();
 });
 
+test.serial("Per-query timeout option interrupts long-running Statement.get()", async (t) => {
+  if (t.context.provider === "sqlite") {
+    t.assert(true);
+    return;
+  }
+
+  const [db, errorType] = await connect(":memory:");
+  const stmt = db.prepare(`
+    WITH RECURSIVE numbers(value) AS (
+      SELECT 1
+      UNION ALL
+      SELECT value + 1 FROM numbers WHERE value < 1000000000
+    )
+    SELECT sum(value) FROM numbers;
+  `);
+
+  const error = t.throws(() => {
+    stmt.get(undefined, { queryTimeout: 100 });
+  });
+  t.truthy(error);
+  t.true(error instanceof Error);
+  t.true(error instanceof errorType);
+  t.true(error.message.toLowerCase().includes("interrupt"));
+  t.is(error.code, "SQLITE_INTERRUPT");
+
+  db.close();
+});
+
+test.serial("Timeout on Statement.get() does not leak into later prepare()/EXPLAIN", async (t) => {
+  if (t.context.provider === "sqlite") {
+    t.assert(true);
+    return;
+  }
+
+  t.timeout(30_000);
+  const [db, errorType] = await connect(":memory:");
+  const longRunningStmt = db.prepare(`
+    WITH RECURSIVE numbers(value) AS (
+      SELECT 1
+      UNION ALL
+      SELECT value + 1 FROM numbers WHERE value < 1000000000
+    )
+    SELECT sum(value) FROM numbers;
+  `);
+
+  for (let i = 0; i < 20; i++) {
+    const error = t.throws(() => {
+      longRunningStmt.get(undefined, { queryTimeout: 50 });
+    });
+    t.truthy(error);
+    t.true(error instanceof errorType);
+    t.true(error.message.toLowerCase().includes("interrupt"));
+
+    const explainStmt = db.prepare("EXPLAIN QUERY PLAN SELECT 1");
+    const explainRows = explainStmt.all();
+    t.true(explainRows.length > 0);
+  }
+
+  db.close();
+});
+
 test.serial("Per-query timeout option is accepted by Database.exec()", async (t) => {
   if (t.context.provider === "sqlite") {
     t.assert(true);