fix(cluster): skip responses after IPC disconnect

CHANGELOG.md

@@ -40,6 +40,7 @@ This release marks our first release under the Prometheus umbrella.
 - perf: Stat aggregation uses similar strategy to collection. 60% faster aggregation
 - chore: Add copyright license headers and test
 - Make cluster and worker-thread metric aggregation order deterministic
+- Avoid sending cluster metric responses after a worker's IPC channel closes
 
 ### Added
 

lib/cluster.js

@@ -190,13 +190,15 @@ function addListeners() {
 			if (message.type === GET_METRICS_REQ) {
 				Promise.all(registries.map(r => r.getMetricsAsJSON()))
 					.then(metrics => {
+						if (!process.connected) return;
 						process.send({
 							type: GET_METRICS_RES,
 							requestId: message.requestId,
 							metrics,
 						});
 					})
 					.catch(error => {
+						if (!process.connected) return;
 						process.send({
 							type: GET_METRICS_RES,
 							requestId: message.requestId,

test/clusterTest.js

@@ -18,6 +18,7 @@ const cluster = require('cluster');
 const process = require('process');
 const Registry = require('../lib/cluster');
 
+const GET_METRICS_REQ = '@prometheus/client:getMetricsReq';
 const GET_METRICS_RES = '@prometheus/client:getMetricsRes';
 
 function metric(value) {
@@ -129,3 +130,61 @@ describe.each([
 		});
 	});
 });
+
+describe('worker message handling', () => {
+	it('does not send metrics after the IPC channel disconnects', async () => {
+		jest.resetModules();
+		jest.doMock('cluster', () => {
+			return { isPrimary: false };
+		});
+
+		const messageListeners = new Set(process.listeners('message'));
+		const connectedDescriptor = Object.getOwnPropertyDescriptor(
+			process,
+			'connected',
+		);
+		const sendDescriptor = Object.getOwnPropertyDescriptor(process, 'send');
+		const send = jest.fn();
+		let listener;
+
+		try {
+			Object.defineProperty(process, 'connected', {
+				configurable: true,
+				value: true,
+				writable: true,
+			});
+			Object.defineProperty(process, 'send', {
+				configurable: true,
+				value: send,
+			});
+
+			const AggregatorRegistry = require('../lib/cluster');
+			new AggregatorRegistry();
+
+			listener = process
+				.listeners('message')
+				.find(candidate => !messageListeners.has(candidate));
+			expect(listener).toBeDefined();
+
+			listener({ type: GET_METRICS_REQ, requestId: 1 });
+			process.connected = false;
+			await new Promise(resolve => setImmediate(resolve));
+
+			expect(send).not.toHaveBeenCalled();
+		} finally {
+			if (listener) process.removeListener('message', listener);
+			if (connectedDescriptor) {
+				Object.defineProperty(process, 'connected', connectedDescriptor);
+			} else {
+				delete process.connected;
+			}
+			if (sendDescriptor) {
+				Object.defineProperty(process, 'send', sendDescriptor);
+			} else {
+				delete process.send;
+			}
+			jest.dontMock('cluster');
+			jest.resetModules();
+		}
+	});
+});