Serialize database destruction with concurrent opens

README.md

@@ -141,7 +141,13 @@ Creates a new database instance.
 ### `db.close()`
 
 Closes a database. This function can be called multiple times and will only close an opened
-database. A database instance can be reopened once its closed.
+database. A database instance can be reopened once it is closed. A flush failure leaves the native
+database quarantined so `shutdown()` can retry without losing unflushed data; an explicit
+`destroy()` can instead delete it. A failure while waiting for compaction to settle is reported
+after native teardown completes; the optional `compactOnClose` pass itself is best-effort and its
+errors do not fail the close, since a skipped compaction loses no data. All native close errors
+emit `database:closeFailed`. The quarantine applies to both writable and read-only opens because
+both modes share the physical path lifecycle.
 
 ```typescript
 const db = RocksDatabase.open('foo');
@@ -183,6 +189,12 @@ Sets global database settings.
     Defaults to 32MB. Set to `0` (zero) disables block cache for future opened databases. Existing
     block cache for any opened databases is resized immediately. Negative values throw an error.
   - `compactOnClose: boolean` When `true`, compacts the database on close. Defaults to `false`.
+  - `lifecycleWaitSeconds: number` How long a synchronous open, destroy, or shutdown waits for a
+    _conflicting_ lifecycle operation already in progress on the same path (e.g. another open or
+    close) before throwing a retryable timeout error. It does not bound the separate, intentionally
+    unbounded wait that `destroy()`/`shutdown()` make for in-flight backups, checkpoints, or other
+    async work still using the database — see [`db.destroy()`](#dbdestroy-void). Defaults to `30`
+    seconds and must be a positive integer.
   - `verificationTableEntries: number` The number of slots in the process-global
     [Verification Table](#verification-table). Each slot is 8 bytes, so the default of `131072`
     (128K) slots is 1 MB. Set to `0` to disable the verification table. This must be configured
@@ -349,7 +361,16 @@ db.compactSync({ bottommost: true });
 ### `db.destroy(): void`
 
 Completely removes a database based on the `db` instance's path including all data, column families,
-and files on disk.
+and files on disk. Destruction owns the physical path for the process: it closes every writable and
+read-only handle for that path, waits for registered backups and checkpoints to stop using the
+native database, and prevents another handle from reopening the path until removal finishes. Those
+waits are synchronous and can outlive `lifecycleWaitSeconds` once destruction has claimed the path,
+because releasing the native database beneath an active copy would be unsafe.
+
+A previously opened instance does not need to remain open, which allows an explicit `destroy()`
+retry after failed physical cleanup. A never-opened or read-only instance cannot destroy the
+database. `shutdown()` reports a pending cleanup tombstone but never retries deletion; only an
+explicit `destroy()` can remove the path.
 
 ```typescript
 db.destroy();
@@ -1961,6 +1982,9 @@ console.log(currentThreadId());
 Returns an array containing that status of all active RocksDB instances.
 
 - `path: string` The database path.
+- `closeError?: string` The native lifecycle error retaining this registry entry.
+- `destroyCleanupPending?: boolean` The native database is closed, but physical path cleanup must
+  finish before the next open. Call `destroy()` to retry cleanup.
 - `refCount: number` The number of JavaScript database instances plus the registry's reference.
 - `columnFamiles: object` A map of column family names and their their info.
   - `userSharedBuffers: number` The count of active user shared buffers.
@@ -1978,11 +2002,20 @@ console.log(registryStatus());
 
 The `shutdown()` will flush all in-memory data to disk and wait for any outstanding compactions to
 finish, for all open databases. It is highly recommended to call this in a `process` `exit` event
-listener (on the main thread), to ensure that all data is flushed to disk before the process exits:
+listener (on the main thread), to ensure that all data is flushed to disk before the process exits.
+It throws the first close failure after attempting every claimed database; call it again to retry
+any descriptor whose native teardown did not complete. It reports pending destroy-cleanup
+tombstones without deleting their paths; retry those with an explicit `destroy()`:
 
 ```typescript
 import { shutdown } from '@harperfast/rocksdb-js';
-process.on('exit', shutdown);
+process.on('exit', () => {
+	try {
+		shutdown();
+	} catch (error) {
+		console.error('rocksdb-js shutdown failed', error);
+	}
+});
 ```
 
 ### `versions: { 'rocksdb': string; 'rocksdb-js': string }`

benchmark/setup.ts

