diff --git a/DESIGN.md b/DESIGN.md index 8b8a19dcbd..42dae75efd 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -203,6 +203,10 @@ interrupted drop instead of resurrecting the table. Without this, surviving cata silently re-opened with create-if-missing on the next start, which resurrects "deleted" tables (with their data, if the column families were never actually removed). +On RocksDB, treating a redundant `dropSync()` as success is safe only after re-listing the live +column families: both the drop and recovery paths remove catalog rows only when none still begin +with `T/`. + ## MCP protocol surface (`components/mcp/`) The MCP Streamable-HTTP transport (spec `2025-06-18`) is served at `/mcp` under **two profiles**: an diff --git a/components/Scope.ts b/components/Scope.ts index b34599df00..c663d74572 100644 --- a/components/Scope.ts +++ b/components/Scope.ts @@ -474,6 +474,8 @@ export class Scope extends EventEmitter { }) .finally(() => pendingOperations.delete(tracked)); pendingOperations.add(tracked); + // Keep the tracked rejection available to waitForInitialLoads(). + void tracked.catch(() => {}); } }; @@ -487,7 +489,7 @@ export class Scope extends EventEmitter { // Track this promise so the component loader can await it this.#pendingInitialLoads.add(initialLoadPromise); - initialLoadPromise.finally(() => this.#pendingInitialLoads.delete(initialLoadPromise)); + void initialLoadPromise.finally(() => this.#pendingInitialLoads.delete(initialLoadPromise)).catch(() => {}); return wrapped; }; diff --git a/resources/Table.ts b/resources/Table.ts index c96c7e2849..5148188da9 100644 --- a/resources/Table.ts +++ b/resources/Table.ts @@ -403,6 +403,7 @@ export function makeTable(options) { // new ones (droppingTable) once a drop has actually started. const pendingSourceCommits = new Set>(); let droppingTable = false; + const tableIsDropping = () => droppingTable || (dbisDb as any)?.getSync?.(tableName + '/')?.dropping === true; let createdTimeProperty: Attribute | undefined, updatedTimeProperty: Attribute | undefined, expiresAtProperty: Attribute | undefined; @@ -1475,21 +1476,32 @@ export function makeTable(options) { while (!rootStore.tryLock('update-attributes')) {} let removed = false; try { - for (const attribute of attributes) { - const index = indices[attribute.name]; - if (index) + const currentPrimary = (dbisDb as any).getSync(TableResource.tableName + '/'); + if (!currentPrimary?.dropping) { + removed = false; + } else { + const columnPrefix = TableResource.tableName + '/'; + for (const columnName of (rootStore as any).columns) { + if (!columnName.startsWith(columnPrefix)) continue; + const columnStore = new RocksDatabase(rootStore.path, { name: columnName }).open(); try { - index.dropSync(); + columnStore.dropSync(); } catch (error) { ignoreAlreadyDropped(error); + } finally { + columnStore.close(); } + } + const remainingColumnFamilies = (rootStore as any).columns.filter((columnName) => + columnName.startsWith(columnPrefix) + ); + if (remainingColumnFamilies.length) { + throw new Error( + `Column families remain after drop of table ${databaseName}.${tableName}: ${remainingColumnFamilies.join(', ')}` + ); + } + removed = removeTombstonedCatalog(); } - try { - primaryStore.dropSync(); - } catch (error) { - ignoreAlreadyDropped(error); - } - removed = removeTombstonedCatalog(); } finally { rootStore.unlock('update-attributes'); } @@ -5969,6 +5981,10 @@ export function makeTable(options) { nodeName: 'source', commit: (txnTime, existingEntry, _retry, transaction: any) => { sourceWrite.skipped = false; // reset on each retry; cleanup happens after commit if still true + if (tableIsDropping()) { + sourceWrite.skipped = true; + return; + } if (existingEntry?.version !== existingVersion) { // don't do anything if the version has changed sourceWrite.skipped = true; @@ -6090,7 +6106,7 @@ export function makeTable(options) { TableResource.userEmbedders ); if (embedBefore) await embedBefore(); - if (droppingTable) { + if (tableIsDropping()) { // Re-check right before staging the write: dropTable() may have started // while we were awaiting the embed step above (harper#1381). sourceContext.transaction.abort(); diff --git a/resources/databases.ts b/resources/databases.ts index 55cd4fba2b..e83cc838c9 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -773,6 +773,9 @@ function initStores( continue; } const generation = tableDef.primary?.dropGeneration; + // A tombstone must remove an already-loaded class even when its storage + // cleanup fails; otherwise this worker can keep serving a dropped table. + definedTables?.delete(tableName); const failedAttempts = getInterruptedDropAttempts(path, tableName, generation); if (failedAttempts < MAX_INTERRUPTED_DROP_ATTEMPTS) { try { @@ -784,7 +787,6 @@ function initStores( // place that sweeps), the prior generation's entry would otherwise never // be cleared. clearInterruptedDropEntries(path, tableName); - definedTables?.delete(tableName); } catch (error) { const attempt = failedAttempts + 1; setInterruptedDropAttempts(path, tableName, generation, attempt); @@ -2292,8 +2294,9 @@ async function runIndexing(Table, attributes, indicesToRemove) { function completeInterruptedDrop(rootStore, attributesDbi, databaseName: string, tableName: string) { logger.debug(`Completing interrupted drop of table ${databaseName}.${tableName}`); if (rootStore instanceof RocksDatabase) { + const columnPrefix = tableName + '/'; for (const columnName of (rootStore as any).columns) { - if (columnName.startsWith(tableName + '/')) { + if (columnName.startsWith(columnPrefix)) { const columnStore = openRocksDatabase(rootStore.path, { name: columnName }); try { columnStore.dropSync(); @@ -2304,6 +2307,14 @@ function completeInterruptedDrop(rootStore, attributesDbi, databaseName: string, } } } + const remainingColumnFamilies = (rootStore as any).columns.filter((columnName) => + columnName.startsWith(columnPrefix) + ); + if (remainingColumnFamilies.length) { + throw new Error( + `Column families remain after interrupted drop of table ${databaseName}.${tableName}: ${remainingColumnFamilies.join(', ')}` + ); + } } else { // LMDB reuses an existing named sub-database on open, so the stores must // be dropped too; removing only the catalog rows would let a same-name diff --git a/unitTests/components/Scope.test.js b/unitTests/components/Scope.test.js index 84aa125d08..d5b8ba6197 100644 --- a/unitTests/components/Scope.test.js +++ b/unitTests/components/Scope.test.js @@ -422,6 +422,74 @@ describe('Scope', () => { await scope.close(); }); + it('reports an async entry-handler failure without an unhandled rejection', async () => { + writeFileSync(this.configFilePath, stringify({ [this.pluginName]: { files: 'test.js' } })); + const scope = new Scope( + this.appName, + this.pluginName, + this.directory, + this.configFilePath, + this.resources, + this.server + ); + await scope.ready; + const unhandled = []; + const recordUnhandled = (error) => unhandled.push(error); + process.on('unhandledRejection', recordUnhandled); + try { + let handlerStarted = false; + let failHandler; + const handlerFailure = new Promise((_, reject) => { + failHandler = reject; + }); + scope.handleEntry(async () => { + handlerStarted = true; + await handlerFailure; + }); + const initialLoads = scope.waitForInitialLoads(); + await waitFor(() => handlerStarted); + failHandler(new Error('injected entry handler failure')); + await assert.rejects(initialLoads, /injected entry handler failure/); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(unhandled, [], 'the handled load failure must not produce an unhandled rejection'); + } finally { + process.off('unhandledRejection', recordUnhandled); + await scope.close(); + } + }); + + it('handles an async entry-handler failure after initial load', async () => { + writeFileSync(this.configFilePath, stringify({ [this.pluginName]: { files: 'test.js' } })); + const scope = new Scope( + this.appName, + this.pluginName, + this.directory, + this.configFilePath, + this.resources, + this.server + ); + await scope.ready; + const unhandled = []; + const recordUnhandled = (error) => unhandled.push(error); + process.on('unhandledRejection', recordUnhandled); + try { + let handlerFailed = false; + const entryHandler = scope.handleEntry(async (entry) => { + if (entry.eventType !== 'change') return; + handlerFailed = true; + throw new Error('injected post-load entry handler failure'); + }); + await scope.waitForInitialLoads(); + entryHandler.emit('all', { eventType: 'change' }); + await waitFor(() => handlerFailed); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(unhandled, [], 'a post-load failure must not produce an unhandled rejection'); + } finally { + process.off('unhandledRejection', recordUnhandled); + await scope.close(); + } + }); + it('should not create entry handler from options change before handleEntry is called (RE-8)', async () => { // Reproduce the race in RE-8: OptionsWatcher fires a `change` event for the // `files` key BEFORE handleApplication (and thus handleEntry) runs. In the diff --git a/unitTests/resources/dropTableGhost.test.js b/unitTests/resources/dropTableGhost.test.js index 646ec41ab2..9bf42730ba 100644 --- a/unitTests/resources/dropTableGhost.test.js +++ b/unitTests/resources/dropTableGhost.test.js @@ -4,6 +4,7 @@ const { setupTestDBPath } = require('../testUtils'); const { table, database, databases, getDatabases, resetDatabases } = require('#src/resources/databases'); const { setMainIsWorker } = require('#js/server/threads/manageThreads'); const harperLogger = require('#src/utility/logging/harper_logger'); +const { RocksDatabase } = require('@harperfast/rocksdb-js'); const TEST_DB = 'test'; @@ -27,6 +28,16 @@ function getDbisDb() { // test engine-agnostic under `test:unit` and `test:unit:lmdb`. Returns a // restore function. function stubFailingDrop(store, error) { + if (store instanceof RocksDatabase) { + const rocksPrototype = Object.getPrototypeOf(Object.getPrototypeOf(store)); + const originalDropSync = rocksPrototype.dropSync; + rocksPrototype.dropSync = () => { + throw error; + }; + return () => { + rocksPrototype.dropSync = originalDropSync; + }; + } const original = { drop: store.drop, dropSync: store.dropSync }; store.dropSync = () => { throw error; @@ -111,18 +122,15 @@ describe('dropTable ghost regression', () => { }); it('tolerates an already-dropped column family and completes the drop', async function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return this.skip(); const Raced = defineTable('GhostAlreadyDropped'); await Raced.put({ id: 1, str: 'data' }); // A concurrent worker already dropped the shared column family (drops are // broadcast to every thread), so the storage engine reports the redundant // drop as "Column family already dropped!". The family being gone is the // intended outcome, so the drop operation must succeed rather than fail. - const restore = stubFailingDrop(Raced.primaryStore, new Error('Invalid argument: Column family already dropped!')); - try { - await Raced.dropTable(); - } finally { - restore(); - } + Raced.primaryStore.dropSync(); + await Raced.dropTable(); // the table is removed from the live schema... assert.equal( databases[TEST_DB]?.GhostAlreadyDropped, @@ -138,6 +146,83 @@ describe('dropTable ghost regression', () => { ); }); + it('keeps a live-drop tombstone when an already-dropped response leaves its column family present', async function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return this.skip(); + const TABLE = `GhostLiveStillPresent_${process.pid}_${Date.now()}`; + const Stuck = defineTable(TABLE); + await Stuck.put({ id: 1, str: 'data' }); + const restore = stubFailingDrop(Stuck.primaryStore, new Error('Invalid argument: Column family already dropped!')); + try { + await assert.rejects(() => Stuck.dropTable(), /Column families remain after drop/); + } finally { + restore(); + } + + assert.equal(getDbisDb().getSync(`${TABLE}/`)?.dropping, true, 'the live-drop tombstone must survive'); + const Fresh = defineTable(TABLE); + await Fresh.put({ id: 2, str: 'fresh' }); + assert.equal(await Fresh.get(1), undefined, 'the recovery must remove the old column family before recreation'); + await Fresh.dropTable(); + }); + + it('drops an orphaned RocksDB column family during a live drop', async function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return this.skip(); + const TABLE = `GhostOrphanedColumn_${process.pid}_${Date.now()}`; + const Orphaned = defineTable(TABLE); + await Orphaned.put({ id: 1, str: 'data' }); + const rootStore = Orphaned.primaryStore.rootStore; + const orphan = new RocksDatabase(rootStore.path, { name: `${TABLE}/formerIndex` }).open(); + orphan.putSync('orphan', 'data'); + orphan.close(); + + await Orphaned.dropTable(); + assert.equal( + rootStore.columns.some((columnName) => columnName.startsWith(`${TABLE}/`)), + false, + 'a live drop must remove every table-prefixed column family, including orphaned indexes' + ); + }); + + it('keeps a tombstone when an already-dropped response leaves its column family present', async function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return this.skip(); + const TABLE = `GhostStillPresent_${process.pid}_${Date.now()}`; + const Stuck = defineTable(TABLE); + await Stuck.put({ id: 1, str: 'data' }); + const dbisDb = getDbisDb(); + const meta = dbisDb.getSync(`${TABLE}/`); + meta.dropping = true; + await dbisDb.put(`${TABLE}/`, meta); + assert.ok(Stuck.primaryStore.rootStore.columns.includes(`${TABLE}/`), 'the test column family should exist'); + + const rocksPrototype = Object.getPrototypeOf(Stuck.primaryStore); + const originalDropSync = rocksPrototype.dropSync; + rocksPrototype.dropSync = () => { + throw new Error('Invalid argument: Column family already dropped!'); + }; + try { + resetDatabases(); + getDatabases(); + } finally { + rocksPrototype.dropSync = originalDropSync; + } + assert.ok( + database({ database: TEST_DB, table: null }).columns.includes(`${TABLE}/`), + 'the mocked redundant drop should leave the column family present' + ); + + assert.equal( + getDbisDb().getSync(`${TABLE}/`)?.dropping, + true, + 'the tombstone must remain until the column family is actually gone' + ); + assert.equal(databases[TEST_DB]?.[TABLE], undefined, 'a failed cleanup must not leave a live table class'); + + const Fresh = defineTable(TABLE); + await Fresh.put({ id: 2, str: 'fresh' }); + assert.equal(await Fresh.get(1), undefined, 'the retry must remove the old column family before recreation'); + await Fresh.dropTable(); + }); + it('does not clobber a same-name table created during a tolerated drop race', async function () { // Exercises the RocksDB drop path (synchronous dropSync under the exclusive // lock) and its tombstone-guarded catalog removal; the LMDB path keeps the @@ -146,21 +231,22 @@ describe('dropTable ghost regression', () => { const Raced = defineTable('GhostDropRaceCreate'); await Raced.put({ id: 1, str: 'data' }); const dbisDb = getDbisDb(); - const originalDrop = Raced.primaryStore.dropSync; + const rocksPrototype = Object.getPrototypeOf(Object.getPrototypeOf(Raced.primaryStore)); + const originalDrop = rocksPrototype.dropSync; // Simulate the race: while this drop holds the lock, the catalog already // carries a fresh, non-tombstoned row for a same-name table (as a concurrent // create's completeInterruptedDrop would have left it). The guard must see // the fresh row on its re-read and skip removal so it is not clobbered. - Raced.primaryStore.dropSync = () => { + rocksPrototype.dropSync = () => { const fresh = { ...dbisDb.getSync('GhostDropRaceCreate/') }; delete fresh.dropping; dbisDb.putSync('GhostDropRaceCreate/', fresh); throw new Error('Column family already dropped!'); }; try { - await Raced.dropTable(); + await assert.rejects(() => Raced.dropTable(), /Column families remain after drop/); } finally { - Raced.primaryStore.dropSync = originalDrop; + rocksPrototype.dropSync = originalDrop; } // the new table's catalog row must survive - cleanup only runs when this // drop's own tombstone is still the live primary row