fix(replication): fail loud on oversized replication frame instead of silent data loss

integrationTests/cluster/oversizedFrameCursorSafety.test.mjs

@@ -0,0 +1,96 @@
+/**
+ * Integration test: an oversized replication transaction must not be silently skipped.
+ *
+ * The production failure (harper-pro#711) was permanent data loss: a queued transaction larger than
+ * replication_maxPayload was dropped with a bare `return`, and later transactions kept advancing the
+ * peer's resume cursor past it. The fix throws instead, which closes the leg so it resumes from the
+ * un-advanced cursor. This pins the cursor-safety invariant: once a transaction is too large to send,
+ * NO later transaction is delivered ahead of it.
+ *
+ * Setup: a small maxPayload on the source. Write a normal record (delivered), then an oversized record
+ * (wedges the send leg), then another normal record. The follower must have the first record but NOT
+ * the third. On base (silent drop) the third arrives while the oversized one is lost, so this fails.
+ *
+ * Keys are ordered k1 < k2 (oversized) < k3 so that neither delivery path can hand out the later record
+ * early: a wedged leg reconnects into a key-ordered full copy, so k3 sorting AFTER the oversized k2
+ * (rather than alphabetically before it) is what makes its absence a real cursor-safety signal.
+ */
+import { suite, test, before, after } from 'node:test';
+import { equal, ok } from 'node:assert/strict';
+import { setTimeout as delay } from 'node:timers/promises';
+import { startHarper, teardownHarper, getNextAvailableLoopbackAddress } from '@harperfast/integration-testing';
+import { join } from 'node:path';
+import { sendOperation } from './clusterShared.mjs';
+
+process.env.HARPER_INTEGRATION_TEST_INSTALL_SCRIPT = join(import.meta.dirname ?? module.path, '..', '..', 'dist', 'bin', 'harper.js');
+
+const MAX_PAYLOAD = 512 * 1024; // 512 KiB send cap
+const OVERSIZED_BYTES = 1024 * 1024; // 1 MiB value -> transaction frame exceeds the cap
+
+suite('oversized replication transaction does not skip later transactions', { timeout: 180000 }, (ctx) => {
+	before(async () => {
+		const hostnameA = await getNextAvailableLoopbackAddress();
+		const hostnameB = await getNextAvailableLoopbackAddress();
+		const makeNodeCtx = (hostname) => ({ name: ctx.name, harper: { hostname } });
+		const commonConfig = (hostname) => ({
+			config: {
+				analytics: { aggregatePeriod: -1 },
+				logging: { colors: false, stdStreams: false, console: true },
+				replication: { port: hostname + ':9933', securePort: null, databases: ['data'], maxPayload: MAX_PAYLOAD },
+			},
+			env: { HARPER_NO_FLUSH_ON_EXIT: true },
+		});
+		const ctxA = makeNodeCtx(hostnameA);
+		const ctxB = makeNodeCtx(hostnameB);
+		await Promise.all([startHarper(ctxA, commonConfig(hostnameA)), startHarper(ctxB, commonConfig(hostnameB))]);
+		ctx.nodeA = ctxA.harper;
+		ctx.nodeB = ctxB.harper;
+
+		await sendOperation(ctx.nodeA, { operation: 'create_table', database: 'data', table: 'oversized_test', primary_key: 'id' });
+		// Establish replication before any oversized write, so the base copy (small) is unaffected.
+		await sendOperation(ctx.nodeB, {
+			operation: 'add_node',
+			hostname: ctx.nodeA.hostname,
+			rejectUnauthorized: false,
+			isLeader: true,
+			authorization: ctx.nodeA.admin,
+		});
+	});
+
+	after(async () => {
+		await Promise.all([ctx.nodeA && teardownHarper({ harper: ctx.nodeA }), ctx.nodeB && teardownHarper({ harper: ctx.nodeB })]);
+	});
+
+	test('the follower receives the pre-oversized record but not the one queued behind it', async () => {
+		const { nodeA, nodeB } = ctx;
+		const hasId = async (id) =>
+			((await sendOperation(nodeB, {
+				operation: 'search_by_value',
+				database: 'data',
+				table: 'oversized_test',
+				search_attribute: 'id',
+				search_value: id,
+				get_attributes: ['id'],
+			}).catch(() => [])) ?? []).length > 0;
+
+		// A normal record, delivered normally before the wedge.
+		await sendOperation(nodeA, { operation: 'upsert', database: 'data', table: 'oversized_test', records: [{ id: 'k1_before', v: 'x' }] });
+		for (let i = 0; i < 60 && !(await hasId('k1_before')); i++) await delay(500);
+		ok(await hasId('k1_before'), 'the pre-oversized record must replicate normally');
+
+		// An oversized transaction: too large to send, so the leg wedges on it.
+		await sendOperation(nodeA, {
+			operation: 'upsert',
+			database: 'data',
+			table: 'oversized_test',
+			records: [{ id: 'k2_oversized', v: 'x'.repeat(OVERSIZED_BYTES) }],
+		});
+		// A later normal transaction: must NOT be delivered ahead of the wedged one.
+		await sendOperation(nodeA, { operation: 'upsert', database: 'data', table: 'oversized_test', records: [{ id: 'k3_after', v: 'x' }] });
+
+		// Give the leg ample time to (wrongly, on base) skip and deliver 'after'.
+		await delay(10000);
+		equal(await hasId('k3_after'), false, 'a transaction queued behind an oversized one must not be delivered');
+		equal(await hasId('k2_oversized'), false, 'the oversized transaction itself cannot be sent under this cap');
+	});
+});

