Reproduce the cross-worker drop_table race behind the blob lifecycle CI failures as a skipped unit test (blocked on rocksdb-js#806)

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();

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");
+	});
+});