Make Database thread-safe via serialized connection access (#9)

The library funnels all access through a single shared sqlite3* connection
behind a process-wide singleton, with no synchronization. Concurrent use
raced: each operation's BEGIN/COMMIT interleaved on the shared connection
("cannot start a transaction within a transaction"), the singleton lifecycle
was unguarded, and query error paths closed the shared connection out from
under other callers.

This serializes access (approach chosen for #9):

- Open the connection in serialized mode (sqlite3_open_v2 + SQLITE_OPEN_FULLMUTEX).
- Guard every operation that touches db_ with a per-instance mutex, so the
  per-operation transaction is never interleaved. The lock in the
  auto-increment path spans the insert and the last_insert_rowid read, so the
  returned id always corresponds to this insert.
- Guard the singleton lifecycle (Initialize/Finalize/Instance) with a static
  mutex, and make Instance() throw a clear error instead of dereferencing null
  when the database is not initialized.
- Stop closing the shared connection from FetchRecordsQuery/FetchMaxIdQuery
  error paths (this left a dangling handle and could double-close on Finalize).

Adds concurrency tests that hammer the shared instance from multiple threads
and assert all operations succeed, every row persists, and ids stay unique.

Author: claude

include/database.h

@@ -23,6 +23,7 @@
 #pragma once
 
 #include <stdexcept>
+#include <mutex>
 #include <string>
 #include <typeinfo>
 #include <vector>
@@ -98,9 +99,7 @@ class REFLECTION_EXPORT Database {
     int64_t GetMaxId() const {
         const auto type_id = typeid(T).name();
         const auto& record = GetRecord(type_id);
-        FetchMaxIdQuery query(db_, record);
-        const auto max_id = query.GetMaxId();
-        return max_id;
+        return GetMaxId(record);
     }
 
     /// Saves a given record in the database.
@@ -208,6 +207,9 @@ class REFLECTION_EXPORT Database {
     /// Returns a record type from its type information, retrieved from typeid(...).name()
     static const Reflection& GetRecord(const std::string& type_id);
 
+    /// Returns the max id currently stored for a given record (SELECT MAX(id) FROM table)
+    int64_t GetMaxId(const Reflection& record) const;
+
     /// Creates concrete record types with initialized members,
     /// based on the textual representation of results from a fetch query
     template <typename T>
@@ -235,6 +237,14 @@ class REFLECTION_EXPORT Database {
     void Delete(const Reflection& record, const QueryPredicateBase* predicate) const;
 
     static Database* instance_;
+
+    /// Guards the singleton lifecycle (Initialize / Finalize / Instance)
+    static std::mutex instance_mutex_;
+
     sqlite3* db_;
+
+    /// Serializes all access to the shared connection db_, so that the per-operation
+    /// BEGIN/COMMIT transaction is never interleaved across threads
+    mutable std::mutex db_mutex_;
 };
 }  // namespace sqlite_reflection

src/database.cc

@@ -22,19 +22,22 @@
 
 #include "database.h"
 
+#include <mutex>
 #include <stdexcept>
 
 #include "internal/sqlite3.h"
 #include "queries.h"
 
 namespace sqlite_reflection {
 Database* Database::instance_ = nullptr;
+std::mutex Database::instance_mutex_;
 
 const ReflectionRegister& GetReflectionRegister() {
     return *GetReflectionRegisterInstance();
 }
 
 void Database::Initialize(const std::string& path) {
+    std::lock_guard<std::mutex> lock(instance_mutex_);
     if (instance_ != nullptr) {
         throw std::invalid_argument("Database has already been initialized");
     }
@@ -44,6 +47,7 @@ void Database::Initialize(const std::string& path) {
 }
 
 void Database::Finalize() {
+    std::lock_guard<std::mutex> lock(instance_mutex_);
     if (instance_ != nullptr) {
         sqlite3_close(instance_->db_);
         delete instance_;
@@ -52,7 +56,10 @@ void Database::Finalize() {
 }
 
 Database::Database(const char* path) : db_(nullptr) {
-    if (sqlite3_open(path, &db_)) {
+    // Open in serialized threading mode so the shared connection is safe to use from
+    // multiple threads; access is additionally serialized through db_mutex_.
+    const int flags = SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX;
+    if (sqlite3_open_v2(path, &db_, flags, nullptr)) {
         throw std::invalid_argument("Database could not be initialized");
     }
 
@@ -65,10 +72,15 @@ Database::Database(const char* path) : db_(nullptr) {
 }
 
 const Database& Database::Instance() {
+    std::lock_guard<std::mutex> lock(instance_mutex_);
+    if (instance_ == nullptr) {
+        throw std::runtime_error("Database has not been initialized; call Database::Initialize() first");
+    }
     return *instance_;
 }
 
 FetchQueryResults Database::Fetch(const Reflection& record, const QueryPredicateBase* predicate) const {
+    std::lock_guard<std::mutex> lock(db_mutex_);
     FetchRecordsQuery query(db_, record, predicate);
     return query.GetResults();
 }
@@ -77,28 +89,41 @@ const Reflection& Database::GetRecord(const std::string& type_id) {
     return GetReflectionRegister().records.at(type_id);
 }
 
+int64_t Database::GetMaxId(const Reflection& record) const {
+    std::lock_guard<std::mutex> lock(db_mutex_);
+    FetchMaxIdQuery query(db_, record);
+    return query.GetMaxId();
+}
+
 void Database::Save(void* p, const Reflection& record) const {
+    std::lock_guard<std::mutex> lock(db_mutex_);
     InsertQuery query(db_, record, p);
     query.Execute();
 }
 
 int64_t Database::SaveAutoIncrement(void* p, const Reflection& record) const {
+    // The lock spans both the insert and the rowid read so that, on the shared connection,
+    // sqlite3_last_insert_rowid() reflects this insert and not one from another thread.
+    std::lock_guard<std::mutex> lock(db_mutex_);
     InsertQuery query(db_, record, p, true);
     query.Execute();
     return sqlite3_last_insert_rowid(db_);
 }
 
 void Database::Update(void* p, const Reflection& record) const {
+    std::lock_guard<std::mutex> lock(db_mutex_);
     UpdateQuery query(db_, record, p);
     query.Execute();
 }
 
 void Database::Delete(const Reflection& record, const QueryPredicateBase* predicate) const {
+    std::lock_guard<std::mutex> lock(db_mutex_);
     DeleteQuery query(db_, record, predicate);
     query.Execute();
 }
 
 void Database::UnsafeSql(const std::string& raw_sql_query) const {
+    std::lock_guard<std::mutex> lock(db_mutex_);
     SqlQuery sql(db_, raw_sql_query);
     sql.Execute();
 }

src/queries.cc

@@ -293,7 +293,6 @@ int64_t FetchMaxIdQuery::GetMaxId() {
     const auto sql = PrepareSql();
 
     if (sqlite3_prepare_v2(db_, sql.data(), -1, &stmt_, nullptr)) {
-        sqlite3_close(db_);
         throw std::runtime_error("Could not retrieve max id for table " + record_.name);
     }
 
@@ -323,7 +322,6 @@ FetchQueryResults FetchRecordsQuery::GetResults() {
     const auto sql = PrepareSql();
 
     if (sqlite3_prepare_v2(db_, sql.data(), -1, &stmt_, nullptr)) {
-        sqlite3_close(db_);
         throw std::runtime_error((sql + ": could not get results").data());
     }
     BindValues(stmt_, predicate_->Bindings());

tests/CMakeLists.txt

@@ -21,8 +21,11 @@ source_group("src" FILES ${TEST_SOURCES})
 # Set Properties->General->Configuration Type to Application
 add_executable (${EXENAME} ${TEST_SOURCES} ${TEST_HEADERS})
 
+# The concurrency tests use std::thread, which needs the platform threading library
+find_package(Threads REQUIRED)
+
 # Properties->Linker->Input->Additional Dependencies
-target_link_libraries(${EXENAME} PUBLIC sqlite_reflection gtest_main)
+target_link_libraries(${EXENAME} PUBLIC sqlite_reflection gtest_main Threads::Threads)
 
 # Creates a folder "executables" and adds target project under it
 set_property(TARGET ${EXENAME} PROPERTY FOLDER "executables")

tests/concurrency_test.cc

@@ -0,0 +1,141 @@
+// MIT License
+//
+// Copyright (c) 2026 Ioannis Kaliakatsos
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+#include "database.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <cstdint>
+#include <set>
+#include <thread>
+#include <vector>
+
+#include "person.h"
+
+using namespace sqlite_reflection;
+
+// These tests exercise the shared singleton connection from many threads at once.
+// Before the connection access was serialized (issue #9), concurrent writers raced
+// on a single sqlite3* and each operation's BEGIN/COMMIT interleaved, producing
+// "cannot start a transaction within a transaction" failures and intermittent
+// crashes. They deterministically pass only once access is serialized.
+class ConcurrencyTest : public ::testing::Test {
+    void SetUp() override {
+        Database::Initialize("");
+    }
+
+    void TearDown() override {
+        Database::Finalize();
+    }
+};
+
+TEST_F(ConcurrencyTest, ConcurrentAutoIncrementInsertsAllSucceedWithUniqueIds) {
+    const auto& db = Database::Instance();
+
+    constexpr int kThreads = 8;
+    constexpr int kInsertsPerThread = 100;
+    constexpr int kExpected = kThreads * kInsertsPerThread;
+
+    std::atomic<int> failures{0};
+    std::vector<std::thread> workers;
+    workers.reserve(kThreads);
+    for (int t = 0; t < kThreads; ++t) {
+        workers.emplace_back([&db, &failures] {
+            for (int i = 0; i < kInsertsPerThread; ++i) {
+                try {
+                    Person p{L"john", L"doe", 30};
+                    db.SaveAutoIncrement(p);
+                } catch (...) {
+                    failures.fetch_add(1, std::memory_order_relaxed);
+                }
+            }
+        });
+    }
+    for (auto& w : workers) {
+        w.join();
+    }
+
+    // No operation may fail (e.g. with a nested-transaction error).
+    EXPECT_EQ(0, failures.load());
+
+    // Every insert must be persisted exactly once.
+    const auto all = db.FetchAll<Person>();
+    EXPECT_EQ(kExpected, static_cast<int>(all.size()));
+
+    // Every row must have received a distinct, database-assigned id.
+    std::set<int64_t> ids;
+    for (const auto& person : all) {
+        ids.insert(person.id);
+    }
+    EXPECT_EQ(kExpected, static_cast<int>(ids.size()));
+}
+
+TEST_F(ConcurrencyTest, ConcurrentReadsAndWritesDoNotThrow) {
+    const auto& db = Database::Instance();
+
+    constexpr int kWriters = 4;
+    constexpr int kReaders = 4;
+    constexpr int kInsertsPerWriter = 100;
+    constexpr int kReadsPerReader = 100;
+    constexpr int kExpected = kWriters * kInsertsPerWriter;
+
+    std::atomic<int> failures{0};
+    std::vector<std::thread> workers;
+    workers.reserve(kWriters + kReaders);
+
+    for (int t = 0; t < kWriters; ++t) {
+        workers.emplace_back([&db, &failures] {
+            for (int i = 0; i < kInsertsPerWriter; ++i) {
+                try {
+                    Person p{L"jane", L"roe", 41};
+                    db.SaveAutoIncrement(p);
+                } catch (...) {
+                    failures.fetch_add(1, std::memory_order_relaxed);
+                }
+            }
+        });
+    }
+
+    for (int t = 0; t < kReaders; ++t) {
+        workers.emplace_back([&db, &failures] {
+            for (int i = 0; i < kReadsPerReader; ++i) {
+                try {
+                    // Reading concurrently with writers must not crash or throw.
+                    volatile auto count = db.FetchAll<Person>().size();
+                    (void)count;
+                } catch (...) {
+                    failures.fetch_add(1, std::memory_order_relaxed);
+                }
+            }
+        });
+    }
+
+    for (auto& w : workers) {
+        w.join();
+    }
+
+    EXPECT_EQ(0, failures.load());
+
+    const auto all = db.FetchAll<Person>();
+    EXPECT_EQ(kExpected, static_cast<int>(all.size()));
+}