diff --git a/unitTests/resources/dropTableCrossWorkerWrite-worker.js b/unitTests/resources/dropTableCrossWorkerWrite-worker.js new file mode 100644 index 0000000000..039b389d2b --- /dev/null +++ b/unitTests/resources/dropTableCrossWorkerWrite-worker.js @@ -0,0 +1,86 @@ +'use strict'; + +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'); +const { onMessageByType, setMainIsWorker } = 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(); + setMainIsWorker(true); + 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': { + 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; + } + case 'ping': + report('pong'); + 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..78a42f1ad7 --- /dev/null +++ b/unitTests/resources/dropTableCrossWorkerWrite.test.js @@ -0,0 +1,150 @@ +'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; +// 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({ + table: name, + database: 'test', + audit: true, + attributes: [ + { name: 'id', isPrimaryKey: true }, + { name: 'blob', type: 'Blob' }, + ], + }); +} + +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.resolve(message); + else queued.push(message); + }; + 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, + threadCount: 2, + 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 }); + const expect = async (event) => { + for (;;) { + const message = await next(); + if (message.event === event) return message; + if (message.event === 'logged-error') loggedErrors.push(message); + else throw new Error(`unexpected worker event ${message.event}: ${JSON.stringify(message)}`); + } + }; + const drain = () => [...loggedErrors.splice(0), ...queued.splice(0)]; + // 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) { + 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(); + setMainIsWorker(true); + onMessageByType(MESSAGE_TYPE, () => {}); + fixture = startFixtureWorker(); + await fixture.expect('booted'); + Probe = defineTable('CrossDropProbe'); + }); + + after(async () => { + await fixture?.worker?.terminate?.(); + setMainIsWorker(false); + }); + + // 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); + 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(); + await fixture.sync(); + assert.deepStrictEqual(fixture.drain(), [], 'unexpected worker events'); + 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 () => { + 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'); + // 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 + .drain() + .filter((message) => message.event !== 'logged-error' || !CONTAINED_COMMIT_LOSS.test(message.message)); + assert.deepStrictEqual(unexpected, [], `iteration ${i}: unexpected worker events`); + 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"); + }); +});