From 6b2612a8e7c3ec87416510f844f9dfa468062cba Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 1 Jul 2026 16:35:16 -0600 Subject: [PATCH 1/4] fix(replication): close on inbound message-handler errors instead of swallowing them (#440) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An error escaping onWSMessage was logged and swallowed: the rest of the failed frame was silently dropped while later frames kept applying and confirming higher sequence ids — a permanent, undetected gap on the receiver (epic #430 Theme B / workstream #440 land-first safety item). Now the catch latches inbound processing off (queued frames on the messageProcessing chain must not commit past the hole) and closes with 1011 as a transient protocol close, so the normal retry path reconnects and resumes from the last durable cursor, re-streaming everything past it. Extracted as closeOnInboundMessageError with unit coverage, following the file's pure-decision-helper idiom. Co-Authored-By: Claude Fable 5 --- replication/replicationConnection.ts | 48 +++++++++++- .../closeOnInboundMessageError.test.mjs | 77 +++++++++++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 unitTests/replication/closeOnInboundMessageError.test.mjs diff --git a/replication/replicationConnection.ts b/replication/replicationConnection.ts index c28a7e804..2f700051a 100644 --- a/replication/replicationConnection.ts +++ b/replication/replicationConnection.ts @@ -327,6 +327,47 @@ export function shouldTerminateIdlePing(idleMs: number, pingTimeout: number, pau return pauseReasons === 0 && idleMs >= pingTimeout; } +/** + * Handle an error that escaped `onWSMessage` (the inbound handler in `replicateOverWS`). + * + * Historically this was logged and swallowed. That silently dropped the rest of the failed frame + * while later frames kept applying and confirming ever-higher sequence ids — a permanent, + * undetected `[error, head]` gap on the receiver (epic harper-pro#430 Theme B; workstream + * harper-pro#440). Recovery instead rides the established transient-close path: + * + * 1. `markInboundClosed` FIRST — frames already queued behind this one on the + * `messageProcessing` chain must not apply (they would commit past the hole), and after + * `ws.close()` the peer can still deliver frames until the close handshake completes. + * 2. `close(1011)` (internal error — distinct from the 1008 policy closes) WITHOUT the + * `intentional` flag, so `NodeReplicationConnection`'s normal retry path reconnects with + * backoff and resumes from the last durable cursor, re-streaming everything past it. + * Records applied before the error redeliver idempotently (version dedup). + * + * A deterministic error (e.g. a genuinely undecodable frame) becomes a visible, backed-off + * reconnect loop instead of silent loss; bounding that with an escalation budget is W2 + * (harper-pro#432) territory. + * + * Exported for unit tests (`closeOnInboundMessageError.test.mjs`); the production caller is the + * catch in `onWSMessage`. + */ +export function closeOnInboundMessageError( + error: unknown, + deps: { + connectionId: string | number; + logger: { error?: (...args: unknown[]) => void }; + markInboundClosed: () => void; + close: (code: number, reason: string) => void; + } +): void { + deps.markInboundClosed(); + deps.logger.error?.( + deps.connectionId, + 'Error handling incoming replication message; closing so replication resumes from the last durable cursor', + error + ); + deps.close(1011, 'Error handling incoming replication message'); +} + /** * Decide whether the empty-subscription delayed close inside `replicateOverWS`'s `scheduleClose` should * be classified as INTENTIONAL/finished (mark `isFinished`/`intentionallyUnsubscribed`, emit `'finished'`, @@ -3681,7 +3722,12 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) }; tableSubscriptionToReplicator.send(endTxnEvent); } catch (error) { - logger.error?.(connectionId, 'Error handling incoming replication message', error); + closeOnInboundMessageError(error, { + connectionId, + logger, + markInboundClosed: () => (wsClosed = true), + close, + }); } } ws.on('ping', resetPingTimer); diff --git a/unitTests/replication/closeOnInboundMessageError.test.mjs b/unitTests/replication/closeOnInboundMessageError.test.mjs new file mode 100644 index 000000000..124785a4c --- /dev/null +++ b/unitTests/replication/closeOnInboundMessageError.test.mjs @@ -0,0 +1,77 @@ +/** + * Coverage for `closeOnInboundMessageError` — the inbound-handler error path in `replicateOverWS`. + * + * An error escaping `onWSMessage` used to be logged and swallowed, silently dropping the rest of + * the failed frame while later frames kept applying and confirming higher sequence ids — a + * permanent, undetected gap (harper-pro#440, epic #430 Theme B). These tests pin the recovery + * contract: inbound processing is latched off BEFORE the socket close (frames already queued on + * the messageProcessing chain must not commit past the hole), and the close is a transient + * 1011 (no `intentional` flag) so the normal retry path reconnects and resumes from the durable + * cursor. + */ + +import { expect } from 'chai'; +import sinon from 'sinon'; +import { closeOnInboundMessageError } from '#src/replication/replicationConnection'; + +describe('closeOnInboundMessageError', () => { + it('latches inbound processing off before closing the socket', () => { + const markInboundClosed = sinon.spy(); + const close = sinon.spy(); + + closeOnInboundMessageError(new Error('boom'), { + connectionId: 7, + logger: { error: sinon.spy() }, + markInboundClosed, + close, + }); + + expect(markInboundClosed.calledOnce).to.equal(true); + expect(close.calledOnce).to.equal(true); + expect(markInboundClosed.calledBefore(close)).to.equal(true); + }); + + it('closes with 1011 (internal error, transient — reconnect path)', () => { + const close = sinon.spy(); + + closeOnInboundMessageError(new Error('boom'), { + connectionId: 7, + logger: {}, + markInboundClosed: () => {}, + close, + }); + + const [code, reason] = close.firstCall.args; + expect(code).to.equal(1011); + expect(reason).to.be.a('string').and.to.have.length.greaterThan(0); + }); + + it('logs the original error with the connection id', () => { + const error = new Error('malformed frame'); + const logError = sinon.spy(); + + closeOnInboundMessageError(error, { + connectionId: 'conn-42', + logger: { error: logError }, + markInboundClosed: () => {}, + close: () => {}, + }); + + expect(logError.calledOnce).to.equal(true); + expect(logError.firstCall.args[0]).to.equal('conn-42'); + expect(logError.firstCall.args).to.include(error); + }); + + it('still closes when the logger has no error level (optional chaining)', () => { + const close = sinon.spy(); + + closeOnInboundMessageError(new Error('boom'), { + connectionId: 7, + logger: {}, + markInboundClosed: () => {}, + close, + }); + + expect(close.calledOnce).to.equal(true); + }); +}); From 297550e32a91c92b597223558923771a7a7be186 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 1 Jul 2026 16:50:44 -0600 Subject: [PATCH 2/4] fix(replication): guard decode-error log dereferences; document the frame-vs-record error boundary (#440) Cross-model review findings on the previous commit: an unknown tableId made the inner decode-catch's own log line throw (unguarded tableDecoder.decoder deref), masking the original error and escaping to the outer close path by accident. Guard the derefs so the intended skip-and-log semantics and truthful diagnostics are restored. Also scope the closeOnInboundMessageError docstring: per-record value-decode failures are still skip-and-logged with the cursor advancing past them (residual tracked in #440 / #432). Co-Authored-By: Claude Fable 5 --- replication/replicationConnection.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/replication/replicationConnection.ts b/replication/replicationConnection.ts index 2f700051a..b5f6ff52d 100644 --- a/replication/replicationConnection.ts +++ b/replication/replicationConnection.ts @@ -347,6 +347,12 @@ export function shouldTerminateIdlePing(idleMs: number, pingTimeout: number, pau * reconnect loop instead of silent loss; bounding that with an escalation budget is W2 * (harper-pro#432) territory. * + * Coverage boundary: this path handles FRAME-level errors (header/command decode, audit-entry + * structure, unexpected rejections from awaited handler work). A per-record VALUE decode failure + * is caught earlier by the inner catch around `decodeBlobsWithWrites` and skip-and-logged, and + * the batch resume cursor still advances past the skipped record — that residual silent-gap + * class is tracked in harper-pro#440 (with W2 #432). + * * Exported for unit tests (`closeOnInboundMessageError.test.mjs`); the production caller is the * catch in `onWSMessage`. */ @@ -3472,10 +3478,13 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) } ); } catch (error) { + // Guarded dereferences: tableDecoder is undefined for an unknown tableId, and an unguarded + // log line here used to throw its own TypeError — masking the original error and escaping + // to the outer catch, so the skip-vs-close outcome was decided by accident (#440). logger.error?.( 'Error decoding replication message, record id: ' + id, - ' typed structures for current decoder' + JSON.stringify(tableDecoder.decoder.typedStructs), - ' structures for current decoder' + JSON.stringify(tableDecoder.decoder.structures), + ' typed structures for current decoder' + JSON.stringify(tableDecoder?.decoder?.typedStructs), + ' structures for current decoder' + JSON.stringify(tableDecoder?.decoder?.structures), 'encoded message', auditRecord.encoded.subarray(0, 1000), auditRecord, From c48788b76774fbbfaed5f9bdf6c61edc1f3481e1 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 1 Jul 2026 16:58:49 -0600 Subject: [PATCH 3/4] fix(replication): harden closeOnInboundMessageError logging (PR #511 review) Gemini findings: guard the logger access fully (the log must never prevent the close) and make the decode-error log readable when the table decoder is unknown. Co-Authored-By: Claude Fable 5 --- replication/replicationConnection.ts | 12 ++++++++---- .../replication/closeOnInboundMessageError.test.mjs | 13 +++++++++++++ 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/replication/replicationConnection.ts b/replication/replicationConnection.ts index b5f6ff52d..0b1d53380 100644 --- a/replication/replicationConnection.ts +++ b/replication/replicationConnection.ts @@ -360,13 +360,15 @@ export function closeOnInboundMessageError( error: unknown, deps: { connectionId: string | number; - logger: { error?: (...args: unknown[]) => void }; + logger?: { error?: (...args: unknown[]) => void }; markInboundClosed: () => void; close: (code: number, reason: string) => void; } ): void { deps.markInboundClosed(); - deps.logger.error?.( + // The log must never prevent the close — this handler is the last line of defense, so the + // logger access is fully guarded rather than trusted. + deps.logger?.error?.( deps.connectionId, 'Error handling incoming replication message; closing so replication resumes from the last durable cursor', error @@ -3483,8 +3485,10 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) // to the outer catch, so the skip-vs-close outcome was decided by accident (#440). logger.error?.( 'Error decoding replication message, record id: ' + id, - ' typed structures for current decoder' + JSON.stringify(tableDecoder?.decoder?.typedStructs), - ' structures for current decoder' + JSON.stringify(tableDecoder?.decoder?.structures), + ' typed structures for current decoder: ' + + (JSON.stringify(tableDecoder?.decoder?.typedStructs) ?? 'unknown table decoder'), + ' structures for current decoder: ' + + (JSON.stringify(tableDecoder?.decoder?.structures) ?? 'unknown table decoder'), 'encoded message', auditRecord.encoded.subarray(0, 1000), auditRecord, diff --git a/unitTests/replication/closeOnInboundMessageError.test.mjs b/unitTests/replication/closeOnInboundMessageError.test.mjs index 124785a4c..60c8c33ee 100644 --- a/unitTests/replication/closeOnInboundMessageError.test.mjs +++ b/unitTests/replication/closeOnInboundMessageError.test.mjs @@ -74,4 +74,17 @@ describe('closeOnInboundMessageError', () => { expect(close.calledOnce).to.equal(true); }); + + it('still closes when the logger itself is missing — the log must never prevent the close', () => { + const close = sinon.spy(); + + closeOnInboundMessageError(new Error('boom'), { + connectionId: 7, + logger: undefined, + markInboundClosed: () => {}, + close, + }); + + expect(close.calledOnce).to.equal(true); + }); }); From fc578afc17e15e3b138b1b1809afbe11035d8186 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Sat, 4 Jul 2026 16:08:23 -0600 Subject: [PATCH 4/4] fix(replication): close on per-record value-decode failure instead of skipping past it The inner catch around decodeBlobsWithWrites in onWSMessage logged and swallowed a per-record value-decode error while the batch resume cursor (maxBatchVersion -> end_txn -> sender COMMITTED_UPDATE) advanced past the un-decoded record regardless of `event` -- a permanent, undetected [error, head] gap on the receiver. This is the #1163/#1453 structure-fork error class (epic harper-pro#430 Theme B; workstream #440). Route it onto the outer transient-close path established by #511: on a resolved-decoder value-decode failure the catch now re-throws (after the existing accepted-blob cleanup, before the cursor advances) so closeOnInboundMessageError closes 1011 -> reconnect + resume from the last durable cursor. The reconnect rebuilds the per-connection table decoder from the peer's re-sent typedStructs/structures (the SET_TABLE handshake), so a forked/stale-structure decode heals on resume; a genuinely undecodable frame degrades to a visible, backed-off reconnect loop rather than silent loss (bounding that loop is W2 / harper-pro#432). An unknown tableId (decoder unresolved -- a distinct, transient schema-propagation case, not a structure fork) stays skip-and-log, since a reconnect wouldn't supply the missing table def. The decision is extracted as the pure helper classifyRecordDecodeError and pinned by recordDecodeErrorRecovery.test.mjs. Co-Authored-By: Claude Opus 4.8 (1M context) --- replication/replicationConnection.ts | 51 ++++++++++++++++--- .../recordDecodeErrorRecovery.test.mjs | 34 +++++++++++++ 2 files changed, 77 insertions(+), 8 deletions(-) create mode 100644 unitTests/replication/recordDecodeErrorRecovery.test.mjs diff --git a/replication/replicationConnection.ts b/replication/replicationConnection.ts index 0b1d53380..81ed92355 100644 --- a/replication/replicationConnection.ts +++ b/replication/replicationConnection.ts @@ -347,11 +347,14 @@ export function shouldTerminateIdlePing(idleMs: number, pingTimeout: number, pau * reconnect loop instead of silent loss; bounding that with an escalation budget is W2 * (harper-pro#432) territory. * - * Coverage boundary: this path handles FRAME-level errors (header/command decode, audit-entry - * structure, unexpected rejections from awaited handler work). A per-record VALUE decode failure - * is caught earlier by the inner catch around `decodeBlobsWithWrites` and skip-and-logged, and - * the batch resume cursor still advances past the skipped record — that residual silent-gap - * class is tracked in harper-pro#440 (with W2 #432). + * Coverage: FRAME-level errors (header/command decode, audit-entry structure, unexpected rejections + * from awaited handler work) reach the outer catch directly. A per-record VALUE decode failure is + * caught first by the inner catch around `decodeBlobsWithWrites` — which logs the decoder's structures + * and the offending bytes for diagnosis, then RE-THROWS (only when the table decoder resolved) so it + * rides this same close path instead of skip-and-logging while the resume cursor advances past the + * hole. That closes the #1163/#1453 structure-fork silent-gap class: the reconnect rebuilds the table + * decoder from the peer's re-sent structures, healing the fork on resume. An unknown tableId (decoder + * unresolved — a transient schema-propagation case, not a fork) stays skip-and-log. * * Exported for unit tests (`closeOnInboundMessageError.test.mjs`); the production caller is the * catch in `onWSMessage`. @@ -376,6 +379,30 @@ export function closeOnInboundMessageError( deps.close(1011, 'Error handling incoming replication message'); } +/** + * Classify how a per-record value-decode failure inside `onWSMessage` should recover — the one decision + * that keeps a decode error from silently, permanently dropping a record. + * + * A RESOLVED table decoder means the offending bytes are a genuine record for a known table whose decoder + * structures forked from the sender's (#1163/#1453 — e.g. an "end of buffer" structon throw). The batch's + * resume cursor (`maxBatchVersion` → end_txn → sender `COMMITTED_UPDATE`) advances regardless of whether the + * record decoded, so skip-and-logging moves the cursor past the hole forever. We `'close'` instead: the + * transient 1011 reconnect rebuilds this table's decoder from the peer's re-sent typedStructs/structures + * (the SET_TABLE handshake) and re-streams from the durable cursor, healing the fork on resume. + * + * An UNRESOLVED decoder is an unknown tableId — a transient schema-propagation case, not a structure fork; + * a reconnect wouldn't supply the missing table def, so closing would only churn. It stays `'skip'`. + * + * (harper-pro#440, epic #430 Theme B. A bounded-retry escalation budget for a deterministically un-healable + * frame — so a genuinely corrupt record can't wedge the link forever — is W2 / harper-pro#432.) + * + * Exported for unit tests (`recordDecodeErrorRecovery.test.mjs`); the production caller is the inner + * value-decode catch in `onWSMessage`. + */ +export function classifyRecordDecodeError(hasTableDecoder: boolean): 'close' | 'skip' { + return hasTableDecoder ? 'close' : 'skip'; +} + /** * Decide whether the empty-subscription delayed close inside `replicateOverWS`'s `scheduleClose` should * be classified as INTENTIONAL/finished (mark `isFinished`/`intentionallyUnsubscribed`, emit `'finished'`, @@ -3455,6 +3482,8 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) const id = auditRecord.recordId; event = undefined; // reset before each decode attempt let receivedBlobs: any[] | undefined; + // Boxed (not the bare error) so the re-throw below can't be defeated by a falsy thrown value. + let decodeFailure: { error: unknown } | undefined; try { decodeBlobsWithWrites( () => { @@ -3480,9 +3509,10 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) } ); } catch (error) { - // Guarded dereferences: tableDecoder is undefined for an unknown tableId, and an unguarded - // log line here used to throw its own TypeError — masking the original error and escaping - // to the outer catch, so the skip-vs-close outcome was decided by accident (#440). + // Guarded dereferences: tableDecoder is undefined for an unknown tableId. An unguarded log + // line here used to throw its own TypeError, masking the original error and escaping to the + // outer catch — closing by accident. The guard keeps the unknown-tableId case skip-and-log; + // the deliberate close for a resolved-decoder value-decode failure is the re-throw below (#440). logger.error?.( 'Error decoding replication message, record id: ' + id, ' typed structures for current decoder: ' + @@ -3494,12 +3524,17 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) auditRecord, error ); + // A resolved-decoder value-decode failure must close, not skip: `maxBatchVersion` (and the + // end_txn resume cursor) advance below regardless of `event`, so skipping loses the record + // permanently. Latch it to re-throw onto the outer close path. See classifyRecordDecodeError. + if (classifyRecordDecodeError(!!tableDecoder) === 'close') decodeFailure = { error }; } if (!event && receivedBlobs) { // decode failed mid-message; the blobs that were already accepted will never be referenced. Give in-flight reads // a window to complete, then unlink the files. (mirrors the pattern at the relocate path above.) setTimeout(() => receivedBlobs.forEach(deleteBlob), 60000).unref(); } + if (decodeFailure) throw decodeFailure.error; // re-throw after blob cleanup, before the resume cursor advances past this record // During a bulk copy, do NOT advance the received-version watermark per copied record: // records arrive in primary-key order carrying their original (possibly newest) versions, so a // single record at the leader's latest timestamp would otherwise let checkSyncStatus mark the diff --git a/unitTests/replication/recordDecodeErrorRecovery.test.mjs b/unitTests/replication/recordDecodeErrorRecovery.test.mjs new file mode 100644 index 000000000..da01b484a --- /dev/null +++ b/unitTests/replication/recordDecodeErrorRecovery.test.mjs @@ -0,0 +1,34 @@ +/** + * Coverage for `classifyRecordDecodeError` — the per-record value-decode recovery decision in + * `onWSMessage`'s inner catch (around `decodeBlobsWithWrites`). + * + * Historically that catch logged and swallowed the error while the batch resume cursor + * (`maxBatchVersion` → end_txn → sender `COMMITTED_UPDATE`) advanced past the un-decoded record — + * a permanent, undetected `[error, head]` gap on the receiver, the #1163/#1453 structure-fork class + * (harper-pro#440, epic #430 Theme B). These tests pin the deliberate decision that replaced it: + * + * - decoder RESOLVED (a real record whose structures forked) → 'close', so the error re-throws + * onto the outer transient-close path (closeOnInboundMessageError → 1011 → reconnect + resume), + * which rebuilds the decoder from the peer's re-sent structures and heals the fork on resume. + * - decoder UNRESOLVED (unknown tableId — transient schema propagation, not a fork) → 'skip', the + * prior skip-and-log behavior, since a reconnect wouldn't supply the missing table def. + */ + +import { expect } from 'chai'; +import { classifyRecordDecodeError } from '#src/replication/replicationConnection'; + +describe('classifyRecordDecodeError', () => { + it('closes on a resolved-decoder value-decode failure (structure-fork class — heals on reconnect, must not skip past)', () => { + expect(classifyRecordDecodeError(true)).to.equal('close'); + }); + + it('skips (skip-and-log) an unresolved decoder — an unknown tableId is transient schema propagation, not a fork', () => { + expect(classifyRecordDecodeError(false)).to.equal('skip'); + }); + + it('returns only the two known dispositions, keyed solely on whether the decoder resolved', () => { + // Guards against a future edit silently introducing a third disposition or flipping the mapping: + // the resume-cursor-safety contract depends on 'close' being the resolved-decoder outcome. + expect([classifyRecordDecodeError(true), classifyRecordDecodeError(false)]).to.deep.equal(['close', 'skip']); + }); +});