sqlite: handle exceptions in filter callback database.applyChangeset()

doc/api/sqlite.md

@@ -241,8 +241,8 @@ added:
 
 * `changeset` {Uint8Array} A binary changeset or patchset.
 * `options` {Object} The configuration options for how the changes will be applied.
-  * `filter` {Function} Skip changes that, when targeted table name is supplied to this function, return a truthy value.
-    By default, all changes are attempted.
+  * `filter` {Function} A table name is provided as an argument to this callback. Returning a truthy value means changes
+    for the table with that table name should be attempted. By default, changes for all tables are attempted.
   * `onConflict` {Function} A function that determines how to handle conflicts. The function receives one argument,
     which can be one of the following values:
 

src/node_sqlite.cc

@@ -743,26 +743,28 @@ void DatabaseSync::CreateSession(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(session->object());
 }
 
+struct ConflictCallbackContext {
+  std::function<bool(std::string)> filterCallback;
+  std::function<int(int)> conflictCallback;
+};
+
 // the reason for using static functions here is that SQLite needs a
 // function pointer
-static std::function<int(int)> conflictCallback;
 
 static int xConflict(void* pCtx, int eConflict, sqlite3_changeset_iter* pIter) {
-  if (!conflictCallback) return SQLITE_CHANGESET_ABORT;
-  return conflictCallback(eConflict);
+  auto ctx = static_cast<ConflictCallbackContext*>(pCtx);
+  if (!ctx->conflictCallback) return SQLITE_CHANGESET_ABORT;
+  return ctx->conflictCallback(eConflict);
 }
 
-static std::function<bool(std::string)> filterCallback;
-
 static int xFilter(void* pCtx, const char* zTab) {
-  if (!filterCallback) return 1;
-
-  return filterCallback(zTab) ? 1 : 0;
+    auto ctx = static_cast<ConflictCallbackContext*>(pCtx);
+    if (!ctx->filterCallback) return 1;
+    return ctx->filterCallback(zTab) ? 1 : 0;
 }
 
 void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
-  conflictCallback = nullptr;
-  filterCallback = nullptr;
+  ConflictCallbackContext context;
 
   DatabaseSync* db;
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
@@ -794,7 +796,7 @@ void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
         return;
       }
       Local<Function> conflictFunc = conflictValue.As<Function>();
-      conflictCallback = [env, conflictFunc](int conflictType) -> int {
+      context.conflictCallback = [env, conflictFunc](int conflictType) -> int {
         Local<Value> argv[] = {Integer::New(env->isolate(), conflictType)};
         TryCatch try_catch(env->isolate());
         Local<Value> result =
@@ -824,14 +826,21 @@ void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
 
       Local<Function> filterFunc = filterValue.As<Function>();
 
-      filterCallback = [env, filterFunc](std::string item) -> bool {
+      context.filterCallback = [env, filterFunc](std::string item) -> bool {
         Local<Value> argv[] = {String::NewFromUtf8(env->isolate(),
-                                                   item.c_str(),
-                                                   NewStringType::kNormal)
-                                   .ToLocalChecked()};
-        Local<Value> result =
-            filterFunc->Call(env->context(), Null(env->isolate()), 1, argv)
-                .ToLocalChecked();
+                                                  item.c_str(),
+                                                  NewStringType::kNormal)
+                                  .ToLocalChecked()};
+        MaybeLocal<Value> maybe_result =
+            filterFunc->Call(env->context(), Null(env->isolate()), 1, argv);
+
+        if (maybe_result.IsEmpty()) {
+          return false;
+        }
+        Local<Value> result;
+        if (!maybe_result.ToLocal(&result)) {
+          return false;
+        }
         return result->BooleanValue(env->isolate());
       };
     }
@@ -844,7 +853,7 @@ void DatabaseSync::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
       const_cast<void*>(static_cast<const void*>(buf.data())),
       xFilter,
       xConflict,
