From 8d2551f50a80a561d33df29c538835f48e666715 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 4 Aug 2026 18:03:25 -0600 Subject: [PATCH 01/14] fix(replay): resync past a mid-log corrupt frame and report it as data loss endIteratorOnCorruptFrame treated every framing break as end-of-log. That is right for a torn tail -- nothing valid follows it -- and wrong for a mid-log break: a partial ENOSPC/EDQUOT append the process survived leaves intact entries appended after it, every one acknowledged to a client. Ending iteration there amputates all of them, and permanently: the containment was per-drain, so each new drain built fresh iterators, re-read from the same resume cursor, and stopped at the same frame. #2016 rolled a table back 2.2 days on a crash-recovery replay; #2063 starved a replication stream for 11 days with cluster_status reporting connected: true throughout. rocksdb-js now reports where valid framing resumes and leaves its reader positioned there, so the wrapper keeps pulling instead of latching, and only the broken frame is lost. Against a rocksdb-js without resyncPosition (or any other RangeError) every break still reads as a torn tail, so this degrades to the previous behavior rather than requiring a version bump in lockstep. Resyncs are capped per iteration: a log with more breaks than that is damaged beyond one interrupted append, and iteration ends as it did before. Breaks are now accumulated in a process-wide report keyed by (log, file, offset) and exposed via getCorruptFrameReports(), so a stream that has lost entries is distinguishable from a healthy one -- the missing half of #2063 -- and a mid-log break logs at error rather than warn, because entries were lost, not merely truncated. The report deliberately does not exclude the log: excludeLogs is keyed on the log *name*, i.e. an entire per-node stream ('local' is this node's own writes), so durable exclusion would stop every future entry on that stream from replicating -- converting one lost frame into a permanently dead stream. Refs #2016, #2063 Co-Authored-By: Claude Opus 5 --- resources/RocksTransactionLogStore.ts | 87 +++++++++++++++++++++++--- resources/replayLogsGuards.ts | 66 ++++++++++++++----- unitTests/resources/replayLogs.test.js | 75 ++++++++++++++++++++++ 3 files changed, 204 insertions(+), 24 deletions(-) diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index 5a35d3277b..52b5b57ab1 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/resources/RocksTransactionLogStore.ts @@ -23,11 +23,84 @@ 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); +/** + * A corrupt transaction-log frame, accumulated across every drain and boot for the life of the + * process. A break is not a per-drain event: the same frame is re-encountered by each new reader + * until every consumer's resume cursor has passed it, and before harper#2063 that condition was + * observable only as repeating log lines. Keyed by (log, file, offset) so the log line is emitted + * once per distinct break and the repeats become a count. + * + * Deliberately a *report*, not an exclusion list. Excluding the log would be worse than the break: + * `excludeLogs` is keyed on the log *name*, i.e. an entire per-node stream (`'local'` is this + * node's own writes), so dropping it stops every future entry on that stream from replicating — + * turning one lost frame into a permanently dead stream. Resyncing past the break keeps the stream + * alive; this makes the loss visible. + */ +export interface CorruptFrameReport { + log: string; + logId?: number; + position?: number; + /** Bytes skipped to resume framing; 0 when nothing valid followed the break. */ + unreadableBytes: number; + /** `false` when the break ended iteration (a torn tail, or the resync cap was hit). */ + resynced: boolean; + firstSeen: number; + lastSeen: number; + occurrences: number; +} + +const corruptFrameReports = new Map(); + +/** + * Every corrupt transaction-log frame seen by this process. Consumed by cluster/health status so a + * stream that has lost entries is distinguishable from a healthy one without grepping logs — + * the field incident behind harper#2063 ran 11 days with `connected: true` throughout. + */ +export function getCorruptFrameReports(): CorruptFrameReport[] { + return [...corruptFrameReports.values()]; +} + +// Test seam; there is no production reason to forget a break. +export function clearCorruptFrameReports() { + corruptFrameReports.clear(); +} + +/** + * Records a corrupt frame and logs it once per distinct break. A mid-log break is an `error`, not a + * `warn`: entries written and acknowledged after it were skipped, so this is data loss, not a + * tolerable end-of-log. A torn tail stays a `warn` — nothing followed it to lose. + */ +function reportCorruptFrame(logName: string) { + return (error: RangeError, resynced: boolean, unreadableBytes: number) => { + const { logId, position } = error as RangeError & { logId?: number; position?: number }; + const key = `${logName}:${logId}:${position}`; + const now = Date.now(); + const existing = corruptFrameReports.get(key); + if (existing) { + existing.occurrences++; + existing.lastSeen = now; + return; + } + corruptFrameReports.set(key, { + log: logName, + logId, + position, + unreadableBytes, + resynced, + firstSeen: now, + lastSeen: now, + occurrences: 1, + }); + if (resynced) { + harperLogger.error( + `Corrupt entry in transaction log "${logName}"; skipped ${unreadableBytes} unreadable byte(s) and resumed reading after it. ` + + `Entries within that span are lost to replay and replication and cannot be recovered from this log.`, + error + ); + } else { + harperLogger.warn(`Stopping transaction log "${logName}" at a corrupt entry during replay`, error); + } + }; } /** @@ -247,7 +320,7 @@ export class RocksTransactionLogStore extends EventEmitter { log = this.rootStore.useLog(options.log); } } - const queryIterator = endIteratorOnCorruptFrame(log.query(options), warnCorruptFrame(log.name)); + const queryIterator = endIteratorOnCorruptFrame(log.query(options), reportCorruptFrame(log.name)); iterable.iterate = () => queryIterator; } else { const onlyKeys = options.onlyKeys; @@ -298,7 +371,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(endIteratorOnCorruptFrame(log.query(queryOptions), reportCorruptFrame(log.name))); } } latestUpdates = this.updates; diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index e4f15c3c57..ce25b5123d 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -132,34 +132,66 @@ 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. + * Maximum corrupt frames to resync past within one iteration of a single log. A log with more + * breaks than this is damaged beyond "one interrupted append", and walking a long chain of them + * per drain is not worth the CPU; iteration ends at that point, exactly as it did before resync + * existed. Every break is still reported, so the count is visible rather than implied. + */ +export const MAX_RESYNCS_PER_ITERATION = 32; + +/** + * Wraps a transaction-log query iterator so a corrupt/torn frame is contained instead of escaping + * as an uncaughtException. + * + * Two shapes, which must be handled differently (harper#2016, harper#2063): + * + * - **Torn tail** — nothing valid follows the break, so the frame really is end-of-log. Iteration + * ends; entries before it were already yielded. This is what open-time recovery in rocksdb-js + * would have truncated to anyway. + * - **Mid-log break** — a partial append (ENOSPC/EDQUOT) the process survived, so valid entries + * were appended *after* it and acknowledged to clients. Treating that as end-of-log amputates + * every one of them: replay silently rolls the table back to the tear and replication starves + * the peer, permanently, because each new drain restarts from the same resume cursor and stops + * at the same frame. So iteration resyncs past the break and keeps delivering. + * + * The two are distinguished by `resyncPosition` on rocksdb-js's `CorruptFrameError`, which is the + * offset where valid framing resumes. Against a rocksdb-js without it (or any other RangeError), + * every break reads as a torn tail — the previous behavior. + * + * `onCorruptFrame` fires once per break (not once per log) and receives the number of bytes lost, + * so a caller can escalate a mid-log break and surface it as a health signal rather than leaving + * a wedged stream indistinguishable from a healthy one. */ export function endIteratorOnCorruptFrame( iterator: Iterator, - onCorruptFrame: (error: RangeError) => void + onCorruptFrame: (error: RangeError, resynced: boolean, unreadableBytes: number) => void ): IterableIterator { let stopped = false; + let resyncs = 0; 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 }; + while (!stopped) { + 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; + // A resync position means intact entries follow, and the reader has already + // positioned itself there — calling next() again resumes past the break. + const { resyncPosition, unreadableBytes } = error as RangeError & { + resyncPosition?: number; + unreadableBytes?: number; + }; + const canResync = resyncPosition !== undefined && ++resyncs <= MAX_RESYNCS_PER_ITERATION; + if (!canResync) stopped = true; + onCorruptFrame(error, canResync, unreadableBytes ?? 0); + } } + 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. diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index f1e9423c11..2a707bdf42 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -7,6 +7,7 @@ const { isUndecodableValidatedWrite, RECORD_BEARING_FLAGS, endIteratorOnCorruptFrame, + MAX_RESYNCS_PER_ITERATION, shouldAbortStalledReplay, REPLAY_NO_PROGRESS_COUNT_LIMIT, REPLAY_NO_PROGRESS_TIME_LIMIT_MS, @@ -135,6 +136,80 @@ describe('endIteratorOnCorruptFrame', () => { assert.strictEqual(reported.length, 1); }); + // harper#2016 / harper#2063: a mid-log break has intact, already-acknowledged entries behind + // it. rocksdb-js reports where framing resumes and leaves the reader positioned there, so the + // wrapper must keep pulling — treating it as end-of-log amputates every later entry, and every + // future drain restarts from the same cursor and stops at the same frame. + it('resyncs past a mid-log corrupt frame and keeps yielding the entries after 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; + } + if (calls === 3) return { done: false, value: 'b' }; + return { done: true, value: undefined }; + }, + }; + const reported = []; + const wrapped = endIteratorOnCorruptFrame(source, (error, resynced, unreadableBytes) => + reported.push({ error, resynced, unreadableBytes }) + ); + + assert.deepStrictEqual([...wrapped], ['a', 'b']); + assert.strictEqual(reported.length, 1); + assert.strictEqual(reported[0].resynced, true); + assert.strictEqual(reported[0].unreadableBytes, 26); + }); + + it('stops resyncing once a log exceeds the per-iteration cap', () => { + let calls = 0; + const source = { + next() { + calls++; + const error = new RangeError('corrupt'); + error.resyncPosition = calls * 100; + error.unreadableBytes = 8; + throw error; + }, + }; + const reported = []; + const wrapped = endIteratorOnCorruptFrame(source, (error, resynced) => reported.push(resynced)); + + assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); + // the cap is what ends iteration, and the break that hit it is reported as not-resynced + assert.strictEqual(calls, MAX_RESYNCS_PER_ITERATION + 1); + assert.strictEqual(reported.length, MAX_RESYNCS_PER_ITERATION + 1); + assert.strictEqual(reported.at(-1), false); + // latched afterwards: no further pulls on the source + assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); + assert.strictEqual(calls, MAX_RESYNCS_PER_ITERATION + 1); + }); + + 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, resynced, unreadableBytes) => + reported.push({ resynced, unreadableBytes }) + ); + + assert.deepStrictEqual([...wrapped], ['a']); + assert.deepStrictEqual(reported, [{ resynced: false, unreadableBytes: 0 }]); + assert.strictEqual(calls, 2); + }); + it('does not swallow non-RangeError failures', () => { const source = { next() { From 1eca0f30ea4c4d3c1639474fd453b66b8558e853 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 4 Aug 2026 19:56:19 -0600 Subject: [PATCH 02/14] fix(replay): key corrupt-frame severity on the break, not on whether we resynced MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-model review found three real defects in the reporting path — the part whose whole job is to stop loss from being silent. Severity keyed off `resynced`, but a mid-log break that merely hit the resync cap also loses entries; it was logged with the benign torn-tail warn ("Stopping transaction log ... at a corrupt entry"), re-hiding the exact condition #2063 exists to surface. Severity now follows the break's own shape (`resyncPosition`), and the wrapper reports whether iteration stopped as a separate fact. The report also latched on first encounter: `occurrences++; return`. A break sitting behind 32 others records `stoppedIteration` on the first drain and never updates, so once a later drain's cursor advances and it resyncs cleanly, the report still says otherwise. Repeat encounters now update the mutable state and can still escalate a break to data loss. The key was `${log}:${logId}:${position}` read off the error, but no released rocksdb-js sets those fields, so every break on a stream collapsed onto `name:undefined:undefined` — the second genuine corruption folded into a count and never logged, which is worse than the per-iterator warn it replaced. It falls back to the message, which carries the offset and file in text. The reporter moves to replayLogsGuards.ts with the logger injected, so it is unit -testable without the Resource/RocksDB graph — it was previously untested, which is why all three of the above were reachable. Also bounds the report map (256 sites, dropped ones still logged, counted via getDroppedCorruptFrameReportCount()). Co-Authored-By: Claude Opus 5 --- resources/RocksTransactionLogStore.ts | 82 +----------- resources/replayLogsGuards.ts | 168 ++++++++++++++++++++----- unitTests/resources/replayLogs.test.js | 149 ++++++++++++++++++++-- 3 files changed, 276 insertions(+), 123 deletions(-) diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index 52b5b57ab1..d91b17d5ea 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 } from './replayLogsGuards.ts'; import { isMainThread } from 'node:worker_threads'; import { EventEmitter } from 'node:events'; import { asBinary } from 'lmdb'; @@ -23,85 +23,7 @@ type TransactionLogIterator = Iterator & { removeLog(logName: string); }; -/** - * A corrupt transaction-log frame, accumulated across every drain and boot for the life of the - * process. A break is not a per-drain event: the same frame is re-encountered by each new reader - * until every consumer's resume cursor has passed it, and before harper#2063 that condition was - * observable only as repeating log lines. Keyed by (log, file, offset) so the log line is emitted - * once per distinct break and the repeats become a count. - * - * Deliberately a *report*, not an exclusion list. Excluding the log would be worse than the break: - * `excludeLogs` is keyed on the log *name*, i.e. an entire per-node stream (`'local'` is this - * node's own writes), so dropping it stops every future entry on that stream from replicating — - * turning one lost frame into a permanently dead stream. Resyncing past the break keeps the stream - * alive; this makes the loss visible. - */ -export interface CorruptFrameReport { - log: string; - logId?: number; - position?: number; - /** Bytes skipped to resume framing; 0 when nothing valid followed the break. */ - unreadableBytes: number; - /** `false` when the break ended iteration (a torn tail, or the resync cap was hit). */ - resynced: boolean; - firstSeen: number; - lastSeen: number; - occurrences: number; -} - -const corruptFrameReports = new Map(); - -/** - * Every corrupt transaction-log frame seen by this process. Consumed by cluster/health status so a - * stream that has lost entries is distinguishable from a healthy one without grepping logs — - * the field incident behind harper#2063 ran 11 days with `connected: true` throughout. - */ -export function getCorruptFrameReports(): CorruptFrameReport[] { - return [...corruptFrameReports.values()]; -} - -// Test seam; there is no production reason to forget a break. -export function clearCorruptFrameReports() { - corruptFrameReports.clear(); -} - -/** - * Records a corrupt frame and logs it once per distinct break. A mid-log break is an `error`, not a - * `warn`: entries written and acknowledged after it were skipped, so this is data loss, not a - * tolerable end-of-log. A torn tail stays a `warn` — nothing followed it to lose. - */ -function reportCorruptFrame(logName: string) { - return (error: RangeError, resynced: boolean, unreadableBytes: number) => { - const { logId, position } = error as RangeError & { logId?: number; position?: number }; - const key = `${logName}:${logId}:${position}`; - const now = Date.now(); - const existing = corruptFrameReports.get(key); - if (existing) { - existing.occurrences++; - existing.lastSeen = now; - return; - } - corruptFrameReports.set(key, { - log: logName, - logId, - position, - unreadableBytes, - resynced, - firstSeen: now, - lastSeen: now, - occurrences: 1, - }); - if (resynced) { - harperLogger.error( - `Corrupt entry in transaction log "${logName}"; skipped ${unreadableBytes} unreadable byte(s) and resumed reading after it. ` + - `Entries within that span are lost to replay and replication and cannot be recovered from this log.`, - error - ); - } else { - harperLogger.warn(`Stopping transaction log "${logName}" at a corrupt entry during replay`, error); - } - }; -} +const reportCorruptFrame = createCorruptFrameReporter(harperLogger); /** * Represents a transaction log store backed by RocksDB. diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index ce25b5123d..ec0ccdd334 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -132,10 +132,20 @@ export function shouldAbortSlowReplay(totalElapsedMs: number, timeLimitMs = REPL } /** - * Maximum corrupt frames to resync past within one iteration of a single log. A log with more - * breaks than this is damaged beyond "one interrupted append", and walking a long chain of them - * per drain is not worth the CPU; iteration ends at that point, exactly as it did before resync - * existed. Every break is still reported, so the count is visible rather than implied. + * A corrupt transaction-log frame from rocksdb-js. `resyncPosition` is set only when intact entries + * follow the break (a mid-log tear); its absence means end-of-log, which is also what every + * rocksdb-js predating the resync support reports. + */ +export type CorruptFrameError = RangeError & { + logId?: number; + position?: number; + resyncPosition?: number; + unreadableBytes?: number; +}; + +/** + * A log with more breaks than this in one pass is damaged beyond one interrupted append, so + * iteration ends rather than walking a long chain of them on every drain. */ export const MAX_RESYNCS_PER_ITERATION = 32; @@ -143,28 +153,15 @@ export const MAX_RESYNCS_PER_ITERATION = 32; * Wraps a transaction-log query iterator so a corrupt/torn frame is contained instead of escaping * as an uncaughtException. * - * Two shapes, which must be handled differently (harper#2016, harper#2063): - * - * - **Torn tail** — nothing valid follows the break, so the frame really is end-of-log. Iteration - * ends; entries before it were already yielded. This is what open-time recovery in rocksdb-js - * would have truncated to anyway. - * - **Mid-log break** — a partial append (ENOSPC/EDQUOT) the process survived, so valid entries - * were appended *after* it and acknowledged to clients. Treating that as end-of-log amputates - * every one of them: replay silently rolls the table back to the tear and replication starves - * the peer, permanently, because each new drain restarts from the same resume cursor and stops - * at the same frame. So iteration resyncs past the break and keeps delivering. - * - * The two are distinguished by `resyncPosition` on rocksdb-js's `CorruptFrameError`, which is the - * offset where valid framing resumes. Against a rocksdb-js without it (or any other RangeError), - * every break reads as a torn tail — the previous behavior. - * - * `onCorruptFrame` fires once per break (not once per log) and receives the number of bytes lost, - * so a caller can escalate a mid-log break and surface it as a health signal rather than leaving - * a wedged stream indistinguishable from a healthy one. + * A torn tail really is end-of-log. A mid-log break is not: a partial append the process survived + * has intact, already-acknowledged entries after it, so ending iteration there amputates all of + * them — permanently, since every later drain restarts from the same resume cursor and stops at + * the same frame (harper#2016, harper#2063). rocksdb-js distinguishes the two with + * `resyncPosition` and leaves its reader positioned there, so resuming is a further `next()`. */ export function endIteratorOnCorruptFrame( iterator: Iterator, - onCorruptFrame: (error: RangeError, resynced: boolean, unreadableBytes: number) => void + onCorruptFrame: (error: CorruptFrameError, stopped: boolean) => void ): IterableIterator { let stopped = false; let resyncs = 0; @@ -180,15 +177,9 @@ export function endIteratorOnCorruptFrame( // 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; - // A resync position means intact entries follow, and the reader has already - // positioned itself there — calling next() again resumes past the break. - const { resyncPosition, unreadableBytes } = error as RangeError & { - resyncPosition?: number; - unreadableBytes?: number; - }; - const canResync = resyncPosition !== undefined && ++resyncs <= MAX_RESYNCS_PER_ITERATION; - if (!canResync) stopped = true; - onCorruptFrame(error, canResync, unreadableBytes ?? 0); + const { resyncPosition } = error as CorruptFrameError; + if (resyncPosition === undefined || ++resyncs > MAX_RESYNCS_PER_ITERATION) stopped = true; + onCorruptFrame(error as CorruptFrameError, stopped); } } return { done: true, value: undefined }; @@ -207,3 +198,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; + /** Bytes skipped to resume framing; 0 when nothing valid followed the break. */ + unreadableBytes: number; + /** Whether the most recent encounter ended iteration (torn tail, or the resync cap). */ + stoppedIteration: boolean; + 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; + +const corruptFrameReports = new Map(); +let droppedCorruptFrameReports = 0; + +/** + * Every corrupt frame seen by this process, for cluster/health status: a stream that has lost + * entries must be distinguishable from a healthy one without grepping logs — the field incident + * behind harper#2063 ran 11 days with `connected: true` throughout. + */ +export function getCorruptFrameReports(): CorruptFrameReport[] { + return [...corruptFrameReports.values()]; +} + +/** Distinct break sites not retained because {@link MAX_CORRUPT_FRAME_REPORTS} was reached. */ +export function getDroppedCorruptFrameReportCount(): number { + return droppedCorruptFrameReports; +} + +// Test seam. +export function clearCorruptFrameReports() { + corruptFrameReports.clear(); + droppedCorruptFrameReports = 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 !== undefined && position !== undefined + ? `${logName}:${logId}:${position}` + : `${logName}:${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 + * unreachable. Severity follows the break's own shape rather than whether this pass resynced — one + * that merely hit the resync cap lost entries too, and reporting it as the benign torn-tail `warn` + * would re-hide what harper#2063 exists to surface. A later encounter can still escalate, since + * the first pass may have hit the cap while a later one resyncs cleanly. + */ +export function createCorruptFrameReporter(logger: { + warn: (message: string, error?: unknown) => void; + error: (message: string, error?: unknown) => void; +}) { + return (logName: string) => (error: CorruptFrameError, stoppedIteration: boolean) => { + const midLog = error.resyncPosition !== undefined; + const unreadableBytes = error.unreadableBytes ?? 0; + const now = Date.now(); + const key = corruptFrameKey(logName, error); + const existing = corruptFrameReports.get(key); + if (existing) { + existing.occurrences++; + existing.lastSeen = now; + existing.stoppedIteration = stoppedIteration; + if (!midLog || existing.midLog) return; + // first time this break has been seen to have entries behind it + existing.midLog = true; + existing.unreadableBytes = unreadableBytes; + } else if (corruptFrameReports.size >= MAX_CORRUPT_FRAME_REPORTS) { + droppedCorruptFrameReports++; + } else { + corruptFrameReports.set(key, { + log: logName, + logId: error.logId, + position: error.position, + midLog, + unreadableBytes, + stoppedIteration, + 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 to replay and replication. ` + + (stoppedIteration + ? 'This log has too many corrupt frames to read past; entries after this point are unreachable until it is repaired or ages out.' + : 'Reading resumed after it.'), + 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 2a707bdf42..1c10eee06e 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -8,6 +8,11 @@ const { RECORD_BEARING_FLAGS, endIteratorOnCorruptFrame, MAX_RESYNCS_PER_ITERATION, + MAX_CORRUPT_FRAME_REPORTS, + createCorruptFrameReporter, + getCorruptFrameReports, + getDroppedCorruptFrameReportCount, + clearCorruptFrameReports, shouldAbortStalledReplay, REPLAY_NO_PROGRESS_COUNT_LIMIT, REPLAY_NO_PROGRESS_TIME_LIMIT_MS, @@ -157,14 +162,12 @@ describe('endIteratorOnCorruptFrame', () => { }, }; const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, resynced, unreadableBytes) => - reported.push({ error, resynced, unreadableBytes }) - ); + const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => reported.push({ error, stopped })); assert.deepStrictEqual([...wrapped], ['a', 'b']); assert.strictEqual(reported.length, 1); - assert.strictEqual(reported[0].resynced, true); - assert.strictEqual(reported[0].unreadableBytes, 26); + assert.strictEqual(reported[0].stopped, false); + assert.strictEqual(reported[0].error.unreadableBytes, 26); }); it('stops resyncing once a log exceeds the per-iteration cap', () => { @@ -179,13 +182,13 @@ describe('endIteratorOnCorruptFrame', () => { }, }; const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, resynced) => reported.push(resynced)); + const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => reported.push(stopped)); assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); - // the cap is what ends iteration, and the break that hit it is reported as not-resynced + // the cap is what ends iteration, and the break that hit it is reported as stopping it assert.strictEqual(calls, MAX_RESYNCS_PER_ITERATION + 1); assert.strictEqual(reported.length, MAX_RESYNCS_PER_ITERATION + 1); - assert.strictEqual(reported.at(-1), false); + assert.strictEqual(reported.at(-1), true); // latched afterwards: no further pulls on the source assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); assert.strictEqual(calls, MAX_RESYNCS_PER_ITERATION + 1); @@ -201,15 +204,33 @@ describe('endIteratorOnCorruptFrame', () => { }, }; const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, resynced, unreadableBytes) => - reported.push({ resynced, unreadableBytes }) + const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => + reported.push({ stopped, resyncPosition: error.resyncPosition }) ); assert.deepStrictEqual([...wrapped], ['a']); - assert.deepStrictEqual(reported, [{ resynced: false, unreadableBytes: 0 }]); + assert.deepStrictEqual(reported, [{ stopped: true, resyncPosition: undefined }]); assert.strictEqual(calls, 2); }); + it('reports the cap-hit break as having stopped iteration, not as a clean resync', () => { + // The reporter keys severity off the error's own shape, so a mid-log break that merely hit + // the cap must still be distinguishable from a torn tail. + const source = { + next() { + const error = new RangeError('corrupt'); + error.resyncPosition = 1; + throw error; + }, + }; + const reported = []; + const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => + reported.push({ stopped, midLog: error.resyncPosition !== undefined }) + ); + wrapped.next(); + assert.deepStrictEqual(reported.at(-1), { stopped: true, midLog: true }); + }); + it('does not swallow non-RangeError failures', () => { const source = { next() { @@ -290,6 +311,112 @@ 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(), false); + + 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, true); + + assert.strictEqual(logs.error.length, 0); + assert.strictEqual(logs.warn.length, 1); + assert.strictEqual(getCorruptFrameReports()[0].midLog, false); + }); + + // A break that hit the resync cap lost entries just the same. Reporting it as the benign + // torn-tail warn would re-hide the condition #2063 exists to surface. + it('logs a capped mid-log break as data loss, naming it unreachable', () => { + const { logs, report } = setup(); + report(midLogError(), true); + + assert.strictEqual(logs.warn.length, 0); + assert.match(logs.error[0].message, /unreachable until it is repaired/); + }); + + it('counts repeats of the same break without re-logging it', () => { + const { logs, report } = setup(); + report(midLogError(), false); + report(midLogError(), false); + report(midLogError(), false); + + assert.strictEqual(logs.error.length, 1); + const [only] = getCorruptFrameReports(); + assert.strictEqual(only.occurrences, 3); + assert.ok(only.lastSeen >= only.firstSeen); + }); + + // The first pass may sit behind 32 other breaks and hit the cap; a later pass, its cursor + // further along, resyncs cleanly. The report and the tracked state must follow. + it('updates the stopped state on a later encounter of the same break', () => { + const { report } = setup(); + report(midLogError(), true); + assert.strictEqual(getCorruptFrameReports()[0].stoppedIteration, true); + + report(midLogError(), false); + assert.strictEqual(getCorruptFrameReports()[0].stoppedIteration, false); + }); + + // 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'), true); + report(new RangeError('Corrupt transaction log entry at position 3bc071 of log 23'), true); + + assert.strictEqual(getCorruptFrameReports().length, 2); + assert.strictEqual(logs.warn.length, 2); + }); + + it('bounds retained break sites and counts the ones it drops', () => { + const { logs, report } = setup(); + for (let i = 0; i < MAX_CORRUPT_FRAME_REPORTS + 5; i++) { + report(midLogError(0x1000 + i * 0x100), false); + } + + assert.strictEqual(getCorruptFrameReports().length, MAX_CORRUPT_FRAME_REPORTS); + assert.strictEqual(getDroppedCorruptFrameReportCount(), 5); + // dropped sites are still logged, so visibility never depends on retention + assert.strictEqual(logs.error.length, MAX_CORRUPT_FRAME_REPORTS + 5); + }); +}); + // 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 From f25159a9eddad5e8f4b1fa26b8be77f380e83c1a Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 4 Aug 2026 20:05:15 -0600 Subject: [PATCH 03/14] fix(replay): scope break reports per database and evict rather than refuse MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 review findings, both in the report's identity and retention. Every database has its own 'local' log and the report map is module-level, so two databases whose local logs broke at the same file and offset merged into one report and one log line. The key is now qualified by database, and its parts are NUL-separated so a name containing the separator can't forge another key. Retention refused new sites once full, which left them undeduplicated — so an overflow site re-logged on every drain, the log spam the report exists to replace, and the counter measured encounters rather than sites. The oldest site is evicted instead, so every retained site still deduplicates. Co-Authored-By: Claude Opus 5 --- resources/RocksTransactionLogStore.ts | 15 ++++++++++++-- resources/replayLogsGuards.ts | 27 ++++++++++++++++---------- unitTests/resources/replayLogs.test.js | 25 ++++++++++++++++++++---- 3 files changed, 51 insertions(+), 16 deletions(-) diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index d91b17d5ea..c3528c11dc 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/resources/RocksTransactionLogStore.ts @@ -37,6 +37,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 @@ -49,6 +50,8 @@ export class RocksTransactionLogStore extends EventEmitter { super(); this.log = rootDatabase.useLog('local'); this.rootStore = rootDatabase; + // every database has its own 'local' log, so break reports must not share a name + this.corruptFrameScope = (rootDatabase as any).databaseName ?? rootDatabase.name ?? ''; } /** @@ -242,7 +245,10 @@ export class RocksTransactionLogStore extends EventEmitter { log = this.rootStore.useLog(options.log); } } - const queryIterator = endIteratorOnCorruptFrame(log.query(options), reportCorruptFrame(log.name)); + const queryIterator = endIteratorOnCorruptFrame( + log.query(options), + reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`) + ); iterable.iterate = () => queryIterator; } else { const onlyKeys = options.onlyKeys; @@ -293,7 +299,12 @@ export class RocksTransactionLogStore extends EventEmitter { // condition of potentially missing an initial update queryOptions = { ...options, start: options.start ?? 0 }; } - iterators.push(endIteratorOnCorruptFrame(log.query(queryOptions), reportCorruptFrame(log.name))); + iterators.push( + endIteratorOnCorruptFrame( + log.query(queryOptions), + reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`) + ) + ); } } latestUpdates = this.updates; diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index ec0ccdd334..d442ed60b0 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -222,27 +222,32 @@ export interface CorruptFrameReport { /** Distinct break sites retained. One physical corruption yields one site, so this is generous. */ export const MAX_CORRUPT_FRAME_REPORTS = 256; +// Insertion-ordered, so the first key is the oldest site. Full means evicting that one rather than +// refusing the new site: a site absent from the map cannot be deduplicated, so it would re-log on +// every drain. const corruptFrameReports = new Map(); -let droppedCorruptFrameReports = 0; +let evictedCorruptFrameReports = 0; /** - * Every corrupt frame seen by this process, for cluster/health status: a stream that has lost + * Every corrupt frame seen by this worker, for cluster/health status: a stream that has lost * entries must be distinguishable from a healthy one without grepping logs — the field incident * behind harper#2063 ran 11 days with `connected: true` throughout. + * + * Per-isolate, so a node-wide signal has to aggregate across worker threads. */ export function getCorruptFrameReports(): CorruptFrameReport[] { return [...corruptFrameReports.values()]; } -/** Distinct break sites not retained because {@link MAX_CORRUPT_FRAME_REPORTS} was reached. */ -export function getDroppedCorruptFrameReportCount(): number { - return droppedCorruptFrameReports; +/** Break sites evicted because {@link MAX_CORRUPT_FRAME_REPORTS} was reached. */ +export function getEvictedCorruptFrameReportCount(): number { + return evictedCorruptFrameReports; } // Test seam. export function clearCorruptFrameReports() { corruptFrameReports.clear(); - droppedCorruptFrameReports = 0; + evictedCorruptFrameReports = 0; } // `logId`/`position` are absent on any rocksdb-js predating the resync support, and every break on @@ -252,8 +257,8 @@ export function clearCorruptFrameReports() { function corruptFrameKey(logName: string, error: CorruptFrameError): string { const { logId, position } = error; return logId !== undefined && position !== undefined - ? `${logName}:${logId}:${position}` - : `${logName}:${error.message}`; + ? `${logName}\u0000${logId}:${position}` + : `${logName}\u0000${error.message}`; } /** @@ -283,9 +288,11 @@ export function createCorruptFrameReporter(logger: { // first time this break has been seen to have entries behind it existing.midLog = true; existing.unreadableBytes = unreadableBytes; - } else if (corruptFrameReports.size >= MAX_CORRUPT_FRAME_REPORTS) { - droppedCorruptFrameReports++; } else { + if (corruptFrameReports.size >= MAX_CORRUPT_FRAME_REPORTS) { + corruptFrameReports.delete(corruptFrameReports.keys().next().value); + evictedCorruptFrameReports++; + } corruptFrameReports.set(key, { log: logName, logId: error.logId, diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index 1c10eee06e..7420f6b411 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -11,7 +11,7 @@ const { MAX_CORRUPT_FRAME_REPORTS, createCorruptFrameReporter, getCorruptFrameReports, - getDroppedCorruptFrameReportCount, + getEvictedCorruptFrameReportCount, clearCorruptFrameReports, shouldAbortStalledReplay, REPLAY_NO_PROGRESS_COUNT_LIMIT, @@ -404,17 +404,34 @@ describe('createCorruptFrameReporter', () => { assert.strictEqual(logs.warn.length, 2); }); - it('bounds retained break sites and counts the ones it drops', () => { + 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), false); } assert.strictEqual(getCorruptFrameReports().length, MAX_CORRUPT_FRAME_REPORTS); - assert.strictEqual(getDroppedCorruptFrameReportCount(), 5); - // dropped sites are still logged, so visibility never depends on retention + 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), false); + } + 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), false); + } + + assert.strictEqual(logs.error.length, logsAfterFill); + assert.strictEqual(getEvictedCorruptFrameReportCount(), 5); + }); }); // Regression tests for HarperFast/harper#1266: a boot replay over a backlog of unwritable entries From 0510c896aab06154a6bc66bc925b5f0ace32897f Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 4 Aug 2026 20:08:58 -0600 Subject: [PATCH 04/14] fix(replay): scope break reports by store path, not databaseName MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `databaseName` is not set on the root store when the transaction-log store is constructed, so the previous commit's scope resolved to the empty string and the per-database key it was meant to add was a no-op. The path is present, and is the right granularity besides: one root store can back several logical databases and they share its logs, so the log — not the database — is what a break belongs to. Co-Authored-By: Claude Opus 5 --- resources/RocksTransactionLogStore.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index c3528c11dc..f62f10b2cf 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/resources/RocksTransactionLogStore.ts @@ -50,8 +50,10 @@ export class RocksTransactionLogStore extends EventEmitter { super(); this.log = rootDatabase.useLog('local'); this.rootStore = rootDatabase; - // every database has its own 'local' log, so break reports must not share a name - this.corruptFrameScope = (rootDatabase as any).databaseName ?? rootDatabase.name ?? ''; + // 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; } /** From 3714e5a143316060d8329ca1a4481fdd6d4a1037 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 17 Aug 2026 09:30:34 -0600 Subject: [PATCH 05/14] fix(replay): treat null corrupt-frame offsets as absent Co-Authored-By: GPT-5 Codex --- resources/replayLogsGuards.ts | 18 +++---- unitTests/resources/replayLogs.test.js | 72 +++++++++++++++++++++++++- 2 files changed, 80 insertions(+), 10 deletions(-) diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index d442ed60b0..72a65f3226 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -137,10 +137,10 @@ export function shouldAbortSlowReplay(totalElapsedMs: number, timeLimitMs = REPL * rocksdb-js predating the resync support reports. */ export type CorruptFrameError = RangeError & { - logId?: number; - position?: number; - resyncPosition?: number; - unreadableBytes?: number; + logId?: number | null; + position?: number | null; + resyncPosition?: number | null; + unreadableBytes?: number | null; }; /** @@ -178,7 +178,7 @@ export function endIteratorOnCorruptFrame( // version-dependent (1.4.2 added hex offsets). Anything else re-throws. if (!(error instanceof RangeError)) throw error; const { resyncPosition } = error as CorruptFrameError; - if (resyncPosition === undefined || ++resyncs > MAX_RESYNCS_PER_ITERATION) stopped = true; + if (resyncPosition == null || ++resyncs > MAX_RESYNCS_PER_ITERATION) stopped = true; onCorruptFrame(error as CorruptFrameError, stopped); } } @@ -256,7 +256,7 @@ export function clearCorruptFrameReports() { // them when the fields can't. function corruptFrameKey(logName: string, error: CorruptFrameError): string { const { logId, position } = error; - return logId !== undefined && position !== undefined + return logId != null && position != null ? `${logName}\u0000${logId}:${position}` : `${logName}\u0000${error.message}`; } @@ -275,7 +275,7 @@ export function createCorruptFrameReporter(logger: { error: (message: string, error?: unknown) => void; }) { return (logName: string) => (error: CorruptFrameError, stoppedIteration: boolean) => { - const midLog = error.resyncPosition !== undefined; + const midLog = error.resyncPosition != null; const unreadableBytes = error.unreadableBytes ?? 0; const now = Date.now(); const key = corruptFrameKey(logName, error); @@ -295,8 +295,8 @@ export function createCorruptFrameReporter(logger: { } corruptFrameReports.set(key, { log: logName, - logId: error.logId, - position: error.position, + logId: error.logId ?? undefined, + position: error.position ?? undefined, midLog, unreadableBytes, stoppedIteration, diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index 7420f6b411..3b2ec749c6 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -213,6 +213,48 @@ describe('endIteratorOnCorruptFrame', () => { 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, stopped) => { + assert.strictEqual(reportedError, error); + assert.strictEqual(stopped, true); + } + ); + + assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); + assert.strictEqual(calls, 1); + }); + + it('keeps zero as a valid resync position', () => { + const error = new RangeError('corrupt frame at offset 0'); + error.resyncPosition = 0; + let calls = 0; + const wrapped = endIteratorOnCorruptFrame( + { + next() { + if (calls++ === 0) throw error; + return { done: true, value: undefined }; + }, + }, + (reportedError, stopped) => { + assert.strictEqual(reportedError, error); + assert.strictEqual(stopped, false); + } + ); + + assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); + assert.strictEqual(calls, 2); + }); + it('reports the cap-hit break as having stopped iteration, not as a clean resync', () => { // The reporter keys severity off the error's own shape, so a mid-log break that merely hit // the cap must still be distinguishable from a torn tail. @@ -225,7 +267,7 @@ describe('endIteratorOnCorruptFrame', () => { }; const reported = []; const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => - reported.push({ stopped, midLog: error.resyncPosition !== undefined }) + reported.push({ stopped, midLog: error.resyncPosition != null }) ); wrapped.next(); assert.deepStrictEqual(reported.at(-1), { stopped: true, midLog: true }); @@ -404,6 +446,34 @@ describe('createCorruptFrameReporter', () => { 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, true); + } + + 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, false); + + 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++) { From 57bdc44dc3d0a5aaf51631dce3e345d0b3e85cbd Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 17 Aug 2026 09:42:45 -0600 Subject: [PATCH 06/14] fix(replay): reset corruption bounds after progress Keep actively recurring report sites resident so bounded telemetry remains deduplicated. Co-Authored-By: GPT-5 Codex --- resources/replayLogsGuards.ts | 6 +++- unitTests/resources/replayLogs.test.js | 40 ++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index 72a65f3226..643bf6a198 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -172,7 +172,9 @@ export function endIteratorOnCorruptFrame( next(): IteratorResult { while (!stopped) { try { - return iterator.next(); + const result = iterator.next(); + resyncs = 0; + return result; } 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. @@ -281,6 +283,8 @@ export function createCorruptFrameReporter(logger: { const key = corruptFrameKey(logName, error); const existing = corruptFrameReports.get(key); if (existing) { + corruptFrameReports.delete(key); + corruptFrameReports.set(key, existing); existing.occurrences++; existing.lastSeen = now; existing.stoppedIteration = stoppedIteration; diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index 3b2ec749c6..12eca1f553 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -194,6 +194,30 @@ describe('endIteratorOnCorruptFrame', () => { assert.strictEqual(calls, MAX_RESYNCS_PER_ITERATION + 1); }); + it('resets the resync cap after the source makes progress', () => { + let frames = 0; + let corrupt = true; + const source = { + next() { + if (frames > MAX_RESYNCS_PER_ITERATION) return { done: true, value: undefined }; + if (corrupt) { + corrupt = false; + const error = new RangeError('corrupt'); + error.resyncPosition = frames; + throw error; + } + corrupt = true; + return { done: false, value: frames++ }; + }, + }; + const reported = []; + const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => reported.push(stopped)); + + assert.strictEqual([...wrapped].length, MAX_RESYNCS_PER_ITERATION + 1); + assert.strictEqual(reported.length, MAX_RESYNCS_PER_ITERATION + 1); + assert.ok(reported.every((stopped) => stopped === false)); + }); + it('treats a RangeError with no resync position as end-of-log (older rocksdb-js)', () => { let calls = 0; const source = { @@ -502,6 +526,22 @@ describe('createCorruptFrameReporter', () => { 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), false); + } + const firstPosition = 0x1000; + const secondPosition = 0x1100; + report(midLogError(firstPosition), false); + report(midLogError(0x1000 + MAX_CORRUPT_FRAME_REPORTS * 0x100), false); + + 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 From 115ceeaedf838dd11d37418e8523ffcd18fe4442 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 17 Aug 2026 09:45:53 -0600 Subject: [PATCH 07/14] docs(replay): clarify corrupt-frame report scope Co-Authored-By: GPT-5 Codex --- resources/replayLogsGuards.ts | 14 +++++--------- 1 file changed, 5 insertions(+), 9 deletions(-) diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index 643bf6a198..8bf433975a 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -224,18 +224,16 @@ export interface CorruptFrameReport { /** Distinct break sites retained. One physical corruption yields one site, so this is generous. */ export const MAX_CORRUPT_FRAME_REPORTS = 256; -// Insertion-ordered, so the first key is the oldest site. Full means evicting that one rather than -// refusing the new site: a site absent from the map cannot be deduplicated, so it would re-log on -// every drain. +// 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, for cluster/health status: a stream that has lost - * entries must be distinguishable from a healthy one without grepping logs — the field incident - * behind harper#2063 ran 11 days with `connected: true` throughout. + * 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. * - * Per-isolate, so a node-wide signal has to aggregate across worker threads. + * Mid-log breaks are immediately logged at error level even without that consumer. */ export function getCorruptFrameReports(): CorruptFrameReport[] { return [...corruptFrameReports.values()]; @@ -246,7 +244,6 @@ export function getEvictedCorruptFrameReportCount(): number { return evictedCorruptFrameReports; } -// Test seam. export function clearCorruptFrameReports() { corruptFrameReports.clear(); evictedCorruptFrameReports = 0; @@ -289,7 +286,6 @@ export function createCorruptFrameReporter(logger: { existing.lastSeen = now; existing.stoppedIteration = stoppedIteration; if (!midLog || existing.midLog) return; - // first time this break has been seen to have entries behind it existing.midLog = true; existing.unreadableBytes = unreadableBytes; } else { From 0369d6b2caa92b71c15a86cf77a110405ee6e755 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 17 Aug 2026 09:52:49 -0600 Subject: [PATCH 08/14] docs(replay): describe the consecutive resync bound Co-Authored-By: GPT-5 Codex --- resources/replayLogsGuards.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index 8bf433975a..6f9fd7360d 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -144,8 +144,8 @@ export type CorruptFrameError = RangeError & { }; /** - * A log with more breaks than this in one pass is damaged beyond one interrupted append, so - * iteration ends rather than walking a long chain of them on every drain. + * A log with more consecutive breaks than this without yielding an entry is damaged beyond one + * interrupted append, so iteration ends rather than walking that chain on every drain. */ export const MAX_RESYNCS_PER_ITERATION = 32; From 52f13bf5618d7c577c0c23d35fd9b991bc064c3d Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 25 Aug 2026 11:28:48 -0600 Subject: [PATCH 09/14] Clarify corrupt-frame restart recovery Co-Authored-By: GPT-5 Codex --- resources/replayLogsGuards.ts | 2 +- unitTests/resources/replayLogs.test.js | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index 6f9fd7360d..f43de534a9 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -309,7 +309,7 @@ export function createCorruptFrameReporter(logger: { logger.error( `Corrupt entry in transaction log "${logName}"; ${unreadableBytes} byte(s) are unreadable and the entries within them are lost to replay and replication. ` + (stoppedIteration - ? 'This log has too many corrupt frames to read past; entries after this point are unreachable until it is repaired or ages out.' + ? 'This log has too many corrupt frames to read past; entries after this point remain unreachable after repair or aging until the worker/store reader is reconstructed, normally by restarting the worker.' : 'Reading resumed after it.'), error ); diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index 12eca1f553..a1bdaf4850 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -433,7 +433,7 @@ describe('createCorruptFrameReporter', () => { report(midLogError(), true); assert.strictEqual(logs.warn.length, 0); - assert.match(logs.error[0].message, /unreachable until it is repaired/); + assert.match(logs.error[0].message, /remain unreachable.*until the worker\/store reader is reconstructed/); }); it('counts repeats of the same break without re-logging it', () => { From 58f9c266c858aba417092dfce76e1a0d1815ba05 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Tue, 25 Aug 2026 11:44:55 -0600 Subject: [PATCH 10/14] Escalate repeated corrupt-frame stops Co-Authored-By: GPT-5 Codex --- resources/replayLogsGuards.ts | 9 ++++++--- unitTests/resources/replayLogs.test.js | 12 ++++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index f43de534a9..b3fe893bd6 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -280,14 +280,17 @@ export function createCorruptFrameReporter(logger: { const key = corruptFrameKey(logName, error); const existing = corruptFrameReports.get(key); if (existing) { + const stoppedEscalated = stoppedIteration && !existing.stoppedIteration; corruptFrameReports.delete(key); corruptFrameReports.set(key, existing); existing.occurrences++; existing.lastSeen = now; existing.stoppedIteration = stoppedIteration; - if (!midLog || existing.midLog) return; - existing.midLog = true; - existing.unreadableBytes = unreadableBytes; + if (!midLog || (existing.midLog && !stoppedEscalated)) return; + if (!existing.midLog) { + existing.midLog = true; + existing.unreadableBytes = unreadableBytes; + } } else { if (corruptFrameReports.size >= MAX_CORRUPT_FRAME_REPORTS) { corruptFrameReports.delete(corruptFrameReports.keys().next().value); diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index a1bdaf4850..94f0c11c4c 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -459,6 +459,18 @@ describe('createCorruptFrameReporter', () => { assert.strictEqual(getCorruptFrameReports()[0].stoppedIteration, false); }); + it('logs when a repeated mid-log break escalates from resumed to stopped', () => { + const { logs, report } = setup(); + report(midLogError(), false); + report(midLogError(), true); + report(midLogError(), true); + + assert.strictEqual(logs.error.length, 2); + assert.match(logs.error[0].message, /Reading resumed/); + assert.match(logs.error[1].message, /remain unreachable/); + assert.strictEqual(getCorruptFrameReports()[0].stoppedIteration, true); + }); + // 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', () => { From ad90fca221e00a4dcbcedd4f0a56d3d768b457ca Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 26 Aug 2026 11:06:22 -0600 Subject: [PATCH 11/14] fix(replay): fail-stop at a corrupt transaction-log frame instead of resyncing past it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resuming iteration past a mid-log break recovered the intact entries behind it, but a transaction-log frame carries no transaction boundary: 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. Iteration now stops at the break whichever shape it has, and the transaction the break truncated is discarded rather than committed in part — the tear was reachable at the stop as well, not only past it. `resyncPosition` is kept as the signal that entries were lost rather than merely truncated, so the report and its error-level log line still distinguish the #2063 signature from a torn tail. Refs #2016, #2063 Co-Authored-By: Claude Opus --- .../replay-transaction-atomicity.test.ts | 149 ++++++++++ resources/RocksTransactionLogStore.ts | 35 ++- resources/replayLogs.ts | 39 ++- resources/replayLogsGuards.ts | 121 +++++---- unitTests/resources/replayLogs.test.js | 257 +++++++++--------- 5 files changed, 405 insertions(+), 196 deletions(-) create mode 100644 integrationTests/server/replay-transaction-atomicity.test.ts diff --git a/integrationTests/server/replay-transaction-atomicity.test.ts b/integrationTests/server/replay-transaction-atomicity.test.ts new file mode 100644 index 0000000000..03dccfff28 --- /dev/null +++ b/integrationTests/server/replay-transaction-atomicity.test.ts @@ -0,0 +1,149 @@ +/** + * 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; // ran past the end / already unframable + 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); + + const tornRows = await countInRange(ctx.harper, 1001, TORN_IDS); + ok( + tornRows === 0 || tornRows === TORN_IDS, + `the truncated transaction must be all-or-nothing, found ${tornRows} of ${TORN_IDS} rows` + ); + // The entries before it are intact, so the transaction that completed ahead of the break is + // unaffected — fail-stop must not cost more than the transaction it landed in. + equal(await countInRange(ctx.harper, 1, EARLIER_IDS), EARLIER_IDS); + }); +}); diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index f62f10b2cf..09dfda10d9 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 { createCorruptFrameReporter, 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,6 +23,11 @@ type TransactionLogIterator = Iterator & { removeLog(logName: string); }; +export type TransactionLogIterable = Iterable & { + /** Corrupt frames that ended a log's iteration during this range. */ + corruptFrameStop: CorruptFrameStop; +}; + const reportCorruptFrame = createCorruptFrameReporter(harperLogger); /** @@ -231,9 +236,20 @@ export class RocksTransactionLogStore extends EventEmitter { startByLog?: Map; startFromLastFlushed?: boolean; readUncommitted?: boolean; - }): Iterable { + }): TransactionLogIterable { let iterable = new ExtendedIterable(); let aggregateIterator: TransactionLogIterator; + // Shared by every per-log iterator of this range so a consumer can tell iteration that ended at + // a corrupt frame from iteration that reached the end of the logs, and can attribute the break + // to the entry it was reading when the break surfaced. + const corruptFrameStop: CorruptFrameStop = { breaks: 0 }; + const onCorruptFrame = (log: TransactionLog) => { + const report = reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`); + return (error) => { + corruptFrameStop.breaks++; + report(error); + }; + }; if (options.log !== undefined) { let log = typeof options.log === 'number' ? this.nodeLogs?.[options.log] : this.logByName.get(options.log); if (!log) { @@ -247,10 +263,7 @@ export class RocksTransactionLogStore extends EventEmitter { log = this.rootStore.useLog(options.log); } } - const queryIterator = endIteratorOnCorruptFrame( - log.query(options), - reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`) - ); + const queryIterator = endIteratorOnCorruptFrame(log.query(options), onCorruptFrame(log)); iterable.iterate = () => queryIterator; } else { const onlyKeys = options.onlyKeys; @@ -301,12 +314,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), - reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`) - ) - ); + iterators.push(endIteratorOnCorruptFrame(log.query(queryOptions), onCorruptFrame(log))); } } latestUpdates = this.updates; @@ -450,7 +458,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..8d90fd6dc1 100644 --- a/resources/replayLogs.ts +++ b/resources/replayLogs.ts @@ -11,6 +11,7 @@ import { isUndecodableValidatedWrite, shouldAbortStalledReplay, shouldAbortSlowReplay, + createTruncatedTransactionTracker, REPLAY_WALL_CLOCK_LIMIT_MS, } from './replayLogsGuards.ts'; import { purgeAgedLogs } from './auditStore.ts'; @@ -78,6 +79,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,13 +97,16 @@ 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 }); + const truncated = createTruncatedTransactionTracker(entries.corruptFrameStop); + 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.` ); break; } + truncated.observe(auditRecord.version); const { type, tableId, @@ -166,13 +174,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 = truncated.wasTruncated(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 +216,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 +300,20 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise }); } } + const finalTorn = truncated.wasTruncated(lastTimestamp, true); 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 b3fe893bd6..010a8d7552 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -133,8 +133,8 @@ export function shouldAbortSlowReplay(totalElapsedMs: number, timeLimitMs = REPL /** * A corrupt transaction-log frame from rocksdb-js. `resyncPosition` is set only when intact entries - * follow the break (a mid-log tear); its absence means end-of-log, which is also what every - * rocksdb-js predating the resync support reports. + * 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; @@ -144,47 +144,80 @@ export type CorruptFrameError = RangeError & { }; /** - * A log with more consecutive breaks than this without yielding an entry is damaged beyond one - * interrupted append, so iteration ends rather than walking that chain on every drain. + * Corrupt frames hit by one `getRange` call, shared by its per-log iterators. + * + * A count, not a flag, because a consumer has to attribute each break to the source transaction it + * truncated: the entries an aggregate iterator has yielded from the broken log stop at the break, + * so the transaction those last entries belong to is the one left incomplete. Replay must discard + * that transaction rather than commit the surviving part of it. */ -export const MAX_RESYNCS_PER_ITERATION = 32; +export interface CorruptFrameStop { + breaks: number; +} /** - * Wraps a transaction-log query iterator so a corrupt/torn frame is contained instead of escaping - * as an uncaughtException. + * Attributes each corrupt frame to the source transaction it truncated, so a replay can drop that + * transaction whole instead of committing the part of it that was still readable. * - * A torn tail really is end-of-log. A mid-log break is not: a partial append the process survived - * has intact, already-acknowledged entries after it, so ending iteration there amputates all of - * them — permanently, since every later drain restarts from the same resume cursor and stops at - * the same frame (harper#2016, harper#2063). rocksdb-js distinguishes the two with - * `resyncPosition` and leaves its reader positioned there, so resuming is a further `next()`. + * An aggregate transaction-log iterator refills a log's lookahead as it hands back that log's + * entry, so a break becomes visible on the very entry whose successors it destroyed — the version + * that entry belongs to is the incomplete transaction. Call {@link observe} for every entry, in + * order, then ask about a version when its transaction is about to be committed. + */ +export function createTruncatedTransactionTracker(corruptFrameStop: CorruptFrameStop) { + let accountedBreaks = 0; + let truncatedVersion: number; + return { + observe(version: number) { + if (corruptFrameStop.breaks > accountedBreaks) { + accountedBreaks = corruptFrameStop.breaks; + truncatedVersion = version; + } + }, + /** + * @param final the transaction staged when iteration ended — a break on the final pull has no + * following entry to be attributed to, so it belongs to this one + */ + wasTruncated(version: number, final = false): boolean { + return truncatedVersion === version || (final && corruptFrameStop.breaks > accountedBreaks); + }, + }; +} + +/** + * 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: CorruptFrameError, stopped: boolean) => void + onCorruptFrame: (error: CorruptFrameError) => void ): IterableIterator { let stopped = false; - let resyncs = 0; return { [Symbol.iterator]() { return this; }, next(): IteratorResult { - while (!stopped) { - try { - const result = iterator.next(); - resyncs = 0; - return result; - } 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; - const { resyncPosition } = error as CorruptFrameError; - if (resyncPosition == null || ++resyncs > MAX_RESYNCS_PER_ITERATION) stopped = true; - onCorruptFrame(error as CorruptFrameError, stopped); - } + 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 as CorruptFrameError); + return { done: true, value: undefined }; } - 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. @@ -212,10 +245,8 @@ export interface CorruptFrameReport { position?: number; /** Intact entries followed the break, so entries were lost rather than merely truncated. */ midLog: boolean; - /** Bytes skipped to resume framing; 0 when nothing valid followed the break. */ + /** Extent of the unreadable region, when the engine reports it; 0 when nothing valid follows it. */ unreadableBytes: number; - /** Whether the most recent encounter ended iteration (torn tail, or the resync cap). */ - stoppedIteration: boolean; firstSeen: number; lastSeen: number; occurrences: number; @@ -264,33 +295,29 @@ function corruptFrameKey(logName: string, error: CorruptFrameError): string { * 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 - * unreachable. Severity follows the break's own shape rather than whether this pass resynced — one - * that merely hit the resync cap lost entries too, and reporting it as the benign torn-tail `warn` - * would re-hide what harper#2063 exists to surface. A later encounter can still escalate, since - * the first pass may have hit the cap while a later one resyncs cleanly. + * 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, stoppedIteration: boolean) => { + 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) { - const stoppedEscalated = stoppedIteration && !existing.stoppedIteration; corruptFrameReports.delete(key); corruptFrameReports.set(key, existing); existing.occurrences++; existing.lastSeen = now; - existing.stoppedIteration = stoppedIteration; - if (!midLog || (existing.midLog && !stoppedEscalated)) return; - if (!existing.midLog) { - existing.midLog = true; - existing.unreadableBytes = unreadableBytes; - } + 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); @@ -302,7 +329,6 @@ export function createCorruptFrameReporter(logger: { position: error.position ?? undefined, midLog, unreadableBytes, - stoppedIteration, firstSeen: now, lastSeen: now, occurrences: 1, @@ -310,10 +336,9 @@ export function createCorruptFrameReporter(logger: { } if (midLog) { logger.error( - `Corrupt entry in transaction log "${logName}"; ${unreadableBytes} byte(s) are unreadable and the entries within them are lost to replay and replication. ` + - (stoppedIteration - ? 'This log has too many corrupt frames to read past; entries after this point remain unreachable after repair or aging until the worker/store reader is reconstructed, normally by restarting the worker.' - : 'Reading resumed after it.'), + `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 { diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index 94f0c11c4c..20f82c8ec2 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -7,7 +7,7 @@ const { isUndecodableValidatedWrite, RECORD_BEARING_FLAGS, endIteratorOnCorruptFrame, - MAX_RESYNCS_PER_ITERATION, + createTruncatedTransactionTracker, MAX_CORRUPT_FRAME_REPORTS, createCorruptFrameReporter, getCorruptFrameReports, @@ -141,11 +141,11 @@ describe('endIteratorOnCorruptFrame', () => { assert.strictEqual(reported.length, 1); }); - // harper#2016 / harper#2063: a mid-log break has intact, already-acknowledged entries behind - // it. rocksdb-js reports where framing resumes and leaves the reader positioned there, so the - // wrapper must keep pulling — treating it as end-of-log amputates every later entry, and every - // future drain restarts from the same cursor and stops at the same frame. - it('resyncs past a mid-log corrupt frame and keeps yielding the entries after it', () => { + // 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() { @@ -157,65 +157,19 @@ describe('endIteratorOnCorruptFrame', () => { error.unreadableBytes = 26; throw error; } - if (calls === 3) return { done: false, value: 'b' }; - return { done: true, value: undefined }; + return { done: false, value: 'b' }; }, }; const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => reported.push({ error, stopped })); + const wrapped = endIteratorOnCorruptFrame(source, (error) => reported.push(error)); - assert.deepStrictEqual([...wrapped], ['a', 'b']); + assert.deepStrictEqual([...wrapped], ['a']); assert.strictEqual(reported.length, 1); - assert.strictEqual(reported[0].stopped, false); - assert.strictEqual(reported[0].error.unreadableBytes, 26); - }); - - it('stops resyncing once a log exceeds the per-iteration cap', () => { - let calls = 0; - const source = { - next() { - calls++; - const error = new RangeError('corrupt'); - error.resyncPosition = calls * 100; - error.unreadableBytes = 8; - throw error; - }, - }; - const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => reported.push(stopped)); - + 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 }); - // the cap is what ends iteration, and the break that hit it is reported as stopping it - assert.strictEqual(calls, MAX_RESYNCS_PER_ITERATION + 1); - assert.strictEqual(reported.length, MAX_RESYNCS_PER_ITERATION + 1); - assert.strictEqual(reported.at(-1), true); - // latched afterwards: no further pulls on the source - assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); - assert.strictEqual(calls, MAX_RESYNCS_PER_ITERATION + 1); - }); - - it('resets the resync cap after the source makes progress', () => { - let frames = 0; - let corrupt = true; - const source = { - next() { - if (frames > MAX_RESYNCS_PER_ITERATION) return { done: true, value: undefined }; - if (corrupt) { - corrupt = false; - const error = new RangeError('corrupt'); - error.resyncPosition = frames; - throw error; - } - corrupt = true; - return { done: false, value: frames++ }; - }, - }; - const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => reported.push(stopped)); - - assert.strictEqual([...wrapped].length, MAX_RESYNCS_PER_ITERATION + 1); - assert.strictEqual(reported.length, MAX_RESYNCS_PER_ITERATION + 1); - assert.ok(reported.every((stopped) => stopped === false)); + assert.strictEqual(calls, 2); }); it('treats a RangeError with no resync position as end-of-log (older rocksdb-js)', () => { @@ -228,12 +182,10 @@ describe('endIteratorOnCorruptFrame', () => { }, }; const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => - reported.push({ stopped, resyncPosition: error.resyncPosition }) - ); + const wrapped = endIteratorOnCorruptFrame(source, (error) => reported.push(error.resyncPosition)); assert.deepStrictEqual([...wrapped], ['a']); - assert.deepStrictEqual(reported, [{ stopped: true, resyncPosition: undefined }]); + assert.deepStrictEqual(reported, [undefined]); assert.strictEqual(calls, 2); }); @@ -248,9 +200,8 @@ describe('endIteratorOnCorruptFrame', () => { throw error; }, }, - (reportedError, stopped) => { + (reportedError) => { assert.strictEqual(reportedError, error); - assert.strictEqual(stopped, true); } ); @@ -258,43 +209,25 @@ describe('endIteratorOnCorruptFrame', () => { 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() { - if (calls++ === 0) throw error; - return { done: true, value: undefined }; + calls++; + throw error; }, }, - (reportedError, stopped) => { - assert.strictEqual(reportedError, error); - assert.strictEqual(stopped, false); - } + (reportedError) => reported.push(reportedError.resyncPosition) ); assert.deepStrictEqual(wrapped.next(), { done: true, value: undefined }); - assert.strictEqual(calls, 2); - }); - - it('reports the cap-hit break as having stopped iteration, not as a clean resync', () => { - // The reporter keys severity off the error's own shape, so a mid-log break that merely hit - // the cap must still be distinguishable from a torn tail. - const source = { - next() { - const error = new RangeError('corrupt'); - error.resyncPosition = 1; - throw error; - }, - }; - const reported = []; - const wrapped = endIteratorOnCorruptFrame(source, (error, stopped) => - reported.push({ stopped, midLog: error.resyncPosition != null }) - ); - wrapped.next(); - assert.deepStrictEqual(reported.at(-1), { stopped: true, midLog: true }); + assert.deepStrictEqual(reported, [0]); + assert.strictEqual(calls, 1); }); it('does not swallow non-RangeError failures', () => { @@ -377,6 +310,71 @@ describe('endIteratorOnCorruptFrame', () => { }); }); +// harper#2016 / harper#2063: replay groups equal-version entries into one source transaction, so a +// break inside such a group leaves the surviving part of a transaction that never committed that +// way at the source. The tracker is what tells replay which staged transaction to drop. +describe('createTruncatedTransactionTracker', () => { + it('reports nothing truncated while no break has been seen', () => { + const stop = { breaks: 0 }; + const tracker = createTruncatedTransactionTracker(stop); + for (const version of [10, 10, 20]) tracker.observe(version); + + assert.strictEqual(tracker.wasTruncated(10), false); + assert.strictEqual(tracker.wasTruncated(20), false); + assert.strictEqual(tracker.wasTruncated(20, true), false); + }); + + // The break surfaces on the entry whose successors it destroyed, because the aggregate iterator + // refills that log's lookahead as it hands the entry back. + it('attributes a break to the version of the entry it surfaced on', () => { + const stop = { breaks: 0 }; + const tracker = createTruncatedTransactionTracker(stop); + tracker.observe(10); + stop.breaks++; + tracker.observe(10); + tracker.observe(20); + + assert.strictEqual(tracker.wasTruncated(10), true); + assert.strictEqual(tracker.wasTruncated(20), false); + }); + + // A break seen as the FIRST entry of the next transaction arrives belongs to that next + // transaction, not to the one just completed — the opposite attribution would commit the torn + // one and discard an intact one. + it('attributes a break on a boundary entry to the transaction that entry starts', () => { + const stop = { breaks: 0 }; + const tracker = createTruncatedTransactionTracker(stop); + tracker.observe(10); + stop.breaks++; + tracker.observe(20); + + assert.strictEqual(tracker.wasTruncated(10), false); + assert.strictEqual(tracker.wasTruncated(20), true); + }); + + it('attributes a break with no entry after it to the transaction still staged at the end', () => { + const stop = { breaks: 0 }; + const tracker = createTruncatedTransactionTracker(stop); + tracker.observe(10); + stop.breaks++; + + assert.strictEqual(tracker.wasTruncated(10), false); + assert.strictEqual(tracker.wasTruncated(10, true), true); + }); + + it('follows the latest break when several logs break during one replay', () => { + const stop = { breaks: 0 }; + const tracker = createTruncatedTransactionTracker(stop); + stop.breaks++; + tracker.observe(10); + stop.breaks++; + tracker.observe(20); + + assert.strictEqual(tracker.wasTruncated(10), false); + assert.strictEqual(tracker.wasTruncated(20), true); + }); +}); + // 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', () => { @@ -401,7 +399,7 @@ describe('createCorruptFrameReporter', () => { it('logs a mid-log break at error level and records the lost bytes', () => { const { logs, report } = setup(); - report(midLogError(), false); + report(midLogError()); assert.strictEqual(logs.warn.length, 0); assert.strictEqual(logs.error.length, 1); @@ -419,28 +417,28 @@ describe('createCorruptFrameReporter', () => { const error = new RangeError('truncated entry header'); error.logId = 2; error.position = 100; - report(error, true); + report(error); assert.strictEqual(logs.error.length, 0); assert.strictEqual(logs.warn.length, 1); assert.strictEqual(getCorruptFrameReports()[0].midLog, false); }); - // A break that hit the resync cap lost entries just the same. Reporting it as the benign - // torn-tail warn would re-hide the condition #2063 exists to surface. - it('logs a capped mid-log break as data loss, naming it unreachable', () => { + // 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(), true); + report(midLogError()); assert.strictEqual(logs.warn.length, 0); - assert.match(logs.error[0].message, /remain unreachable.*until the worker\/store reader is reconstructed/); + 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(), false); - report(midLogError(), false); - report(midLogError(), false); + report(midLogError()); + report(midLogError()); + report(midLogError()); assert.strictEqual(logs.error.length, 1); const [only] = getCorruptFrameReports(); @@ -448,35 +446,36 @@ describe('createCorruptFrameReporter', () => { assert.ok(only.lastSeen >= only.firstSeen); }); - // The first pass may sit behind 32 other breaks and hit the cap; a later pass, its cursor - // further along, resyncs cleanly. The report and the tracked state must follow. - it('updates the stopped state on a later encounter of the same break', () => { - const { report } = setup(); - report(midLogError(), true); - assert.strictEqual(getCorruptFrameReports()[0].stoppedIteration, true); - - report(midLogError(), false); - assert.strictEqual(getCorruptFrameReports()[0].stoppedIteration, false); - }); - - it('logs when a repeated mid-log break escalates from resumed to stopped', () => { + // 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(); - report(midLogError(), false); - report(midLogError(), true); - report(midLogError(), true); + const tornTail = midLogError(); + delete tornTail.resyncPosition; + delete tornTail.unreadableBytes; + report(tornTail); + assert.strictEqual(getCorruptFrameReports()[0].midLog, false); - assert.strictEqual(logs.error.length, 2); - assert.match(logs.error[0].message, /Reading resumed/); - assert.match(logs.error[1].message, /remain unreachable/); - assert.strictEqual(getCorruptFrameReports()[0].stoppedIteration, true); + 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'), true); - report(new RangeError('Corrupt transaction log entry at position 3bc071 of log 23'), true); + 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); @@ -489,7 +488,7 @@ describe('createCorruptFrameReporter', () => { error.logId = null; error.position = null; error.resyncPosition = null; - report(error, true); + report(error); } assert.strictEqual(getCorruptFrameReports().length, 2); @@ -501,7 +500,7 @@ describe('createCorruptFrameReporter', () => { const { logs, report } = setup(); const error = midLogError(0, 0); error.logId = 0; - report(error, false); + report(error); assert.strictEqual(logs.error.length, 1); assert.deepStrictEqual( @@ -513,7 +512,7 @@ describe('createCorruptFrameReporter', () => { 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), false); + report(midLogError(0x1000 + i * 0x100)); } assert.strictEqual(getCorruptFrameReports().length, MAX_CORRUPT_FRAME_REPORTS); @@ -526,13 +525,13 @@ describe('createCorruptFrameReporter', () => { 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), false); + 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), false); + report(midLogError(0x1000 + (MAX_CORRUPT_FRAME_REPORTS + 4) * 0x100)); } assert.strictEqual(logs.error.length, logsAfterFill); @@ -542,12 +541,12 @@ describe('createCorruptFrameReporter', () => { 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), false); + report(midLogError(0x1000 + i * 0x100)); } const firstPosition = 0x1000; const secondPosition = 0x1100; - report(midLogError(firstPosition), false); - report(midLogError(0x1000 + MAX_CORRUPT_FRAME_REPORTS * 0x100), false); + report(midLogError(firstPosition)); + report(midLogError(0x1000 + MAX_CORRUPT_FRAME_REPORTS * 0x100)); const positions = getCorruptFrameReports().map((entry) => entry.position); assert.ok(positions.includes(firstPosition)); From 32d8aee132c197319a72bbc6efea631246109012 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 26 Aug 2026 11:36:21 -0600 Subject: [PATCH 12/14] fix(replay): attribute a corrupt frame to the log that broke, not to the entry stream A break destroys the framing after the last entry the broken log yielded, so the transaction at risk is the one THAT log's last entry belongs to. Counting breaks against the merged entry stream blamed whichever transaction happened to be staged, so a log that broke before yielding anything, or a break in one log while another log's transaction was open, discarded an intact transaction. Co-Authored-By: Claude Opus --- .../replay-transaction-atomicity.test.ts | 12 ++-- resources/RocksTransactionLogStore.ts | 11 ++-- resources/replayLogs.ts | 7 +- resources/replayLogsGuards.ts | 40 ++--------- unitTests/resources/replayLogs.test.js | 66 ------------------- 5 files changed, 21 insertions(+), 115 deletions(-) diff --git a/integrationTests/server/replay-transaction-atomicity.test.ts b/integrationTests/server/replay-transaction-atomicity.test.ts index 03dccfff28..318c0b5a08 100644 --- a/integrationTests/server/replay-transaction-atomicity.test.ts +++ b/integrationTests/server/replay-transaction-atomicity.test.ts @@ -137,13 +137,11 @@ suite('Replay transaction atomicity across a corrupt frame', (ctx: ContextWithHa await startHarper(ctx); - const tornRows = await countInRange(ctx.harper, 1001, TORN_IDS); - ok( - tornRows === 0 || tornRows === TORN_IDS, - `the truncated transaction must be all-or-nothing, found ${tornRows} of ${TORN_IDS} rows` - ); - // The entries before it are intact, so the transaction that completed ahead of the break is - // unaffected — fail-stop must not cost more than the transaction it landed in. + // 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 09dfda10d9..2842daf06a 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/resources/RocksTransactionLogStore.ts @@ -239,14 +239,15 @@ export class RocksTransactionLogStore extends EventEmitter { }): TransactionLogIterable { let iterable = new ExtendedIterable(); let aggregateIterator: TransactionLogIterator; - // Shared by every per-log iterator of this range so a consumer can tell iteration that ended at - // a corrupt frame from iteration that reached the end of the logs, and can attribute the break - // to the entry it was reading when the break surfaced. - const corruptFrameStop: CorruptFrameStop = { breaks: 0 }; + const corruptFrameStop: CorruptFrameStop = { breaks: 0, truncatedVersions: new Set() }; + // The version of the last entry each log yielded in this range, so a break can be attributed to + // the source transaction whose remaining entries it swallowed. + const lastVersionByLog = new Map(); const onCorruptFrame = (log: TransactionLog) => { const report = reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`); return (error) => { corruptFrameStop.breaks++; + if (lastVersionByLog.has(log)) corruptFrameStop.truncatedVersions.add(lastVersionByLog.get(log)); report(error); }; }; @@ -371,6 +372,8 @@ export class RocksTransactionLogStore extends EventEmitter { } } if (earliestIndex >= 0) { + // before the refill, which is where a break surfaces and needs this entry's version + lastVersionByLog.set(logs[earliestIndex], earliest.timestamp); // replace the entry with the next one from the iterator we pulled from nextEntries[earliestIndex] = safeNext(iterators[earliestIndex], logs[earliestIndex]); return { diff --git a/resources/replayLogs.ts b/resources/replayLogs.ts index 8d90fd6dc1..d4824dc640 100644 --- a/resources/replayLogs.ts +++ b/resources/replayLogs.ts @@ -11,7 +11,6 @@ import { isUndecodableValidatedWrite, shouldAbortStalledReplay, shouldAbortSlowReplay, - createTruncatedTransactionTracker, REPLAY_WALL_CLOCK_LIMIT_MS, } from './replayLogsGuards.ts'; import { purgeAgedLogs } from './auditStore.ts'; @@ -98,7 +97,6 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise const replayTimeoutMs = configuredReplayTimeout > 0 ? configuredReplayTimeout : REPLAY_WALL_CLOCK_LIMIT_MS; const txnLog: RocksTransactionLogStore = (rootStore as any).auditStore; const entries = txnLog.getRange({ startFromLastFlushed: true, readUncommitted: true }); - const truncated = createTruncatedTransactionTracker(entries.corruptFrameStop); for (const auditRecord of entries as any) { if (noProgressRun > 0 && shouldAbortStalledReplay(noProgressRun, performance.now() - lastProgressTime)) { logger.fatal( @@ -106,7 +104,6 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise ); break; } - truncated.observe(auditRecord.version); const { type, tableId, @@ -174,7 +171,7 @@ 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 = truncated.wasTruncated(lastTimestamp); + const torn = entries.corruptFrameStop.truncatedVersions.has(lastTimestamp); lastTimestamp = version; try { // commit the last transaction since we are starting a new one, unless a corrupt @@ -300,7 +297,7 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise }); } } - const finalTorn = truncated.wasTruncated(lastTimestamp, true); + const finalTorn = entries.corruptFrameStop.truncatedVersions.has(lastTimestamp); try { if (finalTorn) { writes -= stagedWrites; diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index 010a8d7552..533600649a 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -146,42 +146,16 @@ export type CorruptFrameError = RangeError & { /** * Corrupt frames hit by one `getRange` call, shared by its per-log iterators. * - * A count, not a flag, because a consumer has to attribute each break to the source transaction it - * truncated: the entries an aggregate iterator has yielded from the broken log stop at the break, - * so the transaction those last entries belong to is the one left incomplete. Replay must discard - * that transaction rather than commit the surviving part of it. + * `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. Only an aggregate range attributes + * versions; a single-log range reports the count alone. */ export interface CorruptFrameStop { breaks: number; -} - -/** - * Attributes each corrupt frame to the source transaction it truncated, so a replay can drop that - * transaction whole instead of committing the part of it that was still readable. - * - * An aggregate transaction-log iterator refills a log's lookahead as it hands back that log's - * entry, so a break becomes visible on the very entry whose successors it destroyed — the version - * that entry belongs to is the incomplete transaction. Call {@link observe} for every entry, in - * order, then ask about a version when its transaction is about to be committed. - */ -export function createTruncatedTransactionTracker(corruptFrameStop: CorruptFrameStop) { - let accountedBreaks = 0; - let truncatedVersion: number; - return { - observe(version: number) { - if (corruptFrameStop.breaks > accountedBreaks) { - accountedBreaks = corruptFrameStop.breaks; - truncatedVersion = version; - } - }, - /** - * @param final the transaction staged when iteration ended — a break on the final pull has no - * following entry to be attributed to, so it belongs to this one - */ - wasTruncated(version: number, final = false): boolean { - return truncatedVersion === version || (final && corruptFrameStop.breaks > accountedBreaks); - }, - }; + truncatedVersions: Set; } /** diff --git a/unitTests/resources/replayLogs.test.js b/unitTests/resources/replayLogs.test.js index 20f82c8ec2..57c67ce2d4 100644 --- a/unitTests/resources/replayLogs.test.js +++ b/unitTests/resources/replayLogs.test.js @@ -7,7 +7,6 @@ const { isUndecodableValidatedWrite, RECORD_BEARING_FLAGS, endIteratorOnCorruptFrame, - createTruncatedTransactionTracker, MAX_CORRUPT_FRAME_REPORTS, createCorruptFrameReporter, getCorruptFrameReports, @@ -310,71 +309,6 @@ describe('endIteratorOnCorruptFrame', () => { }); }); -// harper#2016 / harper#2063: replay groups equal-version entries into one source transaction, so a -// break inside such a group leaves the surviving part of a transaction that never committed that -// way at the source. The tracker is what tells replay which staged transaction to drop. -describe('createTruncatedTransactionTracker', () => { - it('reports nothing truncated while no break has been seen', () => { - const stop = { breaks: 0 }; - const tracker = createTruncatedTransactionTracker(stop); - for (const version of [10, 10, 20]) tracker.observe(version); - - assert.strictEqual(tracker.wasTruncated(10), false); - assert.strictEqual(tracker.wasTruncated(20), false); - assert.strictEqual(tracker.wasTruncated(20, true), false); - }); - - // The break surfaces on the entry whose successors it destroyed, because the aggregate iterator - // refills that log's lookahead as it hands the entry back. - it('attributes a break to the version of the entry it surfaced on', () => { - const stop = { breaks: 0 }; - const tracker = createTruncatedTransactionTracker(stop); - tracker.observe(10); - stop.breaks++; - tracker.observe(10); - tracker.observe(20); - - assert.strictEqual(tracker.wasTruncated(10), true); - assert.strictEqual(tracker.wasTruncated(20), false); - }); - - // A break seen as the FIRST entry of the next transaction arrives belongs to that next - // transaction, not to the one just completed — the opposite attribution would commit the torn - // one and discard an intact one. - it('attributes a break on a boundary entry to the transaction that entry starts', () => { - const stop = { breaks: 0 }; - const tracker = createTruncatedTransactionTracker(stop); - tracker.observe(10); - stop.breaks++; - tracker.observe(20); - - assert.strictEqual(tracker.wasTruncated(10), false); - assert.strictEqual(tracker.wasTruncated(20), true); - }); - - it('attributes a break with no entry after it to the transaction still staged at the end', () => { - const stop = { breaks: 0 }; - const tracker = createTruncatedTransactionTracker(stop); - tracker.observe(10); - stop.breaks++; - - assert.strictEqual(tracker.wasTruncated(10), false); - assert.strictEqual(tracker.wasTruncated(10, true), true); - }); - - it('follows the latest break when several logs break during one replay', () => { - const stop = { breaks: 0 }; - const tracker = createTruncatedTransactionTracker(stop); - stop.breaks++; - tracker.observe(10); - stop.breaks++; - tracker.observe(20); - - assert.strictEqual(tracker.wasTruncated(10), false); - assert.strictEqual(tracker.wasTruncated(20), true); - }); -}); - // 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', () => { From 71050b1b25a76b8889a438dfa841f6ae7703d040 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 26 Aug 2026 11:46:30 -0600 Subject: [PATCH 13/14] fix(replay): track the truncated version on a single-log range too A single-log getRange left truncatedVersions empty, so a consumer that discards torn transactions the way replay does would have committed the readable prefix. Co-Authored-By: Claude Opus --- .../server/replay-transaction-atomicity.test.ts | 2 +- resources/RocksTransactionLogStore.ts | 8 +++++++- resources/replayLogsGuards.ts | 3 +-- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/integrationTests/server/replay-transaction-atomicity.test.ts b/integrationTests/server/replay-transaction-atomicity.test.ts index 318c0b5a08..25d47dd75b 100644 --- a/integrationTests/server/replay-transaction-atomicity.test.ts +++ b/integrationTests/server/replay-transaction-atomicity.test.ts @@ -88,7 +88,7 @@ function tearLastTransaction(path: string, readableBefore: number): number { 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 + if (length === 0 || next > buf.length) break; entries.push({ lengthPos, timestamp }); pos = next; } diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index 2842daf06a..dd3249ad03 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/resources/RocksTransactionLogStore.ts @@ -239,6 +239,8 @@ export class RocksTransactionLogStore extends EventEmitter { }): TransactionLogIterable { let iterable = new ExtendedIterable(); let aggregateIterator: TransactionLogIterator; + // Set only for a single-log range, where every entry comes from the same log. + let singleLog: TransactionLog; const corruptFrameStop: CorruptFrameStop = { breaks: 0, truncatedVersions: new Set() }; // The version of the last entry each log yielded in this range, so a break can be attributed to // the source transaction whose remaining entries it swallowed. @@ -264,6 +266,7 @@ export class RocksTransactionLogStore extends EventEmitter { log = this.rootStore.useLog(options.log); } } + singleLog = log; const queryIterator = endIteratorOnCorruptFrame(log.query(options), onCorruptFrame(log)); iterable.iterate = () => queryIterator; } else { @@ -374,7 +377,6 @@ export class RocksTransactionLogStore extends EventEmitter { if (earliestIndex >= 0) { // before the refill, which is where a break surfaces and needs this entry's version lastVersionByLog.set(logs[earliestIndex], earliest.timestamp); - // replace the entry with the next one from the iterator we pulled from nextEntries[earliestIndex] = safeNext(iterators[earliestIndex], logs[earliestIndex]); return { value: onlyKeys ? earliest.timestamp : earliest, @@ -408,6 +410,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 (singleLog) lastVersionByLog.set(singleLog, 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 diff --git a/resources/replayLogsGuards.ts b/resources/replayLogsGuards.ts index 533600649a..5d2496caaa 100644 --- a/resources/replayLogsGuards.ts +++ b/resources/replayLogsGuards.ts @@ -150,8 +150,7 @@ export type CorruptFrameError = RangeError & { * 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. Only an aggregate range attributes - * versions; a single-log range reports the count alone. + * broke before yielding anything truncates no transaction. */ export interface CorruptFrameStop { breaks: number; From 3b81edbf4c59199b77cfb9f03053be751c682dd9 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Wed, 26 Aug 2026 11:51:37 -0600 Subject: [PATCH 14/14] perf(replay): keep the last-yielded version on the iterator, not in a per-log map The attribution write happens for every entry on the replay and broadcast paths; a property write on the iterator that already exists costs nothing there, and it stays with the iterator when a log is spliced out of the aggregate. Co-Authored-By: Claude Opus --- resources/RocksTransactionLogStore.ts | 38 +++++++++++++++------------ 1 file changed, 21 insertions(+), 17 deletions(-) diff --git a/resources/RocksTransactionLogStore.ts b/resources/RocksTransactionLogStore.ts index dd3249ad03..f1d632ab72 100644 --- a/resources/RocksTransactionLogStore.ts +++ b/resources/RocksTransactionLogStore.ts @@ -23,6 +23,8 @@ type TransactionLogIterator = Iterator & { removeLog(logName: string); }; +type TrackedIterator = IterableIterator & { lastVersion?: number }; + export type TransactionLogIterable = Iterable & { /** Corrupt frames that ended a log's iteration during this range. */ corruptFrameStop: CorruptFrameStop; @@ -240,18 +242,20 @@ export class RocksTransactionLogStore extends EventEmitter { let iterable = new ExtendedIterable(); let aggregateIterator: TransactionLogIterator; // Set only for a single-log range, where every entry comes from the same log. - let singleLog: TransactionLog; + let singleLogIterator: TrackedIterator; const corruptFrameStop: CorruptFrameStop = { breaks: 0, truncatedVersions: new Set() }; - // The version of the last entry each log yielded in this range, so a break can be attributed to - // the source transaction whose remaining entries it swallowed. - const lastVersionByLog = new Map(); - const onCorruptFrame = (log: TransactionLog) => { + // 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}`); - return (error) => { + const iterator: TrackedIterator = endIteratorOnCorruptFrame(log.query(queryOptions), (error) => { corruptFrameStop.breaks++; - if (lastVersionByLog.has(log)) corruptFrameStop.truncatedVersions.add(lastVersionByLog.get(log)); + 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); @@ -266,8 +270,8 @@ export class RocksTransactionLogStore extends EventEmitter { log = this.rootStore.useLog(options.log); } } - singleLog = log; - const queryIterator = endIteratorOnCorruptFrame(log.query(options), onCorruptFrame(log)); + const queryIterator = trackCorruptFrames(log, options); + singleLogIterator = queryIterator; iterable.iterate = () => queryIterator; } else { const onlyKeys = options.onlyKeys; @@ -275,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 @@ -318,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), onCorruptFrame(log))); + iterators.push(trackCorruptFrames(log, queryOptions)); } } latestUpdates = this.updates; @@ -376,7 +380,7 @@ export class RocksTransactionLogStore extends EventEmitter { } if (earliestIndex >= 0) { // before the refill, which is where a break surfaces and needs this entry's version - lastVersionByLog.set(logs[earliestIndex], earliest.timestamp); + iterators[earliestIndex].lastVersion = earliest.timestamp; nextEntries[earliestIndex] = safeNext(iterators[earliestIndex], logs[earliestIndex]); return { value: onlyKeys ? earliest.timestamp : earliest, @@ -410,10 +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 (singleLog) lastVersionByLog.set(singleLog, timestamp); + // 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