fix: bound replication receive memory to stop worker OOM crash loops

[kriszyp]

Summary

Stop the receive side of replication from being able to push a worker past its 2 GB old-gen limit during peer catch-up.

Why

A production node was crash-looping with ERR_WORKER_OUT_OF_MEMORY every ~25 seconds (284 OOM events in 2 hours) while catching up a 4-hour prerender-DB backlog from a peer. The chain of cause and effect (verified live with allocation sampling against the 5.0.16 dist):

• onWSMessage decoded every audit record in an incoming transaction batch inside a synchronous do { ... } while (decoder.position < body.byteLength) loop. A single WS message can carry thousands of records; each one created an event holding a decoded value and a closure over the source body buffer, and pushed it onto tableSubscriptionToReplicator with no intra-message backpressure. outstandingCommits was only incremented after the loop ended, so ws.pause() could never fire mid-message.
• Allocation profile against the running worker showed readAuditEntry @ auditStore.js:442 at 84% of all allocations. Heap was stable at ~70 MB between crashes; the OOM is a sub-second burst that GC reclaims completely after the worker dies.
• Every worker death triggered onDatabase(...) reassignments for every database it held, which all fired in the same tick — a fresh worker was instantly slammed with catch-up connections and OOM'd in turn. The log shows "Setting up subscription with leader …" repeating in bursts after each crash.

A BLOB_CHUNK handler bug compounds this: stream.write(blobBody) ignored the return value, so when saveBlob's file-write pipeline fell behind, blob chunks accumulated in the PassThrough buffer (627 MB external memory on surviving workers).

What changed

Three independent fixes, all in replication/:

1. replicationConnection.ts receive loop — convert onWSMessage to async, serialize handler invocations via a messageProcessing promise chain (preserves WS message order), and inside the per-record do/while check tableSubscriptionToReplicator.queue.length. When it exceeds RECEIVE_EVENT_HIGH_WATER_MARK (default 100, env-tunable via replication_receiveEventHighWaterMark) we pause the WS and await tableSubscriptionToReplicator.waitForDrain() before continuing. The existing EVENT_HIGH_WATER_MARK = 100 pattern in core/resources/Table.ts:2753,2836 is reused.
2. replicationConnection.ts BLOB_CHUNK — respect the stream.write() return value. On false, pause the WS and listen on both 'drain' and 'close' (so a destroyed stream doesn't strand the pause reason).
3. subscriptionManager.ts worker-exit stagger — when a worker exits, space its onDatabase reassignments by WORKER_EXIT_REASSIGN_STAGGER_MS (100 ms) using a rolling nextWorkerExitReassignAt timestamp.

Because we now have three independent reasons to pause receive (commit backlog, consumer queue, blob stream), the single replicationPaused boolean is replaced with a pauseReasons refcount + addPauseReason()/removePauseReason() helpers so one source resuming the WS can't override another that still wants it paused.

Where to look

• The async conversion of onWSMessage and the messageProcessing serialization (~line 471): is the .then(..., onReject) form correct for keeping the chain alive through one message's failure? Cross-model review (Gemini) was positive.
• The pauseReasons refcount and the commitBacklogPaused boolean that gates the original commit-backlog pause/resume. The existing semantics — pause once, resume on any onCommit — are preserved.
• BLOB_CHUNK release listener pair (on('drain', release); on('close', release)): each pause reason is decremented exactly once. stream.off removes the exact-same function reference so it's safe.
• The stagger uses module-level state; under concurrent worker-death this is single-thread main, so no synchronization needed.

Testing

• npm run build → same 60 pre-existing TS errors before and after this change, dist emitted.
• npm run lint:required → 0 errors.
• npm run test:integration -- integrationTests/cluster/replicationLoad.test.mjs fails identically on this branch and on main (pre-existing breakage, database 'data' does not exist). Not introduced here.

⚠️ The harper-pro outer repo has no unit-test infrastructure for the replication/ modules. The new code paths are exercised by the existing replication integration tests; a focused stress test that sends a single WS message containing 10k+ records would be the right follow-up to lock in this fix.

Test plan

• CI passes (or, where it doesn't, fails the same way as main)
• Deploy to one node in a cluster, watch for ERR_WORKER_OUT_OF_MEMORY to stop in docker logs
• Confirm system_information.threads[*].utilization stabilizes and the cascade of "Setting up subscription with leader …" log messages stops repeating
• Verify replication catch-up still completes (cluster_status lastReceivedRemoteTime advances)

🤖 Generated with Claude Code

[claude]

Non-blocker, but worth noting: the messageProcessing chain is never reset when the WS closes. If onWSMessage is mid-await (e.g. inside waitForDrain) when the socket closes, the chain continues running until the drain resolves, keeping the body Buffer alive in the closure. This is bounded — tableSubscriptionToReplicator drains on its own — but calling ws.resume() / ws.pause() on an already-closed socket is a side-effect (harmless with the ws library, which no-ops after destroy).

A simple guard to consider for a follow-up:

Suggested change

	);
	let messageProcessing: Promise<void> = Promise.resolve();
	let wsClosed = false;
	ws.on('message', (body: Buffer) => {
	messageProcessing = messageProcessing.then(
	() => wsClosed ? undefined : onWSMessage(body),
	() => wsClosed ? undefined : onWSMessage(body)
	);
	});

and set wsClosed = true in the ws.on('close', ...) handler. Not required for this fix.

[claude]

Reviewed; no blockers found.