src: handle race condition

Author: geeksilva97

src/node_sqlite.cc

@@ -459,6 +459,10 @@ class SQLiteAsyncTask : public ThreadPoolWork {
   }
 
   void DoThreadPoolWork() override {
+    if (db_->IsClosing()) {
+      cancelled_ = true;
+      return;
+    }
     if (work_) {
       result_ = work_();
     }
@@ -470,6 +474,16 @@ class SQLiteAsyncTask : public ThreadPoolWork {
     Local<Promise::Resolver> resolver =
         Local<Promise::Resolver>::New(isolate, resolver_);
 
+    if (cancelled_ || db_->IsClosing()) {
+      resolver
+          ->Reject(env_->context(),
+                   ERR_INVALID_STATE(isolate, "database is closing"))
+          .FromJust();
+      Finalize();
+      db_->ProcessNextAsyncTask();
+      return;
+    }
+
     if (after_) {
       after_(result_, resolver);
       Finalize();
@@ -486,6 +500,7 @@ class SQLiteAsyncTask : public ThreadPoolWork {
   std::function<T()> work_ = nullptr;
   std::function<void(T, Local<Promise::Resolver>)> after_ = nullptr;
   T result_;
+  bool cancelled_ = false;
 };
 
 // TODO(geeksilva97): Replace BackupJob usage with SQLiteAsyncTask
@@ -1204,6 +1219,7 @@ void Database::Close(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
   THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  db->is_closing_.store(true, std::memory_order_release);
   db->FinalizeStatements();
   db->DeleteSessions();
   int r = sqlite3_close_v2(db->connection_);

src/node_sqlite.h

@@ -10,6 +10,7 @@
 #include "threadpoolwork-inl.h"
 #include "util.h"
 
+#include <atomic>
 #include <list>
 #include <map>
 #include <queue>
@@ -200,6 +201,7 @@ class Database : public BaseObject {
   void FinalizeBackups();
   void UntrackStatement(Statement* statement);
   bool IsOpen();
+  bool IsClosing() const { return is_closing_.load(std::memory_order_acquire); }
   bool is_async() { return open_config_.get_async(); }
   bool use_big_ints() const { return open_config_.get_use_big_ints(); }
   bool return_arrays() const { return open_config_.get_return_arrays(); }
@@ -231,6 +233,7 @@ class Database : public BaseObject {
   bool enable_load_extension_;
   sqlite3* connection_;
   bool ignore_next_sqlite_error_;
+  std::atomic<bool> is_closing_{false};
 
   std::set<ThreadPoolWork*> async_tasks_;
   std::queue<ThreadPoolWork*> task_queue_;

test/parallel/test-sqlite-database-async.mjs

@@ -562,3 +562,44 @@ suite('Async mode restrictions', () => {
     });
   });
 });
+
+suite('Database close behavior', () => {
+  test('rejects pending operations when database is closed', async (t) => {
+    const db = new Database(':memory:');
+    await db.exec('CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)');
+
+    // Start an async operation but don't await it yet
+    const pendingOp = db.exec('INSERT INTO test (value) VALUES (\'test\')');
+
+    // Close the database immediately
+    db.close();
+
+    // The pending operation should be rejected
+    await t.assert.rejects(pendingOp, {
+      code: 'ERR_INVALID_STATE',
+      message: /database is closing/,
+    });
+  });
+
+  test('rejects multiple pending operations when database is closed', async (t) => {
+    const db = new Database(':memory:');
+    await db.exec('CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)');
+
+    // Queue multiple operations
+    const ops = [
+      db.exec('INSERT INTO test (value) VALUES (\'test1\')'),
+      db.exec('INSERT INTO test (value) VALUES (\'test2\')'),
+      db.exec('INSERT INTO test (value) VALUES (\'test3\')'),
+    ];
+
+    // Close the database
+    db.close();
+
+    // All operations should be rejected
+    const results = await Promise.allSettled(ops);
+    for (const result of results) {
+      t.assert.strictEqual(result.status, 'rejected');
+      t.assert.strictEqual(result.reason.code, 'ERR_INVALID_STATE');
+    }
+  });
+});