sqlite: ensure statement finalization on db close

This commit adds statement tracking to the DatabaseSync class.
When a database is closed manually or via garbage collection, it
will force all associated prepared statements to be finalized.
This should mitigate "zombie" connections which can introduce
test flakiness in the CI on Windows.

PR-URL: https://github.com/nodejs/node/pull/54014
Fixes: https://github.com/nodejs/node/issues/54006
Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
Reviewed-By: Tobias Nießen <tniessen@tnie.de>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: Richard Lau <rlau@redhat.com>
Reviewed-By: Michaël Zasso <targos@protonmail.com>

src/node_sqlite.cc

@@ -91,8 +91,11 @@ DatabaseSync::DatabaseSync(Environment* env,
 }
 
 DatabaseSync::~DatabaseSync() {
-  sqlite3_close_v2(connection_);
-  connection_ = nullptr;
+  if (IsOpen()) {
+    FinalizeStatements();
+    sqlite3_close_v2(connection_);
+    connection_ = nullptr;
+  }
 }
 
 void DatabaseSync::MemoryInfo(MemoryTracker* tracker) const {
@@ -100,7 +103,7 @@ void DatabaseSync::MemoryInfo(MemoryTracker* tracker) const {
 }
 
 bool DatabaseSync::Open() {
-  if (connection_ != nullptr) {
+  if (IsOpen()) {
     node::THROW_ERR_INVALID_STATE(env(), "database is already open");
     return false;
   }
@@ -112,6 +115,29 @@ bool DatabaseSync::Open() {
   return true;
 }
 
+void DatabaseSync::FinalizeStatements() {
+  for (auto stmt : statements_) {
+    stmt->Finalize();
+  }
+
+  statements_.clear();
+}
+
+void DatabaseSync::UntrackStatement(StatementSync* statement) {
+  auto it = statements_.find(statement);
+  if (it != statements_.end()) {
+    statements_.erase(it);
+  }
+}
+
+inline bool DatabaseSync::IsOpen() {
+  return connection_ != nullptr;
+}
+
+inline sqlite3* DatabaseSync::Connection() {
+  return connection_;
+}
+
 void DatabaseSync::New(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
 
@@ -164,8 +190,8 @@ void DatabaseSync::Close(const FunctionCallbackInfo<Value>& args) {
   DatabaseSync* db;
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
-  THROW_AND_RETURN_ON_BAD_STATE(
-      env, db->connection_ == nullptr, "database is not open");
+  THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
+  db->FinalizeStatements();
   int r = sqlite3_close_v2(db->connection_);
   CHECK_ERROR_OR_THROW(env->isolate(), db->connection_, r, SQLITE_OK, void());
   db->connection_ = nullptr;
@@ -175,8 +201,7 @@ void DatabaseSync::Prepare(const FunctionCallbackInfo<Value>& args) {
   DatabaseSync* db;
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
-  THROW_AND_RETURN_ON_BAD_STATE(
-      env, db->connection_ == nullptr, "database is not open");
+  THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
 
   if (!args[0]->IsString()) {
     node::THROW_ERR_INVALID_ARG_TYPE(env->isolate(),
@@ -188,8 +213,8 @@ void DatabaseSync::Prepare(const FunctionCallbackInfo<Value>& args) {
   sqlite3_stmt* s = nullptr;
   int r = sqlite3_prepare_v2(db->connection_, *sql, -1, &s, 0);
   CHECK_ERROR_OR_THROW(env->isolate(), db->connection_, r, SQLITE_OK, void());
-  BaseObjectPtr<StatementSync> stmt =
-      StatementSync::Create(env, db->connection_, s);
+  BaseObjectPtr<StatementSync> stmt = StatementSync::Create(env, db, s);
+  db->statements_.insert(stmt.get());
   args.GetReturnValue().Set(stmt->object());
 }
 
@@ -197,8 +222,7 @@ void DatabaseSync::Exec(const FunctionCallbackInfo<Value>& args) {
   DatabaseSync* db;
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
-  THROW_AND_RETURN_ON_BAD_STATE(
-      env, db->connection_ == nullptr, "database is not open");
+  THROW_AND_RETURN_ON_BAD_STATE(env, !db->IsOpen(), "database is not open");
 
   if (!args[0]->IsString()) {
     node::THROW_ERR_INVALID_ARG_TYPE(env->isolate(),
@@ -213,7 +237,7 @@ void DatabaseSync::Exec(const FunctionCallbackInfo<Value>& args) {
 
 StatementSync::StatementSync(Environment* env,
                              Local<Object> object,
-                             sqlite3* db,
+                             DatabaseSync* db,
                              sqlite3_stmt* stmt)
     : BaseObject(env, object) {
   MakeWeak();
@@ -227,13 +251,25 @@ StatementSync::StatementSync(Environment* env,
 }
 
 StatementSync::~StatementSync() {
+  if (!IsFinalized()) {
+    db_->UntrackStatement(this);
+    Finalize();
+  }
+}
+
+void StatementSync::Finalize() {
   sqlite3_finalize(statement_);
   statement_ = nullptr;
 }
 
+inline bool StatementSync::IsFinalized() {
+  return statement_ == nullptr;
+}
+
 bool StatementSync::BindParams(const FunctionCallbackInfo<Value>& args) {
   int r = sqlite3_clear_bindings(statement_);
-  CHECK_ERROR_OR_THROW(env()->isolate(), db_, r, SQLITE_OK, false);
+  CHECK_ERROR_OR_THROW(
+      env()->isolate(), db_->Connection(), r, SQLITE_OK, false);
 
   int anon_idx = 1;
   int anon_start = 0;
@@ -364,7 +400,8 @@ bool StatementSync::BindValue(const Local<Value>& value, const int index) {
     return false;
   }
 
-  CHECK_ERROR_OR_THROW(env()->isolate(), db_, r, SQLITE_OK, false);
+  CHECK_ERROR_OR_THROW(
+      env()->isolate(), db_->Connection(), r, SQLITE_OK, false);
   return true;
 }
 
@@ -435,8 +472,11 @@ void StatementSync::All(const FunctionCallbackInfo<Value>& args) {
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
   Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(
+      env, stmt->IsFinalized(), "statement has been finalized");
   int r = sqlite3_reset(stmt->statement_);
-  CHECK_ERROR_OR_THROW(env->isolate(), stmt->db_, r, SQLITE_OK, void());
+  CHECK_ERROR_OR_THROW(
+      env->isolate(), stmt->db_->Connection(), r, SQLITE_OK, void());
 
   if (!stmt->BindParams(args)) {
     return;
@@ -462,7 +502,8 @@ void StatementSync::All(const FunctionCallbackInfo<Value>& args) {
     rows.emplace_back(row);
   }
 
-  CHECK_ERROR_OR_THROW(env->isolate(), stmt->db_, r, SQLITE_DONE, void());
+  CHECK_ERROR_OR_THROW(
+      env->isolate(), stmt->db_->Connection(), r, SQLITE_DONE, void());
   args.GetReturnValue().Set(
       Array::New(env->isolate(), rows.data(), rows.size()));
 }
@@ -471,8 +512,11 @@ void StatementSync::Get(const FunctionCallbackInfo<Value>& args) {
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
   Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(
+      env, stmt->IsFinalized(), "statement has been finalized");
   int r = sqlite3_reset(stmt->statement_);
-  CHECK_ERROR_OR_THROW(env->isolate(), stmt->db_, r, SQLITE_OK, void());
+  CHECK_ERROR_OR_THROW(
+      env->isolate(), stmt->db_->Connection(), r, SQLITE_OK, void());
 
   if (!stmt->BindParams(args)) {
     return;
@@ -482,7 +526,7 @@ void StatementSync::Get(const FunctionCallbackInfo<Value>& args) {
   r = sqlite3_step(stmt->statement_);
   if (r == SQLITE_DONE) return;
   if (r != SQLITE_ROW) {
-    THROW_ERR_SQLITE_ERROR(env->isolate(), stmt->db_);
+    THROW_ERR_SQLITE_ERROR(env->isolate(), stmt->db_->Connection());
     return;
   }
 
@@ -511,8 +555,11 @@ void StatementSync::Run(const FunctionCallbackInfo<Value>& args) {
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
   Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(
+      env, stmt->IsFinalized(), "statement has been finalized");
   int r = sqlite3_reset(stmt->statement_);
-  CHECK_ERROR_OR_THROW(env->isolate(), stmt->db_, r, SQLITE_OK, void());
+  CHECK_ERROR_OR_THROW(
+      env->isolate(), stmt->db_->Connection(), r, SQLITE_OK, void());
 
   if (!stmt->BindParams(args)) {
     return;
@@ -521,7 +568,7 @@ void StatementSync::Run(const FunctionCallbackInfo<Value>& args) {
   auto reset = OnScopeLeave([&]() { sqlite3_reset(stmt->statement_); });
   r = sqlite3_step(stmt->statement_);
   if (r != SQLITE_ROW && r != SQLITE_DONE) {
-    THROW_ERR_SQLITE_ERROR(env->isolate(), stmt->db_);
+    THROW_ERR_SQLITE_ERROR(env->isolate(), stmt->db_->Connection());
     return;
   }
 
@@ -530,8 +577,9 @@ void StatementSync::Run(const FunctionCallbackInfo<Value>& args) {
       FIXED_ONE_BYTE_STRING(env->isolate(), "lastInsertRowid");
   Local<String> changes_string =
       FIXED_ONE_BYTE_STRING(env->isolate(), "changes");
-  sqlite3_int64 last_insert_rowid = sqlite3_last_insert_rowid(stmt->db_);
-  sqlite3_int64 changes = sqlite3_changes64(stmt->db_);
+  sqlite3_int64 last_insert_rowid =
+      sqlite3_last_insert_rowid(stmt->db_->Connection());
+  sqlite3_int64 changes = sqlite3_changes64(stmt->db_->Connection());
   Local<Value> last_insert_rowid_val;
   Local<Value> changes_val;
 
@@ -557,6 +605,8 @@ void StatementSync::SourceSQL(const FunctionCallbackInfo<Value>& args) {
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
   Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(
+      env, stmt->IsFinalized(), "statement has been finalized");
   Local<String> sql;
   if (!String::NewFromUtf8(env->isolate(), sqlite3_sql(stmt->statement_))
            .ToLocal(&sql)) {
@@ -569,6 +619,8 @@ void StatementSync::ExpandedSQL(const FunctionCallbackInfo<Value>& args) {
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
   Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(
+      env, stmt->IsFinalized(), "statement has been finalized");
   char* expanded = sqlite3_expanded_sql(stmt->statement_);
   auto maybe_expanded = String::NewFromUtf8(env->isolate(), expanded);
   sqlite3_free(expanded);
@@ -584,6 +636,8 @@ void StatementSync::SetAllowBareNamedParameters(
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
   Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(
+      env, stmt->IsFinalized(), "statement has been finalized");
 
   if (!args[0]->IsBoolean()) {
     node::THROW_ERR_INVALID_ARG_TYPE(
@@ -599,6 +653,8 @@ void StatementSync::SetReadBigInts(const FunctionCallbackInfo<Value>& args) {
   StatementSync* stmt;
   ASSIGN_OR_RETURN_UNWRAP(&stmt, args.This());
   Environment* env = Environment::GetCurrent(args);
+  THROW_AND_RETURN_ON_BAD_STATE(
+      env, stmt->IsFinalized(), "statement has been finalized");
 
   if (!args[0]->IsBoolean()) {
     node::THROW_ERR_INVALID_ARG_TYPE(
@@ -640,7 +696,7 @@ Local<FunctionTemplate> StatementSync::GetConstructorTemplate(
 }
 
 BaseObjectPtr<StatementSync> StatementSync::Create(Environment* env,
-                                                   sqlite3* db,
+                                                   DatabaseSync* db,
                                                    sqlite3_stmt* stmt) {
   Local<Object> obj;
   if (!GetConstructorTemplate(env)

src/node_sqlite.h

@@ -9,10 +9,13 @@
 #include "util.h"
 
 #include <map>
+#include <unordered_set>
 
 namespace node {
 namespace sqlite {
 
+class StatementSync;
+
 class DatabaseSync : public BaseObject {
  public:
   DatabaseSync(Environment* env,
@@ -25,6 +28,10 @@ class DatabaseSync : public BaseObject {
   static void Close(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Prepare(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Exec(const v8::FunctionCallbackInfo<v8::Value>& args);
+  void FinalizeStatements();
+  void UntrackStatement(StatementSync* statement);
+  bool IsOpen();
+  sqlite3* Connection();
 
   SET_MEMORY_INFO_NAME(DatabaseSync)
   SET_SELF_SIZE(DatabaseSync)
@@ -35,19 +42,20 @@ class DatabaseSync : public BaseObject {
   ~DatabaseSync() override;
   std::string location_;
   sqlite3* connection_;
+  std::unordered_set<StatementSync*> statements_;
 };
 
 class StatementSync : public BaseObject {
  public:
   StatementSync(Environment* env,
                 v8::Local<v8::Object> object,
-                sqlite3* db,
+                DatabaseSync* db,
                 sqlite3_stmt* stmt);
   void MemoryInfo(MemoryTracker* tracker) const override;
   static v8::Local<v8::FunctionTemplate> GetConstructorTemplate(
       Environment* env);
   static BaseObjectPtr<StatementSync> Create(Environment* env,
-                                             sqlite3* db,
+                                             DatabaseSync* db,
                                              sqlite3_stmt* stmt);
   static void All(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Get(const v8::FunctionCallbackInfo<v8::Value>& args);
@@ -57,13 +65,15 @@ class StatementSync : public BaseObject {
   static void SetAllowBareNamedParameters(
       const v8::FunctionCallbackInfo<v8::Value>& args);
   static void SetReadBigInts(const v8::FunctionCallbackInfo<v8::Value>& args);
+  void Finalize();
+  bool IsFinalized();
 
   SET_MEMORY_INFO_NAME(StatementSync)
   SET_SELF_SIZE(StatementSync)
 
  private:
   ~StatementSync() override;
-  sqlite3* db_;
+  DatabaseSync* db_;
   sqlite3_stmt* statement_;
   bool use_big_ints_;
   bool allow_bare_named_params_;