diff --git a/integrationTests/server/replay-stress.test.ts b/integrationTests/server/replay-stress.test.ts index 9fb3e30f16..c559559c1c 100644 --- a/integrationTests/server/replay-stress.test.ts +++ b/integrationTests/server/replay-stress.test.ts @@ -8,9 +8,10 @@ * 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 @@ -18,9 +19,8 @@ */ 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`); }); }); diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index 8cd899dfb0..5096c7b86b 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/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 & { 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; diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index 054f3c7f6f..316738e74a 100644 --- a/resources/replayLogsGuards.ts +++ b/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( + iterator: Iterator, + onCorruptFrame: (error: RangeError) => void +): IterableIterator { + let stopped = false; + return { + [Symbol.iterator]() { + return this; + }, + next(): IteratorResult { + 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 { + stopped = true; + if (typeof iterator.return === 'function') return iterator.return(value); + return { done: true, value }; + }, + throw(error?: any): IteratorResult { + stopped = true; + if (typeof iterator.throw === 'function') return iterator.throw(error); + throw error; + }, + }; +} diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index ca44699e2d..a5bd8f930e 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/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 + ); + }); +});