fix(replay): recover from corrupt transaction-log frames instead of aborting startup

integrationTests/server/replay-stress.test.ts

@@ -8,19 +8,19 @@
  *   1. Clean crash — replay must recover every row inserted before SIGKILL
  *   2. Truncated tails — last bytes of each txnlog stripped (simulates a torn
  *      write at the moment of crash); replay must still come back up
- *   3. Random byte flips — msgpack-shaped corruption sprinkled across every
- *      txnlog; replay must finish without the CPU-spin regression that
- *      replayLogsGuards.ts (commit d0190ff5a) fixed
+ *   3. Corrupt length prefix — the first entry's declared length is forced to
+ *      overrun the log (a torn/corrupt frame). rocksdb-js's reader throws a bounded
+ *      RangeError for this; replay/broadcast must treat it as end-of-log and the
+ *      server must come back up rather than aborting startup (HarperFast/harper#1135)
  *
  * All three scenarios share one suite/ctx to avoid the schema-registry leak
  * that surfaces when multiple suites each call create_database in the same
  * test-runner process.
  */
 import { suite, test, before, after } from 'node:test';
 import { ok, equal } from 'node:assert/strict';
-import { readdirSync, statSync, openSync, readSync, writeSync, truncateSync, closeSync } from 'node:fs';
+import { readdirSync, readFileSync, statSync, openSync, writeSync, truncateSync, closeSync } from 'node:fs';
 import { join } from 'node:path';
-import { setTimeout as sleep } from 'node:timers/promises';
 
 import {
 	startHarper,
@@ -29,6 +29,12 @@ import {
 	type ContextWithHarper,
 	type HarperContext,
 } from '@harperfast/integration-testing';
+import { constants } from '@harperfast/rocksdb-js';
+
+// Transaction-log framing (all big-endian): a fixed-size file header, then a run of entries
+// each shaped [float64 timestamp][uint32 length][flags byte][length bytes of data]. So an
+// entry's declared length lives at entryStart + 8, and the byte there is its most-significant.
+const { TRANSACTION_LOG_FILE_HEADER_SIZE, TRANSACTION_LOG_ENTRY_HEADER_SIZE } = constants;
 
 const DB = 'stress';
 const TABLES = ['orders', 'items', 'events'];
@@ -128,26 +134,30 @@ function truncateTail(path: string, bytes: number) {
 	truncateSync(path, size - bytes);
 }
 
-function flipBytes(path: string, count: number, seed: number) {
-	const size = statSync(path).size;
-	// Skip the 13-byte file header (4 token + 1 version + 8 ts) so we exercise
-	// the per-entry decoder hardening, not file-open validation.
-	const start = 13;
-	if (size <= start + 32) return;
+function corruptLastEntryLength(path: string): boolean {
+	// Force the *last* well-framed entry's big-endian uint32 length to overrun the log (top
+	// byte → 0xff, ≥ 4 GB). The last entry sits in the unflushed tail that replay reads
+	// (replay starts from the last-flushed position), so a flushed prefix isn't skipped over.
+	const buf = readFileSync(path);
+	let pos = TRANSACTION_LOG_FILE_HEADER_SIZE;
+	let lastLengthPos = -1;
+	while (pos + TRANSACTION_LOG_ENTRY_HEADER_SIZE <= buf.length) {
+		if (buf.readDoubleBE(pos) === 0) break; // a zero timestamp marks end-of-log to the reader
+		const lengthPos = pos + 8;
+		const length = buf.readUInt32BE(lengthPos);
+		const next = pos + TRANSACTION_LOG_ENTRY_HEADER_SIZE + length;
+		if (length === 0 || next > buf.length) break; // ran past the end / already unframable
+		lastLengthPos = lengthPos;
+		pos = next;
+	}
+	if (lastLengthPos < 0) return false;
 	const fd = openSync(path, 'r+');
 	try {
-		const buf = Buffer.alloc(1);
-		let s = seed;
-		for (let i = 0; i < count; i++) {
-			s = (s * 1103515245 + 12345) & 0x7fffffff;
-			const pos = start + (s % (size - start));
-			readSync(fd, buf, 0, 1, pos);
-			buf[0] ^= 0xa5;
-			writeSync(fd, buf, 0, 1, pos);
-		}
+		writeSync(fd, Buffer.from([0xff]), 0, 1, lastLengthPos);
 	} finally {
 		closeSync(fd);
 	}
+	return true;
 }
 
 async function crashAndRestart(ctx: ContextWithHarper, mutate?: (dataRootDir: string) => void) {
@@ -197,37 +207,27 @@ suite('Transaction log replay stress', (ctx: ContextWithHarper) => {
 		}
 	});
 
-	test('crash with random byte flips in txnlogs', async () => {
+	test('crash with corrupt length-prefix recovers without aborting startup', async () => {
 		for (const table of TABLES) {
 			const records = [];
 			for (let i = 0; i < 500; i++) records.push(makeRecord(200_000 + i));
 			await op(ctx.harper, { operation: 'insert', database: DB, table, records });
 		}
-		const replayMs = await crashAndRestart(ctx, (dataRootDir) => {
-			// User-DB only: flipping bytes inside system/ txnlogs corrupts
-			// version-tracking and Harper aborts on next boot with an upgrade
-			// error. That's a real failure mode but not what this test is about —
-			// here we want to exercise replay's per-entry decode hardening on
-			// records that replay genuinely needs to skip rather than reject the
-			// whole boot.
-			const files = listTxnLogFiles(dataRootDir, { userOnly: true });
-			files.forEach((f, i) => flipBytes(f, 32, 0xcafe + i));
+		let corrupted = 0;
+		await crashAndRestart(ctx, (dataRootDir) => {
+			// User-DB only: corrupting system/ txnlogs trips the upgrade-abort path on next
+			// boot — a different failure mode than the framing corruption (#1135) under test.
+			for (const f of listTxnLogFiles(dataRootDir, { userOnly: true })) {
+				if (corruptLastEntryLength(f)) corrupted++;
+			}
 		});
-		// Tighter than the 60s global cap. With the per-entry decode guard in place,
-		// healthy startup is ~3s on this corpus. Without it, every corrupt entry
-		// logs a stack trace inside the loop and startup balloons to ~50s. 20s
-		// catches that regression while leaving CI headroom.
-		ok(replayMs < 20_000, `replay took ${replayMs}ms — guard regression?`);
+		// Fail loudly, not vacuously, if the framing ever changes and nothing gets corrupted.
+		ok(corrupted > 0, 'expected to corrupt at least one user-DB txnlog');
+		// Behavioral, not a wall-clock budget: crashAndRestart resolved → the server came back
+		// up; a regression shows up as a failed restart, not a slow one. Confirm tables queryable.
 		for (const t of TABLES) {
 			const c = await countRows(ctx.harper, t);
 			ok(typeof c === 'number' && c >= 0, `count on ${t} should be a number, got ${c}`);
 		}
-		// Confirm Harper isn't in a CPU-spin loop post-replay: rss should be
-		// roughly stable after startup is reported done. The pre-fix bug pinned
-		// a core forever and rss grew steadily under the spin.
-		const rssBefore = ctx.harper.process.resourceUsage?.()?.maxRSS ?? 0;
-		await sleep(500);
-		const rssAfter = ctx.harper.process.resourceUsage?.()?.maxRSS ?? 0;
-		ok(rssAfter - rssBefore < 200 * 1024, `rss grew by ${rssAfter - rssBefore}KB after replay`);
 	});
 });

resources/RocksTransactionLogStore.ts

@@ -2,6 +2,7 @@ import { TransactionLog, RocksDatabase, shutdown, type TransactionEntry } from '
 import { ExtendedIterable } from '@harperfast/extended-iterable';
 import { getIdOfRemoteNode } from './nodeIdMapping.ts';
 import { Decoder, readAuditEntry, ENTRY_DATAVIEW, AuditRecord, createAuditEntry } from './auditStore.ts';
+import { endIteratorOnCorruptFrame } from './replayLogsGuards.ts';
 import { isMainThread } from 'node:worker_threads';
 import { EventEmitter } from 'node:events';
 import { asBinary } from 'lmdb';
@@ -21,6 +22,13 @@ type TransactionLogIterator = Iterator<TransactionEntry | number> & {
 	removeLog(logName: string);
 };
 
+// Logs (once per log) when a corrupt frame ends a query iterator early; see
+// endIteratorOnCorruptFrame in replayLogsGuards.ts for why this is end-of-log, not a crash.
+function warnCorruptFrame(logName: string) {
+	return (error: RangeError) =>
+		harperLogger.warn(`Stopping transaction log "${logName}" at a corrupt entry during replay`, error);
+}
+
 /**
  * Represents a transaction log store backed by RocksDB.
  * This class provides methods that conform to a standard store interface
@@ -190,7 +198,7 @@ export class RocksTransactionLogStore extends EventEmitter {
 					log = this.rootStore.useLog(options.log);
 				}
 			}
-			const queryIterator = log.query(options);
+			const queryIterator = endIteratorOnCorruptFrame(log.query(options), warnCorruptFrame(log.name));
 			iterable.iterate = () => queryIterator;
 		} else {
 			const onlyKeys = options.onlyKeys;
@@ -217,7 +225,7 @@ export class RocksTransactionLogStore extends EventEmitter {
 								// condition of potentially missing an initial update
 								queryOptions = { ...options, start: options.start ?? 0 };
 							}
-							iterators.push(log.query(queryOptions));
+							iterators.push(endIteratorOnCorruptFrame(log.query(queryOptions), warnCorruptFrame(log.name)));
 						}
 					}
 					latestUpdates = this.updates;

resources/replayLogsGuards.ts

@@ -40,3 +40,48 @@ export function classifyAuditEntryForReplay(
 	if ((action & RECORD_BEARING_FLAGS) !== 0 && !hasRecord) return 'missing-record';
 	return null;
 }
+
+/**
+ * Wraps a transaction-log query iterator so a corrupt/torn frame ends that log's iteration
+ * cleanly instead of escaping as an uncaughtException. rocksdb-js throws a bounded RangeError
+ * when an entry's framing is broken; framing loss means the next entry can't be located, so the
+ * frame marks end-of-log (entries before it were already yielded) and startup replay /
+ * replication broadcast continue. `onCorruptFrame` fires once, latched — kept a callback (not a
+ * direct log) so this module stays out of the Harper module graph and is unit-testable.
+ */
+export function endIteratorOnCorruptFrame<T>(
+	iterator: Iterator<T>,
+	onCorruptFrame: (error: RangeError) => void
+): IterableIterator<T> {
+	let stopped = false;
+	return {
+		[Symbol.iterator]() {
+			return this;
+		},
+		next(): IteratorResult<T> {
+			if (stopped) return { done: true, value: undefined };
+			try {
+				return iterator.next();
+			} catch (error) {
+				// Key on the class, not the message: the framing RangeError's wording is
+				// version-dependent (1.4.2 added hex offsets). Anything else re-throws.
+				if (!(error instanceof RangeError)) throw error;
+				stopped = true;
+				onCorruptFrame(error);
+				return { done: true, value: undefined };
+			}
+		},
+		// Forward early termination (for-of break/return/throw) so the source's cleanup runs;
+		// mark stopped first. Current rocksdb-js implements neither — hence the protocol defaults.
+		return(value?: any): IteratorResult<T> {
+			stopped = true;
+			if (typeof iterator.return === 'function') return iterator.return(value);
+			return { done: true, value };
+		},
+		throw(error?: any): IteratorResult<T> {
+			stopped = true;
+			if (typeof iterator.throw === 'function') return iterator.throw(error);
+			throw error;
+		},
+	};
+}

unitTests/resources/replayLogs.test.js

@@ -2,7 +2,11 @@ const assert = require('node:assert');
 
 // The helper lives in a dependency-free module so the test doesn't need to bootstrap
 // the full Resource/RocksDB module graph (which has a circular require chain).
-const { classifyAuditEntryForReplay, RECORD_BEARING_FLAGS } = require('#src/resources/replayLogsGuards');
+const {
+	classifyAuditEntryForReplay,
+	RECORD_BEARING_FLAGS,
+	endIteratorOnCorruptFrame,
+} = require('#src/resources/replayLogsGuards');
 
 // Regression tests for the unclean-shutdown replay guards. Without these, an audit log
 // containing entries with corrupt MessagePack values caused replayLogs to write
@@ -59,3 +63,108 @@ describe('classifyAuditEntryForReplay', () => {
 		assert.strictEqual(RECORD_BEARING_FLAGS, HAS_RECORD | HAS_PARTIAL_RECORD);
 	});
 });
+
+// Regression tests for HarperFast/harper#1135: the wrapper must turn a framing RangeError into
+// a clean end-of-log (so replay/broadcast don't abort the boot) and leave other errors alone.
+describe('endIteratorOnCorruptFrame', () => {
+	it('yields entries up to a corrupt frame, then ends cleanly and reports it once', () => {
+		let calls = 0;
+		const source = {
+			next() {
+				calls++;
+				if (calls === 1) return { done: false, value: 'a' };
+				if (calls === 2) return { done: false, value: 'b' };
+				throw new RangeError('declared length 1778384896 overruns the log (limit=5439)');
+			},
+		};
+		const reported = [];
+		const wrapped = endIteratorOnCorruptFrame(source, (error) => reported.push(error));
+
+		assert.deepStrictEqual([...wrapped], ['a', 'b']);
+		assert.strictEqual(reported.length, 1);
+		assert.ok(reported[0] instanceof RangeError);
+		// Latched: stays done without re-invoking the source (no repeated reporting/spam).
+		assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined });
+		assert.strictEqual(calls, 3);
+		assert.strictEqual(reported.length, 1);
+	});
+
+	it('does not swallow non-RangeError failures', () => {
+		const source = {
+			next() {
+				throw new TypeError('boom');
+			},
+		};
+		let reported = 0;
+		const wrapped = endIteratorOnCorruptFrame(source, () => reported++);
+		assert.throws(() => wrapped.next(), TypeError);
+		assert.strictEqual(reported, 0);
+	});
+
+	it('passes a normal exhaustion through without reporting a corrupt frame', () => {
+		let calls = 0;
+		const source = {
+			next() {
+				calls++;
+				return calls === 1 ? { done: false, value: 1 } : { done: true, value: undefined };
+			},
+		};
+		let reported = 0;
+		const wrapped = endIteratorOnCorruptFrame(source, () => reported++);
+		assert.deepStrictEqual([...wrapped], [1]);
+		assert.strictEqual(reported, 0);
+	});
+
+	it('delegates return()/throw() to the underlying iterator so early-exit cleanup runs', () => {
+		let returnedWith;
+		let threwWith;
+		const source = {
+			next() {
+				return { done: false, value: 1 };
+			},
+			return(value) {
+				returnedWith = value;
+				return { done: true, value };
+			},
+			throw(error) {
+				threwWith = error;
+				return { done: true, value: undefined };
+			},
+		};
+		const wrapped = endIteratorOnCorruptFrame(source, () => {});
+
+		assert.strictEqual(typeof wrapped.return, 'function');
+		assert.deepStrictEqual(wrapped.return('cleanup'), { done: true, value: 'cleanup' });
+		assert.strictEqual(returnedWith, 'cleanup');
+		// after return(), the wrapper is latched done and never touches the source again
+		assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined });
+
+		assert.strictEqual(typeof wrapped.throw, 'function');
+		const boom = new Error('boom');
+		wrapped.throw(boom);
+		assert.strictEqual(threwWith, boom);
+	});
+
+	it('return()/throw() fall back to protocol defaults and latch when the underlying lacks them', () => {
+		let nextCalls = 0;
+		const source = {
+			next() {
+				nextCalls++;
+				return { done: false, value: 1 };
+			},
+		};
+		const wrapped = endIteratorOnCorruptFrame(source, () => {});
+
+		// return() defaults to done and latches without ever pulling the source again
+		assert.deepStrictEqual(wrapped.return('x'), { done: true, value: 'x' });
+		assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined });
+		assert.strictEqual(nextCalls, 0);
+
+		// throw() rethrows when the source can't handle it
+		const boom = new Error('boom');
+		assert.throws(
+			() => endIteratorOnCorruptFrame({ next: source.next }, () => {}).throw(boom),
+			(error) => error === boom
+		);
+	});
+});