Fail a rolling restart job that did not restart every node

DESIGN.md

@@ -187,6 +187,7 @@ Index of the design notes for the harper core: one line per note (closely relate
 - [Component preparation is serialized across worker threads](components/DESIGN.md#component-preparation-is-serialized-across-worker-threads) — One component transaction at a time across workers; the lifecycle broadcast sits outside the lock; the load lock is keyed by application and plugin.
 - [Peer-side deploy_component payload read: retryable blob stalls and `Readable.from()` cancellation](components/DESIGN.md#peer-side-deploy_component-payload-read-retryable-blob-stalls-and-readablefrom-cancellation) — Retry a `BLOB_UNAVAILABLE` read only before any byte reached the consumer; use `Readable.from()` for cancellation, not a hand-rolled `ReadableStream`.
 - [An origin waits for a peer's deploy answer only as long as the peer may take](components/DESIGN.md#an-origin-waits-for-a-peers-deploy-answer-only-as-long-as-the-peer-may-take) — The replicated deploy's per-peer deadline sums the peer's own allowances, so it never undercuts a healthy peer; a missed deadline is a failure with an unknown outcome, not cancellation.
+- [A rolling restart job completes only if every peer restarted (`bin/restart.ts` `restartEachPeer`)](components/DESIGN.md#a-rolling-restart-job-completes-only-if-every-peer-restarted-binrestartts-restarteachpeer) — `ERROR` names each peer that failed or was skipped, with every peer's outcome in `result.replicated`; a failed or possibly-running peer restart halts the roll; only write back-pressure is retried.
 - [A replicated deploy reports each node's install fingerprint, and never acts on it](components/DESIGN.md#a-replicated-deploy-reports-each-nodes-install-fingerprint-and-never-acts-on-it) — Each node records the source identity its resolver already produced and the sha256 of its root lockfiles; the origin compares peers once on the aggregate and reports differences in a warning, the message and `peer_results`, without ever failing the deploy.
 - [A dangling symlink silently truncates the deploy tarball (`components/packageComponent.ts`)](components/DESIGN.md#a-dangling-symlink-silently-truncates-the-deploy-tarball-componentspackagecomponentts) — tar-fs treats a dangling symlink's ENOENT as end-of-stream, so `scanPackageDirectory()` pre-walks the tree · [The packer never ships the loader's links to the install (`components/packageComponent.ts`)](components/DESIGN.md#the-packer-never-ships-the-loaders-links-to-the-install-componentspackagecomponentts) — The application's `node_modules/harper` is skipped by name; a nested one only when it resolves to the running install.
 - [Deploy watcher generations preserve logical entry events](components/DESIGN.md#deploy-watcher-generations-preserve-logical-entry-events) — `EntryHandler` snapshots matching paths across a deploy so consumers see logical change and delete events, not raw rescan adds.

bin/restart.ts

@@ -19,6 +19,7 @@ const { HTTP_STATUS_CODES } = hdbErrors;
 import * as envMgr from '../utility/environment/environmentManager.ts';
 import * as path from 'node:path';
 import { rmSync } from 'node:fs';
+import { setTimeout as delay } from 'node:timers/promises';
 import { getThisNodeName } from '../server/nodeName.ts';
 import { armRestartExitWatchdog } from './restartExitWatchdog.ts';
 import { PRODUCT_NAME } from '../utility/packageUtils.js';
@@ -30,7 +31,7 @@ const ISOLATED_TOPOLOGY_REQUEST_TIMEOUT_MS = 5000;
 
 let calledFromCli;
 
-export { restart, restartService, activateDeploymentOnPeers };
+export { restart, restartService, activateDeploymentOnPeers, restartEachPeer };
 
 // Add ITC event listener to main thread which will be called from child that receives restart request.
 if (isMainThread) {
@@ -205,6 +206,114 @@ async function activateEachPeer(peers: any[], { project, deploymentId, deploymen
 	}
 }
 
+const PEER_RESTART_TIMING = {
+	attempts: 3,
+	retryDelayMs: 1000,
+	dispatchTimeoutMs: 60 * 1000,
+	pollIntervalMs: 250,
+	pollTimeoutMs: 30 * 1000,
+	jobDeadlineMs: 10 * 60 * 1000,
+};
+// A peer that refused the operation for write back-pressure never started it. Errors cross replication as
+// text, and 5.3 peers word it "too long of queue".
+const WRITE_QUEUE_BACK_PRESSURE = /Outstanding write transactions have too long (?:a|of) queue/;
+// Rejections known to precede the request reaching the peer. Anything else with a socket error code, or the
+// replicator's "outcome there is unknown" (a timeout, or the connection closing first) or the session's code-less
+// "Connection closed" rejection of requests pending at close, may have started it.
+const NEVER_CONNECTED = new Set(['ECONNREFUSED', 'ENOTFOUND', 'EAI_AGAIN', 'EHOSTUNREACH', 'ENETUNREACH']);
+const OUTCOME_UNKNOWN = /outcome there is unknown|^Connection closed\b/;
+const PEER_JOB_PENDING: string[] = [hdbTerms.JOB_STATUS_ENUM.CREATED, hdbTerms.JOB_STATUS_ENUM.IN_PROGRESS];
+
+// See components/DESIGN.md, "A rolling restart job completes only if every peer restarted".
+async function restartEachPeer(peers: any[], request: any, timing = PEER_RESTART_TIMING) {
+	const replication = (global as any).server.replication;
+	const replicated = [];
+	const queue = peers.map((node) => ({ node, attempt: 1, retryAt: 0 }));
+	let haltedBy: string | undefined;
+	while (queue.length > 0) {
+		const { node, attempt, retryAt } = queue.shift();
+		if (haltedBy) {
+			replicated.push({ node: node.name, error: `skipped: the restart of ${haltedBy} did not complete` });
+			continue;
+		}
+		if (retryAt > Date.now()) await delay(retryAt - Date.now());
+		let jobId;
+		try {
+			const answer = await replication.sendOperationToNode(node, request, { timeoutMs: timing.dispatchTimeoutMs });
+			jobId = answer?.job_id;
+		} catch (error) {
+			const message = error?.message ?? String(error);
+			if (attempt < timing.attempts && WRITE_QUEUE_BACK_PRESSURE.test(message)) {
+				hdbLogger.warn(`Restart of ${node.name} was refused (${message}); retrying after the other nodes`);
+				queue.push({ node, attempt: attempt + 1, retryAt: Date.now() + timing.retryDelayMs * 2 ** (attempt - 1) });
+			} else {
+				replicated.push({ node: node.name, error: message });
+				if (mayHaveStarted(error, message)) haltedBy = node.name;
+			}
+			continue;
+		}
+		const outcome = jobId
+			? await awaitPeerJob(replication, node, jobId, timing)
+			: { node: node.name, error: 'it answered without a restart job id' };
+		replicated.push(outcome);
+		if (outcome.error !== undefined) haltedBy = node.name;
+	}
+	const failed = replicated.filter((outcome) => outcome.error !== undefined);
+	if (failed.length > 0) {
+		const error: any = new Error(
+			`Restart of ${request.service} did not complete on ${failed.length} of ${replicated.length} peer node(s): ` +
+				failed
+					.map(({ node, error }) => `${node} (${typeof error === 'string' ? error : JSON.stringify(error)})`)
+					.join('; ')
+		);
+		error.result = { replicated };
+		throw error;
+	}
+	return { replicated };
+}
+
+// A peer's own error answer carries no code; a socket error usually does, a session close does not.
+function mayHaveStarted(error: any, message: string) {
+	if (OUTCOME_UNKNOWN.test(message)) return true;
+	const code = error?.code;
+	return code !== undefined && !NEVER_CONNECTED.has(code);
+}
+
+async function awaitPeerJob(replication: any, node: any, jobId: string, timing: typeof PEER_RESTART_TIMING) {
+	const deadline = Date.now() + timing.jobDeadlineMs;
+	let lastPollError;
+	while (Date.now() < deadline) {
+		await delay(timing.pollIntervalMs);
+		let response;
+		try {
+			response = await replication.sendOperationToNode(
+				node,
+				{ operation: 'get_job', id: jobId },
+				{ timeoutMs: Math.max(1, Math.min(timing.pollTimeoutMs, deadline - Date.now())) }
+			);
+		} catch (error) {
+			// The peer drops this connection while it restarts its own workers.
+			lastPollError = error?.message ?? String(error);
+			hdbLogger.debug?.(`Polling restart job ${jobId} on ${node.name} failed, still waiting:`, lastPollError);
+			continue;
+		}
+		lastPollError = undefined;
+		const job = response?.results?.[0];
+		if (!job) return { node: node.name, error: `its restart job ${jobId} was not found` };
+		if (job.status === hdbTerms.JOB_STATUS_ENUM.COMPLETE) return { node: node.name, message: job.message };
+		if (job.status === hdbTerms.JOB_STATUS_ENUM.ERROR)
+			return { node: node.name, error: job.message ?? `its restart job ${jobId} failed` };
+		if (!PEER_JOB_PENDING.includes(job.status))
+			return { node: node.name, error: `its restart job ${jobId} reported status ${job.status}` };
+	}
+	return {
+		node: node.name,
+		error:
+			`Timed out waiting for restart job ${jobId} to complete` +
+			(lastPollError ? `; the last poll failed: ${lastPollError}` : ''),
+	};
+}
+
 /**
  * Used to restart a particular service, services includes - httpWorkers
  * @param req
@@ -291,53 +400,10 @@ async function restartService(req: any) {
 				}
 			});
 		});
-		let replicatedResponses;
 		if (req.replicated) {
-			req.replicated = false; // don't send a replicated flag to the nodes we are sending to
-			replicatedResponses = [];
-			for (let node of (global as any).server.nodes) {
-				if (node.name === getThisNodeName()) continue;
-				// for now, only one at a time
-				let job_id;
-				try {
-					({ job_id } = await (global as any).server.replication.sendOperationToNode(node, req));
-				} catch (err) {
-					// If request to node fails, add the error to the response and continue to the next node
-					replicatedResponses.push({ node: node.name, message: err.message });
-					continue;
-				}
-				// wait for the job to finish by polling for the completion of the job
-				replicatedResponses.push(
-					await new Promise((resolve, reject) => {
-						const RETRY_INTERVAL = 250;
-						let retriesLeft = 2400; // 10 minutes
-						let interval = setInterval(async () => {
-							if (retriesLeft-- <= 0) {
-								clearInterval(interval);
-								let error: any = new Error('Timed out waiting for restart job to complete');
-								error.replicated = replicatedResponses; // report the finished restarts
-								reject(error);
-							}
-							let response = await (global as any).server.replication.sendOperationToNode(node, {
-								operation: 'get_job',
-								id: job_id,
-							});
-							const jobResult = response.results[0];
-							if (jobResult.status === 'COMPLETE') {
-								clearInterval(interval);
-								resolve({ node: node.name, message: jobResult.message });
-							}
-							if (jobResult.status === 'ERROR') {
-								clearInterval(interval);
-								let error: any = new Error(jobResult.message);
-								error.replicated = replicatedResponses; // report the finished restarts
-								reject(error);
-							}
-						}, RETRY_INTERVAL);
-					})
-				);
-			}
-			return { replicated: replicatedResponses };
+			const thisNode = getThisNodeName();
+			const peers = ((global as any).server.nodes ?? []).filter((node) => node.name !== thisNode);
+			return restartEachPeer(peers, { ...req, replicated: false });
 		}
 		return;
 	}

components/DESIGN.md

@@ -770,6 +770,27 @@ may still have them open, and a later drop of the now-absent application discard
 restart replaces is answered just before it, with any retention already decided and `null` for what is not done yet;
 a later failure to remove branches is only logged, and a second drop retries it (harper#3091).
 
+## A rolling restart job completes only if every peer restarted (`bin/restart.ts` `restartEachPeer`)
+
+The job behind an uncertified `restart: 'rolling'` deploy (a replicated `restart_service`) restarts one
+peer at a time and polls each peer's own restart job. It ends `COMPLETE` only when every peer's job did;
+otherwise `ERROR`, naming each peer that failed or was skipped, with one `result.replicated` entry per peer
+in both outcomes (`{ node, message }` restarted, `{ node, error }` not). A certified rolling deploy runs
+`activateDeploymentOnPeers` instead, which reports through `http_resp_msg`.
+
+- A peer that answered with its own error, or whose connection never opened (`ECONNREFUSED` and similar),
+  never started a restart: it is recorded and the roll moves on.
+- Any other dispatch failure halts the roll, so two peers never restart at once: the replicator's "outcome
+  there is unknown" (a timeout, or the connection closed before the answer), any other socket error,
+  or an answer with no job id. So does a peer whose restart failed or timed out, since the release being
+  rolled out may be what failed. Every peer after it is recorded as skipped.
+- Only the write-queue back-pressure refusal is retried (after the other peers, three attempts): it is the
+  one refusal known to be transient. Errors cross replication as text plus, for a socket error, its `code`, so
+  these classifications match wording and codes.
+
+`COMPLETE` means each peer's restart job completed, not that every worker there took new code. Enforced by
+`unitTests/bin/restartEachPeer.test.js` and `replicatedRestartService.test.js` (harper#3468).
+
 ## A replicated deploy reports each node's install fingerprint, and never acts on it
 
 Every node of a replicated `deploy_component` resolves and installs the release itself, so two nodes can end

server/jobs/jobProcess.ts

@@ -83,6 +83,7 @@ const JOB_ID = JOB_NAME.substring(4);
 		// get_job answers a refused bulk load with its structured permission report as the message.
 		const report = err?.http_resp_msg;
 		jobObj.message = report !== null && typeof report === 'object' ? report : err?.message ? err.message : err;
+		if (err?.result !== undefined) jobObj.result = err.result;
 		jobObj.end_datetime = Date.now();
 	} finally {
 		// A rejected updateJob must not skip handle cleanup and exit scheduling below (that would

server/jobs/jobs.ts

@@ -172,14 +172,7 @@ export async function addJob(jsonBody: any) {
 		['id']
 	);
 
-	let foundJob;
-	try {
-		foundJob = Array.from(await pSearchByValue(searchObj));
-	} catch (e) {
-		let message = `There was an error inserting a new job: ${e}`;
-		log.error(message);
-		return result;
-	}
+	let foundJob: any = Array.from(await pSearchByValue(searchObj));
 	//TODO: Once https://harperdb.atlassian.net/browse/HDB-501 is resolved, this check is no longer needed.
 	let foundValues = Array.isArray(foundJob) ? foundJob : Object.keys(foundJob);
 
@@ -188,13 +181,7 @@ export async function addJob(jsonBody: any) {
 	// that we will just throw an error assuming there is bad input causing the issue.
 	if (foundValues && foundValues.length > 0) {
 		newJob.id = uuidV4();
-		try {
-			foundJob = await pSearchByValue(searchObj);
-		} catch (e) {
-			let message = `There was an error inserting a new job: ${e}`;
-			log.error(message);
-			return result;
-		}
+		foundJob = await pSearchByValue(searchObj);
 		//TODO: Once https://harperdb.atlassian.net/browse/HDB-501 is resolved, this check is no longer needed.
 		foundValues = Array.isArray(foundJob) ? foundJob : Object.keys(foundJob);
 		if (foundValues && foundValues.length > 0) {
@@ -216,14 +203,7 @@ export async function addJob(jsonBody: any) {
 		'id',
 		[newJob]
 	);
-	let insertResult;
-	try {
-		insertResult = await pInsert(insertObject);
-	} catch (e) {
-		log.error(`There was an error inserting a job for job type: ${jsonBody.operation} -- ${e}`);
-		result.success = false;
-		return result;
-	}
+	const insertResult = await pInsert(insertObject);
 
 	if (insertResult.inserted_hashes.length === 0) {
 		result.message = `Had a problem creating a job with type ${newJob.operation} and id ${newJob.id}`;

server/serverHelpers/serverUtilities.ts

@@ -596,21 +596,18 @@ interface JobResult {
 export async function executeJob(json: OperationRequestBody): Promise<JobResult> {
 	transformReq(json);
 
-	let newJobObject;
-	let result;
 	try {
-		result = await jobs.addJob(json);
-		if (result) {
-			newJobObject = result.createdJob;
-			operationLog.info('addJob result', result);
-			const jobRunnerMessage = new jobRunner.RunnerMessage(newJobObject, json);
-			const returnMessage = await jobRunner.parseMessage(jobRunnerMessage);
-
-			return {
-				message: returnMessage ?? `Starting job with id ${newJobObject.id}`,
-				job_id: newJobObject.id,
-			};
-		}
+		const result = await jobs.addJob(json);
+		const newJobObject = result.createdJob;
+		if (!newJobObject) throw new Error(result.error || result.message || `Unable to create a ${json.operation} job`);
+		operationLog.info('addJob result', result);
+		const jobRunnerMessage = new jobRunner.RunnerMessage(newJobObject, json);
+		const returnMessage = await jobRunner.parseMessage(jobRunnerMessage);
+
+		return {
+			message: returnMessage ?? `Starting job with id ${newJobObject.id}`,
+			job_id: newJobObject.id,
+		};
 	} catch (err) {
 		// errors that already carry a statusCode (e.g. ClientError from job validation) are
 		// client-facing as-is; wrapping them here would turn a 400/404/409 into a 500

unitTests/bin/fixtures/replicatedRestartHarness.cjs

@@ -0,0 +1,30 @@
+'use strict';
+
+// Runs a replicated restart_service the way its job worker does — off the main thread, waiting for the
+// main thread's restart-complete — against peers whose answers come from workerData.
+const { workerData, parentPort } = require('node:worker_threads');
+require('../../testUtils.js').preTestPrep();
+const { getThisNodeName } = require('#src/server/nodeName');
+
+const unhandled = [];
+process.on('unhandledRejection', (reason) => unhandled.push(String(reason?.message ?? reason)));
+const { dispatchFailures = {}, pollFailures = {} } = workerData;
+global.server = {
+	nodes: [{ name: getThisNodeName() }, { name: 'peer-a' }, { name: 'peer-b' }],
+	replication: {
+		monitorNodeCAs() {},
+		async sendOperationToNode(node, operation) {
+			const failures = operation.operation === 'get_job' ? pollFailures : dispatchFailures;
+			if (failures[node.name]?.length) throw new Error(failures[node.name].shift());
+			if (operation.operation === 'get_job')
+				return { results: [{ status: 'COMPLETE', message: `restarted ${node.name}` }] };
+			return { job_id: `job-${node.name}` };
+		},
+	},
+};
+const { restartService } = require('#src/bin/restart');
+const settle = (outcome) => setImmediate(() => parentPort.postMessage({ ...outcome, unhandled }));
+restartService({ operation: 'restart_service', service: 'http', replicated: true }).then(
+	(value) => settle({ completed: value }),
+	(error) => settle({ failed: { message: error.message, result: error.result } })
+);