From 1c30b2b7275e9b92d59374f18219cb2ab18ee982 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Fri, 28 Aug 2026 10:49:53 -0600 Subject: [PATCH 1/5] fix(components): keep an async entry-handler failure in the initial-load result Scope.handleEntry's wrapper removed a rejected async operation from pendingOperations as soon as it settled, so a failure that settled before the entry handler's `ready` event was gone by the time the initial-load result was computed: waitForInitialLoads() resolved, the component reported a successful load, and the rejection was left with no handler (the "unhandledRejection in worker thread" lines when a blob table's interrupted drop could not be completed, harper#1381). Record the first failure seen before the initial load settles and fail the load with it after every pending operation has finished; never leave the bookkeeping chains rejecting on their own. Also adds a (skipped) worker-thread regression test that reproduces the drop_table race behind the blob.test.mjs failures: a cross-worker source-fill commit landing on a just-dropped column family latches a fatal RocksDB background error under the binding's default parallel OCC validation. Refs #1381 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01R38MPckiquqdMqYbn96qGD --- components/Scope.ts | 29 +++-- unitTests/components/Scope.test.js | 33 ++++++ .../dropTableCrossWorkerWrite-worker.js | 67 +++++++++++ .../dropTableCrossWorkerWrite.test.js | 104 ++++++++++++++++++ 4 files changed, 226 insertions(+), 7 deletions(-) create mode 100644 unitTests/resources/dropTableCrossWorkerWrite-worker.js create mode 100644 unitTests/resources/dropTableCrossWorkerWrite.test.js diff --git a/components/Scope.ts b/components/Scope.ts index b34599df00..2207ee581c 100644 --- a/components/Scope.ts +++ b/components/Scope.ts @@ -462,17 +462,27 @@ export class Scope extends EventEmitter { entryEventHandler: onEntryEventHandler ): onEntryEventHandler => { const pendingOperations = new Set>(); + // The first async failure seen before the initial load completes. A rejected operation + // cannot stay in pendingOperations as the record of that failure: one that settles before + // 'ready' fires would already be gone by the time the load result is computed, letting a + // failed load report success and leaving the rejection with nobody to observe it. + let initialLoadFailure: unknown; + let initialLoadSettled = false; const wrapped: onEntryEventHandler = (entry) => { const result = entryEventHandler(entry); if (result instanceof Promise) { - const tracked = result - .catch((error) => { + const tracked: Promise = result.then( + () => { + pendingOperations.delete(tracked); + }, + (error) => { + pendingOperations.delete(tracked); this.#logger.error?.('Error in async entry handler:', error); this.#handleError(error); - throw error; - }) - .finally(() => pendingOperations.delete(tracked)); + if (!initialLoadSettled && initialLoadFailure === undefined) initialLoadFailure = error; + } + ); pendingOperations.add(tracked); } }; @@ -482,12 +492,17 @@ export class Scope extends EventEmitter { if (pendingOperations.size > 0) { await Promise.all(pendingOperations); } + initialLoadSettled = true; + if (initialLoadFailure !== undefined) throw initialLoadFailure; targetEntryHandler.emit('initialLoadComplete'); }); - // Track this promise so the component loader can await it + // Track this promise so the component loader can await it. Its rejection is delivered + // through waitForInitialLoads(); this bookkeeping chain must settle either way, or the + // same failure escapes a second time as an unhandled rejection of the derived promise. this.#pendingInitialLoads.add(initialLoadPromise); - initialLoadPromise.finally(() => this.#pendingInitialLoads.delete(initialLoadPromise)); + const forgetInitialLoad = () => this.#pendingInitialLoads.delete(initialLoadPromise); + initialLoadPromise.then(forgetInitialLoad, forgetInitialLoad); return wrapped; }; diff --git a/unitTests/components/Scope.test.js b/unitTests/components/Scope.test.js index 84aa125d08..b5f6b2cf1d 100644 --- a/unitTests/components/Scope.test.js +++ b/unitTests/components/Scope.test.js @@ -476,6 +476,39 @@ describe('Scope', () => { await scope.close(); }); + it('reports an async entry handler failure once, through waitForInitialLoads, not as 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, + new ApplicationScope('test', this.resources, this.server) + ); + await scope.ready; + + const unhandled = []; + const collectUnhandled = (reason) => unhandled.push(reason); + process.on('unhandledRejection', collectUnhandled); + const scopeErrors = []; + scope.on('error', (error) => scopeErrors.push(error)); + try { + const failure = new Error('entry handler failed'); + scope.handleEntry(async () => { + throw failure; + }); + await assert.rejects(scope.waitForInitialLoads(), (error) => error === failure); + // Node reports a rejection nobody handled at the end of the turn it was rejected in. + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.deepEqual(scopeErrors, [failure], 'the failure is reported to the scope once'); + assert.deepEqual(unhandled, [], 'the failure must not also escape as an unhandled rejection'); + } finally { + process.off('unhandledRejection', collectUnhandled); + await scope.close(); + } + }); + describe('deploy lifecycle integration', () => { // These cases ensure that when a deploy is in flight for the parent // component, file changes from the deploy itself (extract + npm install) diff --git a/unitTests/resources/dropTableCrossWorkerWrite-worker.js b/unitTests/resources/dropTableCrossWorkerWrite-worker.js new file mode 100644 index 0000000000..d2e549d4ea --- /dev/null +++ b/unitTests/resources/dropTableCrossWorkerWrite-worker.js @@ -0,0 +1,67 @@ +'use strict'; + +require('../testUtils'); +const { parentPort } = require('node:worker_threads'); +const { setupTestDBPath } = require('../testUtils'); +const { table } = require('#src/resources/databases'); +const { createBlob } = require('#src/resources/blob'); +const { logger } = require('#src/utility/logging/logger'); +const { onMessageByType } = require('#js/server/threads/manageThreads'); + +const MESSAGE_TYPE = 'drop-table-cross-worker-test'; +const CONTROL_TYPE = 'drop-table-cross-worker-control'; +const report = (event, details = {}) => parentPort.postMessage({ type: MESSAGE_TYPE, event, ...details }); + +let TestTable; + +function runWorkerFixture() { + onMessageByType(CONTROL_TYPE, () => {}); + setupTestDBPath(); + process.on('unhandledRejection', (error) => report('unhandled-rejection', { error: error?.stack ?? String(error) })); + const originalError = logger.error; + logger.error = (...args) => { + report('logged-error', { + message: args.map((arg) => (arg instanceof Error ? arg.message : String(arg))).join(' '), + }); + return originalError?.apply(logger, args); + }; + parentPort.on('message', async (message) => { + if (message.type !== CONTROL_TYPE) return; + try { + switch (message.command) { + case 'define': + TestTable = table({ + table: message.table, + database: 'test', + audit: true, + attributes: [ + { name: 'id', isPrimaryKey: true }, + { name: 'blob', type: 'Blob' }, + ], + }); + // A blob defers the cache write's native commit until the blob file has been written, + // which is what lets the drop on the other thread land while the commit is in flight. + TestTable.sourcedFrom({ + get: async (id) => ({ id, blob: await createBlob(Buffer.alloc(100000, 1)) }), + available: () => true, + }); + report('defined'); + break; + case 'get': + // getFromSource resolves the caller before its cache write has committed + TestTable.get(message.id, {}).then( + () => report('get-resolved'), + (error) => report('get-rejected', { error: error?.stack ?? String(error) }) + ); + break; + } + } catch (error) { + report('error', { error: error?.stack ?? String(error) }); + } + }); + // keep the thread alive: manageThreads unrefs parentPort + setInterval(() => {}, 1000); + report('booted'); +} + +if (parentPort) runWorkerFixture(); diff --git a/unitTests/resources/dropTableCrossWorkerWrite.test.js b/unitTests/resources/dropTableCrossWorkerWrite.test.js new file mode 100644 index 0000000000..c573367784 --- /dev/null +++ b/unitTests/resources/dropTableCrossWorkerWrite.test.js @@ -0,0 +1,104 @@ +'use strict'; + +require('../testUtils'); +const assert = require('node:assert'); +const path = require('node:path'); +const { setupTestDBPath } = require('../testUtils'); +const { table } = require('#src/resources/databases'); +const { startWorker, onMessageByType, setMainIsWorker } = require('#js/server/threads/manageThreads'); + +const WORKER_FIXTURE = path.join(__dirname, 'dropTableCrossWorkerWrite-worker.js'); +const MESSAGE_TYPE = 'drop-table-cross-worker-test'; +const CONTROL_TYPE = 'drop-table-cross-worker-control'; +const ITERATIONS = 20; + +function defineTable(name) { + return table({ + table: name, + database: 'test', + audit: true, + attributes: [ + { name: 'id', isPrimaryKey: true }, + { name: 'blob', type: 'Blob' }, + ], + }); +} + +function startFixtureWorker() { + const queued = []; + const waiting = []; + const receive = (message) => { + if (message?.type !== MESSAGE_TYPE) return; + const waiter = waiting.shift(); + if (waiter) waiter(message); + else queued.push(message); + }; + const next = () => + queued.length ? Promise.resolve(queued.shift()) : new Promise((resolve) => waiting.push(resolve)); + const worker = startWorker(WORKER_FIXTURE, { + name: 'drop-table-cross-worker-test', + workerIndex: 1, + threadCount: 2, + autoRestart: false, + onStarted(spawned) { + spawned.on('message', receive); + }, + }); + const send = (command, details = {}) => worker.postMessage({ type: CONTROL_TYPE, command, ...details }); + const expect = async (event) => { + for (;;) { + const message = await next(); + if (message.event === event) return message; + if (message.event === 'logged-error' || message.event === 'unhandled-rejection') continue; + throw new Error(`unexpected worker event ${message.event}: ${JSON.stringify(message)}`); + } + }; + const drain = () => queued.splice(0); + return { worker, send, expect, drain }; +} + +// harper#1381: dropTable() on one thread drops the column families while another worker's +// source-fill cache write is still committing. dropTable() only drains its own thread's in-flight +// source commits, so the other worker's commit lands on the dropped column family; under +// RocksDB's default parallel OCC validation that write is admitted past conflict validation and +// fails inside the memtable insert, which latches a fatal background error on the whole +// environment ("Invalid column family specified in write batch" on every later write, including +// this drop's own catalog cleanup). Skipped until rocksdb-js serializes column-family drops +// against in-flight optimistic commits; it reproduces on the first iteration today. +describe('dropTable racing a cross-worker source-fill commit', function () { + this.timeout(120000); + let fixture; + + before(async () => { + setupTestDBPath(); + setMainIsWorker(true); + onMessageByType(MESSAGE_TYPE, () => {}); + fixture = startFixtureWorker(); + await fixture.expect('booted'); + }); + + after(async () => { + await fixture?.worker?.terminate?.(); + }); + + it.skip('leaves the storage environment writable and the catalog clean', async () => { + const Probe = defineTable('CrossDropProbe'); + for (let i = 0; i < ITERATIONS; i++) { + const name = `CrossDrop${i}`; + const Main = defineTable(name); + fixture.send('define', { table: name }); + await fixture.expect('defined'); + fixture.send('get', { id: i }); + await fixture.expect('get-resolved'); + await Main.dropTable(); + await new Promise((resolve) => setTimeout(resolve, 50)); + const workerEvents = fixture.drain().filter((message) => message.event !== 'logged-error'); + assert.deepEqual(workerEvents, [], `iteration ${i}: unexpected worker events`); + assert.doesNotThrow( + () => Probe.primaryStore.putSync('__probe__', { i }), + `iteration ${i}: the environment must still accept writes` + ); + assert.equal(Main.dbisDB.getSync(`${name}/`), undefined, `iteration ${i}: catalog rows must be removed`); + } + }); +}); From 6ee02b15ecbef45ef14f68d6779e13fc6de41c77 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 1 Sep 2026 15:44:38 -0600 Subject: [PATCH 2/5] Wait for the worker's commit to settle instead of sleeping; check the whole catalog prefix From the planning review: a fixed 50 ms wait after dropTable() could let an iteration end before the worker's late commit settled. The fixture now reports commit-settled once getFromSource releases the record lock (the seam caching.test.js already waits on), the test waits for that before probing, unexpected worker events include unhandled rejections again, and the catalog assertion scans every row under the table's prefix instead of the primary row alone. Comment trimmed to the invariant, the skip condition and the issues. Refs #1381 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013yFRAueahXUwcVM5BjKPMq --- .../dropTableCrossWorkerWrite-worker.js | 23 +++++++++++++++---- .../dropTableCrossWorkerWrite.test.js | 22 ++++++++---------- 2 files changed, 29 insertions(+), 16 deletions(-) diff --git a/unitTests/resources/dropTableCrossWorkerWrite-worker.js b/unitTests/resources/dropTableCrossWorkerWrite-worker.js index d2e549d4ea..f29b346fb5 100644 --- a/unitTests/resources/dropTableCrossWorkerWrite-worker.js +++ b/unitTests/resources/dropTableCrossWorkerWrite-worker.js @@ -3,6 +3,7 @@ require('../testUtils'); const { parentPort } = require('node:worker_threads'); const { setupTestDBPath } = require('../testUtils'); +const { waitFor } = require('../waitFor.js'); const { table } = require('#src/resources/databases'); const { createBlob } = require('#src/resources/blob'); const { logger } = require('#src/utility/logging/logger'); @@ -47,13 +48,27 @@ function runWorkerFixture() { }); report('defined'); break; - case 'get': - // getFromSource resolves the caller before its cache write has committed - TestTable.get(message.id, {}).then( - () => report('get-resolved'), + case 'get': { + const { id } = message; + // getFromSource resolves the caller before its cache write has committed, and holds the + // record lock until that write has settled either way. + TestTable.get(id, {}).then( + async () => { + report('get-resolved'); + try { + await waitFor(() => !TestTable.primaryStore.hasLock(id), { + timeout: 30000, + message: 'the source-fill cache write should settle', + }); + report('commit-settled'); + } catch (error) { + report('settle-error', { error: error?.stack ?? String(error) }); + } + }, (error) => report('get-rejected', { error: error?.stack ?? String(error) }) ); break; + } } } catch (error) { report('error', { error: error?.stack ?? String(error) }); diff --git a/unitTests/resources/dropTableCrossWorkerWrite.test.js b/unitTests/resources/dropTableCrossWorkerWrite.test.js index c573367784..a6bdfc784e 100644 --- a/unitTests/resources/dropTableCrossWorkerWrite.test.js +++ b/unitTests/resources/dropTableCrossWorkerWrite.test.js @@ -49,7 +49,7 @@ function startFixtureWorker() { for (;;) { const message = await next(); if (message.event === event) return message; - if (message.event === 'logged-error' || message.event === 'unhandled-rejection') continue; + if (message.event === 'logged-error') continue; throw new Error(`unexpected worker event ${message.event}: ${JSON.stringify(message)}`); } }; @@ -57,14 +57,11 @@ function startFixtureWorker() { return { worker, send, expect, drain }; } -// harper#1381: dropTable() on one thread drops the column families while another worker's -// source-fill cache write is still committing. dropTable() only drains its own thread's in-flight -// source commits, so the other worker's commit lands on the dropped column family; under -// RocksDB's default parallel OCC validation that write is admitted past conflict validation and -// fails inside the memtable insert, which latches a fatal background error on the whole -// environment ("Invalid column family specified in write batch" on every later write, including -// this drop's own catalog cleanup). Skipped until rocksdb-js serializes column-family drops -// against in-flight optimistic commits; it reproduces on the first iteration today. +// harper#1381: a column family must not be dropped while a commit naming it is between conflict +// validation and its write, because RocksDB latches that write's failure as a fatal background +// error on the whole environment. dropTable() drains only its own thread's source-fill commits, +// so a worker's in-flight commit can still land on the dropped family. Skipped until rocksdb-js#806 +// serializes drops against in-flight commits; on the current binding it fails on iteration 0. describe('dropTable racing a cross-worker source-fill commit', function () { this.timeout(120000); let fixture; @@ -91,14 +88,15 @@ describe('dropTable racing a cross-worker source-fill commit', function () { fixture.send('get', { id: i }); await fixture.expect('get-resolved'); await Main.dropTable(); - await new Promise((resolve) => setTimeout(resolve, 50)); + await fixture.expect('commit-settled'); const workerEvents = fixture.drain().filter((message) => message.event !== 'logged-error'); - assert.deepEqual(workerEvents, [], `iteration ${i}: unexpected worker events`); + assert.deepStrictEqual(workerEvents, [], `iteration ${i}: unexpected worker events`); assert.doesNotThrow( () => Probe.primaryStore.putSync('__probe__', { i }), `iteration ${i}: the environment must still accept writes` ); - assert.equal(Main.dbisDB.getSync(`${name}/`), undefined, `iteration ${i}: catalog rows must be removed`); + const catalogRows = [...Main.dbisDB.getRange({ start: `${name}/`, end: `${name}0` })].map(({ key }) => key); + assert.deepStrictEqual(catalogRows, [], `iteration ${i}: catalog rows must be removed`); } }); }); From da959f1bcbf1b223805ffdf6039a26c4df15eca7 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 1 Sep 2026 15:56:25 -0600 Subject: [PATCH 3/5] Gate the reproducer to RocksDB, run a settle-first control every time, and validate worker errors From the round-1 review: the suite now returns early under HARPER_STORAGE_ENGINE=lmdb (no column families to drop there), an always-on control drops the table after the worker's commit has settled so the fixture's settlement signal, probe write and catalog check run on every resources run, the worker reports whether its commit was still in flight when the drop started and the race test asserts at least one iteration caught it, and the only worker error the race test tolerates is a commit that lost to the drop before reaching RocksDB's write path. The worker fixture calls setMainIsWorker like the other thread fixtures. Refs #1381 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013yFRAueahXUwcVM5BjKPMq --- .../dropTableCrossWorkerWrite-worker.js | 5 +- .../dropTableCrossWorkerWrite.test.js | 59 ++++++++++++++----- 2 files changed, 48 insertions(+), 16 deletions(-) diff --git a/unitTests/resources/dropTableCrossWorkerWrite-worker.js b/unitTests/resources/dropTableCrossWorkerWrite-worker.js index f29b346fb5..55117c0c6e 100644 --- a/unitTests/resources/dropTableCrossWorkerWrite-worker.js +++ b/unitTests/resources/dropTableCrossWorkerWrite-worker.js @@ -7,7 +7,7 @@ const { waitFor } = require('../waitFor.js'); const { table } = require('#src/resources/databases'); const { createBlob } = require('#src/resources/blob'); const { logger } = require('#src/utility/logging/logger'); -const { onMessageByType } = require('#js/server/threads/manageThreads'); +const { onMessageByType, setMainIsWorker } = require('#js/server/threads/manageThreads'); const MESSAGE_TYPE = 'drop-table-cross-worker-test'; const CONTROL_TYPE = 'drop-table-cross-worker-control'; @@ -18,6 +18,7 @@ let TestTable; function runWorkerFixture() { onMessageByType(CONTROL_TYPE, () => {}); setupTestDBPath(); + setMainIsWorker(true); process.on('unhandledRejection', (error) => report('unhandled-rejection', { error: error?.stack ?? String(error) })); const originalError = logger.error; logger.error = (...args) => { @@ -54,7 +55,7 @@ function runWorkerFixture() { // record lock until that write has settled either way. TestTable.get(id, {}).then( async () => { - report('get-resolved'); + report('get-resolved', { commitInFlight: TestTable.primaryStore.hasLock(id) }); try { await waitFor(() => !TestTable.primaryStore.hasLock(id), { timeout: 30000, diff --git a/unitTests/resources/dropTableCrossWorkerWrite.test.js b/unitTests/resources/dropTableCrossWorkerWrite.test.js index a6bdfc784e..f000bdf0ef 100644 --- a/unitTests/resources/dropTableCrossWorkerWrite.test.js +++ b/unitTests/resources/dropTableCrossWorkerWrite.test.js @@ -11,6 +11,9 @@ const WORKER_FIXTURE = path.join(__dirname, 'dropTableCrossWorkerWrite-worker.js const MESSAGE_TYPE = 'drop-table-cross-worker-test'; const CONTROL_TYPE = 'drop-table-cross-worker-control'; const ITERATIONS = 20; +// The one error the worker may log: its cache write lost to the drop and was rejected before it +// reached RocksDB's write path. Anything else the worker logs fails the test. +const CONTAINED_COMMIT_LOSS = /^Error committing cache update .*(Could not access column family|column family .*dropp)/; function defineTable(name) { return table({ @@ -27,6 +30,7 @@ function defineTable(name) { function startFixtureWorker() { const queued = []; const waiting = []; + const loggedErrors = []; const receive = (message) => { if (message?.type !== MESSAGE_TYPE) return; const waiter = waiting.shift(); @@ -49,22 +53,24 @@ function startFixtureWorker() { for (;;) { const message = await next(); if (message.event === event) return message; - if (message.event === 'logged-error') continue; - throw new Error(`unexpected worker event ${message.event}: ${JSON.stringify(message)}`); + if (message.event === 'logged-error') loggedErrors.push(message); + else throw new Error(`unexpected worker event ${message.event}: ${JSON.stringify(message)}`); } }; - const drain = () => queued.splice(0); + // Everything the worker reported that no expect() consumed, logged errors included. + const drain = () => [...loggedErrors.splice(0), ...queued.splice(0)]; return { worker, send, expect, drain }; } -// harper#1381: a column family must not be dropped while a commit naming it is between conflict -// validation and its write, because RocksDB latches that write's failure as a fatal background -// error on the whole environment. dropTable() drains only its own thread's source-fill commits, -// so a worker's in-flight commit can still land on the dropped family. Skipped until rocksdb-js#806 -// serializes drops against in-flight commits; on the current binding it fails on iteration 0. +function catalogRows(Table, name) { + return [...Table.dbisDB.getRange({ start: `${name}/`, end: `${name}0` })].map(({ key }) => key); +} + describe('dropTable racing a cross-worker source-fill commit', function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return; this.timeout(120000); let fixture; + let Probe; before(async () => { setupTestDBPath(); @@ -72,31 +78,56 @@ describe('dropTable racing a cross-worker source-fill commit', function () { onMessageByType(MESSAGE_TYPE, () => {}); fixture = startFixtureWorker(); await fixture.expect('booted'); + Probe = defineTable('CrossDropProbe'); }); after(async () => { await fixture?.worker?.terminate?.(); }); + // Exercises the fixture on every run: the worker's settlement signal, the drop, the probe write + // and the catalog check, with the commit already landed so there is no race to lose. + it("drops cleanly once the worker's source-fill commit has settled", async () => { + const name = 'CrossDropSettled'; + const Main = defineTable(name); + fixture.send('define', { table: name }); + await fixture.expect('defined'); + fixture.send('get', { id: 'settled' }); + await fixture.expect('get-resolved'); + await fixture.expect('commit-settled'); + await Main.dropTable(); + assert.deepStrictEqual(fixture.drain(), [], 'unexpected worker events'); + assert.doesNotThrow(() => Probe.primaryStore.putSync('__probe__', { settled: true })); + assert.deepStrictEqual(catalogRows(Main, name), [], 'catalog rows must be removed'); + }); + + // harper#1381: a column family must not be dropped while a commit naming it is between conflict + // validation and its write, because RocksDB latches that write's failure as a fatal background + // error on the whole environment. dropTable() drains only its own thread's source-fill commits, + // so a worker's in-flight commit can still land on the dropped family. Skipped until rocksdb-js#806 + // serializes drops against in-flight commits; on the current binding it fails on iteration 0. it.skip('leaves the storage environment writable and the catalog clean', async () => { - const Probe = defineTable('CrossDropProbe'); + let raced = 0; for (let i = 0; i < ITERATIONS; i++) { const name = `CrossDrop${i}`; const Main = defineTable(name); fixture.send('define', { table: name }); await fixture.expect('defined'); fixture.send('get', { id: i }); - await fixture.expect('get-resolved'); + const { commitInFlight } = await fixture.expect('get-resolved'); + if (commitInFlight) raced++; await Main.dropTable(); await fixture.expect('commit-settled'); - const workerEvents = fixture.drain().filter((message) => message.event !== 'logged-error'); - assert.deepStrictEqual(workerEvents, [], `iteration ${i}: unexpected worker events`); + const unexpected = fixture + .drain() + .filter((message) => message.event !== 'logged-error' || !CONTAINED_COMMIT_LOSS.test(message.message)); + assert.deepStrictEqual(unexpected, [], `iteration ${i}: unexpected worker events`); assert.doesNotThrow( () => Probe.primaryStore.putSync('__probe__', { i }), `iteration ${i}: the environment must still accept writes` ); - const catalogRows = [...Main.dbisDB.getRange({ start: `${name}/`, end: `${name}0` })].map(({ key }) => key); - assert.deepStrictEqual(catalogRows, [], `iteration ${i}: catalog rows must be removed`); + assert.deepStrictEqual(catalogRows(Main, name), [], `iteration ${i}: catalog rows must be removed`); } + assert.ok(raced > 0, 'no iteration caught the worker commit in flight, so the race was never exercised'); }); }); From e0aae4c46ef905cf8a556906020b4983561b3f86 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 1 Sep 2026 16:26:30 -0600 Subject: [PATCH 4/5] Sample the race on the dropping thread, barrier the control's drain, and reset worker state From the adjudicated review: the control now round-trips through the worker before asserting on its events, so anything it reported because of the drop has crossed the thread boundary; each test discards what the worker reported before it began; the race test samples the shared record lock on the dropping thread right before the drop instead of trusting a worker-side sample that predates the message hop; setMainIsWorker is restored after the suite; two narrating comments are gone. Refs #1381 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013yFRAueahXUwcVM5BjKPMq --- .../dropTableCrossWorkerWrite-worker.js | 5 +++- .../dropTableCrossWorkerWrite.test.js | 24 +++++++++++++------ 2 files changed, 21 insertions(+), 8 deletions(-) diff --git a/unitTests/resources/dropTableCrossWorkerWrite-worker.js b/unitTests/resources/dropTableCrossWorkerWrite-worker.js index 55117c0c6e..039b389d2b 100644 --- a/unitTests/resources/dropTableCrossWorkerWrite-worker.js +++ b/unitTests/resources/dropTableCrossWorkerWrite-worker.js @@ -55,7 +55,7 @@ function runWorkerFixture() { // record lock until that write has settled either way. TestTable.get(id, {}).then( async () => { - report('get-resolved', { commitInFlight: TestTable.primaryStore.hasLock(id) }); + report('get-resolved'); try { await waitFor(() => !TestTable.primaryStore.hasLock(id), { timeout: 30000, @@ -70,6 +70,9 @@ function runWorkerFixture() { ); break; } + case 'ping': + report('pong'); + break; } } catch (error) { report('error', { error: error?.stack ?? String(error) }); diff --git a/unitTests/resources/dropTableCrossWorkerWrite.test.js b/unitTests/resources/dropTableCrossWorkerWrite.test.js index f000bdf0ef..937982bb95 100644 --- a/unitTests/resources/dropTableCrossWorkerWrite.test.js +++ b/unitTests/resources/dropTableCrossWorkerWrite.test.js @@ -57,9 +57,13 @@ function startFixtureWorker() { else throw new Error(`unexpected worker event ${message.event}: ${JSON.stringify(message)}`); } }; - // Everything the worker reported that no expect() consumed, logged errors included. const drain = () => [...loggedErrors.splice(0), ...queued.splice(0)]; - return { worker, send, expect, drain }; + // A round-trip through the worker: everything it reported before answering has arrived. + const sync = async () => { + send('ping'); + await expect('pong'); + }; + return { worker, send, expect, drain, sync }; } function catalogRows(Table, name) { @@ -83,10 +87,12 @@ describe('dropTable racing a cross-worker source-fill commit', function () { after(async () => { await fixture?.worker?.terminate?.(); + setMainIsWorker(false); }); - // Exercises the fixture on every run: the worker's settlement signal, the drop, the probe write - // and the catalog check, with the commit already landed so there is no race to lose. + // Whatever the worker reported before a test began is not that test's signal. + beforeEach(() => fixture.drain()); + it("drops cleanly once the worker's source-fill commit has settled", async () => { const name = 'CrossDropSettled'; const Main = defineTable(name); @@ -96,6 +102,7 @@ describe('dropTable racing a cross-worker source-fill commit', function () { await fixture.expect('get-resolved'); await fixture.expect('commit-settled'); await Main.dropTable(); + await fixture.sync(); assert.deepStrictEqual(fixture.drain(), [], 'unexpected worker events'); assert.doesNotThrow(() => Probe.primaryStore.putSync('__probe__', { settled: true })); assert.deepStrictEqual(catalogRows(Main, name), [], 'catalog rows must be removed'); @@ -114,8 +121,11 @@ describe('dropTable racing a cross-worker source-fill commit', function () { fixture.send('define', { table: name }); await fixture.expect('defined'); fixture.send('get', { id: i }); - const { commitInFlight } = await fixture.expect('get-resolved'); - if (commitInFlight) raced++; + await fixture.expect('get-resolved'); + // The record lock is shared across threads and held until the worker's cache write settles. + // Sampled on the dropping thread right before the drop, it shows the race was exercised at + // least once; it cannot show that a given iteration overlapped. + if (Main.primaryStore.hasLock(i)) raced++; await Main.dropTable(); await fixture.expect('commit-settled'); const unexpected = fixture @@ -128,6 +138,6 @@ describe('dropTable racing a cross-worker source-fill commit', function () { ); assert.deepStrictEqual(catalogRows(Main, name), [], `iteration ${i}: catalog rows must be removed`); } - assert.ok(raced > 0, 'no iteration caught the worker commit in flight, so the race was never exercised'); + assert.ok(raced > 0, "no iteration caught the worker's commit in flight when the drop started"); }); }); From d513d701c8e845710ee4f61f01c93efddf5265e0 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 1 Sep 2026 17:06:08 -0600 Subject: [PATCH 5/5] Fail the fixture's pending wait as soon as its worker dies, and call the probe writes directly Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01St6cyvN2KtgxeaboFBFcHN --- .../dropTableCrossWorkerWrite.test.js | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/unitTests/resources/dropTableCrossWorkerWrite.test.js b/unitTests/resources/dropTableCrossWorkerWrite.test.js index 937982bb95..78a42f1ad7 100644 --- a/unitTests/resources/dropTableCrossWorkerWrite.test.js +++ b/unitTests/resources/dropTableCrossWorkerWrite.test.js @@ -31,14 +31,22 @@ function startFixtureWorker() { const queued = []; const waiting = []; const loggedErrors = []; + let died = null; const receive = (message) => { if (message?.type !== MESSAGE_TYPE) return; const waiter = waiting.shift(); - if (waiter) waiter(message); + if (waiter) waiter.resolve(message); else queued.push(message); }; - const next = () => - queued.length ? Promise.resolve(queued.shift()) : new Promise((resolve) => waiting.push(resolve)); + const fail = (error) => { + died = error; + for (const waiter of waiting.splice(0)) waiter.reject(error); + }; + const next = () => { + if (queued.length) return Promise.resolve(queued.shift()); + if (died) return Promise.reject(died); + return new Promise((resolve, reject) => waiting.push({ resolve, reject })); + }; const worker = startWorker(WORKER_FIXTURE, { name: 'drop-table-cross-worker-test', workerIndex: 1, @@ -46,6 +54,8 @@ function startFixtureWorker() { autoRestart: false, onStarted(spawned) { spawned.on('message', receive); + spawned.on('error', fail); + spawned.on('exit', (code) => fail(new Error(`fixture worker exited with code ${code}`))); }, }); const send = (command, details = {}) => worker.postMessage({ type: CONTROL_TYPE, command, ...details }); @@ -104,7 +114,7 @@ describe('dropTable racing a cross-worker source-fill commit', function () { await Main.dropTable(); await fixture.sync(); assert.deepStrictEqual(fixture.drain(), [], 'unexpected worker events'); - assert.doesNotThrow(() => Probe.primaryStore.putSync('__probe__', { settled: true })); + Probe.primaryStore.putSync('__probe__', { settled: true }); assert.deepStrictEqual(catalogRows(Main, name), [], 'catalog rows must be removed'); }); @@ -132,10 +142,7 @@ describe('dropTable racing a cross-worker source-fill commit', function () { .drain() .filter((message) => message.event !== 'logged-error' || !CONTAINED_COMMIT_LOSS.test(message.message)); assert.deepStrictEqual(unexpected, [], `iteration ${i}: unexpected worker events`); - assert.doesNotThrow( - () => Probe.primaryStore.putSync('__probe__', { i }), - `iteration ${i}: the environment must still accept writes` - ); + Probe.primaryStore.putSync('__probe__', { i }); assert.deepStrictEqual(catalogRows(Main, name), [], `iteration ${i}: catalog rows must be removed`); } assert.ok(raced > 0, "no iteration caught the worker's commit in flight when the drop started");