test: fix BroadcastChannel leakage in worker.js test suite

lib/worker.js

@@ -46,6 +46,7 @@ let requestCtr = 0; // Concurrency control
 const requests = new Map(); // Pending requests for workers' local metrics.
 const workers = new Map();
 let historicMetrics = []; // Metrics from dead workers
+let ownChannel; // This thread's own broadcast channel, set once listeners are added.
 
 class WorkerRegistry extends Registry {
 	/**
@@ -231,6 +232,36 @@ class WorkerRegistry extends Registry {
 		return workers.size;
 	}
 
+	/**
+	 * Close every channel and listener opened by this module instance.
+	 *
+	 * `BroadcastChannel`s created here keep receiving messages until closed,
+	 * even once nothing in JS still references them. Test suites that call
+	 * `jest.resetModules()` between cases therefore need this to release the
+	 * previous instance's channels; otherwise its listeners keep reacting to
+	 * later tests' messages on the same channel names. Not for production use.
+	 * @returns {void}
+	 */
+	static resetForTesting() {
+		if (!['test', 'development'].includes(process.env.NODE_ENV)) {
+			throw new Error(
+				'WorkerRegistry.resetForTesting() is only available in test or development mode.',
+			);
+		}
+
+		ANNOUNCEMENT_CHANNEL.close();
+		ownChannel?.close();
+
+		for (const entry of workers.values()) {
+			entry.channel.close();
+		}
+
+		workers.clear();
+		requests.clear();
+		historicMetrics = [];
+		listenersAdded = false;
+	}
+
 	/**
 	 * Sets the registry or registries to be aggregated. Call from workers to
 	 * use a registry/registries other than the default global registry.
@@ -264,7 +295,7 @@ function addListeners(primary) {
 	}
 
 	const name = `@prometheus-io/client:worker:${threadId}`;
-	const channel = new BroadcastChannel(name).unref();
+	ownChannel = new BroadcastChannel(name).unref();
 
 	ANNOUNCEMENT_CHANNEL.addEventListener('message', async event => {
 		const message = event.data;
@@ -279,15 +310,15 @@ function addListeners(primary) {
 			);
 
 			try {
-				channel.postMessage({
+				ownChannel.postMessage({
 					type: GET_METRICS_RES,
 					requestId: message.requestId,
 					threadId,
 					metrics,
 				});
 			} catch (error) {
 				debug(error);
-				channel.postMessage({
+				ownChannel.postMessage({
 					type: GET_METRICS_RES,
 					requestId: message.requestId,
 					error: error.message,
@@ -298,7 +329,7 @@ function addListeners(primary) {
 		}
 	});
 
-	channel.addEventListener('message', async event => {
+	ownChannel.addEventListener('message', async event => {
 		const message = event.data;
 
 		if (message.type === ACK) {

test/workerTest.js

@@ -44,10 +44,15 @@ describe.each([
 
 	describe('WorkerRegistry.workerMetrics()', () => {
 		it('works properly if there are no workers', async () => {
+			jest.resetModules();
 			const WorkerRegistry = require('../lib/worker');
 			const registry = new WorkerRegistry(regType);
-			const metrics = await registry.workerMetrics();
-			expect(metrics).toEqual('');
+			try {
+				const metrics = await registry.workerMetrics();
+				expect(metrics).toEqual('');
+			} finally {
+				WorkerRegistry.resetForTesting();
+			}
 		});
 
 		it('aggregates worker responses in thread id order', async () => {
@@ -58,14 +63,6 @@ describe.each([
 				'@prometheus-io/client:announce',
 			).unref();
 
-			const discovery = new Promise(resolve => {
-				announcementChannel.addEventListener('message', async event => {
-					if (event.data.type === ANNOUNCEMENT && !event.data.primary) {
-						resolve(event);
-					}
-				});
-			});
-
 			const responders = [1, 2, 3].map(threadId => {
 				const name = `@prometheus-io/client:worker:${threadId}`;
 				const channel = new BroadcastChannel(name).unref();
@@ -79,7 +76,13 @@ describe.each([
 				return { threadId, channel };
 			});
 
-			await discovery; // Let announcements arrive
+			// Wait for the primary to register all 3 fake workers, rather than
+			// hoping to catch a message announcing it: a self-posted message on
+			// announcementChannel is never delivered back to announcementChannel
+			// itself, so nothing here guarantees an observable echo.
+			while (AggregatorRegistry.workerCount() < responders.length) {
+				await delay(1);
+			}
 
 			let finishSendingResponses;
 			const responsesSent = new Promise(resolve => {
@@ -111,6 +114,7 @@ describe.each([
 			} finally {
 				announcementChannel.close();
 				for (const responder of responders) responder.channel.close();
+				AggregatorRegistry.resetForTesting();
 			}
 		});
 
@@ -156,6 +160,7 @@ describe.each([
 			} finally {
 				announcementChannel.close();
 				channel.close();
+				AggregatorRegistry.resetForTesting();
 			}
 		});
 	});
@@ -164,7 +169,6 @@ describe.each([
 		let AggregatorRegistry;
 		let announcementChannel;
 		let registry;
-		let discovery;
 
 		beforeEach(async () => {
 			jest.resetModules();
@@ -175,25 +179,22 @@ describe.each([
 			).unref();
 
 			registry = new AggregatorRegistry(regType);
-
-			discovery = new Promise(resolve => {
-				announcementChannel.addEventListener('message', async event => {
-					if (event.data.type === ANNOUNCEMENT && !event.data.primary) {
-						resolve(event);
-					}
-				});
-			});
 		});
 
 		afterEach(() => {
 			announcementChannel.close();
+			AggregatorRegistry.resetForTesting();
 		});
 
 		it('returns immediately on no outstanding requests', async () => {
 			await expect(registry.shutdown()).resolves.not.toThrow();
 		});
 
 		it('sends data back to the primary', async () => {
+			// The beforeEach above set this instance up as the primary; tear it
+			// down before loading a fresh one to act as the worker instead, so
+			// the primary's channels don't leak into later tests.
+			AggregatorRegistry.resetForTesting();
 			jest.resetModules();
 			AggregatorRegistry = require('../lib/worker');
 
@@ -262,7 +263,13 @@ describe.each([
 					}
 				});
 
-				await discovery;
+				// Wait for the primary to register this fake worker, rather than
+				// hoping to catch a message announcing it: a self-posted message on
+				// announcementChannel is never delivered back to announcementChannel
+				// itself, so nothing here guarantees an observable echo.
+				while (AggregatorRegistry.workerCount() < 1) {
+					await delay(1);
+				}
 			});
 
 			afterEach(() => {
@@ -287,6 +294,7 @@ describe.each([
 
 				const AggregatorRegistry = require('../lib/worker');
 				const ar = new AggregatorRegistry(regType);
+				AggregatorRegistry.resetForTesting();
 			}
 		});
 
@@ -318,7 +326,9 @@ describe.each([
 			try {
 				expect(() => channel.postMessage(unexpected)).not.toThrow();
 			} finally {
+				announcementChannel.close();
 				channel.close();
+				WorkerRegistry.resetForTesting();
 			}
 		});
 	});