fix(deploy): collapse hdb_deployment write burst so peers converge

The deploy lifecycle writes the hdb_deployment row ~10 times within a few
hundred ms (create, payload ingest, phase flushes, per-peer results, finish).
These replicate to peers; on a loaded peer the rapid same-key writes can commit
out of order, where an older full update reverts the terminal `success` write —
the peer row stays stuck at `replicating` and never converges (#1170).

Add DeploymentRecorder.seal(), called before the replicate phase: scheduleFlush()
stops issuing puts (state accumulates in memory) and finish() performs a single
terminal write, isolating it from the concurrent same-key burst so the receiver
converges. The ProgressEmitter still emits live SSE events; only the origin's
get_deployment polling view skips the transient `replicating` status and
incremental peer_results during the final phase. Mitigation pending #1170.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: kriszyp

components/deploymentRecorder.ts

@@ -72,6 +72,7 @@ export class DeploymentRecorder {
 	private unsubscribe: (() => void) | null = null;
 	private pendingPut: Promise<void> | null = null;
 	private dirty = false;
+	private sealed = false;
 
 	private constructor(deploymentId: string, initial: Record<string, any>) {
 		this.deploymentId = deploymentId;
@@ -151,6 +152,12 @@ export class DeploymentRecorder {
 	// the record dirty; the chained continuation issues a follow-up put once the prior one
 	// settles. This keeps event_log writes O(1) puts per burst rather than O(N) per event.
 	private scheduleFlush(): void {
+		if (this.sealed) {
+			// Sealed: accumulate state in memory but don't write. finish() does the single
+			// terminal write. See seal() for why. The emitter still emits live SSE events.
+			this.dirty = true;
+			return;
+		}
 		if (this.pendingPut) {
 			this.dirty = true;
 			return;
@@ -264,6 +271,24 @@ export class DeploymentRecorder {
 		for (const result of results) this.recordPeer(result);
 	}
 
+	/**
+	 * Stop persisting intermediate row updates; accumulate them in memory so finish() writes
+	 * the terminal state in a single put. Called before the replicate phase, where the row
+	 * otherwise receives a tight burst of puts (replicate phase + per-peer + finish) within
+	 * a few ms. That burst can commit out of order on a loaded peer, where an older full
+	 * update reverts the terminal `success` write — the row stays stuck at `replicating` and
+	 * never converges (harperdb/harper#1170). Collapsing to one terminal write isolates it
+	 * from any concurrent same-key write so the receiver converges.
+	 *
+	 * Tradeoff: the origin's get_deployment *polling* view skips the transient `replicating`
+	 * status and incremental peer_results during the final phase; live SSE tailing is
+	 * unaffected (the emitter still emits in real time). Once #1170 lands this seal can be
+	 * removed to restore incremental peer_results persistence.
+	 */
+	seal(): void {
+		this.sealed = true;
+	}
+
 	async finish(status: 'success' | 'failed' | 'rolled_back', error?: unknown): Promise<void> {
 		if (this.finished) return;
 		// Send a terminal sentinel through the emitter (if any) BEFORE we unsubscribe and

components/operations.js

@@ -508,6 +508,11 @@ async function deployComponent(req) {
 					emit('peer', result);
 				}
 			: undefined;
+		// Seal the recorder before the replicate phase so the row's terminal write (finish())
+		// isn't part of the tight put burst that can commit out of order on a peer and revert
+		// it (harperdb/harper#1170). onPeerResult/peer_results accumulate in memory and land in
+		// finish()'s single write; live SSE 'peer' events still fire below.
+		recorder?.seal();
 		emit('phase', { phase: 'replicate', status: 'start' });
 		let response = await server.replication.replicateOperation(req, { onPeerResult });
 		emit('phase', { phase: 'replicate', status: 'done' });

unitTests/components/deploymentRecorder.test.js

@@ -161,6 +161,62 @@ describe('DeploymentRecorder.recordPeers (bulk wrapper)', () => {
 	});
 });
 
+describe('DeploymentRecorder.seal', () => {
+	let installed;
+	let putLog;
+	beforeEach(() => {
+		putLog = [];
+		const rows = new Map();
+		const mock = {
+			rows,
+			async get(id) {
+				return rows.get(id);
+			},
+			async put(row) {
+				putLog.push({ status: row.status, peerCount: (row.peer_results ?? []).length });
+				rows.set(row.deployment_id, { ...row, peer_results: [...(row.peer_results ?? [])] });
+			},
+		};
+		if (!databases.system) databases.system = {};
+		const prior = databases.system[DEPLOYMENT_TABLE];
+		databases.system[DEPLOYMENT_TABLE] = mock;
+		installed = {
+			mock,
+			restore() {
+				databases.system[DEPLOYMENT_TABLE] = prior;
+			},
+		};
+	});
+	afterEach(() => installed.restore());
+
+	it('stops persisting intermediate updates once sealed, but finish() writes the terminal state', async () => {
+		const recorder = await DeploymentRecorder.create({ project: 'p' });
+		const putsAfterCreate = putLog.length;
+		recorder.seal();
+		recorder.recordPeer({ node: 'a', status: 'success' });
+		recorder.recordPeer({ node: 'b', status: 'success' });
+		assert.strictEqual(recorder.row.peer_results.length, 2, 'peer_results accumulate in memory while sealed');
+		assert.strictEqual(putLog.length, putsAfterCreate, 'no puts are issued while sealed (pre-finish)');
+
+		await recorder.finish('success');
+		const terminal = putLog[putLog.length - 1];
+		assert.strictEqual(terminal.status, 'success', 'finish() persists the terminal status');
+		assert.strictEqual(terminal.peerCount, 2, 'finish() carries the accumulated peer_results');
+		const persisted = await installed.mock.get(recorder.deploymentId);
+		assert.strictEqual(persisted.status, 'success');
+		assert.strictEqual(persisted.peer_results.length, 2);
+	});
+
+	it('does not affect persistence before seal: recordPeer still flushes incrementally', async () => {
+		const recorder = await DeploymentRecorder.create({ project: 'p' });
+		const putsAfterCreate = putLog.length;
+		recorder.recordPeer({ node: 'a', status: 'success' });
+		// scheduleFlush issues the put asynchronously; let it settle.
+		await new Promise((resolve) => setImmediate(resolve));
+		assert.ok(putLog.length > putsAfterCreate, 'an unsealed recordPeer persists incrementally');
+	});
+});
+
 describe('awaitDeploymentRow', () => {
 	let installed;
 	beforeEach(() => {