@@ -235,6 +235,7 @@ interface WorkerState {
 	benchPromise: ReturnType<typeof withResolvers<void>>;
 	exitPromise: ReturnType<typeof withResolvers<void>>;
 	teardownPromise: ReturnType<typeof withResolvers<void>>;
+	teardownError?: Error;
 }
 
 interface WorkerBenchmarkOptions extends BenchmarkOptions<any, any> {
@@ -381,6 +382,7 @@ export function workerBenchmark(type: string, options: any): void {
 	}
 
 	const workerState: WorkerState[] = [];
+	const dbPath = join('benchmark', 'data', `rocksdb-benchmark-${randomBytes(8).toString('hex')}`);
 	const workerPayload = {
 		suites: workerCurrentSuites.map((suite) => suite.name),
 		benchmark: benchmarkName,
@@ -408,12 +410,6 @@ export function workerBenchmark(type: string, options: any): void {
 				if (mode === 'run') {
 					return;
 				}
-				const path = join(
-					'benchmark',
-					'data',
-					`rocksdb-benchmark-${randomBytes(8).toString('hex')}`
-				);
-
 				let teardownTimeoutId: NodeJS.Timeout;
 				await Promise.race([
 					activeBenchmark,
@@ -436,19 +432,24 @@ export function workerBenchmark(type: string, options: any): void {
 								benchmarkFile: pathToFileURL(benchmarkFile).toString(),
 								benchmarkWorkerId: i + 1,
 								mode,
-								path,
+								path: dbPath,
 							});
 							// important! these promises need to be referenced as
 							// properties of `state` because they are reset by reference
-							const state = {
+							const state: WorkerState = {
 								worker,
 								benchPromise: withResolvers<void>(),
 								exitPromise: withResolvers<void>(),
 								teardownPromise: withResolvers<void>(),
 							};
 							workerState[i] = state;
 							worker.on('error', reject);
-							worker.on('exit', () => {
+							worker.on('exit', (code) => {
+								if (code !== 0 && !state.teardownError) {
+									state.teardownError = new Error(
+										`Benchmark worker ${i + 1} exited with code ${code}`
+									);
+								}
 								state.benchPromise.resolve();
 								state.teardownPromise.resolve();
 								state.exitPromise.resolve();
@@ -461,6 +462,9 @@ export function workerBenchmark(type: string, options: any): void {
 									state.benchPromise.resolve();
 								} else if (event.teardownDone) {
 									state.teardownPromise.resolve();
+								} else if (event.teardownError) {
+									state.teardownError = new Error(event.teardownError);
+									state.teardownPromise.resolve();
 								} else if (event.timeout) {
 									state.teardownPromise.resolve();
 									state.benchPromise.reject(new Error('Benchmark timed out'));
@@ -492,8 +496,19 @@ export function workerBenchmark(type: string, options: any): void {
 						return workerState[i].exitPromise.promise;
 					})
 				);
+				const teardownError = workerState.find((state) => state.teardownError)?.teardownError;
+				if (!teardownError) {
+					try {
+						rmSync(dbPath, { force: true, recursive: true, maxRetries: 3 });
+					} catch (err) {
+						console.warn(`Benchmark teardown failed to delete db path: ${err}`);
+					}
+				} else {
+					console.warn(`Benchmark teardown failed; retaining ${dbPath} for inspection`);
+				}
 
 				resolve();
+				if (teardownError) throw teardownError;
 			},
 		}
 	);
@@ -529,12 +544,14 @@ export async function workerInit(): Promise<void> {
 				await teardown(ctx);
 			}
 			if (ctx.db) {
-				// console.log('workerTeardown', workerData.benchmarkWorkerId, workerData.mode, type, path);
-				ctx.db.close();
 				try {
-					rmSync(path, { force: true, recursive: true, maxRetries: 3 });
-				} catch (err) {
-					console.warn(`Benchmark teardown failed to delete db path: ${err}`);
+					await ctx.db.close();
+				} catch (error) {
+					parentPort!.postMessage({
+						teardownError: error instanceof Error ? error.message : String(error),
+						benchmarkWorkerId,
+					});
+					process.exit(1);
 				}
 			}
 			parentPort!.postMessage({ teardownDone: true, benchmarkWorkerId });

docs/backups.md

@@ -202,8 +202,9 @@ The archive contains the standard RocksDB files for the snapshot (`CURRENT`, a `
   consumer therefore lets obsolete SST files accumulate (transient disk pressure) until the stream
   finishes. This is the cost of streaming with no scratch copy; a fast consumer is unaffected.
 - **The database must stay open for the whole stream.** Closing or destroying the database while a
-  stream is in flight aborts it: the backup promise rejects, and `destroy()` throws rather than
-  tearing the database down underneath the copy.
+  stream is in flight aborts the stream and makes the backup promise reject. `destroy()` waits for
+  the native producer to release its in-flight claim, then closes every handle for the path and
+  removes the database; it never tears the native database down underneath the copy.
 - **A consumer error aborts the backup.** If `stream.write()` rejects (or the stream is aborted), the
   backup promise rejects and the native producer stops.
 

src/binding/binding.cpp

@@ -39,8 +39,21 @@ namespace rocksdb_js {
  * Shutdown function to ensure that we write in-memory data from all databases.
  */
 napi_value Shutdown(napi_env env, napi_callback_info info) {
+	std::string error;
+	try {
+		DBRegistry::Shutdown();
+	} catch (const std::exception& exception) {
+		error = exception.what();
+	} catch (...) {
+		error = "Unknown native database shutdown failure";
+	}
+	// Release global listener threadsafe functions on every path, including a
+	// failed shutdown -- otherwise they outlive this N-API environment.
 	GlobalEvents::Shutdown();
-	DBRegistry::Shutdown();
+	if (!error.empty()) {
+		::napi_throw_error(env, nullptr, error.c_str());
+		return nullptr;
+	}
 	napi_value result;
 	NAPI_STATUS_THROWS(::napi_get_undefined(env, &result));
 	return result;
@@ -154,6 +167,7 @@ napi_value TransactionLogMapCount(napi_env env, napi_callback_info info) {
 static std::atomic<int32_t> moduleRefCount{0};
 
 NAPI_MODULE_INIT() {
+	initializeTestSeams();
 #ifdef DEBUG
 	// disable buffering for stderr to ensure messages are written immediately
 	::setvbuf(stderr, nullptr, _IONBF, 0);
@@ -216,9 +230,26 @@ NAPI_MODULE_INIT() {
 		int32_t newRefCount = --moduleRefCount;
 		if (newRefCount == 0) {
 			DEBUG_LOG("Binding::Init Cleaning up last instance, shutting down all databases\n");
-			rocksdb_js::GlobalEvents::Shutdown();
-			rocksdb_js::TransactionLogStoreRegistry::Shutdown();
-			rocksdb_js::DBRegistry::Shutdown();
+			auto cleanup = [](const char* name, auto shutdown) {
+				try {
+					shutdown();
+				} catch (const std::exception& error) {
+					::fprintf(stderr, "rocksdb-js %s cleanup failed: %s\n", name, error.what());
+				} catch (...) {
+					::fprintf(stderr, "rocksdb-js %s cleanup failed: unknown native error\n", name);
+				}
+			};
+			cleanup("database registry", []() { rocksdb_js::DBRegistry::Shutdown(); });
+			// Shutdown() leaves a descriptor whose close-time flush failed
+			// quarantined in the registry so shutdown()/destroy() can retry it.
+			// The process is exiting, so there is no later retry -- and a
+			// descriptor that survives to the registry singleton's static
+			// destructor closes its rocksdb::DB from an atexit handler, after
+			// RocksDB's own statics are gone, which aborts. Release whatever is
+			// left here, while that is still safe.
+			cleanup("database registry teardown", []() { rocksdb_js::DBRegistry::Teardown(); });
+			cleanup("transaction logs", []() { rocksdb_js::TransactionLogStoreRegistry::Shutdown(); });
+			cleanup("global events", []() { rocksdb_js::GlobalEvents::Shutdown(); });
 			DEBUG_LOG("Binding::Init env cleanup done\n");
 		} else if (newRefCount < 0) {
 			DEBUG_LOG("Binding::Init WARNING: Module ref count went negative!\n");

src/binding/core/closable.h

@@ -6,6 +6,17 @@ namespace rocksdb_js {
 struct Closable {
 	virtual ~Closable() = default;
 	virtual void close() = 0;
+
+	/**
+	 * Publishes cancellation for work that `close()` will later wait on but
+	 * that cannot poll a flag from where it runs -- today only a manual
+	 * RocksDB compaction, which abandons its range solely through the
+	 * `CompactRangeOptions::canceled` pointer it was handed. Teardown calls
+	 * this on every attached closable before its first step that can block on
+	 * such work, which is earlier than the closables sweep that closes them.
+	 * Default no-op: most closables have nothing that outlives a flag check.
+	 */
+	virtual void cancelBlockingWork() {}
 };
 
 } // namespace rocksdb_js

src/binding/core/test_seam.h

@@ -3,19 +3,110 @@
 
 #include <atomic>
 #include <cstdlib>
+#include <mutex>
 
 // Deterministic test seams that widen a race window are gated on a millisecond
 // delay read from an environment variable (0 = disabled). They are inert in
 // production where the env var is unset.
 //
 // Pass the env var name to testDelayMs() at the call site; see
-// EventEmitter::notify, TransactionHandle::close, and TransactionHandle::get
-// for usage.
+// EventEmitter::notify, TransactionHandle::close, TransactionHandle::get, and
+// DBRegistry::DestroyDB for usage.
 inline int testDelayMs(const char* envName) {
 	const char* value = ::getenv(envName);
 	return value ? ::atoi(value) : 0;
 }
 
+// Snapshot native fault flags once; mutating process.env while native workers
+// can read it is unsafe, and process.env deletion does not update MSVC's CRT.
+inline std::atomic<bool>& closeFailureFlag() {
+	static std::atomic<bool> pending{false};
+	return pending;
+}
+
+// Snapshotted for the same reason as closeFailureFlag(): a fault flag must not
+// be re-read from the environment on a native thread.
+inline std::atomic<bool>& destroyFailureFlag() {
+	static std::atomic<bool> pending{false};
+	return pending;
+}
+
+// A count, not a flag: a quarantined descriptor is only reached at process exit
+// when the shutdown() retry fails too, so reproducing that state needs the
+// close-time flush to fail more than once. `=1` behaves exactly as the previous
+// boolean did.
+inline std::atomic<int>& closeFlushFailureFlag() {
+	static std::atomic<int> pending{0};
+	return pending;
+}
+
+// DBIterator::Next() returns one row per call, so its seam is snapshotted here
+// rather than read per call: a getenv() scan per row is a measurable share of
+// the per-row cost for a seam that is unset in production.
+inline std::atomic<int>& iteratorNextDelayMsFlag() {
+	static std::atomic<int> delayMs{0};
+	return delayMs;
+}
+
+// Per-row delay for DBIteratorHandle::countRemaining(), snapshotted for the
+// same reason.
+inline std::atomic<int>& countScanDelayMsFlag() {
+	static std::atomic<int> delayMs{0};
+	return delayMs;
+}
+
+// Upper bound, in milliseconds, that a cancellable manual compactRange() parks
+// before handing the range to RocksDB. It returns as soon as the descriptor's
+// cancel token is armed, so a fixture can hold a compaction across a foreign
+// close claim without depending on how long a real compaction happens to take.
+// Snapshotted here rather than read per call so the production path (unset)
+// costs one relaxed load.
+inline std::atomic<int>& compactCancelDelayMsFlag() {
+	static std::atomic<int> delayMs{0};
+	return delayMs;
+}
+
+// Delays DBHandle::open() after the registry has returned. A handle must
+// already be adopted and attached while this delay is observable.
+inline std::atomic<int>& openAttachDelayMsFlag() {
+	static std::atomic<int> delayMs{0};
+	return delayMs;
+}
+
+inline void initializeTestSeams() {
+	static std::once_flag initialized;
+	std::call_once(initialized, []() {
+		const char* value = ::getenv("ROCKSDB_JS_CLOSE_FAILURE");
+		closeFailureFlag().store(value && ::atoi(value) > 0, std::memory_order_relaxed);
+		value = ::getenv("ROCKSDB_JS_DESTROY_FAILURE");
+		destroyFailureFlag().store(value && ::atoi(value) > 0, std::memory_order_relaxed);
+		value = ::getenv("ROCKSDB_JS_CLOSE_FLUSH_FAILURE");
+		closeFlushFailureFlag().store(value ? ::atoi(value) : 0, std::memory_order_relaxed);
+		iteratorNextDelayMsFlag().store(
+			testDelayMs("ROCKSDB_JS_ITERATOR_NEXT_DELAY_MS"), std::memory_order_relaxed);
+		countScanDelayMsFlag().store(
+			testDelayMs("ROCKSDB_JS_COUNT_DELAY_MS"), std::memory_order_relaxed);
+		compactCancelDelayMsFlag().store(
+			testDelayMs("ROCKSDB_JS_COMPACT_DELAY_MS"), std::memory_order_relaxed);
+		openAttachDelayMsFlag().store(
+			testDelayMs("ROCKSDB_JS_OPEN_ATTACH_DELAY_MS"), std::memory_order_relaxed);
+	});
+}
+
+inline bool testConsumeCloseFailure() {
+	return closeFailureFlag().exchange(false, std::memory_order_relaxed);
+}
+
+inline bool testConsumeCloseFlushFailure() {
+	int cur = closeFlushFailureFlag().load(std::memory_order_relaxed);
+	while (cur > 0) {
+		if (closeFlushFailureFlag().compare_exchange_weak(cur, cur - 1, std::memory_order_relaxed)) {
+			return true;
+		}
+	}
+	return false;
+}
+
 // Deterministic one-shot(-per-N) seam for the stranded-snapshot retry path: forces the next N
 // transaction commits to fail with TryAgain (the caller rolls back so no data is committed),
 // reproducing an ERR_TRY_AGAIN that a real memtable flush would cause but that is finicky to