-      nullptr);
+      static_cast<void*>(&context));
   if (r == SQLITE_OK) {
     args.GetReturnValue().Set(true);
     return;

test/parallel/test-sqlite-session.js

@@ -6,6 +6,9 @@ const {
   constants,
 } = require('node:sqlite');
 const { test, suite } = require('node:test');
+const { nextDb } = require("../sqlite/next-db.js");
+const { Worker } = require('worker_threads');
+const { once } = require('events');
 
 /**
  * Convenience wrapper around assert.deepStrictEqual that sets a null
@@ -361,29 +364,111 @@ suite('conflict resolution', () => {
   });
 });
 
-test('database.createSession() - filter changes', (t) => {
-  const database1 = new DatabaseSync(':memory:');
-  const database2 = new DatabaseSync(':memory:');
-  const createTableSql = 'CREATE TABLE data1(key INTEGER PRIMARY KEY); CREATE TABLE data2(key INTEGER PRIMARY KEY);';
-  database1.exec(createTableSql);
-  database2.exec(createTableSql);
+suite('filter tables', () => {
+  function testFilter(t, options) {
+    const database1 = new DatabaseSync(':memory:');
+    const database2 = new DatabaseSync(':memory:');
+    const createTableSql = 'CREATE TABLE data1(key INTEGER PRIMARY KEY); CREATE TABLE data2(key INTEGER PRIMARY KEY);';
+
+    database1.exec(createTableSql);
+    database2.exec(createTableSql);
+
+    const session = database1.createSession();
+    database1.exec('INSERT INTO data1 (key) VALUES (1), (2), (3)');
+    database1.exec('INSERT INTO data2 (key) VALUES (1), (2), (3), (4), (5)');
+
+    const applyChangeset = () => database2.applyChangeset(session.changeset(), {
+      ...(options.filter ? { filter: options.filter } : {})
+    });
+    if (options.apply) {
+      options.apply(applyChangeset);
+    } else {
+      applyChangeset();
+    }
+
+    t.assert.strictEqual(database2.prepare('SELECT * FROM data1').all().length, options.data1);
+    t.assert.strictEqual(database2.prepare('SELECT * FROM data2').all().length, options.data2);
+  }
 
-  const session = database1.createSession();
+  test('database.createSession() - filter one table', (t) => {
+    testFilter(t, {
+      filter: (tableName) => tableName === 'data2',
+      // Only changes from data2 should be included
+      data1: 0,
+      data2: 5
+    });
+  });
+
+  test('database.createSession() - throw in filter callback', (t) => {
+    const error = Error('hello world');
+    testFilter(t, {
+      filter: () => { throw error; },
+      error: error,
+      // When an exception is thrown in the filter function, no changes should be applied
+      data1: 0,
+      data2: 0
+    });
+  });
+
+  test('database.createSession() - throw sometimes in filter callback', (t) => {
+    testFilter(t, {
+      filter: (tableName) => { if (tableName === 'data2') throw new Error(); else { return true; } },
+      data1: 0,
+      data2: 0,
+      expectError: true
+    });
+  });
+
+  test('database.createSession() - throw sometimes in filter callback', (t) => {
+    testFilter(t, {
+      filter: (tableName) => {
+        if (tableName === "data1")
+          throw new Error(tableName);
+        return true;
+      },
+      data1: 0,
+      data2: 0,
+      expectError: true
+    });
+  });
 
-  database1.exec('INSERT INTO data1 (key) VALUES (1), (2), (3)');
-  database1.exec('INSERT INTO data2 (key) VALUES (1), (2), (3), (4), (5)');
+  test('database.createSession() - do not return anything in filter callback', (t) => {
+    testFilter(t, {
+      filter: () => {},
+      // Undefined is falsy, so it is interpreted as "do not include changes from this table"
+      data1: 0,
+      data2: 0
+    });
+  });
 
-  database2.applyChangeset(session.changeset(), {
-    filter: (tableName) => tableName === 'data2'
+  test('database.createSession() - return true for all tables', (t) => {
+    const tables = new Set();
+    testFilter(t, {
+      filter: (tableName) => { tables.add(tableName); return true; },
+      // Changes from all tables should be included
+      data1: 3,
+      data2: 5
+    });
+    t.assert.deepEqual(tables, new Set(['data1', 'data2']));
   });
 
-  const data1Rows = database2.prepare('SELECT * FROM data1').all();
-  const data2Rows = database2.prepare('SELECT * FROM data2').all();
+  test('database.createSession() - return truthy value for all tables', (t) => {
+    testFilter(t, {
+      filter: () => 'yes',
+      // Truthy, so changes from all tables should be included
+      data1: 3,
+      data2: 5
+    });
+  });
 
-  // Expect no rows since all changes were filtered out
-  t.assert.strictEqual(data1Rows.length, 0);
-  // Expect 5 rows since these changes were not filtered out
-  t.assert.strictEqual(data2Rows.length, 5);
+  test('database.createSession() - no filter callback', (t) => {
+    testFilter(t, {
+      filter: undefined,
+      // All changes should be applied
+      data1: 3,
+      data2: 5
+    });
+  });
 });
 
 test('database.createSession() - specify other database', (t) => {
@@ -538,3 +623,71 @@ test('session.close() - closing twice', (t) => {
     message: 'session is not open'
   });
 });
+
+test('concurrent applyChangeset with workers', async (t) => {
+  // before adding this test, the callbacks were stored in static variables
+  // this could result in a crash
+  // this test is a regression test for that scenario
+
+  function modeToString(mode) {
+    if (mode === constants.SQLITE_CHANGESET_ABORT) return 'SQLITE_CHANGESET_ABORT';
+    if (mode === constants.SQLITE_CHANGESET_OMIT) return 'SQLITE_CHANGESET_OMIT';
+  }
+
+  const dbPath = nextDb();
+  const db1 = new DatabaseSync(dbPath);
+  const db2 = new DatabaseSync(':memory:');
+  const createTable = `
+    CREATE TABLE data(
+      key INTEGER PRIMARY KEY,
+      value TEXT
+    ) STRICT`;
+  db1.exec(createTable);
+  db2.exec(createTable);
+  db1.prepare('INSERT INTO data (key, value) VALUES (?, ?)').run(1, 'hello');
+  db1.close();
+  const session = db2.createSession();
+  db2.prepare('INSERT INTO data (key, value) VALUES (?, ?)').run(1, 'world');
+  const changeset = session.changeset(); // changeset with conflict (for db1)
+
+  const iterations = 10;
+  for (let i = 0; i < iterations; i++) {
+    const workers = [];
+    const expectedResults = new Map([[constants.SQLITE_CHANGESET_ABORT, false], [constants.SQLITE_CHANGESET_OMIT, true]]);
+
+    // Launch two workers (abort and omit modes)
+    for (const mode of [constants.SQLITE_CHANGESET_ABORT, constants.SQLITE_CHANGESET_OMIT]) {
+      const worker = new Worker(`${__dirname}/../sqlite/worker.js`, {
+        workerData: {
+          dbPath,
+          changeset,
+          mode
+        },
+      });
+      workers.push(worker);
+    }
+
+    const results = await Promise.all(workers.map(async (worker) => {
+      const [message] = await once(worker, 'message');
+      return message;
+    }));
+
+    // Verify each result
+    for (const res of results) {
+      if (res.errorMessage) {
+        if (res.errcode === 5) {  // SQLITE_BUSY
+          break; // ignore
+        }
+        t.assert.fail(`Worker error: ${res.error.message}`);
+      }
+      const expected = expectedResults.get(res.mode);
+      t.assert.strictEqual(
+        res.result,
+        expected,
+        `Iteration ${i}: Worker (${modeToString(res.mode)}) expected ${expected} but got ${res.result}`
+      );
+    }
+
+    workers.forEach(worker => worker.terminate()); // Cleanup
+  }
+});

test/parallel/test-sqlite.js

@@ -1,16 +1,8 @@
 'use strict';
 const { spawnPromisified } = require('../common');
-const tmpdir = require('../common/tmpdir');
-const { join } = require('node:path');
 const { DatabaseSync, constants } = require('node:sqlite');
 const { suite, test } = require('node:test');
-let cnt = 0;
-
-tmpdir.refresh();
-
-function nextDb() {
-  return join(tmpdir.path, `database-${cnt++}.db`);
-}
+const { nextDb } = require("../sqlite/next-db.js");
 
 suite('accessing the node:sqlite module', () => {
   test('cannot be accessed without the node: scheme', (t) => {

test/sqlite/next-db.js

@@ -0,0 +1,12 @@
+const tmpdir = require('../common/tmpdir');
+const { join } = require('node:path');
+
+let cnt = 0;
+
+tmpdir.refresh();
+
+function nextDb() {
+  return join(tmpdir.path, `database-${cnt++}.db`);
+}
+
+module.exports = { nextDb };

test/sqlite/worker.js

@@ -0,0 +1,22 @@
+// this worker is used for one of the tests in test-sqlite-session.js
+
+const { parentPort, workerData } = require('worker_threads');
+const { DatabaseSync, constants } = require('node:sqlite');
+const { changeset, mode, dbPath } = workerData;
+
+const db = new DatabaseSync(dbPath);
+
+const options = {}
+if (mode !== constants.SQLITE_CHANGESET_ABORT && mode !== constants.SQLITE_CHANGESET_OMIT) {
+  throw new Error("Unexpected value for mode");
+}
+options.onConflict = () => mode;
+
+try {
+  const result = db.applyChangeset(changeset, options);
+  parentPort.postMessage({ mode, result, error: null });
+} catch (error) {
+  parentPort.postMessage({ mode, result: null, errorMessage: error.message, errcode: error.errcode });
+} finally {
+  db.close();  // just to make sure it is closed ASAP
+}