replication/replicationConnection.ts

@@ -236,6 +236,20 @@
 }
 
 const MAX_PAYLOAD = env.get('replication_maxPayload') ?? 100_000_000;
+// Flush a bulk-copy batch once it approaches the payload cap, so a run of large rows can't assemble a
+// single oversized frame (harper-pro#711) that would throw and wedge the copy with no operator escape
+// (existing rows can't be split). Leaves headroom for the in-progress record; a single row above this
+// still relies on raising replication_maxPayload.
+const COPY_FLUSH_BYTES = Math.floor(MAX_PAYLOAD * 0.9);
+
+// Pure oversized-frame predicate, extracted so it is unit-testable (the frameWriter/copyFlushPacer
+// pattern, harper-pro#440). A frame at exactly the cap is allowed; only a strictly larger one is rejected.
+export function exceedsMaxPayload(messageSize: number, maxPayload: number = MAX_PAYLOAD): boolean {
+	return messageSize > maxPayload;
+}
+// Throttle the oversized-send error: it closes and reconnects the leg, so it would otherwise re-log on
+// every reconnect cycle. Module-level so the throttle survives across the per-cycle connection instances.
+const oversizedSendWarnThrottle = createThrottleState();
 // When receiving a replication message, we apply per-record backpressure to keep a single
 // large batch from synchronously decoding thousands of records and ballooning the worker
 // heap past its limit. If the local replicator queue grows beyond this threshold we pause
@@ -831,7 +845,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
 }
 
@@ -2056,6 +2070,10 @@
 		noDelay: true, // we want to send the data immediately
 		// we set this very high (2x times the v22 default) because it performs better
 		highWaterMark: 128 * 1024,
+		// Match the server's accept cap (replicator.ts) so the configurable send guard (replication_maxPayload)
+		// is the binding limit on client-initiated legs too. Otherwise ws defaults to 100 MiB here, just under
+		// the 100 MB send default, and raising replication_maxPayload to clear a wedge would hit a silent 1009.
+		maxPayload: 10 * 1024 * 1024 * 1024,
 		rejectUnauthorized: rejectUnauthorized !== false,
 		secureContext: undefined,
 	};
@@ -2189,8 +2207,9 @@
 			this.socket._socket.unref();
 			// in normal startup, just use info, but adjust log level to warn if we were previously disconnected, because there was a warn message on the disconnect and we want to keep symmetry
 			logger[this.isConnected ? 'info' : 'warn']?.(`Connected to ${this.url}, db: ${this.databaseName}`);
