From 2b3f1509cdeaa5d98a243c778a46ad76cc451d2f Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 8 Jun 2026 15:42:01 -0600 Subject: [PATCH] fix(deploy): collapse hdb_deployment write burst so peers converge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 (harperdb/harper#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 --- components/deploymentRecorder.ts | 25 +++++++++ components/operations.js | 5 ++ .../components/deploymentRecorder.test.js | 56 +++++++++++++++++++ 3 files changed, 86 insertions(+) diff --git a/components/deploymentRecorder.ts b/components/deploymentRecorder.ts index aa350a56b..631bd2a3a 100644 --- a/components/deploymentRecorder.ts +++ b/components/deploymentRecorder.ts @@ -72,6 +72,7 @@ export class DeploymentRecorder { private unsubscribe: (() => void) | null = null; private pendingPut: Promise | null = null; private dirty = false; + private sealed = false; private constructor(deploymentId: string, initial: Record) { 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 { if (this.finished) return; // Send a terminal sentinel through the emitter (if any) BEFORE we unsubscribe and diff --git a/components/operations.js b/components/operations.js index b2dbba014..982492a16 100644 --- a/components/operations.js +++ b/components/operations.js @@ -507,6 +507,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' }); diff --git a/unitTests/components/deploymentRecorder.test.js b/unitTests/components/deploymentRecorder.test.js index 5575453bb..39b7f5aec 100644 --- a/unitTests/components/deploymentRecorder.test.js +++ b/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(() => {