diff --git a/AGENTS.md b/AGENTS.md index 7daf60f7f..9b7a1740d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -193,6 +193,17 @@ sufficient (env teardown does not honor tsfn acquire counts); see each completion callback (widens teardown race windows) - `ROCKSDB_JS_TXN_GET_DELAY_MS` - Test-only: delay a transaction's cold-cache async get before it reads (exercises orphan cleanup past the async-work wait timeout) +- `ROCKSDB_JS_PARK_TIMEOUT_MS` - Bounded wait (default `5000`) before a + coordinated-retry commit parked on a conflicting holder's VT lock resolves + RETRY_NOW unconditionally, in case the holder never releases (see + "Coordinated retry" note below). Read once per process (a function-local + `static`, like the other two above — `::getenv` is not safe against a + concurrent `::setenv` from a `process.env` write, and a park runs on whichever + env's JS thread owns the transaction), so it must be set in the environment a + process is started with. Values below `50` are clamped up to it, and `0` (an + ambiguous "disable the bound") falls back to the default like any malformed + value. There is no opt-out: a deployment that would rather wait than fail a + legitimately slow holder raises the value instead ## Test Structure @@ -412,7 +423,73 @@ sufficient (env teardown does not honor tsfn acquire counts); see Resolve it only on a break — `getLogFileSize` crosses into native and takes the store mutex, so a per-frame call would tax every healthy read. -12. **A dropped transaction must release itself**: `DBDescriptor::transactionAdd` holds a **strong** +12. **Coordinated retry parks on a lock, bounded by a descriptor-owned timeout**: a `coordinatedRetry` + commit that loses a conflict (`IsBusy`) parks instead of rejecting immediately — + `completeCommitWork` (`src/binding/transaction/transaction.cpp`) registers a wake callback on the + conflicting VT slot's `LockTracker` (via `addWakeCallback`) and resolves `RETRY_NOW` only when + that lock's last holder releases (`VerificationTable::releaseWriteIntent` → `LockTracker::wake`). + A holder that never releases — a leaked/abandoned transaction, or a wake lost to a bug elsewhere — + would otherwise park forever (harper#2001: a worker's write path disabled for 5+ hours until + restart). `ParkTimeoutRegistry` (`db_descriptor.{h,cpp}`) bounds this with + `ROCKSDB_JS_PARK_TIMEOUT_MS` (default `5000` — the top of this fix's requested 2-5s range, to + leave maximum headroom for a holder that is merely slow rather than abandoned, since a timeout + consumes a `coordinatedRetry` attempt exactly like a real wake does and `maxRetries` is finite): + one registry, and one lazily-started timeout thread, **per descriptor** — joined at + `finishClose()` (and again, idempotently, from the destructor as a safety net, matching + `commitWorker`) — tracks every outstanding deadline instead of spawning a thread per park (the + contention path is exactly where an abandoned holder makes parks dense, so per-park threads would + be a resource cliff, not a fix). Deliberately a plain `std::thread`, not a `uv_timer_t`: this addon + ships one prebuilt binary across Node ABI versions via N-API, and libuv's struct layout is not part + of that stable surface. + + **A `LockTracker` wake callback runs under the process-global VT `writerMutex_`, so it must not + block and must not re-enter the VT or `DBRegistry`.** `LockTracker::wake()` invokes its callbacks + inline and both callers (`releaseWriteIntent`, `cancelForDB`) hold that mutex across the whole + function. Re-entering the registry from there self-deadlocks: `DBRegistry::PurgeIfUnreferenced` + can claim the purge and call `finishClose()` → `cancelForDB()` → a second lock of the same + non-recursive `writerMutex_`, wedging every database's write-intent path process-wide — the exact + symptom this note exists to fix. It is also an AB-BA against `finishClose`'s `txnsMutex` → + `writerMutex_` order, and it would run a flush, a manual compaction, `WaitForCompact` and thread + joins under the global VT lock. That is why `ParkTimeoutRegistry` is a standalone object owned by + the descriptor through a `shared_ptr` rather than state on the descriptor itself: the wake closure + captures a **`std::weak_ptr`** and calls only `fire(id)`, which touches one + mutex and one map. Weak, not raw, because a park can end up registered on a tracker installed by a + _different_ database on a colliding VT slot (`VerificationTable::lockSlotForWrite` joins an existing + tracker without retagging its `dbId`), so that lock's eventual release wakes a park whose own + database may already have closed — `cancelForDB()` only wakes trackers tagged with _its own_ + `vtEpoch`, so it cannot be relied on to have resolved a foreign-`dbId` park first. Weak **to the + registry and not to the descriptor** because a `weak_ptr::lock()` is a transient extra + reference, and `PurgeIfUnreferenced` decides on `use_count() <= 1`: a racing close would see the + inflated count, skip the purge, and leak the registry entry plus the open RocksDB — the + HarperFast/rocksdb-js#672 hazard, which the wake path cannot repair by retrying the purge (that is + the re-entrancy above). `.lock()` failing is the expected outcome once the owning database closes: + `ParkTimeoutRegistry::shutdown()` (called from `finishClose()` right after `cancelForDB`, before + the descriptor can be destroyed) unconditionally resolves every park it still holds regardless of + whether the real holder ever wakes it, so by the time the weak reference can fail, the park has + already settled. + + Each park is identified by a monotonic `uint64_t id`, not its entry's address: `LockTracker::wakeCallbacks` + has no removal API (see the gap noted below), so a stale closure can outlive its entry, and an + address-keyed lookup risks resolving a _different_, later park that reused the same freed heap + address. The timeout thread and the LockTracker wake callback race through one heap-allocated + `std::atomic` per park (independent of the per-park `RetryNowContext`, whose refs/TSFN the + winning side's release eventually frees) — whichever fires first calls+releases the TSFN under the + registry's `mutex` and erases the entry; the loser finds it already gone and touches nothing. That + same mutex is what a dying env's `releaseByEnv` (wired into the module env-cleanup + hook next to `ReleaseCommitCompletionsByEnv`) takes to cancel — release without calling — that + env's pending parks before Node frees their tsfns; `retryNowCallJs` also guards `env == nullptr` + like `commitCompletionCallJs` does, for the same tsfn-queue-drained-during-teardown reason. Parks + are indexed twice, by id and by deadline (`std::multimap`): `fire()` needs an O(1) lookup because it + runs under the global VT mutex, and the timeout thread needs the earliest deadline on every wakeup + without an O(N) scan on that same lock. Known gap: `LockTracker::wakeCallbacks` + itself has no removal API. Before this change an abandoned holder accrued one inert callback per + waiter and then everything hung; now each waiter re-parks (and re-registers) every + `ROCKSDB_JS_PARK_TIMEOUT_MS` up to `maxRetries`, so registrations accumulate per _retry_ rather than + per incident for as long as it lasts (each is inert once its own park resolves, so this is a + memory-growth concern, not a correctness one) — deferred rather than risking an unreviewed change to + `verification_table.cpp`'s concurrency invariants under this fix's scope. + +13. **A dropped transaction must release itself**: `DBDescriptor::transactionAdd` holds a **strong** `shared_ptr` (the parallel `closables` entry is weak), so the registry alone keeps a `TransactionHandle` alive and `~TransactionHandle` — hence `close()`, the only `ClearSnapshot()` path — is unreachable while it is registered. The `NativeTransaction` finalizer therefore calls @@ -434,7 +511,7 @@ sufficient (env teardown does not honor tsfn acquire counts); see handle fields that are fixed before publication (`id`, `createdAt`), because `txnsMutex` covers map membership while mutable-field writers hold no lock. -13. **A recovered active transaction-log file ends on a transaction boundary when recovery can +14. **A recovered active transaction-log file ends on a transaction boundary when recovery can prove one**: only a batch's final entry carries `TRANSACTION_LOG_ENTRY_LAST_FLAG`, so a crash mid-batch leaves whole, well-framed entries that are a _prefix_ of a transaction. `recoverTail()` discards them diff --git a/README.md b/README.md index c3a53ce6c..d82b5cd58 100644 --- a/README.md +++ b/README.md @@ -779,6 +779,15 @@ not committed after the configured number of coordinated retries, the transactio an `ERR_TRANSACTION_ABANDONED` error. Coordinated retry requires the column family to be opened with `verificationTable: true`. +That wait is bounded so a conflicting transaction that is never committed or aborted cannot block +the commit forever: if the write intent has not been released after `ROCKSDB_JS_PARK_TIMEOUT_MS` +(default `5000`), the commit resolves anyway and consumes a retry attempt exactly as a real release +would. A conflicting transaction held for longer than roughly `maxRetries` times that timeout +therefore ends in `ERR_TRANSACTION_ABANDONED` rather than waiting indefinitely. Deployments where +waiting is preferable to failing should raise the timeout; there is no way to disable the bound. +In particular `0` does not disable it: `0`, negative, and unparseable values all fall back to the +`5000` default, and a value between `1` and `49` is clamped up to `50`. + ### Class: `Transaction` The transaction callback is passed in a `Transaction` instance which contains all of the same data diff --git a/src/binding/binding.cpp b/src/binding/binding.cpp index b799a05f3..0b88ba46d 100644 --- a/src/binding/binding.cpp +++ b/src/binding/binding.cpp @@ -201,6 +201,11 @@ NAPI_MODULE_INIT() { // tsfns, so the shared commit thread stops marshalling into a torn-down // env (mirrors the listener cleanup above). rocksdb_js::DBRegistry::ReleaseCommitCompletionsByEnv(dyingEnv); + // Same reasoning for a coordinated-retry commit parked on a VT lock: + // cancel this env's pending park timeouts before Node frees their + // tsfns, so the descriptor's park-timeout thread never fires into a + // torn-down env. + rocksdb_js::DBRegistry::ReleaseParkTimeoutsByEnv(dyingEnv); int32_t newRefCount = --moduleRefCount; if (newRefCount == 0) { diff --git a/src/binding/database/db_descriptor.cpp b/src/binding/database/db_descriptor.cpp index 4fbcffa35..b5ed6920e 100644 --- a/src/binding/database/db_descriptor.cpp +++ b/src/binding/database/db_descriptor.cpp @@ -10,6 +10,7 @@ #include "rocksdb/utilities/options_util.h" #include #include +#include #include namespace rocksdb_js { @@ -358,6 +359,9 @@ DBDescriptor::DBDescriptor( DBDescriptor::~DBDescriptor() { DEBUG_LOG("%p DBDescriptor::~DBDescriptor Closing \"%s\"\n", this, this->path.c_str()); this->close(); + // Idempotent safety net, matching commitWorker/logWorker's own + // destructor shutdown. + this->parkTimeouts->shutdown(); } /** @@ -475,6 +479,12 @@ void DBDescriptor::finishClose() { } } + // A park can be registered on a foreign-dbId tracker (colliding VT slot; + // see the ParkTimeout header comment), so cancelForDB() above cannot be + // relied on to have woken everything this descriptor is waiting on. + // ParkTimeoutRegistry::shutdown() resolves whatever is left regardless. + this->parkTimeouts->shutdown(); + // Unregister from transaction log store registry - this will clean up stores // when the last descriptor for this path is closed TransactionLogStoreRegistry::Unregister(this->path); @@ -565,6 +575,178 @@ void DBDescriptor::releaseCommitCompletionsByEnv(napi_env env) { } } +uint64_t ParkTimeoutRegistry::schedule( + napi_env env, + unsigned timeoutMs, + napi_threadsafe_function tsfn, + std::shared_ptr> fired +) { + std::lock_guard lock(this->mutex); + if (this->stopped) { + // Descriptor already closing: the caller must resolve inline without + // registering with the LockTracker at all (see the header comment). + return 0; + } + if (!this->threadStarted) { + try { + this->thread = std::thread([this]() { this->runLoop(); }); + } catch (...) { + // Thread creation failed (e.g. thread/resource exhaustion): leave + // the flag false so the next park retries, and tell the caller to + // resolve inline now rather than register a park nothing will + // ever time out. + return 0; + } + this->threadStarted = true; + } + auto entry = std::make_unique(); + entry->id = this->nextId++; + entry->env = env; + entry->tsfn = tsfn; + entry->fired = std::move(fired); + uint64_t id = entry->id; + auto deadlineIt = this->deadlines.emplace( + std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutMs), + id + ); + entry->deadlineIt = deadlineIt; + this->parks.emplace(id, std::move(entry)); + if (deadlineIt == this->deadlines.begin()) { + // Only the new earliest deadline needs the loop re-armed (this also + // covers waking it out of the indefinite wait when `deadlines` was + // empty); any later one already fires within a wait it will take. + this->cv.notify_all(); + } + return id; +} + +std::unique_ptr ParkTimeoutRegistry::take(uint64_t id) { + auto it = this->parks.find(id); + if (it == this->parks.end()) { + return nullptr; + } + std::unique_ptr owned = std::move(it->second); + this->deadlines.erase(owned->deadlineIt); + this->parks.erase(it); + return owned; +} + +void ParkTimeoutRegistry::resolve(ParkTimeout& park) { + bool expected = false; + if (park.fired->compare_exchange_strong(expected, true)) { + // A closing tsfn (env teardown racing this resolve) must not be + // touched again -- napi_closing means Node may already be freeing it. + napi_status status = ::napi_call_threadsafe_function(park.tsfn, nullptr, napi_tsfn_nonblocking); + if (status == napi_ok) { + ::napi_release_threadsafe_function(park.tsfn, napi_tsfn_release); + } + } +} + +void ParkTimeoutRegistry::runLoop() { + setThreadName("rocksdb-park-timeout"); + std::unique_lock lock(this->mutex); + for (;;) { + if (this->stopped) { + return; + } + if (this->deadlines.empty()) { + this->cv.wait(lock); + continue; + } + auto now = std::chrono::steady_clock::now(); + // Copy the deadline: wait_until releases the lock while parked, during + // which this entry can be erased (a real wake racing the timeout) and + // the map node freed -- a bound reference into it would be a read of + // freed memory once the wait re-checks time. + std::chrono::steady_clock::time_point earliest = this->deadlines.begin()->first; + if (earliest > now) { + this->cv.wait_until(lock, earliest); + continue; + } + // Fire while still holding the mutex, like dispatchCommitCompletion. + while (!this->deadlines.empty() && this->deadlines.begin()->first <= now) { + auto deadlineIt = this->deadlines.begin(); + auto parkIt = this->parks.find(deadlineIt->second); + this->deadlines.erase(deadlineIt); + if (parkIt == this->parks.end()) { + continue; + } + std::unique_ptr due = std::move(parkIt->second); + this->parks.erase(parkIt); + ParkTimeoutRegistry::resolve(*due); + } + } +} + +void ParkTimeoutRegistry::fire(uint64_t id) { + std::lock_guard lock(this->mutex); + std::unique_ptr owned = this->take(id); + if (!owned) { + // Already claimed by the timeout thread, releaseByEnv, or shutdown. + return; + } + ParkTimeoutRegistry::resolve(*owned); +} + +void ParkTimeoutRegistry::releaseByEnv(napi_env env) { + std::lock_guard lock(this->mutex); + for (auto it = this->parks.begin(); it != this->parks.end();) { + if (it->second->env != env) { + ++it; + continue; + } + // Mark fired first so neither the timeout thread nor a later real + // wake ever calls into the tsfn we're about to release -- the + // promise's env is gone, nothing is listening for the resolve. + bool expected = false; + it->second->fired->compare_exchange_strong(expected, true); + if (!expected) { + ::napi_release_threadsafe_function(it->second->tsfn, napi_tsfn_release); + } + this->deadlines.erase(it->second->deadlineIt); + it = this->parks.erase(it); + } +} + +void ParkTimeoutRegistry::shutdown() { + std::thread toJoin; + { + std::lock_guard lock(this->mutex); + if (this->stopped && !this->threadStarted) { + // Already fully shut down (e.g. finishClose() already ran; this is + // the destructor's belt-and-suspenders call) -- nothing left to do. + return; + } + this->stopped = true; + if (this->threadStarted) { + toJoin = std::move(this->thread); + this->threadStarted = false; + } + // Resolve every park still pending, under the same mutex the other + // three methods serialize their tsfn calls on -- draining outside the + // lock would let a concurrent releaseByEnv for a dying env observe + // "nothing to cancel" while this is mid-call on that same env's tsfn, + // racing Node freeing it. + for (auto& entry : this->parks) { + ParkTimeoutRegistry::resolve(*entry.second); + } + this->parks.clear(); + this->deadlines.clear(); + } + // Notify + join outside the lock: the loop's cv.wait_until needs to + // re-acquire the mutex to observe `stopped` and return, so joining while + // still holding it would deadlock. + this->cv.notify_all(); + if (toJoin.joinable()) { + toJoin.join(); + } +} + +ParkTimeoutRegistry::~ParkTimeoutRegistry() { + this->shutdown(); +} + /** * Registers a database resource to be closed when the descriptor is closed. * diff --git a/src/binding/database/db_descriptor.h b/src/binding/database/db_descriptor.h index ff0b27247..5e8ad919e 100644 --- a/src/binding/database/db_descriptor.h +++ b/src/binding/database/db_descriptor.h @@ -4,11 +4,16 @@ #include #include #include +#include +#include +#include #include #include #include +#include #include #include +#include #include "rocksdb/db.h" #include "rocksdb/statistics.h" #include "rocksdb/utilities/transaction_db.h" @@ -59,6 +64,117 @@ struct DBDeleter { } }; +/** + * Bounded waits for coordinated-retry commits parked on a conflicting holder's + * VT lock, so a holder that never releases resolves RETRY_NOW instead of + * parking forever (harper#2001, see AGENTS.md note 12). One instance per + * `DBDescriptor`, with one lazily-started thread tracking every outstanding + * deadline. + * + * Deliberately standalone — it holds no reference back to its descriptor, and + * nothing reachable from `fire()` touches the descriptor or `DBRegistry`. + * `fire()` is called from a `LockTracker` wake callback, which + * `LockTracker::wake()` invokes inline while the *process-global* VT + * `writerMutex_` is held (`VerificationTable::releaseWriteIntent` and + * `cancelForDB` both wake under it), so the standing invariant on that path is: + * do not block, and do not re-enter the VT or the registry. A + * `DBRegistry::PurgeIfUnreferenced` from there can claim the purge and run + * `finishClose()` -> `cancelForDB()` -> a second lock of that same + * non-recursive `writerMutex_`, wedging every database's write-intent path. + * Holding the wake closure's `weak_ptr` on this object rather than on the + * descriptor is also what keeps its transient `.lock()` out of the + * descriptor's `use_count`, which `PurgeIfUnreferenced` keys its "no handles + * left" decision on — so there is no purge-skip window to retry in the first + * place (the HarperFast/rocksdb-js#672 hazard). + * + * Parks are keyed by a monotonic `id`, not the entry's address: + * `LockTracker::wakeCallbacks` has no removal API, so a stale closure can + * outlive its entry and an address-keyed lookup could resolve a later, + * unrelated park that reused the freed address. `fired` is the exactly-once + * gate shared with that park's wake callback — whichever side wins the CAS + * calls+releases `tsfn`, always under `mutex` so a concurrent `releaseByEnv()` + * for a dying env cannot observe "nothing to cancel" while the other side is + * mid-call on that env's tsfn. + */ +class ParkTimeoutRegistry final { +public: + ~ParkTimeoutRegistry(); + + /** + * JS thread (`completeCommitWork`). Registers a bounded wait, lazily + * starting the single timeout thread. Returns the new park's id, or 0 if + * the registry is shut down or thread creation failed -- either way the + * caller resolves inline instead of parking with no timeout behind it. + */ + uint64_t schedule( + napi_env env, + unsigned timeoutMs, + napi_threadsafe_function tsfn, + std::shared_ptr> fired + ); + + /** + * Resolves a specific park early because its VT lock's holder released + * (LockTracker wake callback, any thread, VT `writerMutex_` held). A no-op + * if the id is already gone -- claimed by the timeout thread, by + * `releaseByEnv`, or drained by `shutdown`. + */ + void fire(uint64_t id); + + /** + * Module env-cleanup hook. Cancels every pending park registered for a + * dying env -- released, never called, so neither the timeout thread nor a + * later real wake can fire into a tsfn Node is about to free. + */ + void releaseByEnv(napi_env env); + + /** + * Descriptor close: stop and join the timeout thread, then resolve + * (call+release) every park still pending. `cancelForDB`, called just + * before this, cannot be relied on to have woken everything -- a park can + * be registered on a foreign-`dbId` tracker (colliding VT slot) that only + * wakes a different database. Idempotent (called from both `finishClose()` + * and the descriptor's destructor, matching `commitWorker`). + */ + void shutdown(); + +private: + using DeadlineIndex = std::multimap; + + struct ParkTimeout { + uint64_t id; + napi_env env; + napi_threadsafe_function tsfn; + std::shared_ptr> fired; + DeadlineIndex::iterator deadlineIt; + }; + + /** Detaches `id` from both indexes; null if already claimed. Holds `mutex`. */ + std::unique_ptr take(uint64_t id); + + /** + * Calls+releases a claimed park's tsfn, unless another side already won the + * exactly-once gate. Every caller holds `mutex`. + */ + static void resolve(ParkTimeout& park); + + /** Runs on `thread` until `shutdown()` stops it. */ + void runLoop(); + + std::mutex mutex; + std::condition_variable cv; + std::unordered_map> parks; + // Deadline-ordered view of `parks`: the timeout thread needs the earliest + // deadline on every wakeup, and scanning for it under `mutex` would put an + // O(N) loop on the same lock `fire()` must take while holding the global VT + // `writerMutex_`. + DeadlineIndex deadlines; + uint64_t nextId = 1; + std::thread thread; + bool threadStarted = false; + bool stopped = false; +}; + /** * Descriptor for a RocksDB database, its column families, and any in-flight * transactions. The DBRegistry uses this to track active databases and reuse @@ -293,6 +409,15 @@ struct DBDescriptor final : public std::enable_shared_from_this { */ void releaseCommitCompletionsByEnv(napi_env env); + /** + * Bounded waits for this database's parked coordinated-retry commits. + * Never null; owned by shared_ptr so a LockTracker wake callback can hold + * a weak reference to it without referencing the descriptor (see + * `ParkTimeoutRegistry`). Drained and joined by `finishClose()`. + */ + const std::shared_ptr parkTimeouts = + std::make_shared(); + private: DBDescriptor( const std::string& path, diff --git a/src/binding/database/db_registry.cpp b/src/binding/database/db_registry.cpp index e79a591b6..bafed0a8f 100644 --- a/src/binding/database/db_registry.cpp +++ b/src/binding/database/db_registry.cpp @@ -657,6 +657,33 @@ void DBRegistry::ReleaseCommitCompletionsByEnv(napi_env env) { } } +/** + * Cancels each descriptor's pending park timeouts owned by the given env. + * Called from the module env-cleanup hook so a worker thread exiting does not + * leave a coordinated-retry park's timeout thread calling into a torn-down + * env ~ROCKSDB_JS_PARK_TIMEOUT_MS later. Mirrors ReleaseCommitCompletionsByEnv. + */ +void DBRegistry::ReleaseParkTimeoutsByEnv(napi_env env) { + if (!instance) { + return; + } + + std::vector> registries; + { + std::lock_guard lock(instance->databasesMutex); + registries.reserve(instance->databases.size()); + for (auto& [_key, entry] : instance->databases) { + if (entry.descriptor) { + registries.push_back(entry.descriptor->parkTimeouts); + } + } + } + + for (auto& registry : registries) { + registry->releaseByEnv(env); + } +} + /** * Shutdown will force all databases to flush in-memory data to disk and purge the registry. */ diff --git a/src/binding/database/db_registry.h b/src/binding/database/db_registry.h index afec14325..f99729d61 100644 --- a/src/binding/database/db_registry.h +++ b/src/binding/database/db_registry.h @@ -97,6 +97,7 @@ class DBRegistry final { static napi_value RegistryStatus(napi_env env, napi_callback_info info); static void RemoveListenersByEnv(napi_env env); static void ReleaseCommitCompletionsByEnv(napi_env env); + static void ReleaseParkTimeoutsByEnv(napi_env env); static void Shutdown(); static size_t Size(); }; diff --git a/src/binding/transaction/transaction.cpp b/src/binding/transaction/transaction.cpp index 655d97a9f..31eba32df 100644 --- a/src/binding/transaction/transaction.cpp +++ b/src/binding/transaction/transaction.cpp @@ -1,10 +1,17 @@ +#include +#include +#include +#include #include +#include #include #include +#include #include #include "database/database.h" #include "database/db_descriptor.h" #include "database/db_handle.h" +#include "database/db_registry.h" #include "iterator/db_iterator.h" #include "database/db_settings.h" #include "napi/macros.h" @@ -199,6 +206,13 @@ struct RetryNowContext { }; static void retryNowCallJs(napi_env env, napi_value /*func*/, void* context, void* /*data*/) { + // env is nullptr when the env is tearing down and this tsfn's queue is + // being drained as part of that (same as commitCompletionCallJs above) — + // nothing left to resolve into; retryNowFinalize still runs afterward to + // free ctx. + if (env == nullptr) { + return; + } auto* ctx = reinterpret_cast(context); napi_value global, resolveFn, retryVal; ::napi_get_global(env, &global); @@ -214,6 +228,56 @@ static void retryNowFinalize(napi_env env, void* finalizeData, void* /*hint*/) { delete ctx; } +/** + * Bounded wait (ms) for a coordinated-retry commit parked on a conflicting + * holder's VT lock, selected by ROCKSDB_JS_PARK_TIMEOUT_MS (harper#2001). + * Default is the top of the 2-5s range this was scoped to: a timeout + * consumes a coordinatedRetry attempt like a genuine wake would, so a + * higher default leaves more headroom for a merely-slow (not abandoned) + * holder before maxRetries exhausts -- a deployment that would rather wait + * than fail raises this, it cannot disable the bound. + * + * Read once per process, like commitThreadMode()/commitDelayMs() below: a park + * runs on whichever env's JS thread owns the transaction, and ::getenv is not + * safe against a `process.env` write on another thread (uv_os_setenv -> + * setenv(3) may reallocate `environ`). Varying it therefore means starting a + * process with the value already set -- worker-thread `process.env` writes + * never reach ::getenv at all (core/test_seam.h). + * + * Malformed input falls back to the default rather than producing a + * degenerate effective timeout; a positive value below kMinimum is clamped up + * to it instead, since the intent there is unambiguous. + */ +static unsigned parkTimeoutMs() { + static const unsigned ms = []() -> unsigned { + constexpr unsigned kDefault = 5000; + constexpr unsigned kMinimum = 50; + const char* v = ::getenv("ROCKSDB_JS_PARK_TIMEOUT_MS"); + if (v == nullptr) { + return kDefault; + } + const char* firstNonSpace = v; + while (*firstNonSpace != '\0' && ::isspace(static_cast(*firstNonSpace))) { + ++firstNonSpace; + } + if (*firstNonSpace == '\0' || *firstNonSpace == '-') { + return kDefault; + } + char* end = nullptr; + errno = 0; + unsigned long parsed = ::strtoul(v, &end, 10); + // parsed == 0 is the one ambiguous input: a literal "0" reads as "disable + // the bound", which is exactly the unbounded park this exists to prevent, + // so it falls back to the default like malformed input rather than being + // honored as an opt-out or clamped up to kMinimum. + if (end == v || *end != '\0' || errno == ERANGE || parsed > UINT32_MAX || parsed == 0) { + return kDefault; + } + return std::max(kMinimum, static_cast(parsed)); + }(); + return ms; +} + /** * State for the `Commit` async work. */ @@ -221,15 +285,48 @@ struct TransactionCommitState final : BaseAsyncState*> savedSlots; + std::weak_ptr parkTimeouts; TransactionCommitState( napi_env env, std::shared_ptr handle ) : BaseAsyncState>(env, handle), - hasLog(false) {} + hasLog(false), + parkTimeouts( + handle && handle->coordinatedRetry && handle->dbHandle && handle->dbHandle->descriptor + ? handle->dbHandle->descriptor->parkTimeouts + : std::shared_ptr() + ) {} }; +static void rejectRetryNowSetupFailure( + napi_env env, + TransactionCommitState* state, + napi_status status +) { + napi_value error = nullptr; + bool exceptionPending = false; + if (::napi_is_exception_pending(env, &exceptionPending) == napi_ok && exceptionPending) { + ::napi_get_and_clear_last_exception(env, &error); + } + if (error == nullptr) { + std::string detail = "Failed to initialize coordinated retry: " + + getNapiExtendedError(env, status); + napi_value message; + if (::napi_create_string_utf8(env, detail.c_str(), detail.size(), &message) == napi_ok) { + ::napi_create_error(env, nullptr, message, &error); + } + } + if (error == nullptr) { + ::napi_get_and_clear_last_exception(env, &error); + } + if (error == nullptr) { + ::napi_get_undefined(env, &error); + } + state->callReject(error); +} + /** * Log-lane stage of the commit: validates the handle and writes the * transaction-log batch (recording the committed position). Runs off the JS @@ -410,14 +507,9 @@ static void completeCommitWork(napi_env env, TransactionCommitState* state) { state->handle->close(); } - // Transfer resolve/reject refs from state to a RetryNowContext - // so the TSFN finalize can clean them up. - auto* ctx = new RetryNowContext{state->resolveRef, state->rejectRef}; - state->resolveRef = nullptr; - state->rejectRef = nullptr; - bool parked = false; VerificationTable* vt = DBSettings::getInstance().getVerificationTableRaw(); + std::shared_ptr parkTimeouts = state->parkTimeouts.lock(); for (auto* slot : state->savedSlots) { // refTrackerIfLocked takes a temporary reference under the VT // writer mutex, so the tracker cannot be freed by a concurrent @@ -428,27 +520,95 @@ static void completeCommitWork(napi_env env, TransactionCommitState* state) { // Create a TSFN that calls resolve(RETRY_NOW) when fired. napi_value resource_name; - ::napi_create_string_latin1(env, "transaction.retry", NAPI_AUTO_LENGTH, &resource_name); + napi_status resourceStatus = ::napi_create_string_latin1( + env, "transaction.retry", NAPI_AUTO_LENGTH, &resource_name + ); + if (resourceStatus != napi_ok) { + vt->unrefTracker(t); + rejectRetryNowSetupFailure(env, state, resourceStatus); + return; + } + + auto* ctx = new RetryNowContext{state->resolveRef, state->rejectRef}; + state->resolveRef = nullptr; + state->rejectRef = nullptr; napi_threadsafe_function tsfn; - ::napi_create_threadsafe_function( + napi_status tsfnStatus = ::napi_create_threadsafe_function( env, nullptr, nullptr, resource_name, 0, 1, ctx, retryNowFinalize, ctx, retryNowCallJs, &tsfn ); + if (tsfnStatus != napi_ok) { + state->resolveRef = ctx->resolveRef; + state->rejectRef = ctx->rejectRef; + ctx->resolveRef = nullptr; + ctx->rejectRef = nullptr; + delete ctx; + vt->unrefTracker(t); + rejectRetryNowSetupFailure(env, state, tsfnStatus); + return; + } ::napi_unref_threadsafe_function(env, tsfn); + // Exactly-once gate: independent of `ctx` so a loser arriving after + // the winner's ctx/tsfn are already gone just loses the CAS. + auto fired = std::make_shared>(false); + + // #741 bound (harper#2001): resolves RETRY_NOW after + // ROCKSDB_JS_PARK_TIMEOUT_MS if the holder never releases. 0 both + // when the descriptor is closing and when there is no descriptor + // at all (DBHandle::close() can reset it concurrently) -- either + // way there is no timeout thread behind this park. + uint64_t parkId = parkTimeouts + ? parkTimeouts->schedule(env, parkTimeoutMs(), tsfn, fired) + : 0; + + if (parkId == 0) { + // No timeout thread behind this park -- resolve now rather + // than register with the LockTracker unbounded. No exactly-once + // gate: `schedule` returns 0 only before it publishes `fired`, + // and the wake callback below is not built yet, so this side + // owns the park outright. + vt->unrefTracker(t); + // A closing tsfn (env teardown racing this resolve) must not be + // touched again -- see the matching guard in + // ParkTimeoutRegistry::resolve. + if (::napi_call_threadsafe_function(tsfn, nullptr, napi_tsfn_nonblocking) == napi_ok) { + ::napi_release_threadsafe_function(tsfn, napi_tsfn_release); + } + parked = true; + break; + } + + // weak_ptr, not raw: LockTracker::wakeCallbacks has no removal API, + // so this closure can outlive the park registry (e.g. a foreign-dbId + // tracker from a colliding VT slot). `.lock()` failing means + // ParkTimeoutRegistry::shutdown already resolved this park at close. + // The weak reference is deliberately to the registry and not to the + // descriptor: this runs inline under the process-global VT + // `writerMutex_`, where re-entering DBRegistry can self-deadlock and + // where a transient descriptor ref would perturb the use_count that + // PurgeIfUnreferenced decides on (see the ParkTimeoutRegistry docs). + std::weak_ptr weakParks = parkTimeouts; + std::weak_ptr> weakFired = fired; + auto fireOnce = [weakParks, weakFired, parkId]() { + auto fired = weakFired.lock(); + if (!fired || fired->load(std::memory_order_acquire)) { + return; + } + if (auto parks = weakParks.lock()) { + parks->fire(parkId); + } + }; + // Register wake callback; if the tracker already fired wake() // before we got here, addWakeCallback returns false and we // call+release the TSFN immediately (async on the JS thread). - bool registered = t->addWakeCallback([tsfn]() { - ::napi_call_threadsafe_function(tsfn, nullptr, napi_tsfn_nonblocking); - ::napi_release_threadsafe_function(tsfn, napi_tsfn_release); - }); + bool registered = t->addWakeCallback(fireOnce); if (!registered) { - ::napi_call_threadsafe_function(tsfn, nullptr, napi_tsfn_nonblocking); - ::napi_release_threadsafe_function(tsfn, napi_tsfn_release); + fireOnce(); } vt->unrefTracker(t); @@ -457,18 +617,14 @@ static void completeCommitWork(napi_env env, TransactionCommitState* state) { } if (!parked) { - // No active lock found; resolve RETRY_NOW directly (we are on - // the JS thread in this complete callback). - napi_value global, resolveFn, retryVal; - ::napi_get_global(env, &global); - ::napi_get_reference_value(env, ctx->resolveRef, &resolveFn); - ::napi_create_int32(env, RETRY_NOW_VALUE, &retryVal); - ::napi_call_function(env, global, resolveFn, 1, &retryVal, nullptr); - ::napi_delete_reference(env, ctx->resolveRef); - ::napi_delete_reference(env, ctx->rejectRef); - delete ctx; + napi_value retryVal; + napi_status retryStatus = ::napi_create_int32(env, RETRY_NOW_VALUE, &retryVal); + if (retryStatus == napi_ok) { + state->callResolve(retryVal); + } else { + rejectRetryNowSetupFailure(env, state, retryStatus); + } } - // If parked, ctx is owned by the TSFN finalize; do not free here. } else { // Normal error path: reset to Pending so JS can retry. // Guard: keep Aborted if close() already set it (DB closing diff --git a/test/commit-teardown.test.ts b/test/commit-teardown.test.ts index eda9c2200..a1903fbd0 100644 --- a/test/commit-teardown.test.ts +++ b/test/commit-teardown.test.ts @@ -13,6 +13,8 @@ const COMMIT_THREAD_MODES: Array<{ label: string; mode: string | undefined }> = { label: '2', mode: '2' }, ]; +const retry = process.versions.deno && process.platform === 'darwin' ? 1 : 0; + /** * Runs the repro fixture in a child process so a native abort (SIGABRT/SIGSEGV) * from an async-commit completion racing worker-env teardown surfaces as a @@ -70,8 +72,12 @@ function spawnRepro( describe('Async commit completion vs. worker env teardown', () => { it.each(COMMIT_THREAD_MODES)( 'should survive worker env teardown with commits in flight on the shared commit thread (ROCKSDB_JS_COMMIT_THREAD=$label)', - ({ mode }) => expectSurvives(mode), - // Worker spawn/teardown dominates wall time and is slow on macOS/Windows. - 120_000 + { + // The single retry is limited to the pre-existing Deno/macOS crash in #746. + retry, + // Worker spawn/teardown dominates wall time and is slow on macOS/Windows. + timeout: 120_000, + }, + ({ mode }) => expectSurvives(mode) ); }); diff --git a/test/fixtures/fork-park-timeout.mts b/test/fixtures/fork-park-timeout.mts new file mode 100644 index 000000000..8545f03f0 --- /dev/null +++ b/test/fixtures/fork-park-timeout.mts @@ -0,0 +1,48 @@ +import { RocksDatabase, Transaction } from '../../src/index.ts'; +import { RETRY_NOW } from '../../src/transaction.ts'; + +const dbPath = process.argv[2]; + +if (!dbPath) { + console.error('Usage: fork-park-timeout.mts '); + process.exit(1); +} + +function valueWithVersion(version: number): Buffer { + const value = Buffer.alloc(16); + value.writeDoubleBE(version, 0); + return value; +} + +const db = new RocksDatabase(dbPath, { encoding: false, verificationTable: true }); +const keepAlive = setInterval(() => {}, 1000); + +try { + db.open(); + const key = Buffer.from('park-timeout-abandoned-holder'); + const initialVersion = 1.6e12; + await db.put(key, valueWithVersion(initialVersion)); + db.populateVersion(key, initialVersion); + if (!db.verifyVersion(key, initialVersion)) { + throw new Error('failed to populate the verification-table version'); + } + + const holder = new Transaction(db.store, { coordinatedRetry: true }); + holder.putSync(key, valueWithVersion(2.1e12)); + + const transaction = new Transaction(db.store, { coordinatedRetry: true }); + await transaction.get(key); + await db.put(key, valueWithVersion(2.2e12)); + transaction.putSync(key, valueWithVersion(2.3e12)); + + const start = performance.now(); + const result = await transaction.commit(); + const elapsed = performance.now() - start; + + if (result !== RETRY_NOW || elapsed < 40 || elapsed >= 4000) { + throw new Error(`unexpected park result=${String(result)} elapsed=${elapsed}`); + } +} finally { + clearInterval(keepAlive); + db.close(); +} diff --git a/test/lock-tracker.test.ts b/test/lock-tracker.test.ts index f616ab7a0..df3ee77fb 100644 --- a/test/lock-tracker.test.ts +++ b/test/lock-tracker.test.ts @@ -1,9 +1,13 @@ import { constants } from '../src/load-binding.ts'; import { RETRY_NOW, Transaction } from '../src/transaction.ts'; import { dbRunner } from './lib/util.ts'; +import { spawn } from 'node:child_process'; +import { join } from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; import { describe, expect, it } from 'vitest'; const FRESH_VERSION_FLAG = constants.FRESH_VERSION_FLAG; +const parkTimeoutFixturePath = join(__dirname, 'fixtures', 'fork-park-timeout.mts'); // Builds a value buffer whose first 8 bytes are the big-endian float64 version, // matching VerificationTable::extractVersionFromValue (Harper's record format). @@ -119,6 +123,7 @@ describe('Coordinated retry (Phase 3)', () => { // Force IsBusy by running concurrent transactions writing the same key // under coordinatedRetry: true. database.ts handles RETRY_NOW internally // via immediate retry; callers never see it as a return value. + const start = performance.now(); const results = await Promise.allSettled( Array.from({ length: 4 }, async (_, i) => { const v = Buffer.alloc(16); @@ -131,6 +136,7 @@ describe('Coordinated retry (Phase 3)', () => { ); }) ); + const elapsed = performance.now() - start; // All transactions should eventually succeed (coordinatedRetry retries // without error) or fail gracefully; none should throw unexpectedly. @@ -141,6 +147,12 @@ describe('Coordinated retry (Phase 3)', () => { } } + // A conflict here resolves via LockTracker's real wake (the other + // commit finishing), not the #741 park timeout (default 5000ms) -- + // bound the wall clock well under that so a broken wake path can't + // hide behind the timeout and still pass. + expect(elapsed).toBeLessThan(3000); + // Slot should be 0 (released) after all transactions settle. const newV = 2.5e12; db.populateVersion(key, newV); @@ -168,6 +180,89 @@ describe('Coordinated retry (Phase 3)', () => { })); }); +// Regression coverage for #741: a park behind a leaked/abandoned holder must +// resolve RETRY_NOW after a bounded wait instead of hanging forever +// (harper#2001). The lower bound below is what tells this apart from the +// `!parked` fast path, which also resolves RETRY_NOW near-instantly. +describe('Coordinated retry — bounded park timeout (#741)', () => { + it('a commit parked behind a never-releasing holder settles with RETRY_NOW within the deadline', () => + dbRunner({ skipOpen: true }, async ({ dbPath }) => { + const { code, signal } = await new Promise<{ + code: number | null; + signal: NodeJS.Signals | null; + }>((resolve, reject) => { + const child = spawn(process.execPath, [parkTimeoutFixturePath, dbPath], { + env: { ...process.env, ROCKSDB_JS_PARK_TIMEOUT_MS: '1' }, + }); + // Hang backstop only — the child's own `elapsed` assertion is the + // deadline check, and this also has to cover node boot and addon + // load, so a budget near the child's would report a slow runner as + // a regression. Under vitest's 30s testTimeout so the kill still + // dumps stderr. + const timer = setTimeout(() => child.kill(), 20_000); + let stderr = ''; + child.stderr?.on('data', (chunk) => { + stderr += chunk.toString(); + }); + child.on('close', (childCode, childSignal) => { + clearTimeout(timer); + if (childCode !== 0 || childSignal) { + console.error(`Park timeout child stderr:\n${stderr}`); + } + resolve({ code: childCode, signal: childSignal }); + }); + child.on('error', (error) => { + clearTimeout(timer); + reject(error); + }); + }); + + expect(signal).toBeNull(); + expect(code).toBe(0); + })); + + // Catches a hang on the wake-during-close path: the holder's intents are + // released from inside finishClose(), so the park's LockTracker callback + // runs while the close is in progress. The assertion is a deadline, so it + // cannot see either of the other two close-path behaviours — shutdown() + // resolves pending parks under its mutex before joining the timeout thread, + // so a detached thread settles this just as fast (its cost is a freed- + // registry touch, an ASan-shaped failure), and the drain of still-pending + // parks needs the holder on another database's colliding VT slot, which is + // hash-dependent and not arrangeable from here. + it('a commit parked at db.close() settles instead of hanging', () => + dbRunner({ dbOptions: [{ encoding: false, verificationTable: true }] }, async ({ db }) => { + const key = Buffer.from('park-timeout-close-drain'); + const v0 = 1.6e12; + await db.put(key, valueWithVersion(v0)); + db.populateVersion(key, v0); + + const holder = new Transaction(db.store, { coordinatedRetry: true }); + holder.putSync(key, valueWithVersion(2.1e12)); + + const txn = new Transaction(db.store, { coordinatedRetry: true }); + await txn.get(key); + await db.put(key, valueWithVersion(2.2e12)); + txn.putSync(key, valueWithVersion(2.3e12)); + + const commit = txn.commit(); + // Long enough for the park to be registered with the timeout thread, + // short enough to stay far from its 5000ms deadline. + await delay(250); + + const start = performance.now(); + db.close(); + // Settled, not resolved: `commit()`'s `aftercommit` notify rejects + // once the database is closed, whatever the native result was. + const [settled] = await Promise.allSettled([commit]); + const elapsed = performance.now() - start; + + expect(settled.status).toBe('rejected'); + // Under the deadline, so this can only have come from the close. + expect(elapsed).toBeLessThan(4000); + })); +}); + // Regression coverage for the VT-fast-path / optimistic-snapshot interaction. // A read satisfied entirely from the Verification Table (returning // FRESH_VERSION_FLAG, skipping the RocksDB read) must STILL establish the