From 5783262669f70ae470357c66df41405318e053e0 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 12 Aug 2026 13:32:29 -0600 Subject: [PATCH 1/5] fix(schema): signal database changes once Co-Authored-By: GPT-5 Codex --- dataLayer/harperBridge/ResourceBridge.ts | 7 ------- integrationTests/apiTests/terminology.test.mjs | 5 +---- 2 files changed, 1 insertion(+), 11 deletions(-) diff --git a/dataLayer/harperBridge/ResourceBridge.ts b/dataLayer/harperBridge/ResourceBridge.ts index 847ac6e196..21c80bd332 100644 --- a/dataLayer/harperBridge/ResourceBridge.ts +++ b/dataLayer/harperBridge/ResourceBridge.ts @@ -4,13 +4,10 @@ import { table, getDatabases, database, dropDatabase, type Table } from '../../r import insertUpdateValidate from './bridgeUtility/insertUpdateValidate.js'; import SearchObject from '../SearchObject.ts'; import { - OPERATIONS_ENUM, VALUE_SEARCH_COMPARATORS, VALUE_SEARCH_COMPARATORS_REVERSE_LOOKUP, READ_AUDIT_LOG_SEARCH_TYPES_ENUM, } from '../../utility/hdbTerms.ts'; -import * as signalling from '../../utility/signalling.ts'; -import { SchemaEventMsg } from '../../server/threads/itc.js'; import { asyncSetTimeout } from '../../utility/common_utils.ts'; import { transaction } from '../../resources/transaction.ts'; import type { @@ -179,14 +176,10 @@ export class ResourceBridge extends BridgeMethods { database: createSchemaObj.schema, table: null, }); - return signalling.signalSchemaChange( - new SchemaEventMsg(process.pid, OPERATIONS_ENUM.CREATE_SCHEMA, createSchemaObj.schema) - ); } async dropSchema(dropSchemaObj) { await dropDatabase(dropSchemaObj.schema); - signalling.signalSchemaChange(new SchemaEventMsg(process.pid, OPERATIONS_ENUM.DROP_SCHEMA, dropSchemaObj.schema)); } async updateRecords(updateObj) { diff --git a/integrationTests/apiTests/terminology.test.mjs b/integrationTests/apiTests/terminology.test.mjs index 3e96a9d069..58a49800b1 100644 --- a/integrationTests/apiTests/terminology.test.mjs +++ b/integrationTests/apiTests/terminology.test.mjs @@ -416,7 +416,7 @@ suite('Terminology aliases (database / primary_key)', (ctx) => { // ── async job operations ──────────────────────────────────────────────── - test('create job_guy database and working table for job tests', async () => { + test('runtime-created database immediately starts a job', async () => { await client .req() .send({ operation: 'create_database', database: 'job_guy' }) @@ -427,9 +427,6 @@ suite('Terminology aliases (database / primary_key)', (ctx) => { .send({ operation: 'create_table', database: 'job_guy', table: 'working', primary_key: 'id' }) .expect((r) => assert.equal(r.body.message, "table 'job_guy.working' successfully created.", r.text)) .expect(200); - }); - - test('delete_records_before with database param starts job', async () => { const r = await client .req() .send({ From 7ff0da544c98dba1078fefa55e7e922775f051b8 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 12 Aug 2026 13:39:07 -0600 Subject: [PATCH 2/5] fix(lmdb): clear database alias environments Co-Authored-By: GPT-5 Codex --- utility/lmdb/cleanLMDBMap.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/utility/lmdb/cleanLMDBMap.ts b/utility/lmdb/cleanLMDBMap.ts index 78e7e8bb51..91b81bcf6c 100644 --- a/utility/lmdb/cleanLMDBMap.ts +++ b/utility/lmdb/cleanLMDBMap.ts @@ -17,6 +17,7 @@ async function cleanLMDBMap(msg: any) { let cachedEnvironment = undefined; switch (msg.operation) { + case 'drop_database': case 'drop_schema': for (let x = 0; x < keys.length; x++) { let key = keys[x]; From 646772c3364f44297122ec2821466b04816f1089 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 12 Aug 2026 13:43:37 -0600 Subject: [PATCH 3/5] fix(storage): serialize RocksDB opens Co-Authored-By: GPT-5 Codex --- dataLayer/restoreMarker.ts | 21 +++++++++++++++++++++ resources/databases.ts | 31 +++++++++++++++++++------------ 2 files changed, 40 insertions(+), 12 deletions(-) diff --git a/dataLayer/restoreMarker.ts b/dataLayer/restoreMarker.ts index dca260ce9f..4ce2426c71 100644 --- a/dataLayer/restoreMarker.ts +++ b/dataLayer/restoreMarker.ts @@ -51,6 +51,7 @@ import { tryFileLock, fileLockRelease } from '@harperfast/rocksdb-js'; export const RESTORE_META_DIR = '`restore`'; export const RESTORE_LOCK_SUFFIX = '.lock'; export const RESTORING_MARKER_SUFFIX = '.restoring'; +const DATABASE_OPEN_LOCK_SUFFIX = '.open'; /** * Directory holding the restore metadata for a database — the reserved `` `restore` `` sibling of @@ -81,6 +82,26 @@ export function restoringMarkerPath(dbPath: string): string { return join(restoreMetaDir(dbPath), restoreMetaKey(dbPath) + RESTORING_MARKER_SUFFIX); } +function databaseOpenLockPath(dbPath: string): string { + return join(restoreMetaDir(dbPath), restoreMetaKey(dbPath) + DATABASE_OPEN_LOCK_SUFFIX); +} + +/** + * Serialize native RocksDB opens for one database across worker threads. rocksdb-js normally + * shares opened handles process-wide, but two cold worker-thread opens can still race before + * that registry entry exists and contend for RocksDB's process-local LOCK file. + */ +export function acquireDatabaseOpenLock(dbPath: string): number { + mkdirSync(restoreMetaDir(dbPath), { recursive: true }); + let token = tryFileLock(databaseOpenLockPath(dbPath)); + while (token === 0) token = tryFileLock(databaseOpenLockPath(dbPath)); + return token; +} + +export function releaseDatabaseOpenLock(token: number): void { + fileLockRelease(token); +} + export type RestoreState = 'in-progress' | 'incomplete' | 'clear'; /** diff --git a/resources/databases.ts b/resources/databases.ts index 55cd4fba2b..39bd31a889 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -43,7 +43,9 @@ import { resolveRocksMemoryConfig } from '../utility/rocksMemoryConfig.ts'; import { isProcessRunning } from '../utility/processManagement/processManagement.js'; import { acquireRestoreLock, + acquireDatabaseOpenLock, checkRestoreState, + releaseDatabaseOpenLock, releaseRestoreLock, restoreMarkerPresent, scanBlockedRestores, @@ -325,19 +327,24 @@ function openRocksDatabase(path: string, options: RocksDatabaseOptions & { dupSo } mkdirSync(path, { recursive: true }); } - let db: RocksRootDatabase; - if (options.dupSort) { - db = new RocksIndexStore(path, options).open() as any; - } else { - db = new PrimaryRocksDatabase(path, options).open() as unknown as RocksRootDatabase; - // the RocksDB put and remove return promises, which masks thrown errors in non-awaiting calls to put/remove, - // making them unsafe to replace LMDB methods, which will synchronously throw errors if there is a problem - db.put = db.putSync as any; - db.remove = db.removeSync as any; - (db.encoder as any).name = options.name; + const openLock = acquireDatabaseOpenLock(path); + try { + let db: RocksRootDatabase; + if (options.dupSort) { + db = new RocksIndexStore(path, options).open() as any; + } else { + db = new PrimaryRocksDatabase(path, options).open() as unknown as RocksRootDatabase; + // the RocksDB put and remove return promises, which masks thrown errors in non-awaiting calls to put/remove, + // making them unsafe to replace LMDB methods, which will synchronously throw errors if there is a problem + db.put = db.putSync as any; + db.remove = db.removeSync as any; + (db.encoder as any).name = options.name; + } + db.env = {}; + return db; + } finally { + releaseDatabaseOpenLock(openLock); } - db.env = {}; - return db; } const lmdbDatabaseEnvs = new Map(); From 7120fc911f4c679d2ccca85fd345854595068999 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 12 Aug 2026 13:50:59 -0600 Subject: [PATCH 4/5] fix(storage): bound database open locking Co-Authored-By: GPT-5 Codex --- dataLayer/restoreMarker.ts | 12 ++++++++++-- dataLayer/schema.ts | 4 ++-- resources/databases.ts | 4 ++-- 3 files changed, 14 insertions(+), 6 deletions(-) diff --git a/dataLayer/restoreMarker.ts b/dataLayer/restoreMarker.ts index 4ce2426c71..3c0d7535cd 100644 --- a/dataLayer/restoreMarker.ts +++ b/dataLayer/restoreMarker.ts @@ -91,10 +91,18 @@ function databaseOpenLockPath(dbPath: string): string { * shares opened handles process-wide, but two cold worker-thread opens can still race before * that registry entry exists and contend for RocksDB's process-local LOCK file. */ -export function acquireDatabaseOpenLock(dbPath: string): number { +export function acquireDatabaseOpenLock(dbPath: string, maxWaitMilliseconds = 30_000): number { mkdirSync(restoreMetaDir(dbPath), { recursive: true }); let token = tryFileLock(databaseOpenLockPath(dbPath)); - while (token === 0) token = tryFileLock(databaseOpenLockPath(dbPath)); + const startedAt = Date.now(); + const sleeper = new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT)); + while (token === 0) { + if (Date.now() - startedAt >= maxWaitMilliseconds) { + throw new Error(`Timed out acquiring RocksDB open lock for ${dbPath}`); + } + Atomics.wait(sleeper, 0, 0, 10); + token = tryFileLock(databaseOpenLockPath(dbPath)); + } return token; } diff --git a/dataLayer/schema.ts b/dataLayer/schema.ts index 57d8bed63c..971f065336 100644 --- a/dataLayer/schema.ts +++ b/dataLayer/schema.ts @@ -64,7 +64,7 @@ export async function createSchema(schemaCreateObject: any) { // Await cross-worker propagation so the new schema is visible on every worker before // returning success — otherwise describe_all on a lagging worker reports it missing (#1497). await signalling.signalSchemaChange( - new SchemaEventMsg(process.pid, schemaCreateObject.operation, schemaCreateObject.schema) + new SchemaEventMsg(process.pid, schemaCreateObject.operation ?? hdbTerms.OPERATIONS_ENUM.CREATE_SCHEMA, schemaCreateObject.schema) ); return schemaStructure; @@ -189,7 +189,7 @@ export async function dropSchema(dropSchemaObject: any) { // Await cross-worker propagation before returning success so no worker keeps serving the // dropped schema (#1497). await signalling.signalSchemaChange( - new SchemaEventMsg(process.pid, dropSchemaObject.operation, dropSchemaObject.schema) + new SchemaEventMsg(process.pid, dropSchemaObject.operation ?? hdbTerms.OPERATIONS_ENUM.DROP_SCHEMA, dropSchemaObject.schema) ); let response = await server.replication.replicateOperation(dropSchemaObject); diff --git a/resources/databases.ts b/resources/databases.ts index 39bd31a889..358eded23f 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -327,7 +327,7 @@ function openRocksDatabase(path: string, options: RocksDatabaseOptions & { dupSo } mkdirSync(path, { recursive: true }); } - const openLock = acquireDatabaseOpenLock(path); + const openLock = !options.name && !isReadOnlyMode() ? acquireDatabaseOpenLock(path) : 0; try { let db: RocksRootDatabase; if (options.dupSort) { @@ -343,7 +343,7 @@ function openRocksDatabase(path: string, options: RocksDatabaseOptions & { dupSo db.env = {}; return db; } finally { - releaseDatabaseOpenLock(openLock); + if (openLock) releaseDatabaseOpenLock(openLock); } } From f7306e9edb583510542b479c7d9def7c349361b3 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Fri, 14 Aug 2026 07:17:13 -0600 Subject: [PATCH 5/5] test(storage): cover concurrent database opens Co-Authored-By: GPT-5 Codex --- dataLayer/restoreMarker.ts | 12 +- dataLayer/schema.ts | 12 +- .../apiTests/terminology.test.mjs | 16 +-- resources/databases.ts | 122 ++++++++++++------ unitTests/dataLayer/restoreMarker.test.js | 52 ++++++++ 5 files changed, 161 insertions(+), 53 deletions(-) diff --git a/dataLayer/restoreMarker.ts b/dataLayer/restoreMarker.ts index 3c0d7535cd..2410b079bc 100644 --- a/dataLayer/restoreMarker.ts +++ b/dataLayer/restoreMarker.ts @@ -21,7 +21,7 @@ import { tryFileLock, fileLockRelease } from '@harperfast/rocksdb-js'; * mutate the same directory concurrently. * * Restore metadata lives in an isolated `` `restore` `` directory *beside* the database directory - * (never inside it, since a restore purges the destination). Each database's two files are keyed by + * (never inside it, since a restore purges the destination). Each database's metadata files are keyed by * a hash of the database directory name rather than being suffixed onto the name itself. That keeps * them out of the database-name namespace — a legal database literally named `orders.restoring` * would otherwise be mistaken for the restore marker of `orders`, and a 250-character name plus a @@ -44,6 +44,8 @@ import { tryFileLock, fileLockRelease } from '@harperfast/rocksdb-js'; * successfully, while still holding the lock. Its *existence* means "a restore started and has * not finished successfully". Its first line records the database directory name so the startup * scan can map a marker back to the database it blocks without decoding the hashed key. + * - `/.open` — a short-lived mutex around root RocksDB opens and destroys. The main thread tracks + * managed worker holders and releases their token if the worker exits before it can clean up. */ // The backtick makes this an illegal database name (schemaRegex rejects `/` and backtick only), so @@ -87,19 +89,19 @@ function databaseOpenLockPath(dbPath: string): string { } /** - * Serialize native RocksDB opens for one database across worker threads. rocksdb-js normally - * shares opened handles process-wide, but two cold worker-thread opens can still race before - * that registry entry exists and contend for RocksDB's process-local LOCK file. + * Serialize root RocksDB opens and destroys for one database. rocksdb-js releases its registry + * entry before `DestroyDB`, so a concurrent reopen must wait until destruction is complete. */ export function acquireDatabaseOpenLock(dbPath: string, maxWaitMilliseconds = 30_000): number { mkdirSync(restoreMetaDir(dbPath), { recursive: true }); let token = tryFileLock(databaseOpenLockPath(dbPath)); const startedAt = Date.now(); - const sleeper = new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT)); + let sleeper: Int32Array; while (token === 0) { if (Date.now() - startedAt >= maxWaitMilliseconds) { throw new Error(`Timed out acquiring RocksDB open lock for ${dbPath}`); } + sleeper ??= new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT)); Atomics.wait(sleeper, 0, 0, 10); token = tryFileLock(databaseOpenLockPath(dbPath)); } diff --git a/dataLayer/schema.ts b/dataLayer/schema.ts index 971f065336..030761ecfc 100644 --- a/dataLayer/schema.ts +++ b/dataLayer/schema.ts @@ -64,7 +64,11 @@ export async function createSchema(schemaCreateObject: any) { // Await cross-worker propagation so the new schema is visible on every worker before // returning success — otherwise describe_all on a lagging worker reports it missing (#1497). await signalling.signalSchemaChange( - new SchemaEventMsg(process.pid, schemaCreateObject.operation ?? hdbTerms.OPERATIONS_ENUM.CREATE_SCHEMA, schemaCreateObject.schema) + new SchemaEventMsg( + process.pid, + schemaCreateObject.operation ?? hdbTerms.OPERATIONS_ENUM.CREATE_SCHEMA, + schemaCreateObject.schema + ) ); return schemaStructure; @@ -189,7 +193,11 @@ export async function dropSchema(dropSchemaObject: any) { // Await cross-worker propagation before returning success so no worker keeps serving the // dropped schema (#1497). await signalling.signalSchemaChange( - new SchemaEventMsg(process.pid, dropSchemaObject.operation ?? hdbTerms.OPERATIONS_ENUM.DROP_SCHEMA, dropSchemaObject.schema) + new SchemaEventMsg( + process.pid, + dropSchemaObject.operation ?? hdbTerms.OPERATIONS_ENUM.DROP_SCHEMA, + dropSchemaObject.schema + ) ); let response = await server.replication.replicateOperation(dropSchemaObject); diff --git a/integrationTests/apiTests/terminology.test.mjs b/integrationTests/apiTests/terminology.test.mjs index 58a49800b1..2ff4063c42 100644 --- a/integrationTests/apiTests/terminology.test.mjs +++ b/integrationTests/apiTests/terminology.test.mjs @@ -15,7 +15,7 @@ import { fileURLToPath } from 'node:url'; import path from 'node:path'; import { startHarper, teardownHarper } from '@harperfast/integration-testing'; import { createApiClient } from './utils/client.mjs'; -import { awaitJobCompleted, waitFor } from './utils/operations.mjs'; +import { awaitJobCompleted } from './utils/operations.mjs'; // Resolve the CSV fixture path relative to this file so Harper can read it. const SUPPLIERS_CSV = path.join(path.dirname(fileURLToPath(import.meta.url)), 'data/Suppliers.csv'); @@ -403,15 +403,11 @@ suite('Terminology aliases (database / primary_key)', (ctx) => { }); test('drop_database with database param', async () => { - // The preceding drop_table signals syncSchemaMetadata across worker threads, which - // briefly reopens all schema RocksDB files including tuckerdoodle. If drop_database - // races that reopen it gets a "No locks available" LOCK error. Poll until the drop - // succeeds (workers release the lock quickly once their getDatabases pass finishes). - const r = await waitFor(() => client.req().send({ operation: 'drop_database', database: 'tuckerdoodle' }), { - until: (res) => res?.body?.message != null, - timeoutSeconds: 10, - }); - assert.equal(r?.body?.message, "successfully deleted 'tuckerdoodle'", r?.text); + await client + .req() + .send({ operation: 'drop_database', database: 'tuckerdoodle' }) + .expect((r) => assert.equal(r.body.message, "successfully deleted 'tuckerdoodle'", r.text)) + .expect(200); }); // ── async job operations ──────────────────────────────────────────────── diff --git a/resources/databases.ts b/resources/databases.ts index 358eded23f..601f1d45ac 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -24,7 +24,7 @@ import { _assignPackageExport } from '../globals.js'; import { getIndexedValues } from '../utility/lmdb/commonUtility.ts'; import * as signalling from '../utility/signalling.ts'; import { SchemaEventMsg } from '../server/threads/itc.js'; -import { workerData } from 'worker_threads'; +import { isMainThread, workerData } from 'worker_threads'; import harperLogger from '../utility/logging/harper_logger.ts'; const { forComponent } = harperLogger; import * as manageThreads from '../server/threads/manageThreads.js'; @@ -53,6 +53,39 @@ import { type RestoreLock, } from '../dataLayer/restoreMarker.ts'; +declare const threads: { sendToThread(threadId: number, message: any): boolean }; + +const DATABASE_OPEN_LOCK_ACQUIRED = 'database_open_lock_acquired'; +const DATABASE_OPEN_LOCK_RELEASED = 'database_open_lock_released'; +const workerDatabaseOpenLocks = new Map(); + +if (isMainThread) { + manageThreads.onMessageByType(DATABASE_OPEN_LOCK_ACQUIRED, (message, port) => { + workerDatabaseOpenLocks.set(message.token, port.threadId); + }); + manageThreads.onMessageByType(DATABASE_OPEN_LOCK_RELEASED, (message) => { + workerDatabaseOpenLocks.delete(message.token); + }); + manageThreads.onThreadExit((threadId) => { + for (const [token, ownerThreadId] of workerDatabaseOpenLocks) { + if (ownerThreadId !== threadId) continue; + releaseDatabaseOpenLock(token); + workerDatabaseOpenLocks.delete(token); + } + }); +} + +function acquireTrackedDatabaseOpenLock(path: string): number { + const token = acquireDatabaseOpenLock(path); + if (!isMainThread) threads.sendToThread(0, { type: DATABASE_OPEN_LOCK_ACQUIRED, token }); + return token; +} + +function releaseTrackedDatabaseOpenLock(token: number): void { + releaseDatabaseOpenLock(token); + if (!isMainThread) threads.sendToThread(0, { type: DATABASE_OPEN_LOCK_RELEASED, token }); +} + /** * Check if Harper is running in read-only mode. * Read-only mode can be enabled via: @@ -327,7 +360,7 @@ function openRocksDatabase(path: string, options: RocksDatabaseOptions & { dupSo } mkdirSync(path, { recursive: true }); } - const openLock = !options.name && !isReadOnlyMode() ? acquireDatabaseOpenLock(path) : 0; + const openLock = !isReadOnlyMode() ? acquireTrackedDatabaseOpenLock(path) : 0; try { let db: RocksRootDatabase; if (options.dupSort) { @@ -343,7 +376,7 @@ function openRocksDatabase(path: string, options: RocksDatabaseOptions & { dupSo db.env = {}; return db; } finally { - if (openLock) releaseDatabaseOpenLock(openLock); + if (openLock) releaseTrackedDatabaseOpenLock(openLock); } } @@ -1221,6 +1254,13 @@ function lockDatabaseForDrop(dbPath: string, databaseName: string, held: Restore held.push(lock); } +type DatabaseOpenLock = { dbPath: string; token: number }; + +function lockDatabaseOpenForDrop(dbPath: string, held: DatabaseOpenLock[]): void { + if (held.some((lock) => lock.dbPath === dbPath)) return; + held.push({ dbPath, token: acquireTrackedDatabaseOpenLock(dbPath) }); +} + /** * Delete the database * @param databaseName @@ -1236,53 +1276,63 @@ export async function dropDatabase(databaseName) { // (before writing its marker), so both operations serialize on this one primitive rather than on // a check-then-act marker probe. Released in the finally below. const restoreLocks: RestoreLock[] = []; + const openLocks: DatabaseOpenLock[] = []; try { - for (const tableName in dbTables) { - const table = dbTables[tableName]; - rootStore = table.primaryStore.rootStore; - if (rootStore instanceof RocksDatabase) lockDatabaseForDrop(rootStore.path, databaseName, restoreLocks); - lmdbDatabaseEnvs.delete(rootStore.path); - rocksdbDatabaseEnvs.delete(rootStore.path); - } + try { + for (const tableName in dbTables) { + const table = dbTables[tableName]; + rootStore = table.primaryStore.rootStore; + if (rootStore instanceof RocksDatabase) { + lockDatabaseForDrop(rootStore.path, databaseName, restoreLocks); + lockDatabaseOpenForDrop(rootStore.path, openLocks); + } + lmdbDatabaseEnvs.delete(rootStore.path); + rocksdbDatabaseEnvs.delete(rootStore.path); + } - for (const tableName in dbTables) { - databaseEventsEmitter.emit('dropTable', tableName, databaseName); - } + for (const tableName in dbTables) { + databaseEventsEmitter.emit('dropTable', tableName, databaseName); + } - if (databaseName === 'data') { - for (const tableName in tables) { - delete tables[tableName]; + if (databaseName === 'data') { + for (const tableName in tables) { + delete tables[tableName]; + } + delete tables[DEFINED_TABLES]; } - delete tables[DEFINED_TABLES]; - } - delete databases[databaseName]; + delete databases[databaseName]; - databaseEventsEmitter.emit('dropDatabase', databaseName); + databaseEventsEmitter.emit('dropDatabase', databaseName); - if (rootStore) { - if (rootStore.status === 'open') { + if (rootStore) { + if (rootStore.status === 'open') { + if (rootStore instanceof RocksDatabase) { + rootStore.close(); + rootStore.destroy(); + } else { + await rootStore.close(); + await unlink(rootStore.path); + } + } + } else { + rootStore = database({ database: databaseName, table: null }); + // a tableless database resolves its root store here rather than in the loop above, so take + // the drop lock now (still before any destructive step) + if (rootStore instanceof RocksDatabase) { + lockDatabaseForDrop(rootStore.path, databaseName, restoreLocks); + lockDatabaseOpenForDrop(rootStore.path, openLocks); + } if (rootStore instanceof RocksDatabase) { rootStore.close(); rootStore.destroy(); - } else { + } else if (rootStore.status === 'open') { await rootStore.close(); await unlink(rootStore.path); } } - } else { - rootStore = database({ database: databaseName, table: null }); - // a tableless database resolves its root store here rather than in the loop above, so take - // the drop lock now (still before any destructive step) - if (rootStore instanceof RocksDatabase) lockDatabaseForDrop(rootStore.path, databaseName, restoreLocks); - if (rootStore instanceof RocksDatabase) { - rootStore.close(); - rootStore.destroy(); - } else if (rootStore.status === 'open') { - await rootStore.close(); - await unlink(rootStore.path); - } + } finally { + for (const lock of openLocks) releaseTrackedDatabaseOpenLock(lock.token); } - await deleteRootBlobPathsForDB(rootStore); } finally { for (const lock of restoreLocks) releaseRestoreLock(lock); diff --git a/unitTests/dataLayer/restoreMarker.test.js b/unitTests/dataLayer/restoreMarker.test.js index f2a1948ab0..aedbcf85c7 100644 --- a/unitTests/dataLayer/restoreMarker.test.js +++ b/unitTests/dataLayer/restoreMarker.test.js @@ -4,12 +4,15 @@ const assert = require('node:assert'); const { existsSync, mkdtempSync, mkdirSync, rmSync, writeFileSync } = require('node:fs'); const { basename, dirname, join } = require('node:path'); const { tmpdir } = require('node:os'); +const { Worker } = require('node:worker_threads'); const { tryFileLock, fileLockRelease } = require('@harperfast/rocksdb-js'); const { beginRestore, completeRestore, abandonRestore, acquireRestoreLock, + acquireDatabaseOpenLock, + releaseDatabaseOpenLock, releaseRestoreLock, clearRestoreMarker, checkRestoreState, @@ -34,6 +37,35 @@ describe('restoreMarker', function () { rmSync(tempDir, { recursive: true, force: true }); }); + function acquireOpenLockInWorker(maxWaitMilliseconds) { + const worker = new Worker( + `const { parentPort, workerData } = require('node:worker_threads'); + const { acquireDatabaseOpenLock, releaseDatabaseOpenLock } = require(workerData.modulePath); + try { + const token = acquireDatabaseOpenLock(workerData.dbPath, workerData.maxWaitMilliseconds); + parentPort.postMessage({ acquired: true }); + releaseDatabaseOpenLock(token); + } catch (error) { + parentPort.postMessage({ message: error.message }); + }`, + { + eval: true, + workerData: { + dbPath, + maxWaitMilliseconds, + modulePath: require.resolve('#src/dataLayer/restoreMarker'), + }, + } + ); + return { + worker, + result: new Promise((resolve, reject) => { + worker.once('message', resolve); + worker.once('error', reject); + }), + }; + } + describe('paths', function () { it('keeps restore metadata in an isolated sibling directory, out of the database-name namespace', function () { const metaDir = restoreMetaDir(dbPath); @@ -210,6 +242,26 @@ describe('restoreMarker', function () { }); }); + describe('database open lock', function () { + it('excludes a concurrent worker-thread open until the first open completes', async function () { + const token = acquireDatabaseOpenLock(dbPath); + const blocked = acquireOpenLockInWorker(25); + try { + const result = await blocked.result; + assert.deepStrictEqual(result, { message: `Timed out acquiring RocksDB open lock for ${dbPath}` }); + } finally { + releaseDatabaseOpenLock(token); + await blocked.worker.terminate(); + } + const available = acquireOpenLockInWorker(25); + try { + assert.deepStrictEqual(await available.result, { acquired: true }); + } finally { + await available.worker.terminate(); + } + }); + }); + describe('clearRestoreMarker', function () { it('removes a leftover marker and releases the lock', function () { abandonRestore(beginRestore(dbPath)); // leftover incomplete marker