-			this.retries = 0;
-			this.retryTime = INITIAL_RETRY_TIME;
+			// Reset backoff on first successful SEND (onFrameSent), not here on open: a leg that reopens and
+			// immediately fails to send (an oversized frame throws and closes it) must keep escalating toward
+			// the 30 s cap instead of hot-looping at 500 ms and accumulating native TLS state (harper-pro#339).
 			// if we have already connected, we need to send a reconnected event
 			if (this.nodeSubscriptions) {
 				connectedToNode({
@@ -2331,6 +2350,14 @@
 		// Doubling reaches 30 s in ~6 retries (~62 s total) and resets on success.
 		this.retryTime = Math.min(this.retryTime << 1, 30_000);
 	}
+	// Called by replicateOverWS after a frame is actually sent: real progress, so it is safe to reset the
+	// backoff. Gated on a non-zero retries so the healthy hot path (already reset) does nothing.
+	onFrameSent() {
+		if (this.retries !== 0 || this.retryTime !== INITIAL_RETRY_TIME) {
+			this.retries = 0;
+			this.retryTime = INITIAL_RETRY_TIME;
+		}
+	}
 	// Retire the live replicateOverWS instance: the single enforcement point for "at most one live session
 	// per connection". Every path that supersedes a session (socket replaced in connect(), forceReconnect)
 	// calls this; the socket's own 'close' does the same teardown through retireInstance, which is
@@ -4623,8 +4650,15 @@
 						const sendQueuedData = () => {
 							if (frame.position - frame.encodingStart > 8) {
 								// if we have more than just a txn time, send it
-								if (checkExcessMessageSize(frame.position - frame.encodingStart)) return;
+								// Throw, don't return: a silent return skips this transaction while later ones
+								// keep advancing the peer's resume cursor past it, losing it permanently. Throwing
+								// closes the leg so it resumes from the un-advanced cursor (as the blob path does).
+								if (checkExcessMessageSize(frame.position - frame.encodingStart))
+									throw new Error('Replication message too large to send');
 								ws.send(frame.encodingBuffer.subarray(frame.encodingStart, frame.position));
+								// A frame actually went out: tell the outbound connection so it can reset its reconnect
+								// backoff on genuine progress rather than on bare socket-open (harper-pro#339).
+								options.connection?.onFrameSent?.();
 								logger.debug?.(connectionId, 'Sent message, size:', frame.position - frame.encodingStart);
 								if (databaseName !== 'system') {
 									recordAction(
@@ -4948,7 +4982,8 @@
 													// naturally above; this also bounds same-version bulk data into committable batches.) The
 													// watermark is only advanced to copyStartTime by the single end_txn after the whole copy.
 													if (
-														++recordsSinceCheckpoint >= COPY_CHECKPOINT_RECORDS &&
+														(++recordsSinceCheckpoint >= COPY_CHECKPOINT_RECORDS ||
+															frame.position - frame.encodingStart >= COPY_FLUSH_BYTES) &&
 														frame.position - frame.encodingStart > 8
 													) {
 														recordsSinceCheckpoint = 0;
@@ -6654,16 +6689,19 @@
 	};
 
 	function checkExcessMessageSize(messageSize) {
-		if (messageSize > MAX_PAYLOAD) {
-			logger.error?.(
-				connectionId,
-				'Message too large to send, size:',
-				messageSize,
-				'remote node:',
-				remoteNodeName,
-				'database:',
-				databaseName
-			);
+		if (exceedsMaxPayload(messageSize)) {
+			const { emit, suppressedCount } = decideThrottledWarn(oversizedSendWarnThrottle, Date.now());
+			if (emit)
+				logger.error?.(
+					connectionId,
+					'Message too large to send, size:',
+					messageSize,
+					'remote node:',
+					remoteNodeName,
+					'database:',
+					databaseName,
+					suppressedCount ? `(${suppressedCount} suppressed)` : ''
+				);
 			return true;
 		}
 	}

unitTests/replication/exceedsMaxPayload.test.mjs

@@ -0,0 +1,30 @@
+/**
+ * Coverage for `exceedsMaxPayload` — the pure oversized-frame predicate extracted from the connection
+ * closure (harper-pro#711 review) so the send-or-throw decision can be pinned without a live socket. The
+ * boundary is inclusive: a frame at exactly the cap is allowed; only a strictly larger one is rejected,
+ * matching the ws server's own accept semantics.
+ */
+import { expect } from 'chai';
+import { exceedsMaxPayload } from '#src/replication/replicationConnection';
+
+describe('exceedsMaxPayload', () => {
+	const cap = 100_000_000;
+
+	it('allows a frame below the cap', () => {
+		expect(exceedsMaxPayload(0, cap)).to.equal(false);
+		expect(exceedsMaxPayload(cap - 1, cap)).to.equal(false);
+	});
+
+	it('allows a frame exactly at the cap (boundary is inclusive)', () => {
+		expect(exceedsMaxPayload(cap, cap)).to.equal(false);
+	});
+
+	it('rejects a frame strictly above the cap', () => {
+		expect(exceedsMaxPayload(cap + 1, cap)).to.equal(true);
+	});
+
+	it('honors an explicit smaller cap', () => {
+		expect(exceedsMaxPayload(2_000_000, 1_000_000)).to.equal(true);
+		expect(exceedsMaxPayload(500_000, 1_000_000)).to.equal(false);
+	});
+});