Skip to content

Commit 2b3f150

Browse files
kriszypclaude
andcommitted
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>
1 parent 6cc8ea7 commit 2b3f150

3 files changed

Lines changed: 86 additions & 0 deletions

File tree

components/deploymentRecorder.ts

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@ export class DeploymentRecorder {
7272
private unsubscribe: (() => void) | null = null;
7373
private pendingPut: Promise<void> | null = null;
7474
private dirty = false;
75+
private sealed = false;
7576

7677
private constructor(deploymentId: string, initial: Record<string, any>) {
7778
this.deploymentId = deploymentId;
@@ -151,6 +152,12 @@ export class DeploymentRecorder {
151152
// the record dirty; the chained continuation issues a follow-up put once the prior one
152153
// settles. This keeps event_log writes O(1) puts per burst rather than O(N) per event.
153154
private scheduleFlush(): void {
155+
if (this.sealed) {
156+
// Sealed: accumulate state in memory but don't write. finish() does the single
157+
// terminal write. See seal() for why. The emitter still emits live SSE events.
158+
this.dirty = true;
159+
return;
160+
}
154161
if (this.pendingPut) {
155162
this.dirty = true;
156163
return;
@@ -264,6 +271,24 @@ export class DeploymentRecorder {
264271
for (const result of results) this.recordPeer(result);
265272
}
266273

274+
/**
275+
* Stop persisting intermediate row updates; accumulate them in memory so finish() writes
276+
* the terminal state in a single put. Called before the replicate phase, where the row
277+
* otherwise receives a tight burst of puts (replicate phase + per-peer + finish) within
278+
* a few ms. That burst can commit out of order on a loaded peer, where an older full
279+
* update reverts the terminal `success` write — the row stays stuck at `replicating` and
280+
* never converges (harperdb/harper#1170). Collapsing to one terminal write isolates it
281+
* from any concurrent same-key write so the receiver converges.
282+
*
283+
* Tradeoff: the origin's get_deployment *polling* view skips the transient `replicating`
284+
* status and incremental peer_results during the final phase; live SSE tailing is
285+
* unaffected (the emitter still emits in real time). Once #1170 lands this seal can be
286+
* removed to restore incremental peer_results persistence.
287+
*/
288+
seal(): void {
289+
this.sealed = true;
290+
}
291+
267292
async finish(status: 'success' | 'failed' | 'rolled_back', error?: unknown): Promise<void> {
268293
if (this.finished) return;
269294
// Send a terminal sentinel through the emitter (if any) BEFORE we unsubscribe and

components/operations.js

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -507,6 +507,11 @@ async function deployComponent(req) {
507507
emit('peer', result);
508508
}
509509
: undefined;
510+
// Seal the recorder before the replicate phase so the row's terminal write (finish())
511+
// isn't part of the tight put burst that can commit out of order on a peer and revert
512+
// it (harperdb/harper#1170). onPeerResult/peer_results accumulate in memory and land in
513+
// finish()'s single write; live SSE 'peer' events still fire below.
514+
recorder?.seal();
510515
emit('phase', { phase: 'replicate', status: 'start' });
511516
let response = await server.replication.replicateOperation(req, { onPeerResult });
512517
emit('phase', { phase: 'replicate', status: 'done' });

unitTests/components/deploymentRecorder.test.js

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,62 @@ describe('DeploymentRecorder.recordPeers (bulk wrapper)', () => {
161161
});
162162
});
163163

164+
describe('DeploymentRecorder.seal', () => {
165+
let installed;
166+
let putLog;
167+
beforeEach(() => {
168+
putLog = [];
169+
const rows = new Map();
170+
const mock = {
171+
rows,
172+
async get(id) {
173+
return rows.get(id);
174+
},
175+
async put(row) {
176+
putLog.push({ status: row.status, peerCount: (row.peer_results ?? []).length });
177+
rows.set(row.deployment_id, { ...row, peer_results: [...(row.peer_results ?? [])] });
178+
},
179+
};
180+
if (!databases.system) databases.system = {};
181+
const prior = databases.system[DEPLOYMENT_TABLE];
182+
databases.system[DEPLOYMENT_TABLE] = mock;
183+
installed = {
184+
mock,
185+
restore() {
186+
databases.system[DEPLOYMENT_TABLE] = prior;
187+
},
188+
};
189+
});
190+
afterEach(() => installed.restore());
191+
192+
it('stops persisting intermediate updates once sealed, but finish() writes the terminal state', async () => {
193+
const recorder = await DeploymentRecorder.create({ project: 'p' });
194+
const putsAfterCreate = putLog.length;
195+
recorder.seal();
196+
recorder.recordPeer({ node: 'a', status: 'success' });
197+
recorder.recordPeer({ node: 'b', status: 'success' });
198+
assert.strictEqual(recorder.row.peer_results.length, 2, 'peer_results accumulate in memory while sealed');
199+
assert.strictEqual(putLog.length, putsAfterCreate, 'no puts are issued while sealed (pre-finish)');
200+
201+
await recorder.finish('success');
202+
const terminal = putLog[putLog.length - 1];
203+
assert.strictEqual(terminal.status, 'success', 'finish() persists the terminal status');
204+
assert.strictEqual(terminal.peerCount, 2, 'finish() carries the accumulated peer_results');
205+
const persisted = await installed.mock.get(recorder.deploymentId);
206+
assert.strictEqual(persisted.status, 'success');
207+
assert.strictEqual(persisted.peer_results.length, 2);
208+
});
209+
210+
it('does not affect persistence before seal: recordPeer still flushes incrementally', async () => {
211+
const recorder = await DeploymentRecorder.create({ project: 'p' });
212+
const putsAfterCreate = putLog.length;
213+
recorder.recordPeer({ node: 'a', status: 'success' });
214+
// scheduleFlush issues the put asynchronously; let it settle.
215+
await new Promise((resolve) => setImmediate(resolve));
216+
assert.ok(putLog.length > putsAfterCreate, 'an unsealed recordPeer persists incrementally');
217+
});
218+
});
219+
164220
describe('awaitDeploymentRow', () => {
165221
let installed;
166222
beforeEach(() => {

0 commit comments

Comments
 (0)