fix(transaction): bound the coordinated-retry park with a descriptor-owned timeout

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<ParkTimeoutRegistry>`** 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<DBDescriptor>::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<bool>` 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

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

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) {

src/binding/database/db_descriptor.cpp

@@ -10,6 +10,7 @@
 #include "rocksdb/utilities/options_util.h"
 #include <algorithm>
 #include <memory>
+#include <system_error>
 #include <unordered_map>
 
 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<std::atomic<bool>> fired
+) {
+	std::lock_guard<std::mutex> 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<ParkTimeout>();
+	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::ParkTimeout> ParkTimeoutRegistry::take(uint64_t id) {
+	auto it = this->parks.find(id);
+	if (it == this->parks.end()) {
+		return nullptr;
+	}
+	std::unique_ptr<ParkTimeout> 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<std::mutex> 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<ParkTimeout> due = std::move(parkIt->second);
+			this->parks.erase(parkIt);
+			ParkTimeoutRegistry::resolve(*due);
+		}
+	}
+}
+
+void ParkTimeoutRegistry::fire(uint64_t id) {
+	std::lock_guard<std::mutex> lock(this->mutex);
+	std::unique_ptr<ParkTimeout> 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<std::mutex> 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<std::mutex> 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.
  *