fix(replication): close on per-record value-decode failure instead of skipping past it (#440)

replication/replicationConnection.ts

@@ -327,6 +327,82 @@
 	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'`,
@@ -451,7 +527,7 @@
 	logger.warn?.(`[test] forcing open-but-idle replication wedge for db "${databaseName}" (harper-pro#420)`);
 	ws.terminate = () => {};
 	ws.close = () => {};
 	ws._socket?.pause();
 	return true; // tell the watchdog to treat this connection's byte count as frozen
 }
 
@@ -3406,6 +3482,8 @@
 				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 @@
 						}
 					);
 				} 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 @@
 			};
 			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);

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);
+	});
+});

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']);
+	});
+});