feat(kvdb): expose read-only lmdb transactions

Makefile

@@ -59,7 +59,7 @@ endef
 # Fast bootstrap.
 # wsdb belongs to foundation until disentangled.
 fast-foundation: barretenberg bb-tests \
-		wsdb wsdb-tests lmdblib-tests \
+		wsdb wsdb-tests lmdblib-tests kvdb-tests \
 		l1-contracts l1-contracts-tests \
 		mock-protocol-circuits \
 		noir-protocol-circuits noir-protocol-circuits-tests \
@@ -395,7 +395,7 @@ ipc-runtime-cross: ipc-runtime ipc-runtime-cross-arm64-linux ipc-runtime-cross-a
 
 # lmdblib and kvdb are barretenberg-free: they build against their own deps
 # (lmdb, msgpack-c, node-addon-api) only, never bb.
-.PHONY: lmdblib kvdb lmdblib-tests wsdb-tests
+.PHONY: lmdblib kvdb lmdblib-tests kvdb-tests wsdb-tests
 lmdblib:
 	$(call build,$@,native-packages/lmdblib)
 
@@ -405,12 +405,16 @@ kvdb: lmdblib
 wsdb: lmdblib
 	$(call build,$@,native-packages/wsdb)
 
-# Native-package C++ tests (self-contained gtest binaries). kvdb has no C++ tests
-# (its NAPI is exercised by yarn-project's kv-store tests). wsdb_tests use no bb
-# headers; the bb-header parity/equivalence target is manual (WSDB_BUILD_BB_TESTS).
+# Native-package tests. lmdblib and wsdb are self-contained gtest binaries. kvdb has no
+# C++ tests: its NAPI protocol is driven from node (ts/test) and exercised further by
+# yarn-project's kv-store tests. wsdb_tests use no bb headers; the bb-header
+# parity/equivalence target is manual (WSDB_BUILD_BB_TESTS).
 lmdblib-tests: lmdblib
 	$(call test,$@,native-packages/lmdblib)
 
+kvdb-tests: kvdb
+	$(call test,$@,native-packages/kvdb)
+
 wsdb-tests: wsdb
 	$(call test,$@,native-packages/wsdb)
 

native-packages/kvdb/.rebuild_patterns

@@ -3,4 +3,5 @@
 ^native-packages/kvdb/\.rebuild_patterns$
 ^native-packages/kvdb/cpp/
 ^native-packages/kvdb/ts/src/
+^native-packages/kvdb/ts/test/
 ^native-packages/lmdblib/

native-packages/kvdb/bootstrap.sh

@@ -32,6 +32,18 @@ function build {
   yarn build
 }
 
+# Emit test commands for the CI test engine: the addon's message protocol, driven over msgpack from node.
+function test_cmds {
+  echo "$hash:TIMEOUT=120s cd native-packages/kvdb/ts && node --test test/*.test.mjs"
+}
+
+# Manual: build then run the tests directly.
+function test {
+  echo_header "kvdb test"
+  build
+  test_cmds | filter_test_cmds | parallelize
+}
+
 function clean {
   rm -rf ts/dest ts/build ts/packages ts/node_modules node_modules cpp/build
 }
@@ -47,7 +59,7 @@ function release {
   (cd ts && retry "deploy_npm ${REF_NAME#v}")
 }
 
-export -f build_native build clean release
+export -f build_native build test_cmds test clean release
 
 case "$cmd" in
   "")

native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp

@@ -28,6 +28,9 @@ enum LMDBStoreMessageType {
 
     CLOSE,
     COPY_STORE,
+
+    START_READ_TX,
+    CLOSE_READ_TX,
 };
 
 struct OpenDatabaseRequest {
@@ -39,7 +42,9 @@ struct OpenDatabaseRequest {
 struct GetRequest {
     lmdblib::KeysVector keys;
     std::string db;
-    MSGPACK_DEFINE_MAP(keys, db);
+    // When set, read through the snapshot of this open read transaction instead of a fresh one
+    std::optional<uint64_t> txId;
+    MSGPACK_DEFINE_MAP(keys, db, txId);
 };
 
 struct GetResponse {
@@ -51,7 +56,9 @@ struct HasRequest {
     // std::map<lmdblib::Key, std::optional<lmdblib::Value>> entries;
     lmdblib::KeyOptionalValuesVector entries;
     std::string db;
-    MSGPACK_DEFINE_MAP(entries, db);
+    // When set, check against the snapshot of this open read transaction instead of a fresh one
+    std::optional<uint64_t> txId;
+    MSGPACK_DEFINE_MAP(entries, db, txId);
 };
 
 struct HasResponse {
@@ -78,7 +85,9 @@ struct StartCursorRequest {
     std::optional<uint32_t> count;
     std::optional<bool> onePage;
     std::string db;
-    MSGPACK_DEFINE_MAP(key, reverse, count, onePage, db);
+    // When set, iterate over the snapshot of this open read transaction instead of a fresh one
+    std::optional<uint64_t> txId;
+    MSGPACK_DEFINE_MAP(key, reverse, count, onePage, db, txId);
 };
 
 struct StartCursorResponse {
@@ -139,6 +148,17 @@ struct CopyStoreRequest {
     MSGPACK_DEFINE_MAP(dstPath, compact);
 };
 
+struct StartReadTxResponse {
+    uint64_t txId;
+    MSGPACK_DEFINE_MAP(txId);
+};
+
+struct CloseReadTxRequest {
+    // Optional only so that a request without it is rejected rather than read as an indeterminate id
+    std::optional<uint64_t> txId;
+    MSGPACK_DEFINE_MAP(txId);
+};
+
 } // namespace azteclabs::kvdb::lmdb_store
 
 MSGPACK_ADD_ENUM(azteclabs::kvdb::lmdb_store::LMDBStoreMessageType)

native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp

@@ -6,9 +6,11 @@
 #include <chrono>
 #include <cstdint>
 #include <memory>
+#include <mutex>
 #include <optional>
 #include <ratio>
 #include <stdexcept>
+#include <tuple>
 #include <utility>
 
 using namespace azteclabs::kvdb;
@@ -17,6 +19,9 @@ using namespace azteclabs::kvdb::lmdb_store;
 const uint64_t DEFAULT_MAP_SIZE = 1024UL * 1024;
 const uint64_t DEFAULT_MAX_READERS = 16;
 const uint64_t DEFAULT_CURSOR_PAGE_SIZE = 10;
+// Reader permits a new read transaction must leave free, so that requests which open their own short-lived read
+// transaction can always make progress while long-lived ones are held open
+const uint32_t READ_TX_RESERVED_READERS = 1;
 
 LMDBStoreWrapper::LMDBStoreWrapper(const Napi::CallbackInfo& info)
     : ObjectWrap(info)
@@ -91,6 +96,9 @@ LMDBStoreWrapper::LMDBStoreWrapper(const Napi::CallbackInfo& info)
     _msg_processor.register_handler(LMDBStoreMessageType::CLOSE, this, &LMDBStoreWrapper::close, true);
 
     _msg_processor.register_handler(LMDBStoreMessageType::COPY_STORE, this, &LMDBStoreWrapper::copy_store, true);
+
+    _msg_processor.register_handler(LMDBStoreMessageType::START_READ_TX, this, &LMDBStoreWrapper::start_read_tx);
+    _msg_processor.register_handler(LMDBStoreMessageType::CLOSE_READ_TX, this, &LMDBStoreWrapper::close_read_tx);
 }
 
 Napi::Value LMDBStoreWrapper::call(const Napi::CallbackInfo& info)
@@ -116,27 +124,87 @@ void LMDBStoreWrapper::verify_store() const
     throw std::runtime_error(azteclabs::lmdblib::format("LMDB store unavailable, was close already called?"));
 }
 
+ReadTxData LMDBStoreWrapper::get_read_tx(uint64_t id)
+{
+    std::lock_guard<std::mutex> lock(_read_tx_mutex);
+    auto it = _read_txs.find(id);
+    if (it == _read_txs.end()) {
+        throw std::runtime_error(
+            azteclabs::lmdblib::format("Read transaction ", id, " not found, was it already closed?"));
+    }
+    return it->second;
+}
+
 BoolResponse LMDBStoreWrapper::open_database(const OpenDatabaseRequest& req)
 {
     verify_store();
     _store->open_database(req.db, !req.uniqueKeys.value_or(true));
     return { true };
 }
 
+StartReadTxResponse LMDBStoreWrapper::start_read_tx()
+{
+    verify_store();
+    // Waiting for a permit here could park every libuv worker, starving the CLOSE_READ_TX that would free one
+    auto tx = _store->try_create_shared_read_transaction(READ_TX_RESERVED_READERS);
+    if (!tx) {
+        throw std::runtime_error("No reader slot available to start a read transaction, close an open read "
+                                 "transaction or cursor and retry");
+    }
+    ReadTxData data{ std::move(tx), std::make_shared<std::mutex>() };
+    uint64_t id = data.tx->id();
+    {
+        std::lock_guard<std::mutex> lock(_read_tx_mutex);
+        _read_txs.emplace(id, std::move(data));
+    }
+    return { id };
+}
+
+BoolResponse LMDBStoreWrapper::close_read_tx(const CloseReadTxRequest& req)
+{
+    if (!req.txId.has_value()) {
+        throw std::runtime_error("CLOSE_READ_TX request is missing the read transaction id");
+    }
+    // Moved out so that, if this was the last reference, the transaction is aborted after the registry lock is released
+    ReadTxData closed;
+    {
+        std::lock_guard<std::mutex> lock(_read_tx_mutex);
+        auto it = _read_txs.find(req.txId.value());
+        if (it == _read_txs.end()) {
+            return { false };
+        }
+        closed = std::move(it->second);
+        _read_txs.erase(it);
+    }
+    return { true };
+}
+
 GetResponse LMDBStoreWrapper::get(const GetRequest& req)
 {
     verify_store();
     lmdblib::OptionalValuesVector vals;
     lmdblib::KeysVector keys = req.keys;
-    _store->get(keys, vals, req.db);
+    if (req.txId.has_value()) {
+        ReadTxData data = get_read_tx(req.txId.value());
+        std::lock_guard<std::mutex> tx_lock(*data.mtx);
+        _store->get(keys, vals, req.db, data.tx);
+    } else {
+        _store->get(keys, vals, req.db);
+    }
     return { vals };
 }
 
 HasResponse LMDBStoreWrapper::has(const HasRequest& req)
 {
     verify_store();
     std::vector<bool> exists;
-    _store->has(req.entries, exists, req.db);
+    if (req.txId.has_value()) {
+        ReadTxData data = get_read_tx(req.txId.value());
+        std::lock_guard<std::mutex> tx_lock(*data.mtx);
+        _store->has(req.entries, exists, req.db, data.tx);
+    } else {
+        _store->has(req.entries, exists, req.db);
+    }
     return { exists };
 }
 
@@ -148,37 +216,36 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req
     bool one_page = req.onePage.value_or(false);
     lmdblib::Key key = req.key;
 
-    auto tx = _store->create_shared_read_transaction();
-    lmdblib::LMDBCursor::SharedPtr cursor = _store->create_cursor(tx, req.db);
-    bool start_ok = cursor->set_at_key(key);
+    lmdblib::LMDBReadTransaction::SharedPtr tx;
+    std::shared_ptr<std::mutex> tx_mtx;
+    if (req.txId.has_value()) {
+        ReadTxData data = get_read_tx(req.txId.value());
+        tx = std::move(data.tx);
+        tx_mtx = std::move(data.mtx);
+    } else {
+        tx = _store->create_shared_read_transaction();
+        tx_mtx = std::make_shared<std::mutex>();
+    }
 
-    if (!start_ok) {
-        // we couldn't find exactly the requested key. Find the next biggest one.
-        start_ok = cursor->set_at_key_gte(key);
-        // if we found a key that's greater _and_ we want to go in reverse order
-        // then we're actually outside the requested bounds, we need to go back one position
-        if (start_ok && reverse) {
-            lmdblib::KeyDupValuesVector entries;
-            // read_prev returns `true` if there's nothing more to read
-            // turn this into a "not ok" because there's nothing in the db for this cursor to read
-            start_ok = !cursor->read_prev(1, entries);
-        } else if (!start_ok && reverse) {
-            // we couldn't find a key greater than our starting point _and_ we want to go in reverse..
-            // then we start at the end of the database (the client requested to start at a key greater than anything in
-            // the DB)
-            start_ok = cursor->set_at_end();
+    lmdblib::LMDBCursor::SharedPtr cursor;
+    bool start_ok = false;
+    bool done = false;
+    lmdblib::KeyDupValuesVector first_page;
+    {
+        // Opening a cursor can update the transaction's database state, so it needs the lock as much as reads do
+        std::lock_guard<std::mutex> tx_lock(*tx_mtx);
+        cursor = _store->create_cursor(tx, req.db);
+        start_ok = _set_cursor_start(*cursor, key, reverse);
+        if (start_ok) {
+            std::tie(done, first_page) = _advance_cursor(*cursor, reverse, page_size);
         }
-
-        // in case we're iterating in ascending order and we can't find the exact key or one that's greater than it
-        // then that means theren's nothing in the DB for the cursor to read
     }
 
     // we couldn't find a starting position
     if (!start_ok) {
         return { std::nullopt, {} };
     }
 
-    auto [done, first_page] = _advance_cursor(*cursor, reverse, page_size);
     // cursor finished after reading a single page or client only wanted the first page
     if (done || one_page) {
         return { std::nullopt, first_page };
@@ -187,12 +254,40 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req
     auto cursor_id = cursor->id();
     {
         std::lock_guard<std::mutex> lock(_cursor_mutex);
-        _cursors[cursor_id] = { cursor, reverse };
+        _cursors[cursor_id] = { cursor, reverse, tx_mtx };
     }
 
     return { cursor_id, first_page };
 }
 
+bool LMDBStoreWrapper::_set_cursor_start(const lmdblib::LMDBCursor& cursor, lmdblib::Key& key, bool reverse)
+{
+    bool start_ok = cursor.set_at_key(key);
+
+    if (!start_ok) {
+        // we couldn't find exactly the requested key. Find the next biggest one.
+        start_ok = cursor.set_at_key_gte(key);
+        // if we found a key that's greater _and_ we want to go in reverse order
+        // then we're actually outside the requested bounds, we need to go back one position
+        if (start_ok && reverse) {
+            lmdblib::KeyDupValuesVector entries;
+            // read_prev returns `true` if there's nothing more to read
+            // turn this into a "not ok" because there's nothing in the db for this cursor to read
+            start_ok = !cursor.read_prev(1, entries);
+        } else if (!start_ok && reverse) {
+            // we couldn't find a key greater than our starting point _and_ we want to go in reverse..
+            // then we start at the end of the database (the client requested to start at a key greater than anything in
+            // the DB)
+            start_ok = cursor.set_at_end();
+        }
+
+        // in case we're iterating in ascending order and we can't find the exact key or one that's greater than it
+        // then that means theren's nothing in the DB for the cursor to read
+    }
+
+    return start_ok;
+}
+
 BoolResponse LMDBStoreWrapper::close_cursor(const CloseCursorRequest& req)
 {
     {
@@ -212,6 +307,7 @@ AdvanceCursorResponse LMDBStoreWrapper::advance_cursor(const AdvanceCursorReques
     }
 
     uint32_t page_size = req.count.value_or(DEFAULT_CURSOR_PAGE_SIZE);
+    std::lock_guard<std::mutex> tx_lock(*data.txMtx);
     auto [done, entries] = _advance_cursor(*data.cursor, data.reverse, page_size);
     return { entries, done };
 }
@@ -225,6 +321,7 @@ AdvanceCursorCountResponse LMDBStoreWrapper::advance_cursor_count(const AdvanceC
         data = _cursors.at(req.cursor);
     }
 
+    std::lock_guard<std::mutex> tx_lock(*data.txMtx);
     auto [done, count] = _advance_cursor_count(*data.cursor, data.reverse, req.endKey);
     return { count, done };
 }
@@ -267,6 +364,12 @@ BoolResponse LMDBStoreWrapper::close()
         _cursors.clear();
     }
 
+    {
+        // and all of the read transactions still held open on behalf of the JS side
+        std::lock_guard read_txs(_read_tx_mutex);
+        _read_txs.clear();
+    }
+
     // and finally close the database handle
     _store.reset(nullptr);