Merge pull request #165 from tursodatabase/stmt-interrupt

Add statement interrupt API

Author: penberg

Cargo.lock

@@ -12,7 +12,7 @@ crate-type = ["cdylib"]
 
 [dependencies]
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
-libsql = { git = "https://github.com/tursodatabase/libsql/", rev = "016bea4f6235d39b5f1da5b3a0412af55c4dd0b2", features = ["encryption"] }
+libsql = { git = "https://github.com/tursodatabase/libsql/", rev = "9aa89fee3a096538336d30f3d9e9f2df8ba6677d", features = ["encryption"] }
 tracing = "0.1"
 once_cell = "1.18.0"
 tokio = { version = "1.29.1", features = [ "rt-multi-thread" ] }

Cargo.toml

@@ -39,6 +39,7 @@ const {
   statementIsReader,
   statementGet,
   statementRun,
+  statementInterrupt,
   statementRowsSync,
   statementColumns,
   statementSafeIntegers,
@@ -384,6 +385,13 @@ class Statement {
     }
   }
 
+  /**
+   * Interrupts the statement.
+   */
+  interrupt() {
+    statementInterrupt.call(this.stmt);
+  }
+
   /**
    * Returns the columns in the result set returned by this prepared statement.
    */

index.js

@@ -328,6 +328,21 @@ test.serial("Database.interrupt()", async (t) => {
   });
 });
 
+
+test.serial("Statement.interrupt()", async (t) => {
+  const db = t.context.db;
+  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;");
+  const fut = stmt.all();
+  stmt.interrupt();
+  await t.throwsAsync(async () => {
+    await fut;
+  }, {
+    instanceOf: t.context.errorType,
+    message: 'interrupted',
+    code: 'SQLITE_INTERRUPT'
+  });
+});
+
 test.serial("Concurrent writes over same connection", async (t) => {
   const db = t.context.db;
   await db.exec(`

integration-tests/tests/async.test.js

@@ -52,6 +52,7 @@ const {
   statementIsReader,
   statementGet,
   statementRun,
+  statementInterrupt,
   statementRowsAsync,
   statementColumns,
   statementSafeIntegers,
@@ -386,6 +387,13 @@ class Statement {
     }
   }
 
+  /**
+   * Interrupts the statement.
+   */
+  interrupt() {
+    statementInterrupt.call(this.stmt);
+  }
+
   /**
    * Returns the columns in the result set returned by this prepared statement.
    */

promise.js

@@ -52,6 +52,7 @@ fn main(mut cx: ModuleContext) -> NeonResult<()> {
     cx.export_function("statementRaw", Statement::js_raw)?;
     cx.export_function("statementIsReader", Statement::js_is_reader)?;
     cx.export_function("statementRun", Statement::js_run)?;
+    cx.export_function("statementInterrupt", Statement::js_interrupt)?;
     cx.export_function("statementGet", Statement::js_get)?;
     cx.export_function("statementRowsSync", Statement::js_rows_sync)?;
     cx.export_function("statementRowsAsync", Statement::js_rows_async)?;

src/lib.rs

@@ -1,10 +1,10 @@
 use neon::types::buffer::TypedArray;
 use neon::types::JsPromise;
 use neon::{prelude::*, types::JsBigInt};
-use tokio::time::Instant;
 use std::cell::RefCell;
 use std::sync::Arc;
 use tokio::sync::Mutex;
+use tokio::time::Instant;
 
 use crate::errors::throw_libsql_error;
 use crate::runtime;
@@ -119,6 +119,13 @@ impl Statement {
         Ok(info.upcast())
     }
 
+    pub fn js_interrupt(mut cx: FunctionContext) -> JsResult<JsNull> {
+        let stmt: Handle<'_, JsBox<Statement>> = cx.this()?;
+        let mut raw_stmt = stmt.stmt.blocking_lock();
+        raw_stmt.interrupt();
+        Ok(cx.null())
+    }
+
     pub fn js_get(mut cx: FunctionContext) -> JsResult<JsValue> {
         let stmt: Handle<'_, JsBox<Statement>> = cx.this()?;
         let params = cx.argument::<JsValue>(0)?;