Dedicated per-database commit thread for async transaction commits

AGENTS.md

@@ -122,12 +122,33 @@ C++ code that needs to emit to JS without a database context should call
 `emitGlobalEvent(key, data)` from `napi/global_events.h`. Use namespaced keys
 (`'transactionLog:warning'`) for internal events to avoid collisions with user-defined ones.
 
+### Commit execution
+
+Async `Transaction.commit()` does not use the libuv threadpool: each database
+has a dedicated commit thread (`CommitWorker`, owned by the shared
+`DBDescriptor`) that runs the txn-log write + RocksDB commit in dispatch order,
+so slow commits cannot starve fs/dns/crypto/async-get work sharing the libuv
+pool. `ROCKSDB_JS_COMMIT_THREAD` selects the mode (`0`/`false` = legacy libuv
+path, default = single lane, `2` = experimental two-lane txnlog→commit
+pipeline). Completions are marshalled back to the originating env via per-env
+tsfns held on the descriptor under `commitMutex`; the commit thread calls them
+under that mutex and a dying env's tsfn is released from the module env-cleanup
+hook (`DBRegistry::ReleaseCommitCompletionsByEnv`) — the same env-teardown
+discipline as `EventEmitter::notify` above. A per-commit tsfn acquire is NOT
+sufficient (env teardown does not honor tsfn acquire counts); see
+`test/commit-teardown.test.ts` and the `ROCKSDB_JS_COMMIT_DELAY_MS` test seam.
+
 ## Environment Variables
 
 - `ROCKSDB_VERSION` - Override RocksDB version (default from package.json, or 'latest')
 - `ROCKSDB_PATH` - Build from local RocksDB source instead of prebuilt
 - `MINIFY=1` - Enable minification of TypeScript bundle
 - `KEEP_FILES=1` - Don't delete temporary test databases for debugging purposes
+- `ROCKSDB_JS_COMMIT_THREAD` - Async-commit execution mode: `0`/`false` = legacy
+  libuv threadpool, unset = dedicated per-database commit thread (default),
+  `2` = experimental two-lane pipeline
+- `ROCKSDB_JS_COMMIT_DELAY_MS` - Test-only: delay on the commit thread before
+  each completion callback (widens teardown race windows)
 
 ## Test Structure
 

docs/stats.md

@@ -34,6 +34,8 @@ By default, `db.getStats()` returns properties only.
 
 | Name                                        | Description                                                                                                                                                                                                                   | Type   |
 | ------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------ |
+| `commitPipeline.commitQueueDepth`           | Number of async commits queued on the database's commit lane but not yet started (in the default single-lane mode this covers the whole commit: log write + RocksDB commit).                                                  | gauge  |
+| `commitPipeline.logQueueDepth`              | Number of async commits queued on the database's transaction-log lane but not yet started (always `0` in the default single-lane mode; see `ROCKSDB_JS_COMMIT_THREAD=2`).                                                     | gauge  |
 | `rocksdb.block-cache-capacity`              | Capacity in bytes of the block cache.                                                                                                                                                                                         | gauge  |
 | `rocksdb.block-cache-pinned-usage`          | Bytes occupied by pinned block cache entries.                                                                                                                                                                                 | gauge  |
 | `rocksdb.block-cache-usage`                 | Bytes currently used by block cache entries.                                                                                                                                                                                  | gauge  |

src/binding/binding.cpp

