Prevent job wedges on runtime database opens #2169
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -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 { isMainThread, workerData } from 'worker_threads'; | ||||||
|
Contributor
There was a problem hiding this comment. According to the repository style guide, Node builtins must use the
Suggested change
References
|
||||||
| import harperLogger from '../utility/logging/harper_logger.ts'; | ||||||
| const { forComponent } = harperLogger; | ||||||
| import * as manageThreads from '../server/threads/manageThreads.js'; | ||||||
|
|
@@ -43,14 +43,49 @@ import { resolveRocksMemoryConfig } from '../utility/rocksMemoryConfig.ts'; | |||||
| import { isProcessRunning } from '../utility/processManagement/processManagement.js'; | ||||||
| import { | ||||||
| acquireRestoreLock, | ||||||
| acquireDatabaseOpenLock, | ||||||
| checkRestoreState, | ||||||
| releaseDatabaseOpenLock, | ||||||
| releaseRestoreLock, | ||||||
| restoreMarkerPresent, | ||||||
| scanBlockedRestores, | ||||||
| RESTORE_META_DIR, | ||||||
| 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<number, number>(); | ||||||
|
|
||||||
| 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: | ||||||
|
|
@@ -325,19 +360,24 @@ function openRocksDatabase(path: string, options: RocksDatabaseOptions & { dupSo | |||||
| } | ||||||
| mkdirSync(path, { recursive: true }); | ||||||
| } | ||||||
| const openLock = !isReadOnlyMode() ? acquireTrackedDatabaseOpenLock(path) : 0; | ||||||
| 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 { | ||||||
| if (openLock) releaseTrackedDatabaseOpenLock(openLock); | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| const lmdbDatabaseEnvs = new Map<string, LMDBRootDatabase>(); | ||||||
|
|
@@ -1214,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 | ||||||
|
|
@@ -1229,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 { | ||||||
| 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); | ||||||
| } | ||||||
|
|
||||||
| if (databaseName === 'data') { | ||||||
| for (const tableName in tables) { | ||||||
| delete tables[tableName]; | ||||||
| } | ||||||
| delete tables[DEFINED_TABLES]; | ||||||
| } | ||||||
| delete databases[databaseName]; | ||||||
|
|
||||||
| databaseEventsEmitter.emit('dropDatabase', databaseName); | ||||||
|
|
||||||
| 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 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); | ||||||
|
|
||||||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Main thread can block its own event loop for up to 30s here
What:
acquireDatabaseOpenLockretries with a blockingAtomics.wait(sleeper, 0, 0, 10)in a tight loop (up tomaxWaitMilliseconds, default 30s) with no yield back to the event loop between iterations.resources/databases.ts:363calls this viaacquireTrackedDatabaseOpenLockfor every root open (!options.name) with noisMainThreadexemption, unlike the message-tracking code right above it which is guarded byif (isMainThread)/if (!isMainThread).The main thread does open root databases directly, not just workers:
bin/run.ts→startHTTPThreads()→loadRootComponents()→getTables()/getDatabases()on initial boot, and — more concerningly —manageThreads.js'srestartWorkers()(if (isMainThread) { ... await loadRootComponents(); }, lines ~456-471) re-runs this on every component/app deploy or reload, concurrently with worker threads restarting and opening the same runtime-created database for the first time. That's exactly the contention window this PR is designed to serialize.Why it matters: If the main thread loses the race for this lock during a restart, it synchronously blocks in
Atomics.waitfor up to 30 seconds. The main thread is the process's IPC hub —manageThreads.js's message dispatch, the newonThreadExithandler that releases a worker's stranded open-lock token (the very mechanism this PR adds), and all other worker coordination stop dead for that window. A worker that dies while holding the lock can't be cleaned up until the main thread's own poll loop finishes, and no other IPC traffic (including HTTP routing setup, shutdown signals, etc.) can be serviced meanwhile. This risks reintroducing a wedge on the main thread instead of a worker.Suggested fix: Skip the lock (or use a non-blocking/async wait, e.g. polling via
setTimeout/microtask so the event loop keeps running) whenisMainThreadis true, mirroring the existingisMainThreadbranching already used for the message-tracking calls in this same function.