From 29d6a21e4d139069c088a6bf8b4a4ca4d39a77d8 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Fri, 9 Oct 2026 16:05:50 -0300 Subject: [PATCH 1/6] feat(kvdb): expose read-only lmdb transactions Adds START_READ_TX / CLOSE_READ_TX to the kvdb addon protocol and an optional txId on GET and START_CURSOR, so a client can hold a consistent LMDB snapshot open across many reads without blocking writers. The wrapper keeps a registry of open read transactions, each with a mutex that serializes its use across the libuv worker pool; cursors opened against a shared transaction (including cursor creation) take the same mutex. Closing a read transaction only unregisters it: cursors opened against it keep it alive until they are closed. CLOSE releases every registered read transaction. GET / START_CURSOR with an unknown txId fail; CLOSE_READ_TX on an unknown id returns ok=false. lmdblib::LMDBStore::get gains an overload that reads through a caller-provided read transaction. The new message types are appended to the enum and txId is optional, so existing clients keep working unchanged. --- .../cpp/src/lmdb_store/lmdb_store_message.hpp | 21 ++- .../cpp/src/lmdb_store/lmdb_store_wrapper.cpp | 133 ++++++++++++++---- .../cpp/src/lmdb_store/lmdb_store_wrapper.hpp | 37 +++++ .../lmdblib/cpp/src/lmdblib/lmdb_store.cpp | 17 ++- .../lmdblib/cpp/src/lmdblib/lmdb_store.hpp | 9 +- .../cpp/src/lmdblib/lmdb_store.test.cpp | 111 +++++++++++++++ 6 files changed, 298 insertions(+), 30 deletions(-) diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp index 497b78669a5..eb8b1c1f637 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp +++ b/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 txId; + MSGPACK_DEFINE_MAP(keys, db, txId); }; struct GetResponse { @@ -78,7 +83,9 @@ struct StartCursorRequest { std::optional count; std::optional 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 txId; + MSGPACK_DEFINE_MAP(key, reverse, count, onePage, db, txId); }; struct StartCursorResponse { @@ -139,6 +146,16 @@ struct CopyStoreRequest { MSGPACK_DEFINE_MAP(dstPath, compact); }; +struct StartReadTxResponse { + uint64_t tx; + MSGPACK_DEFINE_MAP(tx); +}; + +struct CloseReadTxRequest { + uint64_t tx; + MSGPACK_DEFINE_MAP(tx); +}; + } // namespace azteclabs::kvdb::lmdb_store MSGPACK_ADD_ENUM(azteclabs::kvdb::lmdb_store::LMDBStoreMessageType) diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp index 7436e37d4a7..979989afef3 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp @@ -6,9 +6,11 @@ #include #include #include +#include #include #include #include +#include #include using namespace azteclabs::kvdb; @@ -91,6 +93,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,6 +121,17 @@ 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 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(); @@ -123,12 +139,46 @@ BoolResponse LMDBStoreWrapper::open_database(const OpenDatabaseRequest& req) return { true }; } +StartReadTxResponse LMDBStoreWrapper::start_read_tx() +{ + verify_store(); + ReadTxData data{ _store->create_shared_read_transaction(), std::make_shared() }; + uint64_t id = data.tx->id(); + { + std::lock_guard lock(_read_tx_mutex); + _read_txs.emplace(id, std::move(data)); + } + return { id }; +} + +BoolResponse LMDBStoreWrapper::close_read_tx(const CloseReadTxRequest& req) +{ + // 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 lock(_read_tx_mutex); + auto it = _read_txs.find(req.tx); + 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 tx_lock(*data.mtx); + _store->get(keys, vals, req.db, data.tx); + } else { + _store->get(keys, vals, req.db); + } return { vals }; } @@ -148,29 +198,29 @@ 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 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(); + } - 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 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 @@ -178,7 +228,6 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req 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 +236,40 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req auto cursor_id = cursor->id(); { std::lock_guard 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 +289,7 @@ AdvanceCursorResponse LMDBStoreWrapper::advance_cursor(const AdvanceCursorReques } uint32_t page_size = req.count.value_or(DEFAULT_CURSOR_PAGE_SIZE); + std::lock_guard tx_lock(*data.txMtx); auto [done, entries] = _advance_cursor(*data.cursor, data.reverse, page_size); return { entries, done }; } @@ -225,6 +303,7 @@ AdvanceCursorCountResponse LMDBStoreWrapper::advance_cursor_count(const AdvanceC data = _cursors.at(req.cursor); } + std::lock_guard tx_lock(*data.txMtx); auto [done, count] = _advance_cursor_count(*data.cursor, data.reverse, req.endKey); return { count, done }; } @@ -267,6 +346,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); diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp index 33a5eea4139..7df21c58059 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp @@ -9,14 +9,28 @@ #include "util/message_processor.hpp" #include #include +#include #include #include +#include namespace azteclabs::kvdb::lmdb_store { +/** + * @brief A read transaction held open on behalf of the JavaScript side, together with the mutex that serializes + * access to it. The environment is opened with `MDB_NOTLS`, so a read transaction may move between the libuv worker + * threads, but it must never be used by two of them at once. + */ +struct ReadTxData { + lmdblib::LMDBReadTransaction::SharedPtr tx; + std::shared_ptr mtx; +}; + struct CursorData { lmdblib::LMDBCursor::SharedPtr cursor; bool reverse; + // Guards the cursor's read transaction, which may be shared with gets and other cursors + std::shared_ptr txMtx; }; /** * @brief Manages the interaction between the JavaScript runtime and the LMDB instance. @@ -38,12 +52,33 @@ class LMDBStoreWrapper : public Napi::ObjectWrap { std::mutex _cursor_mutex; std::unordered_map _cursors; + std::mutex _read_tx_mutex; + std::unordered_map _read_txs; + azteclabs::kvdb::AsyncMessageProcessor _msg_processor; void verify_store() const; + /** + * @brief Returns a copy of the registered read transaction, which keeps it alive even if it is closed concurrently. + * @throws std::runtime_error if no read transaction with this id is open + */ + ReadTxData get_read_tx(uint64_t id); + BoolResponse open_database(const OpenDatabaseRequest& req); + /** + * @brief Opens a read transaction that stays open until CLOSE_READ_TX (or CLOSE) and returns its id. It holds one + * of the environment's reader slots for its whole lifetime and blocks while none is free. + */ + StartReadTxResponse start_read_tx(); + + /** + * @brief Unregisters a read transaction. Cursors already opened against it keep it alive until they are closed. + * @return ok is false if no read transaction with this id was open + */ + BoolResponse close_read_tx(const CloseReadTxRequest& req); + GetResponse get(const GetRequest& req); HasResponse has(const HasRequest& req); @@ -60,6 +95,8 @@ class LMDBStoreWrapper : public Napi::ObjectWrap { BoolResponse copy_store(const CopyStoreRequest& req); + static bool _set_cursor_start(const lmdblib::LMDBCursor& cursor, lmdblib::Key& key, bool reverse); + static std::pair _advance_cursor(const lmdblib::LMDBCursor& cursor, bool reverse, uint64_t page_size); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp index 61d574ab486..4228bbd3550 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp @@ -113,7 +113,16 @@ void LMDBStore::put(std::vector& data) void LMDBStore::get(KeysVector& keys, OptionalValuesVector& values, const std::string& name) { - get(keys, values, get_database(name)); + LMDBDatabase::SharedPtr db = get_database(name); + get(keys, values, db, create_shared_read_transaction()); +} + +void LMDBStore::get(KeysVector& keys, + OptionalValuesVector& values, + const std::string& name, + ReadTransaction::SharedPtr tx) +{ + get(keys, values, get_database(name), std::move(tx)); } void LMDBStore::put(KeyDupValuesVector& toWrite, @@ -136,10 +145,12 @@ void LMDBStore::put(KeyDupValuesVector& toWrite, } } } -void LMDBStore::get(KeysVector& keys, OptionalValuesVector& values, LMDBDatabase::SharedPtr db) +void LMDBStore::get(KeysVector& keys, + OptionalValuesVector& values, + LMDBDatabase::SharedPtr db, + ReadTransaction::SharedPtr tx) { values.reserve(keys.size()); - ReadTransaction::SharedPtr tx = create_read_transaction(); if (!db->duplicate_keys_permitted()) { const LMDBDatabase& dbRef = *db; for (auto& k : keys) { diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp index d55d1d5bd40..8c8a36b91b0 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp @@ -48,6 +48,13 @@ class LMDBStore : public LMDBStoreBase { void put(std::vector& data); void get(KeysVector& keys, OptionalValuesVector& values, const std::string& name); + /** + * @brief Reads the given keys through an already open read transaction, so the values observed belong to that + * transaction's snapshot rather than to whatever is committed at call time. + * @note LMDB read transactions are not thread safe. The caller must ensure that nothing else uses `tx` + * (including opening or advancing cursors against it) while this call runs. + */ + void get(KeysVector& keys, OptionalValuesVector& values, const std::string& name, ReadTransaction::SharedPtr tx); void has(const KeyOptionalValuesVector& entries, std::vector& results, const std::string& name); Cursor::Ptr create_cursor(ReadTransaction::SharedPtr tx, const std::string& dbName); @@ -63,7 +70,7 @@ class LMDBStore : public LMDBStoreBase { KeyOptionalValuesVector& toDelete, const LMDBDatabase& db, LMDBWriteTransaction& tx); - void get(KeysVector& keys, OptionalValuesVector& values, LMDBDatabase::SharedPtr db); + void get(KeysVector& keys, OptionalValuesVector& values, LMDBDatabase::SharedPtr db, ReadTransaction::SharedPtr tx); // Returns the database of the given name Database::SharedPtr get_database(const std::string& name); // Returns all databases diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp index e297e6c3681..9648a2786a8 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp @@ -229,6 +229,117 @@ TEST_F(LMDBStoreTest, can_read_from_database) EXPECT_EQ(data[0].value(), ValuesVector{ expected }); } +TEST_F(LMDBStoreTest, reads_against_a_shared_read_transaction_see_a_stable_snapshot) +{ + LMDBStore::Ptr store = create_store(); + const std::string dbName = "Test Database"; + store->open_database(dbName); + + auto key = get_key(0); + auto original = get_value(0, 1); + auto updated = get_value(0, 2); + auto addedKey = get_key(1); + auto addedValue = get_value(1, 1); + + KeyOptionalValuesVector toDelete; + KeyDupValuesVector toWrite = { { { key, { original } } } }; + std::vector putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + LMDBStore::ReadTransaction::SharedPtr tx = store->create_shared_read_transaction(); + + KeysVector keys = { { key }, { addedKey } }; + OptionalValuesVector snapshot; + store->get(keys, snapshot, dbName, tx); + ASSERT_EQ(snapshot.size(), 2); + EXPECT_EQ(snapshot[0].value(), ValuesVector{ original }); + EXPECT_FALSE(snapshot[1].has_value()); + + // commits while the read transaction is still open + toWrite = { KeyValuesPair{ key, { updated } }, KeyValuesPair{ addedKey, { addedValue } } }; + putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + OptionalValuesVector afterWrite; + store->get(keys, afterWrite, dbName, tx); + ASSERT_EQ(afterWrite.size(), 2); + EXPECT_EQ(afterWrite[0].value(), ValuesVector{ original }); + EXPECT_FALSE(afterWrite[1].has_value()); + + OptionalValuesVector latest; + store->get(keys, latest, dbName); + ASSERT_EQ(latest.size(), 2); + EXPECT_EQ(latest[0].value(), ValuesVector{ updated }); + EXPECT_EQ(latest[1].value(), ValuesVector{ addedValue }); +} + +TEST_F(LMDBStoreTest, can_read_duplicates_against_a_shared_read_transaction) +{ + LMDBStore::Ptr store = create_store(); + const std::string dbName = "Test Database"; + store->open_database(dbName, true); + + int64_t numKeys = 5; + int64_t numValues = 3; + write_test_data({ dbName }, numKeys, numValues, *store); + + LMDBStore::ReadTransaction::SharedPtr tx = store->create_shared_read_transaction(); + + // adds a further duplicate to the key once the read transaction is open + KeyOptionalValuesVector toDelete; + KeyDupValuesVector toWrite = { { { get_key(2), { get_value(2, numValues) } } } }; + std::vector putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + KeysVector keys = { { get_key(2) } }; + OptionalValuesVector values; + store->get(keys, values, dbName, tx); + + ValuesVector expected; + for (int64_t i = 0; i < numValues; i++) { + expected.emplace_back(get_value(2, i)); + } + ASSERT_EQ(values.size(), 1); + ASSERT_TRUE(values[0].has_value()); + EXPECT_EQ(values[0].value(), expected); + + OptionalValuesVector latest; + store->get(keys, latest, dbName); + expected.emplace_back(get_value(2, numValues)); + ASSERT_TRUE(latest[0].has_value()); + EXPECT_EQ(latest[0].value(), expected); +} + +TEST_F(LMDBStoreTest, can_share_a_read_transaction_between_gets_and_cursors) +{ + LMDBStore::Ptr store = create_store(); + const std::string dbName = "Test Database"; + store->open_database(dbName); + + int64_t numKeys = 5; + write_test_data({ dbName }, numKeys, 1, *store); + + LMDBStore::ReadTransaction::SharedPtr tx = store->create_shared_read_transaction(); + LMDBStore::Cursor::Ptr cursor = store->create_cursor(tx, dbName); + + KeyOptionalValuesVector toDelete = { { get_key(0), std::nullopt } }; + KeyDupValuesVector toWrite; + std::vector putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + KeysVector keys = { { get_key(0) } }; + OptionalValuesVector values; + store->get(keys, values, dbName, tx); + ASSERT_TRUE(values[0].has_value()); + EXPECT_EQ(values[0].value(), ValuesVector{ get_value(0, 0) }); + + EXPECT_TRUE(cursor->set_at_start()); + KeyDupValuesVector entries; + cursor->read_next(static_cast(numKeys), entries); + ASSERT_EQ(entries.size(), static_cast(numKeys)); + EXPECT_EQ(entries[0].first, get_key(0)); +} + TEST_F(LMDBStoreTest, can_not_read_from_non_existent_database) { LMDBStore::Ptr store = create_store(); From 109365dc9b7c35a280ed71e270684c41b6ff21b9 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Fri, 9 Oct 2026 16:16:42 -0300 Subject: [PATCH 2/6] fix(lmdblib): return permits when a transaction fails to begin Each transaction factory took its reader or writer permit before constructing the transaction, and only the transaction's destructor gave it back. If mdb_txn_begin threw (e.g. MDB_READERS_FULL because another process holds the reader table), the permit leaked, and repeated failures eventually left every later reader or writer waiting forever. The factories now release the permit when construction throws. Also adds try_create_shared_read_transaction, which takes a reader permit without waiting and can be told to leave some permits free. --- .../cpp/src/lmdblib/lmdb_environment.cpp | 5 ++ .../cpp/src/lmdblib/lmdb_environment.hpp | 17 +++++ .../cpp/src/lmdblib/lmdb_store.test.cpp | 76 +++++++++++++++++++ .../cpp/src/lmdblib/lmdb_store_base.cpp | 42 +++++++++- .../cpp/src/lmdblib/lmdb_store_base.hpp | 6 ++ 5 files changed, 142 insertions(+), 4 deletions(-) diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp index 50bcbabf474..b896509f21a 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp @@ -43,6 +43,11 @@ void LMDBEnvironment::wait_for_reader() _readGuard.wait(); } +bool LMDBEnvironment::try_wait_for_reader(uint32_t keepFree) +{ + return _readGuard.try_wait(keepFree); +} + void LMDBEnvironment::release_reader() { _readGuard.release(); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp index 8b42fb43d09..795247f6c79 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp @@ -49,6 +49,13 @@ class LMDBEnvironment { void wait_for_reader(); + /** + * @brief Takes a reader permit without blocking. + * @param keepFree How many permits must remain available after this one is taken + * @return false, taking nothing, if fewer than `keepFree + 1` permits are available + */ + bool try_wait_for_reader(uint32_t keepFree = 0); + void release_reader(); void wait_for_writer(); @@ -86,6 +93,16 @@ class LMDBEnvironment { ++_current; } + bool try_wait(uint32_t keepFree) + { + std::unique_lock lock(_lock); + if (static_cast(_current) + keepFree >= _maxAllowed) { + return false; + } + ++_current; + return true; + } + void release() { std::unique_lock lock(_lock); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp index 9648a2786a8..438381fcdc4 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp @@ -9,7 +9,9 @@ #include #include #include +#include #include +#include #include #include "lmdblib/fixtures.hpp" @@ -340,6 +342,80 @@ TEST_F(LMDBStoreTest, can_share_a_read_transaction_between_gets_and_cursors) EXPECT_EQ(entries[0].first, get_key(0)); } +TEST_F(LMDBStoreTest, try_create_shared_read_transaction_does_not_wait_and_keeps_permits_free) +{ + const uint32_t maxReaders = 3; + LMDBStore::Ptr store = std::make_unique(_directory, _mapSize, maxReaders, 1); + + std::vector held; + for (uint32_t i = 0; i < maxReaders - 1; i++) { + held.push_back(store->try_create_shared_read_transaction(1)); + ASSERT_NE(held.back(), nullptr); + } + EXPECT_EQ(store->try_create_shared_read_transaction(1), nullptr); + + held.push_back(store->try_create_shared_read_transaction()); + ASSERT_NE(held.back(), nullptr); + EXPECT_EQ(store->try_create_shared_read_transaction(), nullptr); + + held.pop_back(); + held.pop_back(); + EXPECT_NE(store->try_create_shared_read_transaction(1), nullptr); +} + +TEST_F(LMDBStoreTest, reader_permit_is_returned_when_a_read_transaction_fails_to_begin) +{ + const uint32_t maxReaders = 2; + int toParent[2]; + int toChild[2]; + ASSERT_EQ(pipe(toParent), 0); + ASSERT_EQ(pipe(toChild), 0); + + // Another process occupies every slot in the environment's reader table, so mdb_txn_begin fails here + pid_t pid = fork(); + ASSERT_GE(pid, 0); + if (pid == 0) { + { + auto env = std::make_shared(_directory, _mapSize, 1, maxReaders); + std::vector> txs; + for (uint32_t i = 0; i < maxReaders; i++) { + env->wait_for_reader(); + txs.push_back(std::make_unique(env)); + } + char c = 'r'; + if (write(toParent[1], &c, 1) != 1 || read(toChild[0], &c, 1) != 1) { + _exit(1); + } + } + _exit(0); + } + + char c = 0; + ASSERT_EQ(read(toParent[0], &c, 1), 1); + + LMDBStore::Ptr store = std::make_unique(_directory, _mapSize, maxReaders, 1); + // more failed attempts than there are permits, so a leaked permit would surface as a nullptr here + EXPECT_THROW(store->create_shared_read_transaction(), std::runtime_error); + for (uint32_t i = 0; i < maxReaders; i++) { + EXPECT_THROW(store->try_create_shared_read_transaction(), std::runtime_error); + } + + ASSERT_EQ(write(toChild[1], &c, 1), 1); + int status = 0; + ASSERT_EQ(waitpid(pid, &status, 0), pid); + ASSERT_TRUE(WIFEXITED(status)); + ASSERT_EQ(WEXITSTATUS(status), 0); + for (int fd : { toParent[0], toParent[1], toChild[0], toChild[1] }) { + close(fd); + } + + std::vector held; + for (uint32_t i = 0; i < maxReaders; i++) { + held.push_back(store->try_create_shared_read_transaction()); + EXPECT_NE(held.back(), nullptr); + } +} + TEST_F(LMDBStoreTest, can_not_read_from_non_existent_database) { LMDBStore::Ptr store = create_store(); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp index efc85c1de91..1e2acaae0cc 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp @@ -7,28 +7,62 @@ LMDBStoreBase::LMDBStoreBase( , _environment(std::make_shared(_dbDirectory, mapSizeKb, maxDbs, maxNumReaders, ephemeral)) {} LMDBStoreBase::~LMDBStoreBase() = default; +// A transaction returns its permit from its destructor, which never runs if its constructor (mdb_txn_begin) throws LMDBStoreBase::ReadTransaction::Ptr LMDBStoreBase::create_read_transaction() const { _environment->wait_for_reader(); - return std::make_unique(_environment); + try { + return std::make_unique(_environment); + } catch (...) { + _environment->release_reader(); + throw; + } } LMDBStoreBase::ReadTransaction::SharedPtr LMDBStoreBase::create_shared_read_transaction() const { _environment->wait_for_reader(); - return std::make_shared(_environment); + try { + return std::make_shared(_environment); + } catch (...) { + _environment->release_reader(); + throw; + } +} + +LMDBStoreBase::ReadTransaction::SharedPtr LMDBStoreBase::try_create_shared_read_transaction(uint32_t keepFree) const +{ + if (!_environment->try_wait_for_reader(keepFree)) { + return nullptr; + } + try { + return std::make_shared(_environment); + } catch (...) { + _environment->release_reader(); + throw; + } } LMDBStoreBase::DBCreationTransaction::Ptr LMDBStoreBase::create_db_transaction() const { _environment->wait_for_writer(); - return std::make_unique(_environment); + try { + return std::make_unique(_environment); + } catch (...) { + _environment->release_writer(); + throw; + } } LMDBStoreBase::WriteTransaction::Ptr LMDBStoreBase::create_write_transaction() const { _environment->wait_for_writer(); - return std::make_unique(_environment); + try { + return std::make_unique(_environment); + } catch (...) { + _environment->release_writer(); + throw; + } } void LMDBStoreBase::copy_store(const std::string& dstPath, bool compact) diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp index 234721b6db2..90dfa624cdb 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp @@ -20,6 +20,12 @@ class LMDBStoreBase { virtual ~LMDBStoreBase() = 0; ReadTransaction::Ptr create_read_transaction() const; ReadTransaction::SharedPtr create_shared_read_transaction() const; + /** + * @brief Like create_shared_read_transaction, but never waits for a reader permit. + * @param keepFree How many reader permits must remain available once this transaction holds one + * @return nullptr if no permit could be taken + */ + ReadTransaction::SharedPtr try_create_shared_read_transaction(uint32_t keepFree = 0) const; WriteTransaction::Ptr create_write_transaction() const; LMDBDatabaseCreationTransaction::Ptr create_db_transaction() const; void copy_store(const std::string& dstPath, bool compact); From d84a273ab29e39867fbc1c38d4e81b2cf9bbe17d Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Fri, 9 Oct 2026 16:16:42 -0300 Subject: [PATCH 3/6] fix(kvdb): reject START_READ_TX instead of waiting for a reader slot START_READ_TX waited for a reader permit on a libuv worker while holding the dispatcher's shared lock. With every slot held by open read transactions, a few such requests parked every worker, so the CLOSE_READ_TX that would free a slot never ran and CLOSE could never take the dispatcher's exclusive lock. START_READ_TX now fails immediately when no permit is available, and it never takes the last one, so requests that open their own short-lived read transaction (GET/HAS/STATS and cursors without txId) can always make progress while long-lived read transactions are held. CLOSE_READ_TX now rejects a request without a tx id instead of reading an uninitialized integer. --- .../cpp/src/lmdb_store/lmdb_store_message.hpp | 3 ++- .../cpp/src/lmdb_store/lmdb_store_wrapper.cpp | 16 ++++++++++++++-- .../cpp/src/lmdb_store/lmdb_store_wrapper.hpp | 4 +++- 3 files changed, 19 insertions(+), 4 deletions(-) diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp index eb8b1c1f637..377dfed7978 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp @@ -152,7 +152,8 @@ struct StartReadTxResponse { }; struct CloseReadTxRequest { - uint64_t tx; + // Optional only so that a request without it is rejected rather than read as an indeterminate id + std::optional tx; MSGPACK_DEFINE_MAP(tx); }; diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp index 979989afef3..c8207ed3736 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp @@ -19,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) @@ -142,7 +145,13 @@ BoolResponse LMDBStoreWrapper::open_database(const OpenDatabaseRequest& req) StartReadTxResponse LMDBStoreWrapper::start_read_tx() { verify_store(); - ReadTxData data{ _store->create_shared_read_transaction(), std::make_shared() }; + // 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() }; uint64_t id = data.tx->id(); { std::lock_guard lock(_read_tx_mutex); @@ -153,11 +162,14 @@ StartReadTxResponse LMDBStoreWrapper::start_read_tx() BoolResponse LMDBStoreWrapper::close_read_tx(const CloseReadTxRequest& req) { + if (!req.tx.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 lock(_read_tx_mutex); - auto it = _read_txs.find(req.tx); + auto it = _read_txs.find(req.tx.value()); if (it == _read_txs.end()) { return { false }; } diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp index 7df21c58059..c627bc89c4d 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp @@ -69,13 +69,15 @@ class LMDBStoreWrapper : public Napi::ObjectWrap { /** * @brief Opens a read transaction that stays open until CLOSE_READ_TX (or CLOSE) and returns its id. It holds one - * of the environment's reader slots for its whole lifetime and blocks while none is free. + * of the environment's reader slots for its whole lifetime. + * @throws std::runtime_error without waiting if taking a slot would leave none free for other reads */ StartReadTxResponse start_read_tx(); /** * @brief Unregisters a read transaction. Cursors already opened against it keep it alive until they are closed. * @return ok is false if no read transaction with this id was open + * @throws std::runtime_error if the request carries no id */ BoolResponse close_read_tx(const CloseReadTxRequest& req); From 585da76255b6b1275943ede7753064a2128a9e3b Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Fri, 9 Oct 2026 16:16:43 -0300 Subject: [PATCH 4/6] test(kvdb): drive the read transaction protocol from node Adds node:test coverage that loads the built addon and talks to it over msgpack the way the kv-store client does: requests without txId, snapshot gets and cursor scans, concurrent use of one read transaction, cursors outliving CLOSE_READ_TX, unknown and missing ids, reader slot exhaustion and recovery, and CLOSE with live read transactions. Wired into CI as kvdb-tests. --- Makefile | 14 +- native-packages/kvdb/.rebuild_patterns | 1 + native-packages/kvdb/bootstrap.sh | 14 +- native-packages/kvdb/ts/package.json | 1 + native-packages/kvdb/ts/test/read_tx.test.mjs | 291 ++++++++++++++++++ native-packages/kvdb/yarn.lock | 287 +++++++++++++++++ 6 files changed, 602 insertions(+), 6 deletions(-) create mode 100644 native-packages/kvdb/ts/test/read_tx.test.mjs diff --git a/Makefile b/Makefile index 3a864de4cc3..aa529e8e62a 100644 --- a/Makefile +++ b/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) diff --git a/native-packages/kvdb/.rebuild_patterns b/native-packages/kvdb/.rebuild_patterns index 9cb9042b35b..44afd463a93 100644 --- a/native-packages/kvdb/.rebuild_patterns +++ b/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/ diff --git a/native-packages/kvdb/bootstrap.sh b/native-packages/kvdb/bootstrap.sh index 894e2fb4481..692fc06bf60 100755 --- a/native-packages/kvdb/bootstrap.sh +++ b/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 "") diff --git a/native-packages/kvdb/ts/package.json b/native-packages/kvdb/ts/package.json index 72caccac35b..5a8ec4d5b91 100644 --- a/native-packages/kvdb/ts/package.json +++ b/native-packages/kvdb/ts/package.json @@ -28,6 +28,7 @@ }, "devDependencies": { "@types/node": "^22.15.17", + "msgpackr": "1.11.2", "typescript": "^5.3.3" } } diff --git a/native-packages/kvdb/ts/test/read_tx.test.mjs b/native-packages/kvdb/ts/test/read_tx.test.mjs new file mode 100644 index 00000000000..47dfe4059b7 --- /dev/null +++ b/native-packages/kvdb/ts/test/read_tx.test.mjs @@ -0,0 +1,291 @@ +// Exercises the addon's read transaction protocol over the same msgpack wire format the kv-store client uses. +// Run after building the addon: `node --test test/read_tx.test.mjs` (AZTEC_KVDB_NAPI_PATH overrides the binary). +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createRequire } from "node:module"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, describe, it } from "node:test"; + +import { Encoder } from "msgpackr"; + +import { findNapiBinary } from "../dest/index.js"; + +const MSG = { + OPEN_DATABASE: 100, + GET: 101, + START_CURSOR: 103, + ADVANCE_CURSOR: 104, + CLOSE_CURSOR: 106, + BATCH: 107, + CLOSE: 109, + START_READ_TX: 111, + CLOSE_READ_TX: 112, +}; + +const napiPath = findNapiBinary(); +assert.ok(napiPath, "kvdb NAPI binary not found, build it first"); +const { LMDBStore } = createRequire(import.meta.url)(napiPath); + +const encoder = new Encoder({ useRecords: false, int64AsType: "bigint" }); +const b = (s) => Buffer.from(s); +const str = (v) => Buffer.from(v).toString(); +const values = (resp) => + resp.values.map((v) => (v == null ? null : v.map(str))); +const keysOf = (entries) => entries.map(([k]) => str(k)); + +let messageId = 1; +function openStore(dir, maxReaders) { + const store = new LMDBStore(dir, 10 * 1024, maxReaders); + const send = async (msgType, value) => { + const msg = { msgType, header: { messageId: messageId++, requestId: 0 } }; + if (value !== undefined) { + msg.value = value; + } + return encoder.unpack(await store.call(encoder.encode(msg))).value; + }; + const batch = (data = [], index = []) => + send(MSG.BATCH, { + batches: new Map([ + ["data", { addEntries: data, removeEntries: [] }], + ["index", { addEntries: index, removeEntries: [] }], + ]), + }); + const startCursor = (key, txId, opts = {}) => + send(MSG.START_CURSOR, { + key: b(key), + reverse: false, + count: 1, + onePage: false, + db: "data", + ...opts, + txId, + }); + return { send, batch, startCursor }; +} + +async function drainCursor(send, start) { + const keys = keysOf(start.entries); + let cursor = start.cursor; + while (cursor != null) { + const next = await send(MSG.ADVANCE_CURSOR, { cursor, count: 1 }); + keys.push(...keysOf(next.entries)); + if (next.done) { + await send(MSG.CLOSE_CURSOR, { cursor }); + cursor = null; + } + } + return keys; +} + +describe("kvdb read transactions", () => { + let dir; + let send; + let batch; + let startCursor; + const maxReaders = 4; + + before(async () => { + dir = mkdtempSync(join(tmpdir(), "kvdb-read-tx-")); + ({ send, batch, startCursor } = openStore(dir, maxReaders)); + await send(MSG.OPEN_DATABASE, { db: "data", uniqueKeys: true }); + await send(MSG.OPEN_DATABASE, { db: "index", uniqueKeys: false }); + await batch( + [ + [b("a"), [b("a1")]], + [b("b"), [b("b1")]], + [b("c"), [b("c1")]], + ], + [[b("k"), [b("v1"), b("v2")]]], + ); + }); + + after(() => rmSync(dir, { recursive: true, force: true })); + + it("serves requests that omit txId or set it to null as before", async () => { + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a")], db: "data" })), + [["a1"]], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a")], db: "data", txId: null })), + [["a1"]], + ); + const legacy = await send(MSG.START_CURSOR, { + key: b("a"), + reverse: false, + count: null, + onePage: null, + db: "data", + }); + assert.equal(legacy.cursor, null); + assert.deepEqual(keysOf(legacy.entries), ["a", "b", "c"]); + }); + + it("reads and iterates a snapshot that later writes do not change", async () => { + const { tx } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", tx); + assert.notEqual(cursor.cursor, null); + + await batch( + [ + [b("a"), [b("a2")]], + [b("d"), [b("d2")]], + ], + [[b("k"), [b("v3")]]], + ); + + assert.deepEqual( + values( + await send(MSG.GET, { keys: [b("a"), b("d")], db: "data", txId: tx }), + ), + [["a1"], null], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("k")], db: "index", txId: tx })), + [["v1", "v2"]], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a"), b("d")], db: "data" })), + [["a2"], ["d2"]], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("k")], db: "index" })), + [["v1", "v2", "v3"]], + ); + + const reverse = await startCursor("z", tx, { reverse: true, count: 100 }); + assert.deepEqual(keysOf(reverse.entries), ["c", "b", "a"]); + assert.deepEqual(await drainCursor(send, cursor), ["a", "b", "c"]); + + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); + }); + + it("keeps cursors usable after their read transaction is closed", async () => { + const { tx } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", tx); + await batch([[b("e"), [b("e3")]]]); + + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: false }); + assert.deepEqual(await drainCursor(send, cursor), ["a", "b", "c", "d"]); + }); + + it("rejects unknown and missing read transaction ids", async () => { + const { tx } = await send(MSG.START_READ_TX); + await send(MSG.CLOSE_READ_TX, { tx }); + await assert.rejects( + send(MSG.GET, { keys: [b("a")], db: "data", txId: tx }), + /Read transaction .* not found/, + ); + await assert.rejects( + startCursor("a", 987654), + /Read transaction 987654 not found/, + ); + await assert.rejects( + send(MSG.CLOSE_READ_TX, {}), + /missing the read transaction id/, + ); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx: 987654 }), { + ok: false, + }); + }); + + it("serializes concurrent gets and cursor scans on one read transaction", async () => { + const { tx } = await send(MSG.START_READ_TX); + const expected = keysOf( + (await startCursor("a", tx, { count: 100 })).entries, + ); + const work = []; + for (let i = 0; i < 200; i++) { + if (i % 20 === 0) { + work.push(batch([[b(`n${i}`), [b("x")]]], [[b("k"), [b(`z${i}`)]]])); + } + work.push( + send(MSG.GET, { keys: [b("a"), b("k")], db: "data", txId: tx }).then( + (r) => assert.deepEqual(values(r), [["a2"], null]), + ), + send(MSG.GET, { keys: [b("k")], db: "index", txId: tx }).then((r) => + assert.deepEqual(values(r), [["v1", "v2", "v3"]]), + ), + startCursor("a", tx) + .then((start) => drainCursor(send, start)) + .then((keys) => assert.deepEqual(keys, expected)), + ); + } + await Promise.all(work); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); + }); + + it("rejects new read transactions instead of waiting when reader slots run out", async () => { + const held = []; + // one slot always stays free for requests that open their own read transaction + for (let i = 0; i < maxReaders - 1; i++) { + held.push((await send(MSG.START_READ_TX)).tx); + } + const attempts = await Promise.allSettled( + Array.from({ length: 8 }, () => send(MSG.START_READ_TX)), + ); + for (const attempt of attempts) { + assert.equal(attempt.status, "rejected"); + assert.match(attempt.reason.message, /No reader slot available/); + } + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a")], db: "data" })), + [["a2"]], + ); + + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx: held.pop() }), { + ok: true, + }); + held.push((await send(MSG.START_READ_TX)).tx); + for (const tx of held) { + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); + } + }); + + it("counts closed read transactions kept alive by cursors against the reader slots", async () => { + const { tx } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", tx); + await send(MSG.CLOSE_READ_TX, { tx }); + + const held = []; + for (let i = 0; i < maxReaders - 2; i++) { + held.push((await send(MSG.START_READ_TX)).tx); + } + await assert.rejects(send(MSG.START_READ_TX), /No reader slot available/); + + await send(MSG.CLOSE_CURSOR, { cursor: cursor.cursor }); + held.push((await send(MSG.START_READ_TX)).tx); + for (const id of held) { + await send(MSG.CLOSE_READ_TX, { tx: id }); + } + }); + + it("closes the store with live read transactions and cursors, releasing the environment", async () => { + const { tx } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", tx); + assert.notEqual(cursor.cursor, null); + + assert.deepEqual(await send(MSG.CLOSE), { ok: true }); + await assert.rejects( + send(MSG.GET, { keys: [b("a")], db: "data", txId: tx }), + /closed/, + ); + + const reopened = openStore(dir, maxReaders); + await reopened.send(MSG.OPEN_DATABASE, { db: "data", uniqueKeys: true }); + const { tx: fresh } = await reopened.send(MSG.START_READ_TX); + assert.deepEqual( + values( + await reopened.send(MSG.GET, { + keys: [b("d")], + db: "data", + txId: fresh, + }), + ), + [["d2"]], + ); + assert.deepEqual(await reopened.send(MSG.CLOSE), { ok: true }); + }); +}); diff --git a/native-packages/kvdb/yarn.lock b/native-packages/kvdb/yarn.lock index 621e736e478..a071444928a 100644 --- a/native-packages/kvdb/yarn.lock +++ b/native-packages/kvdb/yarn.lock @@ -44,6 +44,7 @@ __metadata: "@aztec-labs/kvdb-linux-arm64": "npm:0.1.0" "@aztec-labs/kvdb-linux-x64": "npm:0.1.0" "@types/node": "npm:^22.15.17" + msgpackr: "npm:1.11.2" tslib: "npm:^2.4.0" typescript: "npm:^5.3.3" dependenciesMeta: @@ -58,6 +59,57 @@ __metadata: languageName: unknown linkType: soft +"@isaacs/fs-minipass@npm:^4.0.0": + version: 4.0.1 + resolution: "@isaacs/fs-minipass@npm:4.0.1" + dependencies: + minipass: "npm:^7.0.4" + checksum: 10c0/c25b6dc1598790d5b55c0947a9b7d111cfa92594db5296c3b907e2f533c033666f692a3939eadac17b1c7c40d362d0b0635dc874cbfe3e70db7c2b07cc97a5d2 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-darwin-arm64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-darwin-arm64@npm:3.0.4" + conditions: os=darwin & cpu=arm64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-darwin-x64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-darwin-x64@npm:3.0.4" + conditions: os=darwin & cpu=x64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-linux-arm64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-linux-arm64@npm:3.0.4" + conditions: os=linux & cpu=arm64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-linux-arm@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-linux-arm@npm:3.0.4" + conditions: os=linux & cpu=arm + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-linux-x64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-linux-x64@npm:3.0.4" + conditions: os=linux & cpu=x64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-win32-x64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-win32-x64@npm:3.0.4" + conditions: os=win32 & cpu=x64 + languageName: node + linkType: hard + "@types/node@npm:^22.15.17": version: 22.20.0 resolution: "@types/node@npm:22.20.0" @@ -67,6 +119,216 @@ __metadata: languageName: node linkType: hard +"abbrev@npm:^5.0.0": + version: 5.0.0 + resolution: "abbrev@npm:5.0.0" + checksum: 10c0/8e88f5c798ea4562d28c5a3e9ad69e3879890bc5d695d8f2dffb8609be4c890aacc8f80ef4553fdd2c6a62d70c2ce8bc57b38074e383beb7487bdafa9ed42ea5 + languageName: node + linkType: hard + +"chownr@npm:^3.0.0": + version: 3.0.0 + resolution: "chownr@npm:3.0.0" + checksum: 10c0/43925b87700f7e3893296c8e9c56cc58f926411cce3a6e5898136daaf08f08b9a8eb76d37d3267e707d0dcc17aed2e2ebdf5848c0c3ce95cf910a919935c1b10 + languageName: node + linkType: hard + +"detect-libc@npm:^2.0.1": + version: 2.1.2 + resolution: "detect-libc@npm:2.1.2" + checksum: 10c0/acc675c29a5649fa1fb6e255f993b8ee829e510b6b56b0910666949c80c364738833417d0edb5f90e4e46be17228b0f2b66a010513984e18b15deeeac49369c4 + languageName: node + linkType: hard + +"env-paths@npm:^2.2.0": + version: 2.2.1 + resolution: "env-paths@npm:2.2.1" + checksum: 10c0/285325677bf00e30845e330eec32894f5105529db97496ee3f598478e50f008c5352a41a30e5e72ec9de8a542b5a570b85699cd63bd2bc646dbcb9f311d83bc4 + languageName: node + linkType: hard + +"exponential-backoff@npm:^3.1.1": + version: 3.1.3 + resolution: "exponential-backoff@npm:3.1.3" + checksum: 10c0/77e3ae682b7b1f4972f563c6dbcd2b0d54ac679e62d5d32f3e5085feba20483cf28bd505543f520e287a56d4d55a28d7874299941faf637e779a1aa5994d1267 + languageName: node + linkType: hard + +"fdir@npm:^6.5.0": + version: 6.5.0 + resolution: "fdir@npm:6.5.0" + peerDependencies: + picomatch: ^3 || ^4 + peerDependenciesMeta: + picomatch: + optional: true + checksum: 10c0/e345083c4306b3aed6cb8ec551e26c36bab5c511e99ea4576a16750ddc8d3240e63826cc624f5ae17ad4dc82e68a253213b60d556c11bfad064b7607847ed07f + languageName: node + linkType: hard + +"graceful-fs@npm:^4.2.6": + version: 4.2.11 + resolution: "graceful-fs@npm:4.2.11" + checksum: 10c0/386d011a553e02bc594ac2ca0bd6d9e4c22d7fa8cfbfc448a6d148c59ea881b092db9dbe3547ae4b88e55f1b01f7c4a2ecc53b310c042793e63aa44cf6c257f2 + languageName: node + linkType: hard + +"isexe@npm:^4.0.0": + version: 4.0.0 + resolution: "isexe@npm:4.0.0" + checksum: 10c0/5884815115bceac452877659a9c7726382531592f43dc29e5d48b7c4100661aed54018cb90bd36cb2eaeba521092570769167acbb95c18d39afdccbcca06c5ce + languageName: node + linkType: hard + +"minipass@npm:^7.0.4, minipass@npm:^7.1.2": + version: 7.1.3 + resolution: "minipass@npm:7.1.3" + checksum: 10c0/539da88daca16533211ea5a9ee98dc62ff5742f531f54640dd34429e621955e91cc280a91a776026264b7f9f6735947629f920944e9c1558369e8bf22eb33fbb + languageName: node + linkType: hard + +"minizlib@npm:^3.1.0": + version: 3.1.0 + resolution: "minizlib@npm:3.1.0" + dependencies: + minipass: "npm:^7.1.2" + checksum: 10c0/5aad75ab0090b8266069c9aabe582c021ae53eb33c6c691054a13a45db3b4f91a7fb1bd79151e6b4e9e9a86727b522527c0a06ec7d45206b745d54cd3097bcec + languageName: node + linkType: hard + +"msgpackr-extract@npm:^3.0.2": + version: 3.0.4 + resolution: "msgpackr-extract@npm:3.0.4" + dependencies: + "@msgpackr-extract/msgpackr-extract-darwin-arm64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-darwin-x64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-linux-arm": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-linux-arm64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-linux-x64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-win32-x64": "npm:3.0.4" + node-gyp: "npm:latest" + node-gyp-build-optional-packages: "npm:5.2.2" + dependenciesMeta: + "@msgpackr-extract/msgpackr-extract-darwin-arm64": + optional: true + "@msgpackr-extract/msgpackr-extract-darwin-x64": + optional: true + "@msgpackr-extract/msgpackr-extract-linux-arm": + optional: true + "@msgpackr-extract/msgpackr-extract-linux-arm64": + optional: true + "@msgpackr-extract/msgpackr-extract-linux-x64": + optional: true + "@msgpackr-extract/msgpackr-extract-win32-x64": + optional: true + bin: + download-msgpackr-prebuilds: bin/download-prebuilds.js + checksum: 10c0/582a9d17abbf3019e600e948736695056280ce401fd0235ee2474e95f9952208b9f6cce4d0e355b03b7d3c5630e6c3d11fe5fc27fdedb2311cce48de464338d8 + languageName: node + linkType: hard + +"msgpackr@npm:1.11.2": + version: 1.11.2 + resolution: "msgpackr@npm:1.11.2" + dependencies: + msgpackr-extract: "npm:^3.0.2" + dependenciesMeta: + msgpackr-extract: + optional: true + checksum: 10c0/7d2e81ca82c397b2352d470d6bc8f4a967fe4fe14f8fc1fc9906b23009fdfb543999b1ad29c700b8861581e0b6bf903d6f0fefb69a09375cbca6d4d802e6c906 + languageName: node + linkType: hard + +"node-gyp-build-optional-packages@npm:5.2.2": + version: 5.2.2 + resolution: "node-gyp-build-optional-packages@npm:5.2.2" + dependencies: + detect-libc: "npm:^2.0.1" + bin: + node-gyp-build-optional-packages: bin.js + node-gyp-build-optional-packages-optional: optional.js + node-gyp-build-optional-packages-test: build-test.js + checksum: 10c0/c81128c6f91873381be178c5eddcbdf66a148a6a89a427ce2bcd457593ce69baf2a8662b6d22cac092d24aa9c43c230dec4e69b3a0da604503f4777cd77e282b + languageName: node + linkType: hard + +"node-gyp@npm:latest": + version: 13.1.0 + resolution: "node-gyp@npm:13.1.0" + dependencies: + env-paths: "npm:^2.2.0" + exponential-backoff: "npm:^3.1.1" + graceful-fs: "npm:^4.2.6" + nopt: "npm:^10.0.0" + proc-log: "npm:^7.0.0" + semver: "npm:^7.3.5" + tar: "npm:^7.5.7" + tinyglobby: "npm:^0.2.12" + undici: "npm:^8.4.1" + which: "npm:^7.0.0" + bin: + node-gyp: bin/node-gyp.js + checksum: 10c0/c6693e52e0e111b21f5cd8ff303b41b51f40338adcc79ac5c6a02399eba7039d50dc816dcafcc485870291a641f7cb75db8887a9bdd0de537fe635a3a81cba5b + languageName: node + linkType: hard + +"nopt@npm:^10.0.0": + version: 10.0.1 + resolution: "nopt@npm:10.0.1" + dependencies: + abbrev: "npm:^5.0.0" + bin: + nopt: bin/nopt.js + checksum: 10c0/980d89257f9587f3e1f77877ddbf905d6aa3b738ec33e49a4fa1a059a0dd82eb28063982b150654a7ae9de386f2ead60e56172db7d37cf56de545f7392a2a26a + languageName: node + linkType: hard + +"picomatch@npm:^4.0.4": + version: 4.0.7 + resolution: "picomatch@npm:4.0.7" + checksum: 10c0/beb6ae02c43ae44e84883b90830196d9046b1726ead292adcf7f57945e0bb0d992d68563d87e03b484b6f3c9a5c6defda7523477f047d7f0e663f126cc01787f + languageName: node + linkType: hard + +"proc-log@npm:^7.0.0": + version: 7.0.0 + resolution: "proc-log@npm:7.0.0" + checksum: 10c0/b89c2d862604f35fec795477b0c7e376feab3ba0d4f4d291c4e959567442697cf451ac557d0623c1cc38af45a78128b983410f397a10c5d3a67f76c33de4754b + languageName: node + linkType: hard + +"semver@npm:^7.3.5": + version: 7.8.5 + resolution: "semver@npm:7.8.5" + bin: + semver: bin/semver.js + checksum: 10c0/b1f3127a5be8125a94f37188b361c212466c292c6910adce3ec106cff5dc211ccaedc4739c11bb70fda59d6fc1f040a9bca289f4e093451521a2372e5231fe0c + languageName: node + linkType: hard + +"tar@npm:^7.5.7": + version: 7.5.22 + resolution: "tar@npm:7.5.22" + dependencies: + "@isaacs/fs-minipass": "npm:^4.0.0" + chownr: "npm:^3.0.0" + minipass: "npm:^7.1.2" + minizlib: "npm:^3.1.0" + yallist: "npm:^5.0.0" + checksum: 10c0/1311f6be85a8157ac4c9147bae43e13923d2a1aae15e4aa1bd5239e4e03d2cf53cfe103dde7f35832fbb4c938b042856bc8e9a0afd29abd05e2d1608788c4fea + languageName: node + linkType: hard + +"tinyglobby@npm:^0.2.12": + version: 0.2.17 + resolution: "tinyglobby@npm:0.2.17" + dependencies: + fdir: "npm:^6.5.0" + picomatch: "npm:^4.0.4" + checksum: 10c0/7f7bb0f197c88bc4b20c231e0deca4240ca3bf313a88f5a7fee93a872b84966a4d50220947c0455ad07a60b3b360961c5b7fd979222aeb716a9f99b412002e4c + languageName: node + linkType: hard + "tslib@npm:^2.4.0": version: 2.8.1 resolution: "tslib@npm:2.8.1" @@ -100,3 +362,28 @@ __metadata: checksum: 10c0/c01ed51829b10aa72fc3ce64b747f8e74ae9b60eafa19a7b46ef624403508a54c526ffab06a14a26b3120d055e1104d7abe7c9017e83ced038ea5cf52f8d5e04 languageName: node linkType: hard + +"undici@npm:^8.4.1": + version: 8.11.2 + resolution: "undici@npm:8.11.2" + checksum: 10c0/106f6991178fb7562fd90cdec7993c5f807a9428682306091f0c4b27fc474ec291f868e29972c6252988bf061ad193df943b3b99e775be157a28d92bbb95699d + languageName: node + linkType: hard + +"which@npm:^7.0.0": + version: 7.0.0 + resolution: "which@npm:7.0.0" + dependencies: + isexe: "npm:^4.0.0" + bin: + node-which: bin/which.js + checksum: 10c0/ca0b54f198f78bbc4b7c02e34bda8d335cb352e0adb4cbca1c37b1a957af3a879a82c4c27ca6525bc942f548d8b64f816ef6528360af9f3de55ffb9b979b620d + languageName: node + linkType: hard + +"yallist@npm:^5.0.0": + version: 5.0.0 + resolution: "yallist@npm:5.0.0" + checksum: 10c0/a499c81ce6d4a1d260d4ea0f6d49ab4da09681e32c3f0472dee16667ed69d01dae63a3b81745a24bd78476ec4fcf856114cb4896ace738e01da34b2c42235416 + languageName: node + linkType: hard From 3a0d6d1468ba1af020fc7ca48b99c409f5f79fa2 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Wed, 14 Oct 2026 09:04:37 -0300 Subject: [PATCH 5/6] refactor(lmdblib): return transaction permits through one helper --- .../cpp/src/lmdblib/lmdb_store_base.cpp | 48 +++++++++---------- 1 file changed, 23 insertions(+), 25 deletions(-) diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp index 1e2acaae0cc..acd29e21ced 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp @@ -7,27 +7,34 @@ LMDBStoreBase::LMDBStoreBase( , _environment(std::make_shared(_dbDirectory, mapSizeKb, maxDbs, maxNumReaders, ephemeral)) {} LMDBStoreBase::~LMDBStoreBase() = default; -// A transaction returns its permit from its destructor, which never runs if its constructor (mdb_txn_begin) throws -LMDBStoreBase::ReadTransaction::Ptr LMDBStoreBase::create_read_transaction() const +namespace { +// A transaction returns its permit from its destructor, which never runs if its constructor (mdb_txn_begin) throws, so +// a permit taken before construction has to be given back here +template auto with_permit(LMDBEnvironment& env, void (LMDBEnvironment::*release)(), Make make) { - _environment->wait_for_reader(); try { - return std::make_unique(_environment); + return make(); } catch (...) { - _environment->release_reader(); + (env.*release)(); throw; } } +} // namespace + +LMDBStoreBase::ReadTransaction::Ptr LMDBStoreBase::create_read_transaction() const +{ + _environment->wait_for_reader(); + return with_permit(*_environment, &LMDBEnvironment::release_reader, [&] { + return std::make_unique(_environment); + }); +} LMDBStoreBase::ReadTransaction::SharedPtr LMDBStoreBase::create_shared_read_transaction() const { _environment->wait_for_reader(); - try { + return with_permit(*_environment, &LMDBEnvironment::release_reader, [&] { return std::make_shared(_environment); - } catch (...) { - _environment->release_reader(); - throw; - } + }); } LMDBStoreBase::ReadTransaction::SharedPtr LMDBStoreBase::try_create_shared_read_transaction(uint32_t keepFree) const @@ -35,34 +42,25 @@ LMDBStoreBase::ReadTransaction::SharedPtr LMDBStoreBase::try_create_shared_read_ if (!_environment->try_wait_for_reader(keepFree)) { return nullptr; } - try { + return with_permit(*_environment, &LMDBEnvironment::release_reader, [&] { return std::make_shared(_environment); - } catch (...) { - _environment->release_reader(); - throw; - } + }); } LMDBStoreBase::DBCreationTransaction::Ptr LMDBStoreBase::create_db_transaction() const { _environment->wait_for_writer(); - try { + return with_permit(*_environment, &LMDBEnvironment::release_writer, [&] { return std::make_unique(_environment); - } catch (...) { - _environment->release_writer(); - throw; - } + }); } LMDBStoreBase::WriteTransaction::Ptr LMDBStoreBase::create_write_transaction() const { _environment->wait_for_writer(); - try { + return with_permit(*_environment, &LMDBEnvironment::release_writer, [&] { return std::make_unique(_environment); - } catch (...) { - _environment->release_writer(); - throw; - } + }); } void LMDBStoreBase::copy_store(const std::string& dstPath, bool compact) From 068d85cd833c088f52440a01a223ad925c4b0b09 Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Wed, 14 Oct 2026 09:04:37 -0300 Subject: [PATCH 6/6] feat(kvdb): read HAS through a read transaction and name its id txId everywhere HAS takes the same optional txId as GET and START_CURSOR. START_READ_TX now returns { txId } and CLOSE_READ_TX takes { txId }, so a client cannot misspell the field on a read and silently get a live read. Documents that START_READ_TX only checks the free reader slot when it opens, so callers must count read transactions with their cursors. --- .../cpp/src/lmdb_store/lmdb_store_message.hpp | 12 +-- .../cpp/src/lmdb_store/lmdb_store_wrapper.cpp | 12 ++- .../cpp/src/lmdb_store/lmdb_store_wrapper.hpp | 5 +- native-packages/kvdb/ts/test/read_tx.test.mjs | 90 +++++++++++-------- .../lmdblib/cpp/src/lmdblib/lmdb_store.cpp | 10 ++- .../lmdblib/cpp/src/lmdblib/lmdb_store.hpp | 8 ++ 6 files changed, 92 insertions(+), 45 deletions(-) diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp index 377dfed7978..03bd02fb0d2 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp @@ -56,7 +56,9 @@ struct HasRequest { // std::map> 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 txId; + MSGPACK_DEFINE_MAP(entries, db, txId); }; struct HasResponse { @@ -147,14 +149,14 @@ struct CopyStoreRequest { }; struct StartReadTxResponse { - uint64_t tx; - MSGPACK_DEFINE_MAP(tx); + 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 tx; - MSGPACK_DEFINE_MAP(tx); + std::optional txId; + MSGPACK_DEFINE_MAP(txId); }; } // namespace azteclabs::kvdb::lmdb_store diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp index c8207ed3736..d627990ad5a 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp @@ -162,14 +162,14 @@ StartReadTxResponse LMDBStoreWrapper::start_read_tx() BoolResponse LMDBStoreWrapper::close_read_tx(const CloseReadTxRequest& req) { - if (!req.tx.has_value()) { + 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 lock(_read_tx_mutex); - auto it = _read_txs.find(req.tx.value()); + auto it = _read_txs.find(req.txId.value()); if (it == _read_txs.end()) { return { false }; } @@ -198,7 +198,13 @@ HasResponse LMDBStoreWrapper::has(const HasRequest& req) { verify_store(); std::vector exists; - _store->has(req.entries, exists, req.db); + if (req.txId.has_value()) { + ReadTxData data = get_read_tx(req.txId.value()); + std::lock_guard tx_lock(*data.mtx); + _store->has(req.entries, exists, req.db, data.tx); + } else { + _store->has(req.entries, exists, req.db); + } return { exists }; } diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp index c627bc89c4d..024531bdcc2 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp @@ -1,11 +1,11 @@ #pragma once +#include "lmdb_store/lmdb_store_message.hpp" #include "lmdblib/lmdb_cursor.hpp" #include "lmdblib/lmdb_store.hpp" #include "lmdblib/types.hpp" #include "messaging/dispatcher.hpp" #include "messaging/header.hpp" -#include "lmdb_store/lmdb_store_message.hpp" #include "util/message_processor.hpp" #include #include @@ -70,6 +70,9 @@ class LMDBStoreWrapper : public Napi::ObjectWrap { /** * @brief Opens a read transaction that stays open until CLOSE_READ_TX (or CLOSE) and returns its id. It holds one * of the environment's reader slots for its whole lifetime. + * @note The free slot is only checked here: a START_CURSOR without txId still waits for a slot and holds it until + * the cursor closes. Callers must keep open read transactions plus open cursors below the reader count, or plain + * reads can wait on every libuv worker at once, leaving no worker to run the close that would free a slot. * @throws std::runtime_error without waiting if taking a slot would leave none free for other reads */ StartReadTxResponse start_read_tx(); diff --git a/native-packages/kvdb/ts/test/read_tx.test.mjs b/native-packages/kvdb/ts/test/read_tx.test.mjs index 47dfe4059b7..54621db9190 100644 --- a/native-packages/kvdb/ts/test/read_tx.test.mjs +++ b/native-packages/kvdb/ts/test/read_tx.test.mjs @@ -14,6 +14,7 @@ import { findNapiBinary } from "../dest/index.js"; const MSG = { OPEN_DATABASE: 100, GET: 101, + HAS: 102, START_CURSOR: 103, ADVANCE_CURSOR: 104, CLOSE_CURSOR: 106, @@ -123,8 +124,8 @@ describe("kvdb read transactions", () => { }); it("reads and iterates a snapshot that later writes do not change", async () => { - const { tx } = await send(MSG.START_READ_TX); - const cursor = await startCursor("a", tx); + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); assert.notEqual(cursor.cursor, null); await batch( @@ -137,12 +138,12 @@ describe("kvdb read transactions", () => { assert.deepEqual( values( - await send(MSG.GET, { keys: [b("a"), b("d")], db: "data", txId: tx }), + await send(MSG.GET, { keys: [b("a"), b("d")], db: "data", txId }), ), [["a1"], null], ); assert.deepEqual( - values(await send(MSG.GET, { keys: [b("k")], db: "index", txId: tx })), + values(await send(MSG.GET, { keys: [b("k")], db: "index", txId })), [["v1", "v2"]], ); assert.deepEqual( @@ -154,28 +155,47 @@ describe("kvdb read transactions", () => { [["v1", "v2", "v3"]], ); - const reverse = await startCursor("z", tx, { reverse: true, count: 100 }); + const entries = [ + [b("d"), null], + [b("a"), [b("a1")]], + ]; + assert.deepEqual(await send(MSG.HAS, { entries, db: "data", txId }), { + exists: [false, true], + }); + assert.deepEqual(await send(MSG.HAS, { entries, db: "data" }), { + exists: [true, false], + }); + assert.deepEqual( + await send(MSG.HAS, { + entries: [[b("k"), [b("v3")]]], + db: "index", + txId, + }), + { exists: [false] }, + ); + + const reverse = await startCursor("z", txId, { reverse: true, count: 100 }); assert.deepEqual(keysOf(reverse.entries), ["c", "b", "a"]); assert.deepEqual(await drainCursor(send, cursor), ["a", "b", "c"]); - assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); }); it("keeps cursors usable after their read transaction is closed", async () => { - const { tx } = await send(MSG.START_READ_TX); - const cursor = await startCursor("a", tx); + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); await batch([[b("e"), [b("e3")]]]); - assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); - assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: false }); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: false }); assert.deepEqual(await drainCursor(send, cursor), ["a", "b", "c", "d"]); }); it("rejects unknown and missing read transaction ids", async () => { - const { tx } = await send(MSG.START_READ_TX); - await send(MSG.CLOSE_READ_TX, { tx }); + const { txId } = await send(MSG.START_READ_TX); + await send(MSG.CLOSE_READ_TX, { txId }); await assert.rejects( - send(MSG.GET, { keys: [b("a")], db: "data", txId: tx }), + send(MSG.GET, { keys: [b("a")], db: "data", txId }), /Read transaction .* not found/, ); await assert.rejects( @@ -186,15 +206,15 @@ describe("kvdb read transactions", () => { send(MSG.CLOSE_READ_TX, {}), /missing the read transaction id/, ); - assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx: 987654 }), { + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId: 987654 }), { ok: false, }); }); it("serializes concurrent gets and cursor scans on one read transaction", async () => { - const { tx } = await send(MSG.START_READ_TX); + const { txId } = await send(MSG.START_READ_TX); const expected = keysOf( - (await startCursor("a", tx, { count: 100 })).entries, + (await startCursor("a", txId, { count: 100 })).entries, ); const work = []; for (let i = 0; i < 200; i++) { @@ -202,26 +222,26 @@ describe("kvdb read transactions", () => { work.push(batch([[b(`n${i}`), [b("x")]]], [[b("k"), [b(`z${i}`)]]])); } work.push( - send(MSG.GET, { keys: [b("a"), b("k")], db: "data", txId: tx }).then( + send(MSG.GET, { keys: [b("a"), b("k")], db: "data", txId }).then( (r) => assert.deepEqual(values(r), [["a2"], null]), ), - send(MSG.GET, { keys: [b("k")], db: "index", txId: tx }).then((r) => + send(MSG.GET, { keys: [b("k")], db: "index", txId }).then((r) => assert.deepEqual(values(r), [["v1", "v2", "v3"]]), ), - startCursor("a", tx) + startCursor("a", txId) .then((start) => drainCursor(send, start)) .then((keys) => assert.deepEqual(keys, expected)), ); } await Promise.all(work); - assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); }); it("rejects new read transactions instead of waiting when reader slots run out", async () => { const held = []; // one slot always stays free for requests that open their own read transaction for (let i = 0; i < maxReaders - 1; i++) { - held.push((await send(MSG.START_READ_TX)).tx); + held.push((await send(MSG.START_READ_TX)).txId); } const attempts = await Promise.allSettled( Array.from({ length: 8 }, () => send(MSG.START_READ_TX)), @@ -235,47 +255,47 @@ describe("kvdb read transactions", () => { [["a2"]], ); - assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx: held.pop() }), { + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId: held.pop() }), { ok: true, }); - held.push((await send(MSG.START_READ_TX)).tx); - for (const tx of held) { - assert.deepEqual(await send(MSG.CLOSE_READ_TX, { tx }), { ok: true }); + held.push((await send(MSG.START_READ_TX)).txId); + for (const txId of held) { + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); } }); it("counts closed read transactions kept alive by cursors against the reader slots", async () => { - const { tx } = await send(MSG.START_READ_TX); - const cursor = await startCursor("a", tx); - await send(MSG.CLOSE_READ_TX, { tx }); + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); + await send(MSG.CLOSE_READ_TX, { txId }); const held = []; for (let i = 0; i < maxReaders - 2; i++) { - held.push((await send(MSG.START_READ_TX)).tx); + held.push((await send(MSG.START_READ_TX)).txId); } await assert.rejects(send(MSG.START_READ_TX), /No reader slot available/); await send(MSG.CLOSE_CURSOR, { cursor: cursor.cursor }); - held.push((await send(MSG.START_READ_TX)).tx); + held.push((await send(MSG.START_READ_TX)).txId); for (const id of held) { - await send(MSG.CLOSE_READ_TX, { tx: id }); + await send(MSG.CLOSE_READ_TX, { txId: id }); } }); it("closes the store with live read transactions and cursors, releasing the environment", async () => { - const { tx } = await send(MSG.START_READ_TX); - const cursor = await startCursor("a", tx); + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); assert.notEqual(cursor.cursor, null); assert.deepEqual(await send(MSG.CLOSE), { ok: true }); await assert.rejects( - send(MSG.GET, { keys: [b("a")], db: "data", txId: tx }), + send(MSG.GET, { keys: [b("a")], db: "data", txId }), /closed/, ); const reopened = openStore(dir, maxReaders); await reopened.send(MSG.OPEN_DATABASE, { db: "data", uniqueKeys: true }); - const { tx: fresh } = await reopened.send(MSG.START_READ_TX); + const { txId: fresh } = await reopened.send(MSG.START_READ_TX); assert.deepEqual( values( await reopened.send(MSG.GET, { diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp index 4228bbd3550..3c036ed6b33 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp @@ -190,6 +190,14 @@ void LMDBStore::get(KeysVector& keys, } void LMDBStore::has(const KeyOptionalValuesVector& entries, std::vector& results, const std::string& name) +{ + has(entries, results, name, create_shared_read_transaction()); +} + +void LMDBStore::has(const KeyOptionalValuesVector& entries, + std::vector& results, + const std::string& name, + ReadTransaction::SharedPtr tx) { auto string_cmp = [](const Key& a, const Key& b) { return std::lexicographical_compare(a.begin(), a.end(), b.begin(), b.end()); @@ -202,7 +210,7 @@ void LMDBStore::has(const KeyOptionalValuesVector& entries, std::vector& r KeysVector keys(key_set.begin(), key_set.end()); OptionalValuesVector vals; - get(keys, vals, name); + get(keys, vals, name, std::move(tx)); results.reserve(entries.size()); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp index 8c8a36b91b0..a4113632a37 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp @@ -56,6 +56,14 @@ class LMDBStore : public LMDBStoreBase { */ void get(KeysVector& keys, OptionalValuesVector& values, const std::string& name, ReadTransaction::SharedPtr tx); void has(const KeyOptionalValuesVector& entries, std::vector& results, const std::string& name); + /** + * @brief Checks the given entries through an already open read transaction, like the `get` overload above. + * @note The caller must ensure that nothing else uses `tx` while this call runs. + */ + void has(const KeyOptionalValuesVector& entries, + std::vector& results, + const std::string& name, + ReadTransaction::SharedPtr tx); Cursor::Ptr create_cursor(ReadTransaction::SharedPtr tx, const std::string& dbName);