@@ -160,6 +160,10 @@ NAPI_MODULE_INIT() {
 		napi_env dyingEnv = static_cast<napi_env>(data);
 		rocksdb_js::GlobalEvents::getInstance().removeListenersByEnv(dyingEnv);
 		rocksdb_js::DBRegistry::RemoveListenersByEnv(dyingEnv);
+		// Release this env's commit-completion tsfns before Node frees the env's
+		// tsfns, so the shared commit thread stops marshalling into a torn-down
+		// env (mirrors the listener cleanup above).
+		rocksdb_js::DBRegistry::ReleaseCommitCompletionsByEnv(dyingEnv);
 
 		int32_t newRefCount = --moduleRefCount;
 		if (newRefCount == 0) {

src/binding/core/platform.cpp

@@ -1,8 +1,10 @@
 #include <cmath>
+#include <cstring>
 #include <atomic>
 #include <chrono>
 #include <filesystem>
 #include <limits>
+#include <string>
 #include <thread>
 #include "core/debug.h"
 #include "core/exception.h"
@@ -11,6 +13,7 @@
 #include <windows.h>
 #elif defined(__linux__)
 #include <sys/syscall.h>
+#include <pthread.h>
 #elif defined(__APPLE__)
 #include <pthread.h>
 #endif
@@ -31,6 +34,28 @@ size_t getThreadId() {
 #endif
 }
 
+void setThreadName(const char* name) {
+#if defined(__linux__)
+	// Linux caps thread names at 16 bytes including the null terminator;
+	// longer names fail with ERANGE (not truncated), so truncate ourselves.
+	char truncated[16];
+	::strncpy(truncated, name, sizeof(truncated) - 1);
+	truncated[sizeof(truncated) - 1] = '\0';
+	::pthread_setname_np(::pthread_self(), truncated);
+#elif defined(__APPLE__)
+	::pthread_setname_np(name);
+#elif defined(_WIN32)
+	int len = ::MultiByteToWideChar(CP_UTF8, 0, name, -1, nullptr, 0);
+	if (len > 0) {
+		std::wstring wide(static_cast<size_t>(len), L'\0');
+		::MultiByteToWideChar(CP_UTF8, 0, name, -1, wide.data(), len);
+		::SetThreadDescription(::GetCurrentThread(), wide.c_str());
+	}
+#else
+	(void)name;
+#endif
+}
+
 std::chrono::system_clock::time_point convertFileTimeToSystemTime(
 	const std::filesystem::file_time_type& fileTime
 ) {

src/binding/core/platform.h

@@ -9,6 +9,13 @@ namespace rocksdb_js {
 
 size_t getThreadId();
 
+/**
+ * Sets the current thread's name for diagnostics (visible in top, gdb, etc.).
+ * Best-effort and platform-guarded; names longer than the OS limit (15 chars
+ * on Linux) are truncated by the platform.
+ */
+void setThreadName(const char* name);
+
 std::chrono::system_clock::time_point convertFileTimeToSystemTime(const std::filesystem::file_time_type& fileTime);
 
 double getMonotonicTimestamp();

src/binding/database/commit_worker.h

@@ -0,0 +1,140 @@
+#ifndef __COMMIT_WORKER_H__
+#define __COMMIT_WORKER_H__
+
+#include <condition_variable>
+#include <deque>
+#include <functional>
+#include <mutex>
+#include <thread>
+#include "core/debug.h"
+#include "core/platform.h"
+
+namespace rocksdb_js {
+
+/**
+ * A dedicated worker thread with a task queue, used for the per-database
+ * commit pipeline. Async transaction commits execute on these instead of the
+ * libuv threadpool so that slow commits (write stalls, large batches,
+ * transaction-log contention) cannot occupy libuv slots and starve unrelated
+ * async work (fs, dns, crypto, async gets). Completion is marshalled back to
+ * the calling env via a threadsafe function.
+ *
+ * Each database has a commit lane and, in two-lane mode, a transaction-log
+ * lane that writes the log batch before forwarding to the commit lane (see
+ * commitThreadMode() in transaction.cpp). Each lane preserves dispatch order —
+ * which keeps per-database commit order stable and serializes the
+ * transaction-log write mutex for free. RocksDB commits never block on one
+ * another's locks (pessimistic locks are acquired at put time, optimistic
+ * validation does not block), so a single commit lane cannot deadlock.
+ *
+ * The thread is started lazily on the first task and joined on shutdown after
+ * draining any queued tasks. Each run-loop wakeup drains the entire queue
+ * snapshot in one pass — the structural hook that later lets the log lane see
+ * every pending batch and aggregate them into one writev.
+ */
+struct CommitWorker final {
+	const char* threadName;
+	std::mutex mutex;
+	std::condition_variable cv;
+	std::deque<std::function<void()>> queue;
+	std::thread thread;
+	bool started = false;
+	bool stopped = false;
+
+	explicit CommitWorker(const char* threadName) : threadName(threadName) {}
+
+	~CommitWorker() {
+		this->shutdown();
+	}
+
+	/**
+	 * Enqueues a task, lazily starting the worker thread. If the worker has
+	 * already been shut down (descriptor closing), the task runs inline on the
+	 * calling thread; a commit will fail fast on the closing checks.
+	 */
+	void enqueue(std::function<void()> task) {
+		bool runInline = false;
+		bool wasEmpty = false;
+		{
+			std::lock_guard<std::mutex> lock(this->mutex);
+			if (this->stopped) {
+				runInline = true;
+			} else {
+				if (!this->started) {
+					this->started = true;
+					this->thread = std::thread([this]() { this->run(); });
+				}
+				wasEmpty = this->queue.empty();
+				this->queue.push_back(std::move(task));
+			}
+		}
+		if (runInline) {
+			DEBUG_LOG("%p CommitWorker::enqueue Worker stopped, running task inline\n", this);
+			task();
+		} else if (wasEmpty) {
+			// Only the empty->non-empty transition needs a wakeup: the worker
+			// drains the whole queue per wakeup and its wait predicate re-checks
+			// emptiness before blocking, so tasks appended to a non-empty queue
+			// are picked up without a signal. Profiling showed the per-enqueue
+			// pthread_cond_signal as a measurable JS-thread cost under load.
+			this->cv.notify_one();
+		}
+	}
+
+	/**
+	 * Number of queued (not yet started) tasks. Diagnostic only — the value is
+	 * stale the moment it is read.
+	 */
+	size_t depth() {
+		std::lock_guard<std::mutex> lock(this->mutex);
+		return this->queue.size();
+	}
+
+	/**
+	 * Drains any remaining queued tasks and joins the thread. Idempotent;
+	 * called from DBDescriptor::finishClose() and the destructor.
+	 */
+	void shutdown() {
+		std::thread toJoin;
+		{
+			std::lock_guard<std::mutex> lock(this->mutex);
+			this->stopped = true;
+			if (this->started) {
+				toJoin = std::move(this->thread);
+				this->started = false;
+			}
+		}
+		this->cv.notify_all();
+		if (toJoin.joinable()) {
+			DEBUG_LOG("%p CommitWorker::shutdown Draining and joining worker thread\n", this);
+			toJoin.join();
+		}
+	}
+
+private:
+	void run() {
+		setThreadName(this->threadName);
+		std::unique_lock<std::mutex> lock(this->mutex);
+		for (;;) {
+			this->cv.wait(lock, [this] { return this->stopped || !this->queue.empty(); });
+			if (this->queue.empty()) {
+				// stopped and fully drained
+				return;
+			}
+			// Drain the whole queue in one pass so tasks run without retaking
+			// the mutex per item (and so a future log-lane aggregation step can
+			// see the full pending batch).
+			std::deque<std::function<void()>> batch;
+			batch.swap(this->queue);
+			lock.unlock();
+			for (auto& task : batch) {
+				task();
+			}
+			lock.lock();
+		}
+	}
+};
+
+} // namespace rocksdb_js
+
+#endif

src/binding/database/db_descriptor.cpp

@@ -162,6 +162,33 @@ void DBDescriptor::finishClose() {
 	}
 	DEBUG_LOG("%p DBDescriptor::close All operations complete \"%s\"\n", this, this->path.c_str());
 
+	// Drain the commit pipeline before flushing so its data is included in
+	// the flush. The log lane feeds the commit lane, so it must drain first;
+	// its final tasks enqueue onto the still-running commit lane (or run
+	// inline once that lane stops).
+	this->logWorker.shutdown();
+	this->commitWorker.shutdown();
+
+	// Release any remaining per-env commit-completion tsfns. An in-flight
+	// commit pins this descriptor (state -> txnHandle -> dbHandle -> descriptor),
+	// so reaching here means no commit is in flight; only idle (unref'd) tsfns
+	// for still-living envs can remain, and those envs will issue no further
+	// commits to this descriptor. Queued completions already handed to a tsfn
+	// are still delivered (napi_tsfn_release, not abort).
+	{
+		std::lock_guard<std::mutex> lock(this->commitMutex);
+		for (auto& [env, completion] : this->commitCompletions) {
+			if (completion.tsfn) {
+				::napi_release_threadsafe_function(completion.tsfn, napi_tsfn_release);
+			}
+		}
+		this->commitCompletions.clear();
+		// Block any later registerCommitCompletion (a commit racing this close
+		// from another env) from re-creating a tsfn that would never be
+		// released; such commits fall back to the legacy libuv path.
+		this->commitCompletionsClosed = true;
+	}
+
 	// We want to ensure that all in-memory data is written to disk
 	this->flush();
 
@@ -237,6 +264,81 @@ void DBDescriptor::finishClose() {
 	this->db.reset();
 }
 
+napi_status DBDescriptor::registerCommitCompletion(napi_env env, napi_threadsafe_function_call_js callJs, bool& closed) {
+	std::lock_guard<std::mutex> lock(this->commitMutex);
+	closed = this->commitCompletionsClosed;
+	if (closed) {
+		return napi_ok;
+	}
+	CommitCompletion& completion = this->commitCompletions[env];
+	if (completion.tsfn == nullptr) {
+		napi_value resourceName;
+		napi_status status = ::napi_create_string_utf8(env, "rocksdb.commit", NAPI_AUTO_LENGTH, &resourceName);
+		if (status != napi_ok) {
+			return status;
+		}
+		// Created ref'd (thread count 1 for the commit thread), which is what we
+		// want with a commit about to be dispatched.
+		status = ::napi_create_threadsafe_function(
+			env,
+			nullptr,   // func: callJs does all the work
+			nullptr,   // async_resource
+			resourceName,
+			0,         // unlimited queue
+			1,         // initial thread count: the commit thread
+			nullptr,   // finalize data
+			nullptr,   // finalize cb
+			nullptr,   // context
+			callJs,
+			&completion.tsfn
+		);
+		if (status != napi_ok) {
+			this->commitCompletions.erase(env);
+			return status;
+		}
+	} else if (completion.pending == 0) {
+		// Waking from idle: keep the event loop alive until completion.
+		napi_status status = ::napi_ref_threadsafe_function(env, completion.tsfn);
+		if (status != napi_ok) {
+			return status;
+		}
+	}
+	completion.pending++;
+	return napi_ok;
+}
+
+bool DBDescriptor::dispatchCommitCompletion(napi_env env, void* state) {
+	std::lock_guard<std::mutex> lock(this->commitMutex);
+	auto it = this->commitCompletions.find(env);
+	if (it == this->commitCompletions.end() || it->second.tsfn == nullptr) {
+		// env was torn down / released; the caller drops the state.
+		return false;
+	}
+	napi_status status = ::napi_call_threadsafe_function(it->second.tsfn, state, napi_tsfn_nonblocking);
+	return status == napi_ok;
+}
+
+void DBDescriptor::finishCommitCompletion(napi_env env) {
+	std::lock_guard<std::mutex> lock(this->commitMutex);
+	auto it = this->commitCompletions.find(env);
+	if (it != this->commitCompletions.end() && --it->second.pending == 0 && it->second.tsfn != nullptr) {
+		// Idle: allow the event loop to exit.
+		::napi_unref_threadsafe_function(env, it->second.tsfn);
+	}
+}
+
+void DBDescriptor::releaseCommitCompletionsByEnv(napi_env env) {
+	std::lock_guard<std::mutex> lock(this->commitMutex);
+	auto it = this->commitCompletions.find(env);
+	if (it != this->commitCompletions.end()) {
+		if (it->second.tsfn != nullptr) {
+			// Queued completions are still delivered before the tsfn finalizes.
+			::napi_release_threadsafe_function(it->second.tsfn, napi_tsfn_release);
+		}
+		this->commitCompletions.erase(it);
+	}
+}
+
 /**
  * Registers a database resource to be closed when the descriptor is closed.
  *