diff --git a/integrationTests/server/replay-transaction-atomicity.test.ts b/integrationTests/server/replay-transaction-atomicity.test.ts new file mode 100644 index 0000000000..25d47dd75b --- /dev/null +++ b/integrationTests/server/replay-transaction-atomicity.test.ts @@ -0,0 +1,147 @@ +/** + * A corrupt transaction-log frame in the middle of a source transaction must not leave that + * transaction half-applied. + * + * Replay groups every equal-version entry into one transaction and commits it at the next version + * boundary, so a break inside such a group used to commit whatever part of it was still readable — + * a transaction that never committed that way at the source becoming durable here. This drives a + * real multi-record insert, tears the log inside it, and requires the replayed table to hold all of + * that insert or none of it. See HarperFast/harper#2016 and #2063. + */ +import { suite, test, before, after } from 'node:test'; +import { ok, strictEqual as equal } from 'node:assert'; +import { readdirSync, readFileSync, openSync, writeSync, closeSync } from 'node:fs'; +import { join } from 'node:path'; + +import { + startHarper, + teardownHarper, + sendOperation, + type ContextWithHarper, + type HarperContext, +} from '@harperfast/integration-testing'; +import { constants } from '@harperfast/rocksdb-js'; + +// Transaction-log framing (big-endian): a fixed-size file header, then entries shaped +// [float64 timestamp][uint32 length][flags byte][length bytes of data]. +const { TRANSACTION_LOG_FILE_HEADER_SIZE, TRANSACTION_LOG_ENTRY_HEADER_SIZE } = constants; + +const DB = 'atomicity'; +const TABLE = 'orders'; +const EARLIER_IDS = 60; +const TORN_IDS = 60; +// Entries of the torn transaction left readable before the break, so the test proves the readable +// part is discarded rather than proving the whole transaction was unreachable anyway. +const READABLE_BEFORE_BREAK = 10; + +async function op(ctx: HarperContext, body: any) { + return await sendOperation(ctx, { ...body, authorization: ctx.admin }); +} + +function records(start: number, count: number) { + const out = []; + for (let i = 0; i < count; i++) out.push({ id: start + i, payload: 'x'.repeat(256), n: i }); + return out; +} + +async function countInRange(ctx: HarperContext, start: number, count: number): Promise { + const rows = await op(ctx, { + operation: 'sql', + sql: `select count(*) as c from ${DB}.${TABLE} where id >= ${start} and id < ${start + count}`, + }); + return rows[0]?.c ?? 0; +} + +function userTxnLogFiles(dataRootDir: string): string[] { + const out: string[] = []; + const dbRoot = join(dataRootDir, 'database'); + for (const db of readdirSync(dbRoot)) { + if (db === 'system') continue; + const tlogRoot = join(dbRoot, db, 'transaction_logs'); + let nodes: string[]; + try { + nodes = readdirSync(tlogRoot); + } catch { + continue; + } + for (const node of nodes) { + for (const file of readdirSync(join(tlogRoot, node))) { + if (file.endsWith('.txnlog')) out.push(join(tlogRoot, node, file)); + } + } + } + return out; +} + +/** + * Break the framing partway through the log's LAST transaction — the run of trailing entries that + * share the highest timestamp — leaving `readableBefore` of its entries intact ahead of the break. + * Returns how many entries that transaction has, or 0 if the log has no such run to tear. + */ +function tearLastTransaction(path: string, readableBefore: number): number { + const buf = readFileSync(path); + const entries: { lengthPos: number; timestamp: number }[] = []; + let pos = TRANSACTION_LOG_FILE_HEADER_SIZE; + while (pos + TRANSACTION_LOG_ENTRY_HEADER_SIZE <= buf.length) { + const timestamp = buf.readDoubleBE(pos); + if (timestamp === 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; + entries.push({ lengthPos, timestamp }); + pos = next; + } + if (entries.length === 0) return 0; + const lastTimestamp = entries.at(-1).timestamp; + let first = entries.length - 1; + while (first > 0 && entries[first - 1].timestamp === lastTimestamp) first--; + const transactionEntries = entries.length - first; + if (transactionEntries <= readableBefore) return 0; + // Force this entry's declared length to overrun the log (top byte → 0xff, ≥ 4 GB): the reader + // throws a bounded RangeError there and cannot locate any entry after it. + const fd = openSync(path, 'r+'); + try { + writeSync(fd, Buffer.from([0xff]), 0, 1, entries[first + readableBefore].lengthPos); + } finally { + closeSync(fd); + } + return transactionEntries; +} + +suite('Replay transaction atomicity across a corrupt frame', (ctx: ContextWithHarper) => { + before(async () => { + // Don't flush on exit: the crash must leave these writes recoverable only from the txn log. + await startHarper(ctx, { env: { HARPER_NO_FLUSH_ON_EXIT: true } }); + await op(ctx.harper, { operation: 'create_database', database: DB }); + await op(ctx.harper, { operation: 'create_table', database: DB, table: TABLE, primary_key: 'id' }); + }); + after(async () => teardownHarper(ctx)); + + test('discards a transaction the corrupt frame truncated instead of applying part of it', async () => { + await op(ctx.harper, { operation: 'insert', database: DB, table: TABLE, records: records(1, EARLIER_IDS) }); + // The last insert is one source transaction, and is the one the tear lands inside. + await op(ctx.harper, { operation: 'insert', database: DB, table: TABLE, records: records(1001, TORN_IDS) }); + + const dataRootDir = ctx.harper.dataRootDir; + await new Promise((resolve) => { + ctx.harper.process.once('exit', () => resolve()); + ctx.harper.process.kill('SIGKILL'); + }); + let torn = 0; + for (const file of userTxnLogFiles(dataRootDir)) { + torn = Math.max(torn, tearLastTransaction(file, READABLE_BEFORE_BREAK)); + } + // Fail loudly, not vacuously, if the framing or the transaction grouping ever changes. + ok(torn > READABLE_BEFORE_BREAK, `expected to tear a multi-entry transaction, tore ${torn} entries`); + + await startHarper(ctx); + + // None of it: the break makes the rest of that insert unreadable, so the readable prefix is + // discarded rather than committed as a transaction the source never committed. + equal(await countInRange(ctx.harper, 1001, TORN_IDS), 0, 'the truncated transaction must not be applied in part'); + // The transactions that completed ahead of the break are unaffected: fail-stop costs the + // transaction the break landed in, not the log up to it. + equal(await countInRange(ctx.harper, 1, EARLIER_IDS), EARLIER_IDS); + }); +}); diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index 5a35d3277b..f1d632ab72 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/resources/RocksTransactionLogStore.ts @@ -3,7 +3,7 @@ import { ExtendedIterable } from '@harperfast/extended-iterable'; import { getIdOfRemoteNode } from './nodeIdMapping.ts'; import { Decoder, readAuditEntry, ENTRY_DATAVIEW, AuditRecord, createAuditEntry } from './auditStore.ts'; import { HAS_STRUCTURE_UPDATE } from './RecordEncoder.ts'; -import { endIteratorOnCorruptFrame } from './replayLogsGuards.ts'; +import { createCorruptFrameReporter, endIteratorOnCorruptFrame, type CorruptFrameStop } from './replayLogsGuards.ts'; import { isMainThread } from 'node:worker_threads'; import { EventEmitter } from 'node:events'; import { asBinary } from 'lmdb'; @@ -23,12 +23,14 @@ 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); -} +type TrackedIterator = IterableIterator & { lastVersion?: number }; + +export type TransactionLogIterable = Iterable & { + /** Corrupt frames that ended a log's iteration during this range. */ + corruptFrameStop: CorruptFrameStop; +}; + +const reportCorruptFrame = createCorruptFrameReporter(harperLogger); /** * Represents a transaction log store backed by RocksDB. @@ -42,6 +44,7 @@ export class RocksTransactionLogStore extends EventEmitter { logByName: Map = new Map(); updates = 0; // the number of updates to the list of logs that have occurred rootStore: RocksDatabase; + corruptFrameScope: string; reusableIterable = true; // flag indicating that iterable can be reused to resume iterating through audit log // Highest structureVersion appended to each per-node TransactionLog, tracked per tableId. Drives the // per-log HAS_STRUCTURE_UPDATE flag in put(). Keyed by (log, tableId): a per-node log interleaves entries @@ -54,6 +57,10 @@ export class RocksTransactionLogStore extends EventEmitter { super(); this.log = rootDatabase.useLog('local'); this.rootStore = rootDatabase; + // Break reports are keyed per log name, but every store has its own 'local' log, so they + // need a scope. The path, not `databaseName`: one root store can back several logical + // databases and they share these logs, and `databaseName` is not set on it yet here. + this.corruptFrameScope = rootDatabase.path; } /** @@ -231,9 +238,25 @@ export class RocksTransactionLogStore extends EventEmitter { startByLog?: Map; startFromLastFlushed?: boolean; readUncommitted?: boolean; - }): Iterable { + }): TransactionLogIterable { let iterable = new ExtendedIterable(); let aggregateIterator: TransactionLogIterator; + // Set only for a single-log range, where every entry comes from the same log. + let singleLogIterator: TrackedIterator; + const corruptFrameStop: CorruptFrameStop = { breaks: 0, truncatedVersions: new Set() }; + // Each log's iterator carries the version of the last entry it yielded, so a break can be + // attributed to the source transaction whose remaining entries it swallowed. On the iterator + // itself, not in a map keyed by log: this is written per entry on the replay/broadcast path, + // and it stays attached when a removed log is spliced out of the aggregate. + const trackCorruptFrames = (log: TransactionLog, queryOptions: typeof options): TrackedIterator => { + const report = reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`); + const iterator: TrackedIterator = endIteratorOnCorruptFrame(log.query(queryOptions), (error) => { + corruptFrameStop.breaks++; + if (iterator.lastVersion !== undefined) corruptFrameStop.truncatedVersions.add(iterator.lastVersion); + report(error); + }); + return iterator; + }; if (options.log !== undefined) { let log = typeof options.log === 'number' ? this.nodeLogs?.[options.log] : this.logByName.get(options.log); if (!log) { @@ -247,7 +270,8 @@ export class RocksTransactionLogStore extends EventEmitter { log = this.rootStore.useLog(options.log); } } - const queryIterator = endIteratorOnCorruptFrame(log.query(options), warnCorruptFrame(log.name)); + const queryIterator = trackCorruptFrames(log, options); + singleLogIterator = queryIterator; iterable.iterate = () => queryIterator; } else { const onlyKeys = options.onlyKeys; @@ -255,7 +279,7 @@ export class RocksTransactionLogStore extends EventEmitter { // holds the queue of next entries from each iterator let nextEntries: any[]; let latestUpdates: number; - const iterators: IterableIterator[] = []; + const iterators: TrackedIterator[] = []; // Iterators that have permanently failed (corrupt entry stuck at the same // position). Tracked by identity so the retry-poll path in next() and // updateIterators() never calls .next() on them again — otherwise every @@ -298,7 +322,7 @@ export class RocksTransactionLogStore extends EventEmitter { // condition of potentially missing an initial update queryOptions = { ...options, start: options.start ?? 0 }; } - iterators.push(endIteratorOnCorruptFrame(log.query(queryOptions), warnCorruptFrame(log.name))); + iterators.push(trackCorruptFrames(log, queryOptions)); } } latestUpdates = this.updates; @@ -355,7 +379,8 @@ export class RocksTransactionLogStore extends EventEmitter { } } if (earliestIndex >= 0) { - // replace the entry with the next one from the iterator we pulled from + // before the refill, which is where a break surfaces and needs this entry's version + iterators[earliestIndex].lastVersion = earliest.timestamp; nextEntries[earliestIndex] = safeNext(iterators[earliestIndex], logs[earliestIndex]); return { value: onlyKeys ? earliest.timestamp : earliest, @@ -389,6 +414,10 @@ export class RocksTransactionLogStore extends EventEmitter { iterable.iterate = () => aggregateIterator; } const mappedAggregateIterable = iterable.map(({ timestamp, data, endTxn }: TransactionEntry) => { + // A break surfaces on the pull after this entry, so recording it here is in time to attribute + // that break to this entry's transaction. The aggregate branch records its own, per source + // log, because there this callback cannot tell which log an entry came from. + if (singleLogIterator) singleLogIterator.lastVersion = timestamp; // Per-entry try/catch: a corrupt rocks prelude (first 4-16 bytes) would otherwise // throw a raw `RangeError: Offset is outside the bounds of the DataView` out // through `iterable.map`, escape the for-of consumer, and land as an @@ -442,7 +471,8 @@ export class RocksTransactionLogStore extends EventEmitter { mappedAggregateIterable.addLog = aggregateIterator.addLog; mappedAggregateIterable.removeLog = aggregateIterator.removeLog; } - return mappedAggregateIterable; + mappedAggregateIterable.corruptFrameStop = corruptFrameStop; + return mappedAggregateIterable as TransactionLogIterable; } getKeys(_options?: any) { return []; // TODO: implement this diff --git a/resources/replayLogs.ts b/resources/replayLogs.ts index b0caecad24..d4824dc640 100644 --- a/resources/replayLogs.ts +++ b/resources/replayLogs.ts @@ -78,6 +78,10 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise let transaction: DatabaseTransaction; let lastTimestamp = 0; let writes = 0; + // Writes staged on the currently-open transaction, so a discard can be taken back out of `writes`. + let stagedWrites = 0; + // Records dropped because a corrupt frame truncated the transaction they belong to. + let discardedWrites = 0; let skipped = 0; // Track forward progress so a backlog of unwritable entries can't grind the boot thread // forever (harper#1266). `noProgressRun` counts every entry processed without a successful @@ -92,7 +96,8 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise const configuredReplayTimeout = Number(envGet(CONFIG_PARAMS.REPLICATION_REPLAYTIMEOUT)); const replayTimeoutMs = configuredReplayTimeout > 0 ? configuredReplayTimeout : REPLAY_WALL_CLOCK_LIMIT_MS; const txnLog: RocksTransactionLogStore = (rootStore as any).auditStore; - for (const auditRecord of txnLog.getRange({ startFromLastFlushed: true, readUncommitted: true }) as any) { + const entries = txnLog.getRange({ startFromLastFlushed: true, readUncommitted: true }); + for (const auditRecord of entries as any) { if (noProgressRun > 0 && shouldAbortStalledReplay(noProgressRun, performance.now() - lastProgressTime)) { logger.fatal( `Aborting transaction-log replay in ${(rootStore as any).databaseName} database: ${noProgressRun} consecutive audit entries with no successful write (${skipped} skipped as unrecoverable, ${writes} replayed so far). This backlog is making no forward progress and was blocking startup (harper#1266) — typically a peer transaction log whose values reference unresolvable shared structures (harper#1163), or a backlog for a dropped table. Continuing boot without replaying the remainder; shed or relocate the oversized/undecodable peer transaction log(s), or re-clone this node, to recover the unreplayed data.` @@ -166,13 +171,21 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise console.warn('Harper was not properly shutdown, replaying transaction logs to synchronize database'); } if (lastTimestamp !== version) { + const torn = entries.corruptFrameStop.truncatedVersions.has(lastTimestamp); lastTimestamp = version; try { - // commit the last transaction since we are starting a new one - transaction?.directCommitSync(); + // commit the last transaction since we are starting a new one, unless a corrupt + // frame swallowed the rest of it — half of a source transaction must never become + // durable, so it is dropped whole and stays in the log for a repaired retry + if (torn) { + writes -= stagedWrites; + discardedWrites += stagedWrites; + transaction?.abort(); + } else transaction?.directCommitSync(); } catch (error) { - logger.error('Error committing replay transaction', error); + logger.error(`Error ${torn ? 'discarding a torn' : 'committing'} replay transaction`, error); } + stagedWrites = 0; // Abort if replay has exceeded the total wall-clock budget even while making progress // (harper#1316, facet a). shouldAbortStalledReplay resets its counters on every write, // so a slow-but-progressing replay (deep out-of-order audit chain walk per entry) can @@ -200,6 +213,7 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise context.transaction = transaction; const options = { context, residencyId, nodeId, originatingOperation }; writes++; + stagedWrites++; switch (type) { case 'put': tableInstance._writeUpdate(recordId, record, true, options); @@ -283,10 +297,20 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise }); } } + const finalTorn = entries.corruptFrameStop.truncatedVersions.has(lastTimestamp); try { - transaction?.directCommitSync(); + if (finalTorn) { + writes -= stagedWrites; + discardedWrites += stagedWrites; + transaction?.abort(); + } else transaction?.directCommitSync(); } catch (error) { - logger.error('Error committing replay transaction', error); + logger.error(`Error ${finalTorn ? 'discarding a torn' : 'committing'} replay transaction`, error); + } + if (entries.corruptFrameStop.breaks > 0) { + logger.error( + `Transaction-log replay in ${(rootStore as any).databaseName} database stopped at a corrupt entry after replaying ${writes} records. Every entry after the break is quarantined — neither replayed nor replicated — and ${discardedWrites} record(s) of the transaction the break truncated were discarded rather than applied in part. Repair the transaction log or re-clone this node to recover them.` + ); } if (writes > 0) logger.warn(`Replayed ${writes} records in ${(rootStore as any).databaseName} database`); if (skipped > 0) diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index e4f15c3c57..5d2496caaa 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -132,16 +132,47 @@ export function shouldAbortSlowReplay(totalElapsedMs: number, timeLimitMs = REPL } /** - * 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. + * A corrupt transaction-log frame from rocksdb-js. `resyncPosition` is set only when intact entries + * follow the break, which is what distinguishes lost entries from a merely truncated tail; its + * absence is also all that any rocksdb-js predating that field can report. + */ +export type CorruptFrameError = RangeError & { + logId?: number | null; + position?: number | null; + resyncPosition?: number | null; + unreadableBytes?: number | null; +}; + +/** + * Corrupt frames hit by one `getRange` call, shared by its per-log iterators. + * + * `truncatedVersions` is what a consumer has to act on: a break destroys the framing after the last + * entry the broken log yielded, and whether the entries it swallowed continued that entry's version + * is exactly what can no longer be read. That version's transaction is therefore incomplete, and + * replay must discard it rather than commit the part of it that was still readable. A log that + * broke before yielding anything truncates no transaction. + */ +export interface CorruptFrameStop { + breaks: number; + truncatedVersions: Set; +} + +/** + * Wraps a transaction-log query iterator so a corrupt/torn frame ends that log's iteration cleanly + * instead of escaping as an uncaughtException. + * + * Iteration stops at the break whether or not intact entries follow it. Resuming past a mid-log + * break would recover those entries, but a transaction log carries no frame-level transaction + * boundaries: replay groups equal-version entries into one source transaction, so skipping a frame + * inside such a group applies and checkpoints the surviving subset of a transaction that never + * committed that way at the source. Availability loss is visible and recoverable; a silently torn + * transaction is neither. Recovery is deferred until the engine can resume at a proven boundary + * (harper#2016, harper#2063); `resyncPosition` is used only to report the break as lost entries + * rather than as a truncated tail. */ export function endIteratorOnCorruptFrame( iterator: Iterator, - onCorruptFrame: (error: RangeError) => void + onCorruptFrame: (error: CorruptFrameError) => void ): IterableIterator { let stopped = false; return { @@ -157,7 +188,7 @@ export function endIteratorOnCorruptFrame( // version-dependent (1.4.2 added hex offsets). Anything else re-throws. if (!(error instanceof RangeError)) throw error; stopped = true; - onCorruptFrame(error); + onCorruptFrame(error as CorruptFrameError); return { done: true, value: undefined }; } }, @@ -175,3 +206,116 @@ export function endIteratorOnCorruptFrame( }, }; } + +/** + * A corrupt frame, accumulated for the life of the process. The same frame is re-encountered by + * every reader until each consumer's resume cursor has passed it, so this is deduplicated by + * location and the repeats become a count. + */ +export interface CorruptFrameReport { + log: string; + logId?: number; + position?: number; + /** Intact entries followed the break, so entries were lost rather than merely truncated. */ + midLog: boolean; + /** Extent of the unreadable region, when the engine reports it; 0 when nothing valid follows it. */ + unreadableBytes: number; + firstSeen: number; + lastSeen: number; + occurrences: number; +} + +/** Distinct break sites retained. One physical corruption yields one site, so this is generous. */ +export const MAX_CORRUPT_FRAME_REPORTS = 256; + +// Re-encounters move a site to the end, so the first key is the least recently seen. Full means +// evicting that one rather than refusing a new site, which could not then be deduplicated. +const corruptFrameReports = new Map(); +let evictedCorruptFrameReports = 0; + +/** + * Every corrupt frame seen by this worker. This is not yet consumed by cluster/health status; + * callers must aggregate it across worker threads before exposing a node-wide signal. + * + * Mid-log breaks are immediately logged at error level even without that consumer. + */ +export function getCorruptFrameReports(): CorruptFrameReport[] { + return [...corruptFrameReports.values()]; +} + +/** Break sites evicted because {@link MAX_CORRUPT_FRAME_REPORTS} was reached. */ +export function getEvictedCorruptFrameReportCount(): number { + return evictedCorruptFrameReports; +} + +export function clearCorruptFrameReports() { + corruptFrameReports.clear(); + evictedCorruptFrameReports = 0; +} + +// `logId`/`position` are absent on any rocksdb-js predating the resync support, and every break on +// a stream would then collapse onto one key — folding genuinely different corruptions into a single +// count that only ever logs once. The message carries the offset and file in text, so it separates +// them when the fields can't. +function corruptFrameKey(logName: string, error: CorruptFrameError): string { + const { logId, position } = error; + return logId != null && position != null + ? `${logName}\u0000${logId}:${position}` + : `${logName}\u0000${error.message}`; +} + +/** + * Records a corrupt frame and logs it once per distinct break. + * + * A mid-log break is an `error`, not a `warn`: entries after it were acknowledged and are now + * quarantined behind it. Severity follows the break's own shape, never this pass's outcome — every + * pass stops, so keying on that would report the #2063 signature with the benign torn-tail `warn`. + * A site first seen as a torn tail can escalate later, since only a rocksdb-js that reports + * `resyncPosition` can tell the two apart. + */ +export function createCorruptFrameReporter(logger: { + warn: (message: string, error?: unknown) => void; + error: (message: string, error?: unknown) => void; +}) { + return (logName: string) => (error: CorruptFrameError) => { + const midLog = error.resyncPosition != null; + const unreadableBytes = error.unreadableBytes ?? 0; + const now = Date.now(); + const key = corruptFrameKey(logName, error); + const existing = corruptFrameReports.get(key); + if (existing) { + corruptFrameReports.delete(key); + corruptFrameReports.set(key, existing); + existing.occurrences++; + existing.lastSeen = now; + if (!midLog || existing.midLog) return; + existing.midLog = true; + existing.unreadableBytes = unreadableBytes; + } else { + if (corruptFrameReports.size >= MAX_CORRUPT_FRAME_REPORTS) { + corruptFrameReports.delete(corruptFrameReports.keys().next().value); + evictedCorruptFrameReports++; + } + corruptFrameReports.set(key, { + log: logName, + logId: error.logId ?? undefined, + position: error.position ?? undefined, + midLog, + unreadableBytes, + firstSeen: now, + lastSeen: now, + occurrences: 1, + }); + } + if (midLog) { + logger.error( + `Corrupt entry in transaction log "${logName}"; ${unreadableBytes} byte(s) are unreadable and the entries within them are lost. ` + + 'Intact entries follow the break, but reading stops there rather than skipping the frame, which could tear a source transaction: ' + + 'they are quarantined until the log is repaired or this node is re-cloned, and they are neither replayed nor replicated meanwhile.', + error + ); + } else { + logger.warn(`Stopping transaction log "${logName}" at a corrupt entry during replay`, error); + } + }; +} diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index f1e9423c11..57c67ce2d4 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -7,6 +7,11 @@ const { isUndecodableValidatedWrite, RECORD_BEARING_FLAGS, endIteratorOnCorruptFrame, + MAX_CORRUPT_FRAME_REPORTS, + createCorruptFrameReporter, + getCorruptFrameReports, + getEvictedCorruptFrameReportCount, + clearCorruptFrameReports, shouldAbortStalledReplay, REPLAY_NO_PROGRESS_COUNT_LIMIT, REPLAY_NO_PROGRESS_TIME_LIMIT_MS, @@ -135,6 +140,95 @@ describe('endIteratorOnCorruptFrame', () => { assert.strictEqual(reported.length, 1); }); + // harper#2016 / harper#2063: intact, already-acknowledged entries follow a mid-log break, and + // skipping the frame would recover them — but the frame carries no transaction boundary, so the + // surviving part of a torn source transaction would be applied. Iteration stops on either shape; + // only the report distinguishes them. + it('stops at a mid-log break rather than resuming past it', () => { + let calls = 0; + const source = { + next() { + calls++; + if (calls === 1) return { done: false, value: 'a' }; + if (calls === 2) { + const error = new RangeError('declared length 1778384896 overruns the log (limit=5439)'); + error.resyncPosition = 0x7d3f; + error.unreadableBytes = 26; + throw error; + } + return { done: false, value: 'b' }; + }, + }; + const reported = []; + const wrapped = endIteratorOnCorruptFrame(source, (error) => reported.push(error)); + + assert.deepStrictEqual([...wrapped], ['a']); + assert.strictEqual(reported.length, 1); + assert.strictEqual(reported[0].resyncPosition, 0x7d3f); + assert.strictEqual(reported[0].unreadableBytes, 26); + // latched: the entries after the break are never pulled, on this or any later call + assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); + assert.strictEqual(calls, 2); + }); + + it('treats a RangeError with no resync position as end-of-log (older rocksdb-js)', () => { + let calls = 0; + const source = { + next() { + calls++; + if (calls === 1) return { done: false, value: 'a' }; + throw new RangeError('truncated entry header'); + }, + }; + const reported = []; + const wrapped = endIteratorOnCorruptFrame(source, (error) => reported.push(error.resyncPosition)); + + assert.deepStrictEqual([...wrapped], ['a']); + assert.deepStrictEqual(reported, [undefined]); + assert.strictEqual(calls, 2); + }); + + it('treats a null resync position from the native addon as end-of-log', () => { + const error = new RangeError('truncated entry header'); + error.resyncPosition = null; + let calls = 0; + const wrapped = endIteratorOnCorruptFrame( + { + next() { + calls++; + throw error; + }, + }, + (reportedError) => { + assert.strictEqual(reportedError, error); + } + ); + + assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); + assert.strictEqual(calls, 1); + }); + + // 0 is a legitimate resync position, so the mid-log signal must not be a truthiness test. + it('keeps zero as a valid resync position', () => { + const error = new RangeError('corrupt frame at offset 0'); + error.resyncPosition = 0; + let calls = 0; + const reported = []; + const wrapped = endIteratorOnCorruptFrame( + { + next() { + calls++; + throw error; + }, + }, + (reportedError) => reported.push(reportedError.resyncPosition) + ); + + assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); + assert.deepStrictEqual(reported, [0]); + assert.strictEqual(calls, 1); + }); + it('does not swallow non-RangeError failures', () => { const source = { next() { @@ -215,6 +309,186 @@ describe('endIteratorOnCorruptFrame', () => { }); }); +// harper#2063: the reporter is what makes a lossy stream distinguishable from a healthy one, so +// its severity choice, deduplication, and key derivation are the load-bearing parts. +describe('createCorruptFrameReporter', () => { + function setup() { + clearCorruptFrameReports(); + const logs = { warn: [], error: [] }; + const reporter = createCorruptFrameReporter({ + warn: (message, error) => logs.warn.push({ message, error }), + error: (message, error) => logs.error.push({ message, error }), + }); + return { logs, report: reporter('local') }; + } + + function midLogError(position = 0x7d20bb, unreadableBytes = 26) { + const error = new RangeError(`Corrupt transaction log entry at position ${position.toString(16)} of log 2`); + error.logId = 2; + error.position = position; + error.resyncPosition = position + unreadableBytes; + error.unreadableBytes = unreadableBytes; + return error; + } + + it('logs a mid-log break at error level and records the lost bytes', () => { + const { logs, report } = setup(); + report(midLogError()); + + assert.strictEqual(logs.warn.length, 0); + assert.strictEqual(logs.error.length, 1); + assert.match(logs.error[0].message, /26 byte\(s\) are unreadable/); + const reports = getCorruptFrameReports(); + assert.strictEqual(reports.length, 1); + assert.deepStrictEqual( + { log: reports[0].log, midLog: reports[0].midLog, unreadableBytes: reports[0].unreadableBytes }, + { log: 'local', midLog: true, unreadableBytes: 26 } + ); + }); + + it('logs a torn tail at warn level', () => { + const { logs, report } = setup(); + const error = new RangeError('truncated entry header'); + error.logId = 2; + error.position = 100; + report(error); + + assert.strictEqual(logs.error.length, 0); + assert.strictEqual(logs.warn.length, 1); + assert.strictEqual(getCorruptFrameReports()[0].midLog, false); + }); + + // The whole point of #2063 is that the lossy case is distinguishable from the benign one, and + // that the operator is told what state the rest of the log is in. + it('names the entries behind a mid-log break as quarantined', () => { + const { logs, report } = setup(); + report(midLogError()); + + assert.strictEqual(logs.warn.length, 0); + assert.match(logs.error[0].message, /quarantined until the log is repaired or this node is re-cloned/); + }); + + it('counts repeats of the same break without re-logging it', () => { + const { logs, report } = setup(); + report(midLogError()); + report(midLogError()); + report(midLogError()); + + assert.strictEqual(logs.error.length, 1); + const [only] = getCorruptFrameReports(); + assert.strictEqual(only.occurrences, 3); + assert.ok(only.lastSeen >= only.firstSeen); + }); + + // The same site reads as a torn tail on a rocksdb-js that cannot report `resyncPosition` and as + // data loss on one that can, so a report that latched on the first encounter would keep the + // benign classification after the engine bump that makes the loss visible. + it('escalates a site first seen as a torn tail once it is known to be mid-log', () => { + const { logs, report } = setup(); + const tornTail = midLogError(); + delete tornTail.resyncPosition; + delete tornTail.unreadableBytes; + report(tornTail); + assert.strictEqual(getCorruptFrameReports()[0].midLog, false); + + report(midLogError()); + assert.deepStrictEqual( + { midLog: getCorruptFrameReports()[0].midLog, unreadableBytes: getCorruptFrameReports()[0].unreadableBytes }, + { midLog: true, unreadableBytes: 26 } + ); + assert.strictEqual(logs.warn.length, 1); + assert.strictEqual(logs.error.length, 1); + + // and it does not re-log on every later drain + report(midLogError()); + assert.strictEqual(logs.error.length, 1); + }); + + // Against a rocksdb-js with no logId/position, keying on those fields alone collapses every + // break on the stream onto one entry, so the second real corruption is never logged. + it('separates breaks by message when the error carries no logId/position', () => { + const { logs, report } = setup(); + report(new RangeError('Corrupt transaction log entry at position 7d20bb of log 2')); + report(new RangeError('Corrupt transaction log entry at position 3bc071 of log 23')); + + assert.strictEqual(getCorruptFrameReports().length, 2); + assert.strictEqual(logs.warn.length, 2); + }); + + it('treats null native position fields as absent when keying and classifying breaks', () => { + const { logs, report } = setup(); + for (const message of ['corrupt frame at offset 1', 'corrupt frame at offset 2']) { + const error = new RangeError(message); + error.logId = null; + error.position = null; + error.resyncPosition = null; + report(error); + } + + assert.strictEqual(getCorruptFrameReports().length, 2); + assert.strictEqual(logs.warn.length, 2); + assert.strictEqual(logs.error.length, 0); + }); + + it('keeps zero native position fields as valid report metadata', () => { + const { logs, report } = setup(); + const error = midLogError(0, 0); + error.logId = 0; + report(error); + + assert.strictEqual(logs.error.length, 1); + assert.deepStrictEqual( + { logId: getCorruptFrameReports()[0].logId, position: getCorruptFrameReports()[0].position }, + { logId: 0, position: 0 } + ); + }); + + it('bounds retained break sites by evicting the oldest', () => { + const { logs, report } = setup(); + for (let i = 0; i < MAX_CORRUPT_FRAME_REPORTS + 5; i++) { + report(midLogError(0x1000 + i * 0x100)); + } + + assert.strictEqual(getCorruptFrameReports().length, MAX_CORRUPT_FRAME_REPORTS); + assert.strictEqual(getEvictedCorruptFrameReportCount(), 5); + assert.strictEqual(logs.error.length, MAX_CORRUPT_FRAME_REPORTS + 5); + }); + + // Refusing a new site once full would leave it undeduplicated, so it would re-log on every + // drain — the log spam this report exists to replace. + it('keeps deduplicating the newest sites after the bound is reached', () => { + const { logs, report } = setup(); + for (let i = 0; i < MAX_CORRUPT_FRAME_REPORTS + 5; i++) { + report(midLogError(0x1000 + i * 0x100)); + } + const logsAfterFill = logs.error.length; + + // re-encounter the most recent site repeatedly, as every later drain would + for (let i = 0; i < 10; i++) { + report(midLogError(0x1000 + (MAX_CORRUPT_FRAME_REPORTS + 4) * 0x100)); + } + + assert.strictEqual(logs.error.length, logsAfterFill); + assert.strictEqual(getEvictedCorruptFrameReportCount(), 5); + }); + + it('retains recently encountered sites when evicting old reports', () => { + const { logs, report } = setup(); + for (let i = 0; i < MAX_CORRUPT_FRAME_REPORTS; i++) { + report(midLogError(0x1000 + i * 0x100)); + } + const firstPosition = 0x1000; + const secondPosition = 0x1100; + report(midLogError(firstPosition)); + report(midLogError(0x1000 + MAX_CORRUPT_FRAME_REPORTS * 0x100)); + + const positions = getCorruptFrameReports().map((entry) => entry.position); + assert.ok(positions.includes(firstPosition)); + assert.ok(!positions.includes(secondPosition)); + assert.strictEqual(logs.error.length, MAX_CORRUPT_FRAME_REPORTS + 1); + }); +}); + // Regression tests for HarperFast/harper#1266: a boot replay over a backlog of unwritable entries // (undecodable peer-log entries, or entries for a dropped table) must give up once it is making no // forward progress, instead of grinding the main thread for minutes. A healthy replay (which keeps