Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions components/deploymentRecorder.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
5 changes: 5 additions & 0 deletions components/operations.js
Original file line number Diff line number Diff line change
Expand Up @@ -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' });
Expand Down
56 changes: 56 additions & 0 deletions unitTests/components/deploymentRecorder.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -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(() => {
Expand Down
Loading