diff --git a/AGENTS.md b/AGENTS.md index 60734d8fa..4f1f3dec1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,10 +34,12 @@ GitHub Copilot, and other AI coding assistants when working with code in this re - `pnpm type-check` - TypeScript type checking only **Run `pnpm fmt` before every commit** (or `pnpm check` to also type-check and lint) — CI runs -`pnpm fmt:check` and fails the build on unformatted code. Note the scope: oxfmt formats **TS/JS/JSON -only**. It does **not** touch C++ or Markdown, so changes to `src/binding/**` and to docs like this -file (`AGENTS.md`) are not auto-formatted and must be checked by hand — a mis-numbered invariant or a -stray C++ indent will pass `fmt:check` untouched. +`pnpm fmt:check` and fails the build on unformatted code. Note the scope: oxfmt formats TS/JS/JSON +**and Markdown**, but **not C++**, so a stray indent under `src/binding/**` passes `fmt:check` +untouched and must be checked by hand. Markdown is checked, which includes the ordered list of +invariants below: a branch that adds an invariant while `main` adds another **renumbers cleanly in +git and still fails `fmt:check` on the merge ref**, because both sides claim the same number. Rebase +onto `main` and renumber before pushing rather than reading the red check as unrelated. ### Development Workflow @@ -739,8 +741,40 @@ sufficient (env teardown does not honor tsfn acquire counts); see written remains frozen across retries, though reapplying the same timestamp is idempotent while the transaction remains pending. rocksdb-js does not define record value layouts: a producer that copies `getTimestamp()` into record bytes must call `setTimestamp()` first. - -20. **A WriteBufferManager stall is a second, entirely separate stall mechanism, and nothing in +20. **Transactional ranges keep the caller's column family and close before the transaction**: + `Store.getRange()` routes `options.transaction` to native by transaction ID, where the caller + database descriptor resolves it and supplies the caller's `DBHandle` to `DBIteratorHandle`. + Replacing the context with `transaction._context` is incorrect for cross-column-family scans: + that native transaction carries the column family on which it was created. Transaction ids are + allocated per `DBDescriptor` (`nextTransactionId`), so an id from another database resolves in + the caller's descriptor to an unrelated transaction of the same number; `Store.getTxnId()` + rejects that by comparing `NativeDatabase.identityPath`, the resolved identity the registry + keyed the descriptor on (`resolveIdentityPath`), cached on the `Store` at open. Never compare + the path a caller passed to `open()`: it is a spelling, so `data` and `./data` — one database + and one id space — would be rejected, while one relative path can name two databases across a + `chdir`. Column families of a database share the identity, so cross-column-family reads pass. + Transaction-backed iterators establish and pass the transaction snapshot, seek explicitly, and + enforce their encoded + bounds in `valid()` rather than trusting RocksDB alone: `iterate_lower_bound` is inclusive, so the + exclusive lower bound of a reverse range (`exclusiveStart`) has to be applied by the handle when + the iterator reaches it, and a transaction's write batch ignored the read-option bounds before + RocksDB 8.10.0, so a build linked against an older release (`ROCKSDB_VERSION` / `ROCKSDB_PATH`) + checks both bounds on transaction iterators (a compile-time `ROCKSDB_MAJOR`/`ROCKSDB_MINOR` + check); the pinned 11.8.1 only pays the reverse `exclusiveStart` compare, like a plain iterator. + `closeIterators()` waits for a handle that is mid-destruction on another thread to reset its + RocksDB iterator before the transaction is freed; it does not serialize a cross-environment + close against a `next()` in flight on the owning thread (the descriptor's closables sweep never + did either). They register weakly with `TransactionHandle`; commit, + abort, the coordinated-retry reset (`resetTransaction`), and forced teardown close every + registered iterator before committing, rolling back, resetting, or deleting the RocksDB + transaction, so a later `next()` deterministically reports an uninitialized iterator rather than + reading freed write-batch state (`return()`/`throw()` stay idempotent so loop cleanup after that + close cannot throw), and `createIterator` rejects a range or count once the transaction is no + longer pending. The reverse seek always steps off a key equal to the encoded end + bound: `inclusiveEnd` appends a NUL to that bound, so the bound itself is exclusive in both + directions and a staged key that lands exactly on it must be excluded like a committed one. + +21. **A WriteBufferManager stall is a second, entirely separate stall mechanism, and nothing in RocksDB reports it**: `DBImpl::WriteBufferManagerStallWrites` parks writers on the manager's own queue (`WBMStallInterface::Block`) without touching the `WriteController`, so `rocksdb.stall.micros`, the `WRITE_STALL` histogram, `OnStallConditionsChanged` — and therefore the `'writeStall'` event diff --git a/README.md b/README.md index fe615a589..bf8721b65 100644 --- a/README.md +++ b/README.md @@ -884,6 +884,23 @@ for (const { key, value } of db.getRange({ start: 'a', end: 'z' })) { } ``` +Pass `transaction: txn` to iterate this column family through that transaction, exactly as +`get()` does with the same option: the iterator sees the transaction's staged writes and reads on its +snapshot, and range bounds apply to staged keys too. `getKeys()` and `getKeysCount()` accept it as +well. On a transaction (`txn.getRange()`) the transaction itself is the context and takes precedence +over a `transaction` option, as it does for `txn.get()`. Such an iterator is closed when the +transaction commits or aborts: a later `next()` throws, while `return()` stays a no-op. Opening a +range or counting through a transaction that has already started committing throws as well. + +```typescript +await db.transaction(async (txn) => { + await txn.put('c', 'staged'); + for (const { key, value } of db.getRange({ start: 'a', end: 'z', transaction: txn })) { + console.log({ key, value }); // includes { key: "c", value: "staged" } + } +}); +``` + ### `db.getUserSharedBuffer(key: Key, defaultBuffer: ArrayBuffer, options?)` Creates a new buffer with the contents of `defaultBuffer` that can be accessed across threads. This @@ -2574,7 +2591,9 @@ Options for `get()`, `getSync()`, and the `getBinary*` methods. - `tailing: boolean` When `true`, creates a "tailing iterator" which is a special iterator that has a view of the complete database including newly added data and is optimized for sequential reads. This will return records that were inserted into the database after the creation of the - iterator. Defaults to `false`. + iterator. Defaults to `false`. A tailing iterator ignores a transaction's snapshot: through a + transaction it still reads the latest committed state, merged with that transaction's staged + writes. ### `RangeOptions` diff --git a/src/binding/binding.cpp b/src/binding/binding.cpp index d8e54e014..d6f8ab409 100644 --- a/src/binding/binding.cpp +++ b/src/binding/binding.cpp @@ -349,6 +349,7 @@ NAPI_MODULE_INIT() { EXPORT_CONSTANT(constants, ITERATOR_INCLUDE_VALUES_FLAG) EXPORT_CONSTANT(constants, ITERATOR_NEEDS_STABLE_VALUE_BUFFER_FLAG) EXPORT_CONSTANT(constants, ITERATOR_CONTEXT_IS_TRANSACTION_FLAG) + EXPORT_CONSTANT(constants, ITERATOR_HAS_TRANSACTION_ID_FLAG) EXPORT_CONSTANT(constants, ITERATOR_RESULT_DONE) EXPORT_CONSTANT(constants, ITERATOR_RESULT_FAST) NAPI_STATUS_THROWS(::napi_set_named_property(env, exports, "constants", constants)); diff --git a/src/binding/database/database.cpp b/src/binding/database/database.cpp index 0e5234cd2..6b5dd577c 100644 --- a/src/binding/database/database.cpp +++ b/src/binding/database/database.cpp @@ -237,6 +237,29 @@ napi_value Database::Columns(napi_env env, napi_callback_info info) { return result; } +/** + * The database's resolved filesystem identity — the registry key that two + * spellings of one directory (`data` and `./data`, a symlink and its target) + * share and that a repointed symlink or a `chdir` cannot change afterwards. + * `undefined` until the handle has been opened; retained after close. + * + * Callers comparing two handles for "same database" must use this and never + * the path they passed to `open()`, which is a spelling, not an identity. + */ +napi_value Database::IdentityPath(napi_env env, napi_callback_info info) { + NAPI_METHOD(); + UNWRAP_DB_HANDLE(); + + if (dbHandle == nullptr || (*dbHandle)->identityPath.empty()) { + NAPI_RETURN_UNDEFINED(); + } + + const std::string& identityPath = (*dbHandle)->identityPath; + napi_value result; + NAPI_STATUS_THROWS(::napi_create_string_utf8(env, identityPath.c_str(), identityPath.size(), &result)); + return result; +} + /** * Compacts the entire key range of the database asynchronously. * This triggers manual compaction which removes tombstones and reclaims space. @@ -1327,12 +1350,22 @@ napi_value Database::GetCount(napi_env env, napi_callback_info info) { ::napi_throw_error(env, nullptr, errorMsg.c_str()); NAPI_RETURN_UNDEFINED(); } - txnHandle->getCount(itOptions, count, *dbHandle); + try { + txnHandle->getCount(itOptions, count, *dbHandle); + } catch (const std::exception& e) { + ::napi_throw_error(env, nullptr, e.what()); + NAPI_RETURN_UNDEFINED(); + } } else { - std::unique_ptr itHandle = std::make_unique(*dbHandle, itOptions); - while (itHandle->iterator->Valid()) { - ++count; - itHandle->iterator->Next(); + try { + std::unique_ptr itHandle = std::make_unique(*dbHandle, itOptions); + while (itHandle->valid()) { + ++count; + itHandle->advance(); + } + } catch (const std::exception& e) { + ::napi_throw_error(env, nullptr, e.what()); + NAPI_RETURN_UNDEFINED(); } } @@ -2726,6 +2759,7 @@ void Database::Init(napi_env env, napi_value exports) { { "getSync", nullptr, GetSync, nullptr, nullptr, nullptr, napi_default, nullptr }, { "getUserSharedBuffer", nullptr, GetUserSharedBuffer, nullptr, nullptr, nullptr, napi_default, nullptr }, { "hasLock", nullptr, HasLock, nullptr, nullptr, nullptr, napi_default, nullptr }, + { "identityPath", nullptr, nullptr, IdentityPath, nullptr, nullptr, napi_default, nullptr }, { "listeners", nullptr, Listeners, nullptr, nullptr, nullptr, napi_default, nullptr }, { "listLogs", nullptr, ListLogs, nullptr, nullptr, nullptr, napi_default, nullptr }, { "notify", nullptr, Notify, nullptr, nullptr, nullptr, napi_default, nullptr }, diff --git a/src/binding/database/database.h b/src/binding/database/database.h index 3c56e5e90..09e45ca10 100644 --- a/src/binding/database/database.h +++ b/src/binding/database/database.h @@ -329,6 +329,7 @@ struct Database final { static napi_value GetSync(napi_env env, napi_callback_info info); static napi_value GetUserSharedBuffer(napi_env env, napi_callback_info info); static napi_value HasLock(napi_env env, napi_callback_info info); + static napi_value IdentityPath(napi_env env, napi_callback_info info); static napi_value IsOpen(napi_env env, napi_callback_info info); static napi_value Listeners(napi_env env, napi_callback_info info); static napi_value ListLogs(napi_env env, napi_callback_info info); diff --git a/src/binding/iterator/db_iterator.cpp b/src/binding/iterator/db_iterator.cpp index d2ac626e3..67cd0525a 100644 --- a/src/binding/iterator/db_iterator.cpp +++ b/src/binding/iterator/db_iterator.cpp @@ -76,9 +76,10 @@ napi_status DBIteratorOptions::initFromNapiObject(napi_env env, napi_value optio * argv[4] - endKeyEnd (uint32) - end position of end key; if equal to * endKeyStart there is no end key * argv[5] - optional advanced ReadOptions object (rare path) + * argv[6] - optional transaction ID for a Database context */ napi_value DBIterator::Constructor(napi_env env, napi_callback_info info) { - NAPI_CONSTRUCTOR_ARGV("Iterator", 6); + NAPI_CONSTRUCTOR_ARGV("Iterator", 7); uint32_t flags = 0; NAPI_STATUS_THROWS(::napi_get_value_uint32(env, argv[1], &flags)); @@ -117,6 +118,7 @@ napi_value DBIterator::Constructor(napi_env env, napi_callback_info info) { // napi_get_reference_value + napi_get_named_property + napi_instanceof // per iterator construction. const bool isTransaction = (flags & ITERATOR_CONTEXT_IS_TRANSACTION_FLAG) != 0; + const bool hasTransactionId = (flags & ITERATOR_HAS_TRANSACTION_ID_FLAG) != 0; std::shared_ptr* itHandle = nullptr; std::shared_ptr* dbHandle = nullptr; @@ -130,6 +132,14 @@ napi_value DBIterator::Constructor(napi_env env, napi_callback_info info) { return nullptr; } txnHandle = *wrappedTxnHandle; + if (!txnHandle->dbHandle) { + ::napi_throw_error(env, nullptr, "Transaction is not in pending state"); + return nullptr; + } + if (!txnHandle->dbHandle->opened()) { + ::napi_throw_error(env, nullptr, "Database not open"); + return nullptr; + } dbHandle = &txnHandle->dbHandle; DEBUG_LOG("DBIterator::Constructor txnHandle=%p dbHandle=%p\n", txnHandle.get(), dbHandle->get()); } else { @@ -139,6 +149,21 @@ napi_value DBIterator::Constructor(napi_env env, napi_callback_info info) { return nullptr; } DEBUG_LOG("DBIterator::Constructor Initializing iterator handle with Database instance (dbHandle=%p)\n", (*dbHandle).get()); + + if (hasTransactionId) { + if (argc <= 6) { + ::napi_throw_type_error(env, nullptr, "Invalid transaction"); + return nullptr; + } + uint32_t txnId = 0; + NAPI_STATUS_THROWS(::napi_get_value_uint32(env, argv[6], &txnId)); + txnHandle = (*dbHandle)->descriptor->transactionGet(txnId); + if (!txnHandle) { + std::string errorMsg = "Iterator failed: Transaction not found (txnId: " + std::to_string(txnId) + ")"; + ::napi_throw_error(env, nullptr, errorMsg.c_str()); + return nullptr; + } + } } // Resolve start/end key pointers from the shared default key buffer @@ -163,11 +188,13 @@ napi_value DBIterator::Constructor(napi_env env, napi_callback_info info) { } try { + std::shared_ptr iteratorHandle; if (txnHandle) { - itHandle = new std::shared_ptr(std::make_shared(txnHandle, itOptions)); + iteratorHandle = txnHandle->createIterator(itOptions, isTransaction ? nullptr : *dbHandle); } else { - itHandle = new std::shared_ptr(std::make_shared(*dbHandle, itOptions)); + iteratorHandle = std::make_shared(*dbHandle, itOptions); } + itHandle = new std::shared_ptr(std::move(iteratorHandle)); } catch (const std::exception& e) { ::napi_throw_error(env, nullptr, e.what()); return nullptr; @@ -218,6 +245,16 @@ napi_value DBIterator::Constructor(napi_env env, napi_callback_info info) { } \ } while (0) +#define CLOSE_ITERATOR_HANDLE(fnName) \ + do { \ + std::shared_ptr* itHandle = nullptr; \ + NAPI_STATUS_THROWS(::napi_unwrap(env, jsThis, reinterpret_cast(&itHandle))); \ + if (itHandle && *itHandle) { \ + DEBUG_LOG("%p DBIterator::" fnName " Closing iterator handle\n", (*itHandle).get()); \ + (*itHandle)->close(); \ + } \ + } while (0) + /** * Builds a slow-path object `{ key: Buffer, value?: Buffer }` for the rare case * where the shared key/value buffers cannot be used (oversized data or stable @@ -266,8 +303,8 @@ napi_value DBIterator::Next(napi_env env, napi_callback_info info) { auto& it = *itHandle; napi_value result; - if (!it->iterator->Valid()) { - if (!it->iterator->status().ok()) { + if (!it->valid()) { + if (it->iterator && !it->iterator->status().ok()) { DEBUG_LOG("%p DBIterator::Next iterator not valid/ok: %s\n", itHandle, it->iterator->status().ToString().c_str()); } else { DEBUG_LOG("%p DBIterator::Next iterator no keys found in range\n", it.get()); @@ -278,21 +315,6 @@ napi_value DBIterator::Next(napi_env env, napi_callback_info info) { rocksdb::Slice keySlice = it->iterator->key(); - // Edge case: reverse + exclusiveStart and we just landed on the start key - // (which is the lower bound for reverse iteration). We need to peek ahead - // to determine whether this is the last item. - if (it->reverse && it->exclusiveStart && - it->startKey.size() > 0 && keySlice.compare(it->startKey) == 0) { - it->iterator->Prev(); - if (!it->iterator->Valid()) { - NAPI_STATUS_THROWS(::napi_create_uint32(env, ITERATOR_RESULT_DONE, &result)); - return result; - } - // not the last item; restore position and continue normally below - it->iterator->Next(); - keySlice = it->iterator->key(); - } - // Try the fast path: copy key (and optionally value) into the shared // default buffers and write lengths to the iterator state buffer. auto& dbHandle = it->dbHandle; @@ -321,22 +343,14 @@ napi_value DBIterator::Next(napi_env env, napi_callback_info info) { ::memcpy(valueBuffer, valueSlice.data(), valueSlice.size()); state[1] = static_cast(valueSlice.size()); } - if (it->reverse) { - it->iterator->Prev(); - } else { - it->iterator->Next(); - } + it->advance(); NAPI_STATUS_THROWS(::napi_create_uint32(env, ITERATOR_RESULT_FAST, &result)); return result; } // Slow path: at least one of key or value can't go in the shared buffer. napi_value slowResult = buildSlowResult(env, keySlice, it->values, valueSlice); - if (it->reverse) { - it->iterator->Prev(); - } else { - it->iterator->Next(); - } + it->advance(); return slowResult; } @@ -347,11 +361,7 @@ napi_value DBIterator::Next(napi_env env, napi_callback_info info) { */ napi_value DBIterator::Return(napi_env env, napi_callback_info info) { NAPI_METHOD(); - UNWRAP_ITERATOR_HANDLE("Return"); - - DEBUG_LOG("%p DBIterator::Return Closing iterator handle\n", (*itHandle).get()); - (*itHandle)->close(); - + CLOSE_ITERATOR_HANDLE("Return"); NAPI_RETURN_UNDEFINED(); } @@ -361,11 +371,7 @@ napi_value DBIterator::Return(napi_env env, napi_callback_info info) { */ napi_value DBIterator::Throw(napi_env env, napi_callback_info info) { NAPI_METHOD(); - UNWRAP_ITERATOR_HANDLE("Throw"); - - DEBUG_LOG("%p DBIterator::Throw Closing iterator handle\n", (*itHandle).get()); - (*itHandle)->close(); - + CLOSE_ITERATOR_HANDLE("Throw"); NAPI_RETURN_UNDEFINED(); } diff --git a/src/binding/iterator/db_iterator.h b/src/binding/iterator/db_iterator.h index 8536f4ef5..575b57d2a 100644 --- a/src/binding/iterator/db_iterator.h +++ b/src/binding/iterator/db_iterator.h @@ -19,6 +19,7 @@ namespace rocksdb_js { // constructor relies on this flag to skip the expensive // napi_get_named_property + napi_instanceof type checks. #define ITERATOR_CONTEXT_IS_TRANSACTION_FLAG 0x20 +#define ITERATOR_HAS_TRANSACTION_ID_FLAG 0x40 // Iterator Next() return signals #define ITERATOR_RESULT_DONE 0 diff --git a/src/binding/iterator/db_iterator_handle.cpp b/src/binding/iterator/db_iterator_handle.cpp index 9ccfc589e..c5d29cf06 100644 --- a/src/binding/iterator/db_iterator_handle.cpp +++ b/src/binding/iterator/db_iterator_handle.cpp @@ -1,9 +1,24 @@ #include "iterator/db_iterator_handle.h" #include "database/db_descriptor.h" +#include #include namespace rocksdb_js { +namespace { + +// A transaction's write batch ignored ReadOptions bounds before RocksDB 8.10.0 +// (facebook/rocksdb#11680). Either way `iterate_lower_bound` is inclusive, so +// an exclusive lower bound reached by a reverse scan is the handle's to apply. +constexpr bool WRITE_BATCH_HONORS_BOUNDS = ROCKSDB_MAJOR > 8 || (ROCKSDB_MAJOR == 8 && ROCKSDB_MINOR >= 10); + +bool needsBoundCheck(const DBIteratorOptions& options, bool writeBatch) { + return (writeBatch && !WRITE_BATCH_HONORS_BOUNDS) + || (options.reverse && options.exclusiveStart && options.startKeyStr != nullptr); +} + +} + DBIteratorHandle::DBIteratorHandle( std::shared_ptr dbHandle, DBIteratorOptions& options @@ -13,7 +28,8 @@ DBIteratorHandle::DBIteratorHandle( inclusiveEnd(options.inclusiveEnd), reverse(options.reverse), values(options.values), - needsStableValueBuffer(options.needsStableValueBuffer) + needsStableValueBuffer(options.needsStableValueBuffer), + enforceBounds(needsBoundCheck(options, false)) { DEBUG_LOG("%p DBIteratorHandle::Constructor dbHandle=%p\n", this, dbHandle.get()); this->init(options); @@ -39,9 +55,14 @@ DBIteratorHandle::DBIteratorHandle( inclusiveEnd(options.inclusiveEnd), reverse(options.reverse), values(options.values), - needsStableValueBuffer(options.needsStableValueBuffer) + needsStableValueBuffer(options.needsStableValueBuffer), + enforceBounds(needsBoundCheck(options, true)) { DEBUG_LOG("DBIteratorHandle::Constructor txnHandle=%p dbDescriptor=%p\n", this->txnHandle.get(), dbHandle->descriptor.get()); + this->txnHandle->ensureSnapshot(); + if (this->txnHandle->snapshotSet) { + options.readOptions.snapshot = this->txnHandle->txn->GetSnapshot(); + } this->init(options); this->iterator = std::unique_ptr( @@ -52,7 +73,6 @@ DBIteratorHandle::DBIteratorHandle( ); this->seek(options); - this->txnHandle->registerIterator(); } DBIteratorHandle::~DBIteratorHandle() { @@ -60,14 +80,16 @@ DBIteratorHandle::~DBIteratorHandle() { } void DBIteratorHandle::close() { + std::lock_guard lock(this->closeMutex); DEBUG_LOG("%p DBIteratorHandle::close dbHandle=%p dbDescriptor=%p\n", this, this->dbHandle.get(), this->dbHandle->descriptor.get()); if (this->iterator) { this->iterator->Reset(); this->iterator.reset(); } - if (this->txnHandle) { + if (this->txnHandle && this->transactionRegistered) { + this->transactionRegistered = false; auto txnHandle = std::move(this->txnHandle); - txnHandle->unregisterIterator(); + txnHandle->unregisterIterator(this); } } @@ -100,9 +122,20 @@ void DBIteratorHandle::init(DBIteratorOptions& options) { void DBIteratorHandle::seek(DBIteratorOptions& options) { if (options.reverse) { - this->iterator->SeekToLast(); + if (this->endKey.size() > 0) { + this->iterator->SeekForPrev(this->endKey); + if (this->iterator->Valid() && this->iterator->key().compare(this->endKey) == 0) { + this->iterator->Prev(); + } + } else { + this->iterator->SeekToLast(); + } } else { - this->iterator->SeekToFirst(); + if (this->startKey.size() > 0) { + this->iterator->Seek(this->startKey); + } else { + this->iterator->SeekToFirst(); + } } if (options.exclusiveStart && options.startKeyStr != nullptr && this->iterator->Valid()) { @@ -117,4 +150,23 @@ void DBIteratorHandle::seek(DBIteratorOptions& options) { } } +bool DBIteratorHandle::valid() const { + if (!this->iterator || !this->iterator->Valid()) { + return false; + } + if (!this->enforceBounds) { + return true; + } + + const rocksdb::Slice key = this->iterator->key(); + if (this->reverse) { + if (this->startKey.size() == 0) { + return true; + } + const int comparison = key.compare(this->startKey); + return comparison > 0 || (comparison == 0 && !this->exclusiveStart); + } + return this->endKey.size() == 0 || key.compare(this->endKey) < 0; +} + } diff --git a/src/binding/iterator/db_iterator_handle.h b/src/binding/iterator/db_iterator_handle.h index 06e237275..5c763b45b 100644 --- a/src/binding/iterator/db_iterator_handle.h +++ b/src/binding/iterator/db_iterator_handle.h @@ -47,11 +47,24 @@ struct DBIteratorHandle final : Closable, public std::enable_shared_from_thisreverse) { + this->iterator->Prev(); + } else { + this->iterator->Next(); + } + } + std::shared_ptr dbHandle; std::shared_ptr txnHandle; bool exclusiveStart; @@ -59,11 +72,14 @@ struct DBIteratorHandle final : Closable, public std::enable_shared_from_this iterator; std::string startKeyStr; std::string endKeyStr; rocksdb::Slice startKey; rocksdb::Slice endKey; + std::mutex closeMutex; + bool transactionRegistered = false; private: /** diff --git a/src/binding/transaction/transaction.cpp b/src/binding/transaction/transaction.cpp index ee13024fc..8f4d22b35 100644 --- a/src/binding/transaction/transaction.cpp +++ b/src/binding/transaction/transaction.cpp @@ -152,6 +152,7 @@ napi_value Transaction::Abort(napi_env env, napi_callback_info info) { } bool hadLogWrites = (*txnHandle)->committedPosition.logSequenceNumber > 0; + (*txnHandle)->closeIterators(); (*txnHandle)->state = TransactionState::Aborted; ROCKSDB_STATUS_THROWS_ERROR_LIKE((*txnHandle)->txn->Rollback(), "Transaction rollback failed"); @@ -764,6 +765,7 @@ napi_value Transaction::Commit(napi_env env, napi_callback_info info) { NAPI_STATUS_THROWS(::napi_call_function(env, global, resolve, 0, nullptr, nullptr)); return nullptr; } + (*txnHandle)->closeIterators(); TransactionCommitState* state = new TransactionCommitState(env, *txnHandle); NAPI_STATUS_THROWS(::napi_create_reference(env, resolve, 1, &state->resolveRef)); @@ -896,6 +898,7 @@ napi_value Transaction::CommitSync(napi_env env, napi_callback_info info) { if (txnState == TransactionState::Committing || txnState == TransactionState::Committed) { NAPI_RETURN_UNDEFINED(); } + (*txnHandle)->closeIterators(); (*txnHandle)->state = TransactionState::Committing; std::shared_ptr store = nullptr; @@ -1056,7 +1059,12 @@ napi_value Transaction::GetCount(napi_env env, napi_callback_info info) { itOptions.values = false; uint64_t count = 0; - (*txnHandle)->getCount(itOptions, count); + try { + (*txnHandle)->getCount(itOptions, count); + } catch (const std::exception& e) { + ::napi_throw_error(env, nullptr, e.what()); + NAPI_RETURN_UNDEFINED(); + } napi_value result; NAPI_STATUS_THROWS(::napi_create_int64(env, count, &result)); diff --git a/src/binding/transaction/transaction_handle.cpp b/src/binding/transaction/transaction_handle.cpp index a0e346130..a12f93111 100644 --- a/src/binding/transaction/transaction_handle.cpp +++ b/src/binding/transaction/transaction_handle.cpp @@ -96,6 +96,7 @@ TransactionHandle::TransactionHandle(std::shared_ptr dbHandle, bool di void TransactionHandle::resetTransaction(){ // clear/delete the previous transaction and create a new transaction so that it can be retried + this->closeIterators(); if (this->txn) { this->txn->ClearSnapshot(); delete this->txn; @@ -250,11 +251,38 @@ void TransactionHandle::onWrapperCollected() { this->closeOrphanIfUnused(); } -void TransactionHandle::registerIterator() { +std::shared_ptr TransactionHandle::createIterator( + DBIteratorOptions& options, + std::shared_ptr dbHandleOverride +) { + std::lock_guard lock(this->iteratorsMutex); + if (this->closed.load() || !this->txn || this->state != TransactionState::Pending) { + throw std::runtime_error("Transaction is not in pending state"); + } + const std::shared_ptr& targetHandle = dbHandleOverride ? dbHandleOverride : this->dbHandle; + if (!targetHandle || !targetHandle->opened()) { + throw std::runtime_error("Database not open"); + } + std::shared_ptr iterator = std::make_shared( + this->shared_from_this(), + options, + dbHandleOverride + ); + const bool inserted = this->activeIterators.emplace(iterator.get(), iterator).second; + assert(inserted && "Transaction iterator registered twice"); + (void)inserted; + iterator->transactionRegistered = true; this->activeIteratorCount.fetch_add(1, std::memory_order_relaxed); + return iterator; } -void TransactionHandle::unregisterIterator() { +void TransactionHandle::unregisterIterator(DBIteratorHandle* iterator) { + { + std::lock_guard lock(this->iteratorsMutex); + if (this->activeIterators.erase(iterator) == 0) { + return; + } + } const uint32_t previous = this->activeIteratorCount.fetch_sub(1, std::memory_order_relaxed); assert(previous > 0 && "Transaction iterator count underflow"); if (previous == 1) { @@ -262,6 +290,33 @@ void TransactionHandle::unregisterIterator() { } } +void TransactionHandle::closeIterators() { + // Never allocates: teardown paths cannot throw. A handle mid-destruction on + // another thread cannot be pinned, but its destructor resets the RocksDB + // iterator before erasing its entry, so wait for that before `txn` is freed. + for (;;) { + std::shared_ptr iterator; + bool expired = false; + { + std::lock_guard lock(this->iteratorsMutex); + for (const auto& [_iterator, weakIterator] : this->activeIterators) { + iterator = weakIterator.lock(); + if (iterator) { + break; + } + expired = true; + } + } + if (iterator) { + iterator->close(); + } else if (expired) { + std::this_thread::yield(); + } else { + return; + } + } +} + void TransactionHandle::closeOrphanIfUnused() { if (!this->wrapperCollected.load() || this->closed.load()) { return; @@ -302,6 +357,7 @@ void TransactionHandle::close() { if (this->closed.exchange(true)) { return; } + this->closeIterators(); if (this->dbHandle && this->dbHandle->descriptor) { this->dbHandle->descriptor->transactionRemove(shared_from_this()); @@ -595,16 +651,11 @@ void TransactionHandle::getCount( uint64_t& count, std::shared_ptr dbHandleOverride ) { - this->ensureSnapshot(); - if (this->snapshotSet) { - itOptions.readOptions.snapshot = this->txn->GetSnapshot(); - } - - std::unique_ptr itHandle = - std::make_unique(this->shared_from_this(), itOptions, dbHandleOverride); - for (count = 0; itHandle->iterator->Valid(); ++count) { - itHandle->iterator->Next(); + std::shared_ptr itHandle = this->createIterator(itOptions, std::move(dbHandleOverride)); + for (count = 0; itHandle->valid(); ++count) { + itHandle->advance(); } + itHandle->close(); } /** diff --git a/src/binding/transaction/transaction_handle.h b/src/binding/transaction/transaction_handle.h index 85c01e33a..d8b6c0805 100644 --- a/src/binding/transaction/transaction_handle.h +++ b/src/binding/transaction/transaction_handle.h @@ -21,6 +21,7 @@ namespace rocksdb_js { struct DBHandle; +struct DBIteratorHandle; struct DBIteratorOptions; struct TransactionLogStore; @@ -44,8 +45,9 @@ enum class TransactionState { * This handle contains `get()`, `put()`, and `remove()` methods which are * shared between the `Database` and `Transaction` classes. * - * Each instance of this class is bound to a JavaScript `Transaction` instance. - * Since a JS instance is bound to a single thread, we don't need any mutexes. + * Each instance of this class is bound to a JavaScript `Transaction` instance, + * but descriptor teardown can close it from another environment. State used by + * those cross-thread teardown paths must be synchronized explicitly. */ struct TransactionHandle final : Closable, AsyncWorkHandle, std::enable_shared_from_this { /** @@ -125,9 +127,13 @@ struct TransactionHandle final : Closable, AsyncWorkHandle, std::enable_shared_f /** * Transaction-backed iterators whose RocksDB iterator still depends on txn. - * Orphan cleanup waits for this to reach zero before destroying txn. + * Orphan cleanup waits for the count to reach zero. The weak map lets an + * explicit commit/abort close those iterators before RocksDB mutates or + * destroys txn without creating an ownership cycle. */ std::atomic activeIteratorCount{0}; + std::mutex iteratorsMutex; + std::unordered_map> activeIterators; /** * A batch of log entries to write to the transaction log. It can only be @@ -199,11 +205,20 @@ struct TransactionHandle final : Closable, AsyncWorkHandle, std::enable_shared_f void onWrapperCollected(); /** - * Registers/releases a transaction-backed iterator dependency. The final - * release retries a deferred orphan close. + * Builds a transaction-backed iterator and registers it while the + * transaction is still pending. The handle is marked registered only after + * the insert, so a failed insert unwinds without re-entering + * `iteratorsMutex` from its destructor. `unregisterIterator` releases the + * dependency (the final release retries a deferred orphan close); + * `closeIterators` closes every registered handle before `txn` is consumed, + * reset, or deleted. */ - void registerIterator(); - void unregisterIterator(); + std::shared_ptr createIterator( + DBIteratorOptions& options, + std::shared_ptr dbHandleOverride = nullptr + ); + void unregisterIterator(DBIteratorHandle* iterator); + void closeIterators(); /** * Closes a collected wrapper once no async work or iterator still depends diff --git a/src/load-binding.ts b/src/load-binding.ts index fc2ae1f04..1e55307d1 100644 --- a/src/load-binding.ts +++ b/src/load-binding.ts @@ -260,7 +260,8 @@ export declare class NativeIteratorCls { startKeyEnd: number, endKeyStart: number, endKeyEnd: number, - options?: NativeIteratorAdvancedOptions + options?: NativeIteratorAdvancedOptions, + transactionId?: number ); next(): NativeIteratorResult; return(): void; @@ -478,6 +479,7 @@ export type NativeDatabase = { callback?: UserSharedBufferCallback ): ArrayBuffer; hasLock(key: BufferWithDataView): boolean; + identityPath: string | undefined; listeners(event: string | BufferWithDataView): number; listLogs(): string[]; opened: boolean; @@ -796,6 +798,7 @@ export const constants: { ITERATOR_INCLUDE_VALUES_FLAG: number; ITERATOR_NEEDS_STABLE_VALUE_BUFFER_FLAG: number; ITERATOR_CONTEXT_IS_TRANSACTION_FLAG: number; + ITERATOR_HAS_TRANSACTION_ID_FLAG: number; ITERATOR_RESULT_DONE: number; ITERATOR_RESULT_FAST: number; } = binding.constants; diff --git a/src/store.ts b/src/store.ts index 99b654e25..b58741e41 100644 --- a/src/store.ts +++ b/src/store.ts @@ -45,6 +45,7 @@ const { ITERATOR_INCLUDE_VALUES_FLAG, ITERATOR_NEEDS_STABLE_VALUE_BUFFER_FLAG, ITERATOR_CONTEXT_IS_TRANSACTION_FLAG, + ITERATOR_HAS_TRANSACTION_ID_FLAG, } = constants; const KEY_BUFFER_SIZE = 4096; @@ -563,6 +564,12 @@ export class Store { */ path: string; + /** + * The open database's resolved filesystem identity, read once from + * `NativeDatabase.identityPath`. `undefined` until opened. + */ + identityPath?: string; + /** * Whether to use pessimistic locking for transactions. When `true`, * transactions will fail as soon as a conflict is detected. When `false`, @@ -1056,6 +1063,7 @@ export class Store { const includeValues = options?.values ?? true; const reverse = options?.reverse ?? false; + const txnId = this.getTxnId(options); let exclusiveStart = options?.exclusiveStart ?? false; let inclusiveEnd = options?.inclusiveEnd ?? false; @@ -1116,6 +1124,9 @@ export class Store { if (context !== this.db) { flags |= ITERATOR_CONTEXT_IS_TRANSACTION_FLAG; } + if (txnId !== undefined) { + flags |= ITERATOR_HAS_TRANSACTION_ID_FLAG; + } // Only pass the advanced ReadOptions object on the rare path where any // of the underlying RocksDB iterator options are actually overridden. @@ -1134,7 +1145,15 @@ export class Store { return new ExtendedIterable( // @ts-expect-error ExtendedIterable v1 constructor type definition is incorrect new DBIterator( - new NativeIterator(context, flags, startKeyEnd, endKeyStart, endKeyEnd, advancedOptions), + new NativeIterator( + context, + flags, + startKeyEnd, + endKeyStart, + endKeyEnd, + advancedOptions, + txnId + ), this, includeValues, options?.limit @@ -1180,11 +1199,21 @@ export class Store { */ getTxnId(options?: DBITransactional | unknown): number | undefined { let txnId: number | undefined; - if (!this.readOnly && (options as DBITransactional)?.transaction) { - txnId = (options as DBITransactional).transaction!.id; + const transaction = (options as DBITransactional)?.transaction; + if (!this.readOnly && transaction) { + txnId = transaction.id; if (txnId === undefined) { throw new TypeError('Invalid transaction'); } + if (transaction.store === undefined) { + throw new TypeError('Invalid transaction'); + } + // Ids are per database, so one from elsewhere resolves here to an + // unrelated transaction of the same number. Identity, never the path + // the caller spelled. + if (this.identityPath !== undefined && transaction.store.identityPath !== this.identityPath) { + throw new TypeError('Transaction belongs to a different database'); + } } return txnId; } @@ -1265,6 +1294,7 @@ export class Store { */ open(): boolean { if (this.db.opened) { + this.identityPath = this.db.identityPath; return true; } @@ -1299,6 +1329,8 @@ export class Store { writeBufferSize: this.writeBufferSize, }); + this.identityPath = this.db.identityPath; + return false; } diff --git a/test/fixtures/transaction-orphan-dependents.mts b/test/fixtures/transaction-orphan-dependents.mts index 5cb2cf813..56c1c09e1 100644 --- a/test/fixtures/transaction-orphan-dependents.mts +++ b/test/fixtures/transaction-orphan-dependents.mts @@ -6,9 +6,13 @@ import { setTimeout as delay } from 'node:timers/promises'; const mode = process.argv[2]; const dbPath = process.argv[3]; -if ((mode !== 'async-get' && mode !== 'iterator') || !dbPath || !globalThis.gc) { +if ( + (mode !== 'async-get' && mode !== 'iterator' && mode !== 'routed-iterator') || + !dbPath || + !globalThis.gc +) { console.error( - 'Usage: node --expose-gc transaction-orphan-dependents.mts ' + 'Usage: node --expose-gc transaction-orphan-dependents.mts ' ); process.exit(1); } @@ -55,14 +59,16 @@ async function testDelayedAsyncGet(db: RocksDatabase): Promise { await waitForTransactionClose(); } -async function testLiveIterator(db: RocksDatabase): Promise { +async function testLiveIterator(db: RocksDatabase, routed: boolean): Promise { for (const key of ['a', 'b', 'c']) { await db.put(key, `value-${key}`); } const iterator = (() => { const txn = new Transaction(db.store); - const iterator = txn.getRange()[Symbol.iterator](); + const iterator = (routed ? db.getRange({ transaction: txn }) : txn.getRange())[ + Symbol.iterator + ](); assert.deepEqual(iterator.next(), { done: false, value: { key: 'a', value: 'value-a' } }); return iterator; })(); @@ -79,7 +85,7 @@ try { if (mode === 'async-get') { await testDelayedAsyncGet(db); } else { - await testLiveIterator(db); + await testLiveIterator(db, mode === 'routed-iterator'); } } finally { db.close(); diff --git a/test/ranges.test.ts b/test/ranges.test.ts index 1abde0857..fb40ac782 100644 --- a/test/ranges.test.ts +++ b/test/ranges.test.ts @@ -1,6 +1,8 @@ import type { IteratorOptions } from '../src/dbi.ts'; import type { Key } from '../src/encoding.ts'; -import { dbRunner } from './lib/util.ts'; +import { Transaction } from '../src/transaction.ts'; +import { dbRunner, generateDBPath } from './lib/util.ts'; +import { basename, sep } from 'node:path'; import { describe, expect, it } from 'vitest'; describe('Ranges', () => { @@ -111,6 +113,376 @@ describe('Ranges', () => { }); })); + it('should honor the transaction option across range APIs', () => + dbRunner(async ({ db }) => { + await db.put('committed', 'before'); + const txn = new Transaction(db.store); + try { + await txn.put('staged', 'in-batch'); + const options = { start: 'a', end: 'z', transaction: txn }; + + expect(await db.get('staged', { transaction: txn })).toBe('in-batch'); + expect(db.getRange(options).asArray).toEqual([ + { key: 'committed', value: 'before' }, + { key: 'staged', value: 'in-batch' }, + ]); + expect(db.getKeys(options).asArray).toEqual(['committed', 'staged']); + expect(db.getKeysCount(options)).toBe(2); + expect(db.store.getCount(db._context, options)).toBe(2); + expect(txn.getRange({ start: 'a', end: 'z' }).asArray).toEqual([ + { key: 'committed', value: 'before' }, + { key: 'staged', value: 'in-batch' }, + ]); + } finally { + txn.abort(); + } + })); + + it('should use the transaction snapshot for routed and direct ranges', () => + dbRunner(async ({ db }) => { + await db.put('committed', 'before'); + const txn = new Transaction(db.store); + try { + expect(await db.get('committed', { transaction: txn })).toBe('before'); + await db.transaction(async (writer) => { + await writer.put('later', 'after'); + }); + + const options = { start: 'a', end: 'z', transaction: txn }; + const expected = [{ key: 'committed', value: 'before' }]; + expect(db.getRange(options).asArray).toEqual(expected); + expect(txn.getRange({ start: 'a', end: 'z' }).asArray).toEqual(expected); + } finally { + txn.abort(); + } + })); + + it('should enforce bounds and limits for staged keys in both directions', () => + dbRunner( + { dbOptions: [{}, { path: generateDBPath() }] }, + async ({ db }, { db: reference }) => { + for (const key of ['a', 'b', 'c', 'd', 'e', 'f', 'g']) { + await reference.put(key, `value ${key}`); + } + for (const key of ['b', 'd', 'f']) { + await db.put(key, `value ${key}`); + } + + const txn = new Transaction(db.store); + try { + for (const key of ['a', 'c', 'e', 'g']) { + await txn.put(key, `value ${key}`); + } + + const forward = { start: 'a', end: 'e', exclusiveStart: true, inclusiveEnd: true }; + const reverse = { start: 'e', end: 'a', reverse: true }; + const reverseOpen = { ...reverse, exclusiveStart: false, inclusiveEnd: false }; + expect(reference.getKeys(forward).asArray).toEqual(['b', 'c', 'd', 'e']); + expect(reference.getKeys(reverse).asArray).toEqual(['e', 'd', 'c', 'b']); + expect(reference.getKeys(reverseOpen).asArray).toEqual(['d', 'c', 'b', 'a']); + + const variants: IteratorOptions[] = [ + { start: 'a', end: 'e' }, + forward, + { start: 'a', end: 'e', limit: 2 }, + { end: 'c' }, + { key: 'c' }, + { key: 'e' }, + reverse, + reverseOpen, + { ...reverse, limit: 2 }, + { start: 'g', reverse: true }, + ]; + for (const variant of variants) { + const expected = reference.getKeys(variant).asArray; + expect(expected.length).toBeGreaterThan(0); + expect(db.getKeys({ ...variant, transaction: txn }).asArray).toEqual(expected); + expect(txn.getKeys(variant).asArray).toEqual(expected); + if (!variant.reverse && variant.limit === undefined && variant.key === undefined) { + expect(db.getKeysCount({ ...variant, transaction: txn })).toBe(expected.length); + expect(db.store.getCount(db._context, { ...variant, transaction: txn })).toBe( + expected.length + ); + } + } + } finally { + txn.abort(); + } + } + )); + + it('should exclude a staged key that lands exactly on the encoded end bound', () => + dbRunner( + { + dbOptions: [{ keyEncoding: 'binary' }, { keyEncoding: 'binary', path: generateDBPath() }], + }, + async ({ db }, { db: reference }) => { + // `inclusiveEnd` makes the native bound `b\0`, so a key equal to `b\0` sits on it + const [a, b, bNul, c] = ['a', 'b', 'b\0', 'c'].map((key) => Buffer.from(key, 'latin1')); + for (const key of [a, b, bNul, c]) { + await reference.put(key, key.toString('latin1')); + } + await db.put(a, 'a'); + await db.put(c, 'c'); + + const txn = new Transaction(db.store); + try { + await txn.put(b, 'b'); + await txn.put(bNul, 'b\0'); + + const decode = (keys: Uint8Array[]) => + keys.map((key) => Buffer.from(key).toString('latin1')); + const variants: IteratorOptions[] = [ + { start: b, end: a, reverse: true }, + { start: a, end: b, inclusiveEnd: true }, + ]; + for (const variant of variants) { + const expected = decode(reference.getKeys(variant).asArray); + expect(expected).toEqual(variant.reverse ? ['b'] : ['a', 'b']); + expect(decode(db.getKeys({ ...variant, transaction: txn }).asArray)).toEqual( + expected + ); + expect(decode(txn.getKeys(variant).asArray)).toEqual(expected); + } + } finally { + txn.abort(); + } + } + )); + + it('should reflect staged overwrites and deletes in routed ranges', () => + dbRunner(async ({ db }) => { + await db.put('a', 'committed-a'); + await db.put('b', 'committed-b'); + const txn = new Transaction(db.store); + try { + await txn.put('a', 'staged-a'); + await txn.remove('b'); + await txn.put('c', 'staged-c'); + + const expected = [ + { key: 'a', value: 'staged-a' }, + { key: 'c', value: 'staged-c' }, + ]; + expect(db.getRange({ transaction: txn }).asArray).toEqual(expected); + expect(db.getKeysCount({ transaction: txn })).toBe(2); + expect(txn.getRange().asArray).toEqual(expected); + expect(db.getRange().asArray).toEqual([ + { key: 'a', value: 'committed-a' }, + { key: 'b', value: 'committed-b' }, + ]); + } finally { + txn.abort(); + } + })); + + it('should iterate staged keys over an empty database', () => + dbRunner(async ({ db }) => { + const txn = new Transaction(db.store); + try { + await txn.put('b', 'staged-b'); + await txn.put('a', 'staged-a'); + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['a', 'b']); + expect(db.getKeys({ transaction: txn, reverse: true }).asArray).toEqual(['b', 'a']); + expect(db.getKeysCount({ transaction: txn })).toBe(2); + expect(db.getKeys().asArray).toEqual([]); + } finally { + txn.abort(); + } + })); + + it('should establish the transaction snapshot on a first routed range read', () => + dbRunner(async ({ db }) => { + await db.put('committed', 'before'); + const txn = new Transaction(db.store); + try { + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed']); + await db.transaction(async (writer) => { + await writer.put('later', 'after'); + }); + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed']); + expect(db.getKeysCount({ transaction: txn })).toBe(1); + expect(await db.get('later', { transaction: txn })).toBeUndefined(); + expect(db.getKeys().asArray).toEqual(['committed', 'later']); + } finally { + txn.abort(); + } + })); + + it('should read the latest committed state through a disableSnapshot transaction', () => + dbRunner(async ({ db }) => { + await db.put('committed', 'before'); + const txn = new Transaction(db.store, { disableSnapshot: true }); + try { + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed']); + await db.transaction(async (writer) => { + await writer.put('later', 'after'); + }); + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed', 'later']); + expect(db.getKeysCount({ transaction: txn })).toBe(2); + expect(txn.getKeys().asArray).toEqual(['committed', 'later']); + } finally { + txn.abort(); + } + })); + + it('should read the latest committed state through a tailing transactional range', () => + dbRunner(async ({ db }) => { + await db.put('committed', 'before'); + const txn = new Transaction(db.store); + try { + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed']); + await db.transaction(async (writer) => { + await writer.put('later', 'after'); + }); + expect(db.getKeys({ transaction: txn, tailing: true }).asArray).toEqual([ + 'committed', + 'later', + ]); + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed']); + } finally { + txn.abort(); + } + })); + + it('should honor the transaction option on a pessimistic database', () => + dbRunner({ dbOptions: [{ pessimistic: true }] }, async ({ db }) => { + await db.put('committed', 'before'); + const txn = new Transaction(db.store); + try { + await txn.put('staged', 'in-batch'); + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed', 'staged']); + expect(db.getKeysCount({ transaction: txn })).toBe(2); + await db.transaction(async (writer) => { + await writer.put('later', 'after'); + }); + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['committed', 'staged']); + } finally { + txn.abort(); + } + })); + + it('should reject routed ranges and counts once the transaction starts committing', () => + dbRunner(async ({ db }) => { + const txn = new Transaction(db.store); + await txn.put('a', 'staged'); + const committing = txn.commit(); + expect(() => db.getRange({ transaction: txn })).toThrow('not in pending state'); + expect(() => db.getKeysCount({ transaction: txn })).toThrow('not in pending state'); + expect(() => txn.getRange()).toThrow('not in pending state'); + expect(() => txn.getKeysCount()).toThrow('not in pending state'); + await committing; + expect(() => db.getRange({ transaction: txn })).toThrow('Transaction not found'); + expect(db.getKeys().asArray).toEqual(['a']); + })); + + it('should reject direct ranges and counts on a finished transaction', () => + dbRunner(async ({ db }) => { + const aborted = new Transaction(db.store); + await aborted.put('a', 'staged'); + aborted.abort(); + expect(() => aborted.getRange({ start: 'a' })).toThrow('not in pending state'); + expect(() => aborted.getKeysCount()).toThrow('not in pending state'); + + const committed = new Transaction(db.store); + await committed.put('b', 'staged'); + await committed.commit(); + expect(() => committed.getRange({ start: 'a' })).toThrow('not in pending state'); + expect(() => committed.getKeysCount()).toThrow('not in pending state'); + })); + + it('should reject direct ranges and counts once the transaction database handle closes', () => + dbRunner(async ({ db }, { db: other }) => { + const txn = new Transaction(db.store); + await txn.put('a', 'staged'); + db.close(); + try { + expect(() => txn.getRange({ start: 'a' })).toThrow('Database not open'); + expect(() => txn.getKeysCount()).toThrow('Database not open'); + expect(other.getKeys({ transaction: txn }).asArray).toEqual(['a']); + } finally { + txn.abort(); + } + })); + + it('should reject a transaction that belongs to another database', () => + dbRunner({ dbOptions: [{}, { path: generateDBPath() }] }, async ({ db }, { db: other }) => { + const txn = new Transaction(db.store); + const otherTxn = new Transaction(other.store); + try { + // ids are per database: the first transaction of each shares one + expect(otherTxn.id).toBe(txn.id); + await txn.put('staged', 'in-batch'); + await otherTxn.put('other-staged', 'other-batch'); + + const rejected = 'Transaction belongs to a different database'; + expect(() => other.getRange({ transaction: txn })).toThrow(rejected); + expect(() => other.getKeysCount({ transaction: txn })).toThrow(rejected); + expect(() => other.getSync('other-staged', { transaction: txn })).toThrow(rejected); + expect(db.getKeys({ transaction: txn }).asArray).toEqual(['staged']); + expect(other.getKeys({ transaction: otherTxn }).asArray).toEqual(['other-staged']); + } finally { + otherTxn.abort(); + txn.abort(); + } + })); + + it('should accept a transaction from another spelling of the same database', () => { + const dbPath = generateDBPath(); + const alias = `${dbPath}${sep}..${sep}${basename(dbPath)}`; + return dbRunner( + { dbOptions: [{ path: dbPath }, { path: alias }] }, + async ({ db }, { db: aliased }) => { + const txn = new Transaction(db.store); + try { + await txn.put('staged', 'in-batch'); + expect(aliased.store.path).not.toBe(db.store.path); + expect(aliased.store.identityPath).toBe(db.store.identityPath); + expect(aliased.getKeys({ transaction: txn }).asArray).toEqual(['staged']); + expect(aliased.getKeysCount({ transaction: txn })).toBe(1); + expect(aliased.getSync('staged', { transaction: txn })).toBe('in-batch'); + } finally { + txn.abort(); + } + } + ); + }); + + for (const action of ['commit', 'commitSync', 'abort'] as const) { + it(`should close routed and direct iterators on transaction ${action}`, () => + dbRunner(async ({ db }) => { + await db.put('a', 'committed-a'); + await db.put('b', 'committed-b'); + const txn = new Transaction(db.store); + await txn.put('c', 'staged-c'); + + const routed = db.getRange({ transaction: txn })[Symbol.iterator](); + const direct = txn.getRange()[Symbol.iterator](); + const limited = db.getRange({ transaction: txn, limit: 1 })[Symbol.iterator](); + const thrown = txn.getRange()[Symbol.iterator](); + expect(routed.next().done).toBe(false); + expect(direct.next().done).toBe(false); + expect(limited.next().done).toBe(false); + expect(thrown.next().done).toBe(false); + + if (action === 'commit') { + await txn.commit(); + } else if (action === 'commitSync') { + txn.commitSync(); + } else { + txn.abort(); + } + + expect(() => routed.next()).toThrow('Iterator not initialized'); + expect(() => direct.next()).toThrow('Iterator not initialized'); + // loop cleanup (break, limit, error) must not throw after the transaction closed it + expect(routed.return!().done).toBe(true); + expect(direct.return!().done).toBe(true); + expect(limited.next().done).toBe(true); + expect(() => thrown.throw!(new Error('consumer error'))).toThrow('consumer error'); + })); + } + it('should iterate over a range asynchronously', () => dbRunner(async ({ db }) => { for (const key of ['a', 'b', 'c', 'd', 'e']) { diff --git a/test/transaction-cross-column-family.test.ts b/test/transaction-cross-column-family.test.ts index 7a747bb96..d5544955f 100644 --- a/test/transaction-cross-column-family.test.ts +++ b/test/transaction-cross-column-family.test.ts @@ -178,4 +178,23 @@ describe('transaction reads across column families', () => { db.close(); } }); + + it("getRange uses the caller's column family with another column family's transaction", async () => { + const { db, other, third } = await seedAndReopen(); + try { + await db.transaction(async (txn: Transaction) => { + await other.put('staged-other', 'staged-value', { transaction: txn }); + const entries = other.getRange({ transaction: txn }).asArray; + + expect(entries).toHaveLength(26); + expect(entries).toContainEqual({ key: 'key-0', value: 'value-0' }); + expect(entries).toContainEqual({ key: 'staged-other', value: 'staged-value' }); + expect(entries).not.toContainEqual({ key: 'anchor', value: 'anchor-value' }); + }); + } finally { + third.close(); + other.close(); + db.close(); + } + }); }); diff --git a/test/transaction-orphan-gc.test.ts b/test/transaction-orphan-gc.test.ts index adc895b13..350a04341 100644 --- a/test/transaction-orphan-gc.test.ts +++ b/test/transaction-orphan-gc.test.ts @@ -39,7 +39,7 @@ const itWithGC = it.skipIf(!forceGC); const itWithNodeGC = it.skipIf(Boolean(process.versions.bun || process.versions.deno)); function runDependentFixture( - mode: 'async-get' | 'iterator', + mode: 'async-get' | 'iterator' | 'routed-iterator', dbPath: string ): Promise<{ code: number | null; signal: NodeJS.Signals | null; stderr: string }> { return new Promise((resolve, reject) => { @@ -58,7 +58,9 @@ function runDependentFixture( }); } -async function expectDependentFixtureSurvives(mode: 'async-get' | 'iterator'): Promise { +async function expectDependentFixtureSurvives( + mode: 'async-get' | 'iterator' | 'routed-iterator' +): Promise { const dbPath = join(process.cwd(), `.transaction-orphan-${mode}-${process.pid}-${Date.now()}`); try { const { code, signal, stderr } = await runDependentFixture(mode, dbPath); @@ -97,6 +99,12 @@ describe('orphaned transactions', () => { 15_000 ); + itWithNodeGC( + 'should keep a transaction alive until a routed iterator closes', + () => expectDependentFixtureSurvives('routed-iterator'), + 15_000 + ); + itWithGC('should release a transaction dropped without commit or abort', () => dbRunner(async ({ db, dbPath }) => { await db.put('foo', 'bar'); diff --git a/test/transactions.test.ts b/test/transactions.test.ts index 83facf62a..a36ccb546 100644 --- a/test/transactions.test.ts +++ b/test/transactions.test.ts @@ -780,12 +780,19 @@ for (const { name, options, txnOptions } of testOptions) { await expect(db.get('foo', { transaction: 'bar' as any })).rejects.toThrow( 'Invalid transaction' ); + expect(() => db.getRange({ transaction: 'bar' as any })).toThrow('Invalid transaction'); })); it('should error if transaction is not found', () => dbRunner({ dbOptions: [options] }, async ({ db }) => { + const txn = new Transaction(db.store, txnOptions); + await txn.put('foo', 'bar'); + await txn.commit(); + await expect(db.get('foo', { transaction: txn })).rejects.toThrow('Transaction not found'); + expect(() => db.getRange({ transaction: txn })).toThrow('Transaction not found'); + // a bare id is not a transaction: only a Transaction carries the store that owns it await expect(db.get('foo', { transaction: { id: 9926 } as any })).rejects.toThrow( - 'Transaction not found' + 'Invalid transaction' ); })); });