Merge pull request #8 from HarperFast/fix/shutdown-review

fix(browser): address shutdown-drain review (v1.4.2)

Author: harper-joseph

packages/browser/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@harperfast/prerender-browser",
-	"version": "1.4.1",
+	"version": "1.4.2",
 	"type": "module",
 	"description": "Headless-browser render library for Harper Prerender: claims render jobs from the @harperfast/prerender queue, renders pages in headless Chrome (Puppeteer), and posts the HTML back. Embedded by a render service and configured entirely via startWorker() options.",
 	"keywords": [

packages/browser/src/RenderQueueConsumer.ts

@@ -27,55 +27,67 @@ interface ProducerState {
 export async function* RenderQueueConsumer(signal?: AbortSignal) {
 	const mqttClient = await connectMqtt();
 
-	const producerStates = new Map<string, ProducerState>();
-
-	mqttClient.on('message', (_topic, payload) => {
-		const queueState = JSON.parse(payload.toString());
-		producerStates.set(queueState.hostname, queueState);
-	});
-
-	await mqttClient.subscribeAsync(Topic.queueState, { qos: 1, rh: 0 });
-
-	const pickAvailableQueueHost: () => string | null = () => {
-		const eligibleHosts: string[] = [];
-
-		producerStates.forEach((producer) => {
-			if (producer.status === ProducerStatus.queued) {
-				eligibleHosts.push(producer.hostname);
+	// try wraps ALL setup after connectMqtt (message handler, subscribe, loop) so a failure
+	// during setup — e.g. subscribeAsync throwing — still closes the MQTT client in `finally`
+	// instead of leaking the connection.
+	try {
+		const producerStates = new Map<string, ProducerState>();
+
+		mqttClient.on('message', (_topic, payload) => {
+			// This runs async in the event loop, so a JSON.parse throw would bypass the
+			// surrounding try/finally and crash the worker — guard it.
+			try {
+				const queueState = JSON.parse(payload.toString());
+				producerStates.set(queueState.hostname, queueState);
+			} catch (err) {
+				logger.error({ err }, 'failed to parse queue_status message');
 			}
 		});
 
-		if (eligibleHosts.length === 0) return null;
-
-		return eligibleHosts[Math.floor(Math.random() * eligibleHosts.length)];
-	};
-
-	const claimJobs = async (host: string, limit: number): Promise<RenderJob[]> => {
-		try {
-			const res = await request(`http://${host}:${settings.queuePort}`, {
-				method: 'POST',
-				path: '/render_queue/claim',
-				body: JSON.stringify({
-					limit,
-				}),
-				headers: {
-					'content-type': 'application/json',
-				},
+		await mqttClient.subscribeAsync(Topic.queueState, { qos: 1, rh: 0 });
+
+		const pickAvailableQueueHost: () => string | null = () => {
+			const eligibleHosts: string[] = [];
+
+			producerStates.forEach((producer) => {
+				if (producer.status === ProducerStatus.queued) {
+					eligibleHosts.push(producer.hostname);
+				}
 			});
-			if (res.statusCode === 200) {
-				const jobs: any = await res.body.json();
-				return jobs.map((job: any) => new RenderJob(job));
-			} else {
-				logger.error(res.body.json());
+
+			if (eligibleHosts.length === 0) return null;
+
+			return eligibleHosts[Math.floor(Math.random() * eligibleHosts.length)];
+		};
+
+		const claimJobs = async (host: string, limit: number): Promise<RenderJob[]> => {
+			try {
+				const res = await request(`http://${host}:${settings.queuePort}`, {
+					method: 'POST',
+					path: '/render_queue/claim',
+					body: JSON.stringify({
+						limit,
+					}),
+					headers: {
+						'content-type': 'application/json',
+					},
+				});
+				if (res.statusCode === 200) {
+					const jobs: any = await res.body.json();
+					return jobs.map((job: any) => new RenderJob(job));
+				} else {
+					// res.body.json() is a Promise (and can reject on a non-JSON error body) — await
+					// text() so we log the actual response, not a pending Promise / unhandled rejection.
+					const body = await res.body.text().catch(() => '');
+					logger.error({ statusCode: res.statusCode, body }, 'failed to claim jobs');
+					return [];
+				}
+			} catch (e) {
+				logger.error(e);
 				return [];
 			}
-		} catch (e) {
-			logger.error(e);
-			return [];
-		}
-	};
+		};
 
-	try {
 		while (!signal?.aborted) {
 			const host = pickAvailableQueueHost();
 

packages/browser/src/Worker.ts

@@ -72,9 +72,11 @@ export default class RenderWorker {
 		}, 10000);
 		this.browserCleanupInterval.unref();
 
-		process.on('uncaughtException', (err) => {
+		process.on('uncaughtException', async (err) => {
 			logger.error({ err }, 'uncaught exception');
-			this.destroy();
+			// Await so browsers are actually closed before the event loop dies (otherwise Chrome
+			// is orphaned). ManagedBrowser.close() has its own SIGKILL fallback if it hangs.
+			await this.destroy().catch(() => {});
 			process.exit(1);
 		});
 
@@ -175,26 +177,37 @@ export default class RenderWorker {
 		await Promise.race([Promise.allSettled([...this.inflight]), deadline]);
 		ac.abort();
 
-		this.destroy();
+		await this.destroy();
 		logger.info('worker shutdown complete');
 	}
 
-	destroy() {
+	// Async so callers can AWAIT it before process.exit() — otherwise the event loop dies before
+	// the browser .close() promises run and Chrome is orphaned (the whole point of closing here).
+	async destroy() {
 		clearInterval(this.logStatsInterval);
 		if (this.browserCleanupInterval !== null) {
 			clearInterval(this.browserCleanupInterval);
 			this.browserCleanupInterval = null;
 		}
 
-		// close all browsers
+		const closing: Promise<void>[] = [];
 		if (this.browser) {
-			this.browser.close().catch(noop);
+			closing.push(this.browser.close().catch(noop));
+		}
+		// A browser mid-launch isn't in `this.browser` yet; close it once it resolves so a
+		// destroy() during launch (shutdown drain deadline, or an uncaught exception) doesn't
+		// orphan the Chrome process.
+		if (this.browserPromise) {
+			closing.push(this.browserPromise.then((b) => b.close().catch(noop)).catch(noop));
 		}
 		this.retiredBrowsers.forEach((browser) => {
-			browser.close().catch(noop);
+			closing.push(browser.close().catch(noop));
 		});
+
 		this.browser = null;
 		this.retiredBrowsers.clear();
+
+		await Promise.all(closing);
 	}
 
 	closeRetiredBrowsers() {

packages/browser/src/errorHandler.ts

@@ -53,20 +53,29 @@ export class ErrorHandler {
 	}
 
 	private async handleTermination(signal: string) {
-		if (this.terminating) return; // ignore a second SIGTERM/SIGINT while already draining
+		if (this.terminating) {
+			// A second signal (e.g. an impatient Ctrl+C) forces an immediate, unclean exit
+			// instead of waiting out the drain.
+			logger.warn({ signal }, 'Second termination signal — forcing exit');
+			process.exit(1);
+		}
 		this.terminating = true;
 		logger.info({ signal }, 'Termination signal received');
 
-		// Hard backstop: if the drain hangs, still exit before the supervisor SIGKILLs us.
-		const backstop = setTimeout(() => process.exit(0), this.shutdownDeadlineMs);
+		// Hard backstop: if the drain hangs, exit before the supervisor SIGKILLs us — with a
+		// NON-zero code so the orchestrator sees an unclean/timed-out shutdown, not a clean one.
+		const backstop = setTimeout(() => process.exit(1), this.shutdownDeadlineMs);
 		backstop.unref();
+		let exitCode = 0;
 		try {
 			await this.onTerminate?.();
 		} catch (err) {
+			// The drain failed/was incomplete — signal an unclean shutdown to the orchestrator.
 			logger.error({ err }, 'error during graceful shutdown');
+			exitCode = 1;
 		}
 		clearTimeout(backstop);
-		process.exit(0);
+		process.exit(exitCode);
 	}
 
 	private gracefulShutdown(exitCode: number) {