diff --git a/replication/replicationConnection.ts b/replication/replicationConnection.ts index c28a7e804..81ed92355 100644 --- a/replication/replicationConnection.ts +++ b/replication/replicationConnection.ts @@ -327,6 +327,82 @@ 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. + * + * 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`. + */ +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(); + // 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 + ); + 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'`, @@ -3406,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( () => { @@ -3431,21 +3509,32 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) } ); } catch (error) { + // 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' + 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, 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 @@ -3681,7 +3770,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..60c8c33ee --- /dev/null +++ b/unitTests/replication/closeOnInboundMessageError.test.mjs @@ -0,0 +1,90 @@ +/** + * 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); + }); + + 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); + }); +}); 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']); + }); +});