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,33 @@ 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 {
+	// Walk the entry frames to the *last* well-framed entry, then force its declared length
+	// to overrun the log by setting the most-significant byte of its big-endian uint32 length
+	// to 0xff (≥ 4 GB). Targeting the last entry puts the corruption in the unflushed tail
+	// that replay actually reads (replay starts from the last-flushed position), rather than a
+	// flushed prefix it skips over. Deterministic — same frame, same corruption — unlike the
+	// old random byte flips, whose effect depended on what RocksDB had flushed before SIGKILL.
+	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 +210,33 @@ 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 version-tracking
+			// upgrade-abort path on next boot — a real but different failure mode.
+			// Here we want the framing-corruption case (#1135): a declared length that
+			// overruns the log used to throw an uncaught RangeError out of the txnlog
+			// iterator and abort startup; replay must now treat it as end-of-log.
+			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 rather than vacuously pass if the framing ever changes and we corrupt
+		// nothing — otherwise this would silently stop being a regression test for #1135.
+		ok(corrupted > 0, 'expected to corrupt at least one user-DB txnlog');
+		// Behavioral assertion, not a wall-clock budget: crashAndRestart resolving above
+		// means the server came back up (startHarper saw 'successfully started'); a
+		// regression resurfaces as a failed/timed-out restart, not a slow one — so there
+		// is nothing to tune per-runner. Confirm every table is still 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,49 @@ export function classifyAuditEntryForReplay(
 	if ((action & RECORD_BEARING_FLAGS) !== 0 && !hasRecord) return 'missing-record';
 	return null;
 }
+
+/**
+ * Wraps a transaction-log query iterator so a framing-level corruption error ends
+ * iteration of that log cleanly instead of escaping as an uncaughtException.
+ *
+ * rocksdb-js's txnlog reader throws a bounded `RangeError` when an entry's declared
+ * length overruns the log or its header is truncated (intentional hardening — it would
+ * otherwise OOM on `allocUnsafe(bogusLength)` or deref an undefined buffer). A torn write
+ * at SIGKILL time, or a flipped byte in an unflushed/corrupt log, looks exactly like that.
+ * Once the framing is lost we can't locate the next entry, so the corrupt frame marks the
+ * usable end of this log (torn-write semantics): entries before it have already been
+ * yielded, and startup replay / replication broadcast must continue rather than abort the
+ * boot. The latch means a persistently-corrupt log is reported once, not on every re-poll.
+ *
+ * `onCorruptFrame` is invoked once, with the error, when a corrupt frame is hit — kept as
+ * a callback (rather than logging here) so this module stays free of the Harper module
+ * graph and the behavior is unit-testable. Non-`RangeError` failures propagate unchanged.
+ */
+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) {
+				// rocksdb-js's txnlog reader signals frame corruption with a RangeError; the
+				// message wording is version-dependent (1.4.2 added hex offsets), so we key on
+				// the class, not the text. Anything else is unexpected and propagates. Treating
+				// a stray non-framing RangeError as end-of-log is the deliberate tradeoff: on
+				// this cold replay/boot path it still beats an uncaughtException aborting
+				// startup, and onCorruptFrame logs every occurrence so it is never silent.
+				if (!(error instanceof RangeError)) throw error;
+				stopped = true;
+				onCorruptFrame(error);
+				return { done: true, value: undefined };
+			}
+		},
+	};
+}

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,57 @@ describe('classifyAuditEntryForReplay', () => {
 		assert.strictEqual(RECORD_BEARING_FLAGS, HAS_RECORD | HAS_PARTIAL_RECORD);
 	});
 });
+
+// Regression tests for HarperFast/harper#1135: rocksdb-js's txnlog reader throws a bounded
+// RangeError when an entry's declared length overruns the log (a torn/corrupt frame). That
+// used to escape uncaught out of the replay/broadcast iterator and abort startup. The wrapper
+// must turn it into a clean end-of-log instead, while leaving every other failure untouched.
+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);
+	});
+});