fix(replay): fail-stop at a corrupt transaction-log frame and discard the transaction it truncated

integrationTests/server/replay-transaction-atomicity.test.ts

@@ -0,0 +1,147 @@
+/**
+ * A corrupt transaction-log frame in the middle of a source transaction must not leave that
+ * transaction half-applied.
+ *
+ * Replay groups every equal-version entry into one transaction and commits it at the next version
+ * boundary, so a break inside such a group used to commit whatever part of it was still readable —
+ * a transaction that never committed that way at the source becoming durable here. This drives a
+ * real multi-record insert, tears the log inside it, and requires the replayed table to hold all of
+ * that insert or none of it. See HarperFast/harper#2016 and #2063.
+ */
+import { suite, test, before, after } from 'node:test';
+import { ok, strictEqual as equal } from 'node:assert';
+import { readdirSync, readFileSync, openSync, writeSync, closeSync } from 'node:fs';
+import { join } from 'node:path';
+
+import {
+	startHarper,
+	teardownHarper,
+	sendOperation,
+	type ContextWithHarper,
+	type HarperContext,
+} from '@harperfast/integration-testing';
+import { constants } from '@harperfast/rocksdb-js';
+
+// Transaction-log framing (big-endian): a fixed-size file header, then entries shaped
+// [float64 timestamp][uint32 length][flags byte][length bytes of data].
+const { TRANSACTION_LOG_FILE_HEADER_SIZE, TRANSACTION_LOG_ENTRY_HEADER_SIZE } = constants;
+
+const DB = 'atomicity';
+const TABLE = 'orders';
+const EARLIER_IDS = 60;
+const TORN_IDS = 60;
+// Entries of the torn transaction left readable before the break, so the test proves the readable
+// part is discarded rather than proving the whole transaction was unreachable anyway.
+const READABLE_BEFORE_BREAK = 10;
+
+async function op(ctx: HarperContext, body: any) {
+	return await sendOperation(ctx, { ...body, authorization: ctx.admin });
+}
+
+function records(start: number, count: number) {
+	const out = [];
+	for (let i = 0; i < count; i++) out.push({ id: start + i, payload: 'x'.repeat(256), n: i });
+	return out;
+}
+
+async function countInRange(ctx: HarperContext, start: number, count: number): Promise<number> {
+	const rows = await op(ctx, {
+		operation: 'sql',
+		sql: `select count(*) as c from ${DB}.${TABLE} where id >= ${start} and id < ${start + count}`,
+	});
+	return rows[0]?.c ?? 0;
+}
+
+function userTxnLogFiles(dataRootDir: string): string[] {
+	const out: string[] = [];
+	const dbRoot = join(dataRootDir, 'database');
+	for (const db of readdirSync(dbRoot)) {
+		if (db === 'system') continue;
+		const tlogRoot = join(dbRoot, db, 'transaction_logs');
+		let nodes: string[];
+		try {
+			nodes = readdirSync(tlogRoot);
+		} catch {
+			continue;
+		}
+		for (const node of nodes) {
+			for (const file of readdirSync(join(tlogRoot, node))) {
+				if (file.endsWith('.txnlog')) out.push(join(tlogRoot, node, file));
+			}
+		}
+	}
+	return out;
+}
+
+/**
+ * Break the framing partway through the log's LAST transaction — the run of trailing entries that
+ * share the highest timestamp — leaving `readableBefore` of its entries intact ahead of the break.
+ * Returns how many entries that transaction has, or 0 if the log has no such run to tear.
+ */
+function tearLastTransaction(path: string, readableBefore: number): number {
+	const buf = readFileSync(path);
+	const entries: { lengthPos: number; timestamp: number }[] = [];
+	let pos = TRANSACTION_LOG_FILE_HEADER_SIZE;
+	while (pos + TRANSACTION_LOG_ENTRY_HEADER_SIZE <= buf.length) {
+		const timestamp = buf.readDoubleBE(pos);
+		if (timestamp === 0) break; // a zero timestamp marks end-of-log to the reader
+		const lengthPos = pos + 8;
+		const length = buf.readUInt32BE(lengthPos);
+		const next = pos + TRANSACTION_LOG_ENTRY_HEADER_SIZE + length;
+		if (length === 0 || next > buf.length) break;
+		entries.push({ lengthPos, timestamp });
+		pos = next;
+	}
+	if (entries.length === 0) return 0;
+	const lastTimestamp = entries.at(-1).timestamp;
+	let first = entries.length - 1;
+	while (first > 0 && entries[first - 1].timestamp === lastTimestamp) first--;
+	const transactionEntries = entries.length - first;
+	if (transactionEntries <= readableBefore) return 0;
+	// Force this entry's declared length to overrun the log (top byte → 0xff, ≥ 4 GB): the reader
+	// throws a bounded RangeError there and cannot locate any entry after it.
+	const fd = openSync(path, 'r+');
+	try {
+		writeSync(fd, Buffer.from([0xff]), 0, 1, entries[first + readableBefore].lengthPos);
+	} finally {
+		closeSync(fd);
+	}
+	return transactionEntries;
+}
+
+suite('Replay transaction atomicity across a corrupt frame', (ctx: ContextWithHarper) => {
+	before(async () => {
+		// Don't flush on exit: the crash must leave these writes recoverable only from the txn log.
+		await startHarper(ctx, { env: { HARPER_NO_FLUSH_ON_EXIT: true } });
+		await op(ctx.harper, { operation: 'create_database', database: DB });
+		await op(ctx.harper, { operation: 'create_table', database: DB, table: TABLE, primary_key: 'id' });
+	});
+	after(async () => teardownHarper(ctx));
+
+	test('discards a transaction the corrupt frame truncated instead of applying part of it', async () => {
+		await op(ctx.harper, { operation: 'insert', database: DB, table: TABLE, records: records(1, EARLIER_IDS) });
+		// The last insert is one source transaction, and is the one the tear lands inside.
+		await op(ctx.harper, { operation: 'insert', database: DB, table: TABLE, records: records(1001, TORN_IDS) });
+
+		const dataRootDir = ctx.harper.dataRootDir;
+		await new Promise<void>((resolve) => {
+			ctx.harper.process.once('exit', () => resolve());
+			ctx.harper.process.kill('SIGKILL');
+		});
+		let torn = 0;
+		for (const file of userTxnLogFiles(dataRootDir)) {
+			torn = Math.max(torn, tearLastTransaction(file, READABLE_BEFORE_BREAK));
+		}
+		// Fail loudly, not vacuously, if the framing or the transaction grouping ever changes.
+		ok(torn > READABLE_BEFORE_BREAK, `expected to tear a multi-entry transaction, tore ${torn} entries`);
+
+		await startHarper(ctx);
+
+		// None of it: the break makes the rest of that insert unreadable, so the readable prefix is
+		// discarded rather than committed as a transaction the source never committed.
+		equal(await countInRange(ctx.harper, 1001, TORN_IDS), 0, 'the truncated transaction must not be applied in part');
+		// The transactions that completed ahead of the break are unaffected: fail-stop costs the
+		// transaction the break landed in, not the log up to it.
+		equal(await countInRange(ctx.harper, 1, EARLIER_IDS), EARLIER_IDS);
+	});
+});

resources/RocksTransactionLogStore.ts

@@ -3,7 +3,7 @@ import { ExtendedIterable } from '@harperfast/extended-iterable';
 import { getIdOfRemoteNode } from './nodeIdMapping.ts';
 import { Decoder, readAuditEntry, ENTRY_DATAVIEW, AuditRecord, createAuditEntry } from './auditStore.ts';
 import { HAS_STRUCTURE_UPDATE } from './RecordEncoder.ts';
-import { endIteratorOnCorruptFrame } from './replayLogsGuards.ts';
+import { createCorruptFrameReporter, endIteratorOnCorruptFrame, type CorruptFrameStop } from './replayLogsGuards.ts';
 import { isMainThread } from 'node:worker_threads';
 import { EventEmitter } from 'node:events';
 import { asBinary } from 'lmdb';
@@ -23,12 +23,14 @@ type TransactionLogIterator = Iterator<TransactionEntry | number> & {
 	removeLog(logName: string);
 };
 
-// Logs (once per log) when a corrupt frame ends a query iterator early; see
-// endIteratorOnCorruptFrame in replayLogsGuards.ts for why this is end-of-log, not a crash.
-function warnCorruptFrame(logName: string) {
-	return (error: RangeError) =>
-		harperLogger.warn(`Stopping transaction log "${logName}" at a corrupt entry during replay`, error);
-}
+type TrackedIterator = IterableIterator<TransactionEntry> & { lastVersion?: number };
+
+export type TransactionLogIterable = Iterable<AuditRecord> & {
+	/** Corrupt frames that ended a log's iteration during this range. */
+	corruptFrameStop: CorruptFrameStop;
+};
+
+const reportCorruptFrame = createCorruptFrameReporter(harperLogger);
 
 /**
  * Represents a transaction log store backed by RocksDB.
@@ -42,6 +44,7 @@ export class RocksTransactionLogStore extends EventEmitter {
 	logByName: Map<string, TransactionLog> = new Map();
 	updates = 0; // the number of updates to the list of logs that have occurred
 	rootStore: RocksDatabase;
+	corruptFrameScope: string;
 	reusableIterable = true; // flag indicating that iterable can be reused to resume iterating through audit log
 	// Highest structureVersion appended to each per-node TransactionLog, tracked per tableId. Drives the
 	// per-log HAS_STRUCTURE_UPDATE flag in put(). Keyed by (log, tableId): a per-node log interleaves entries
@@ -54,6 +57,10 @@ export class RocksTransactionLogStore extends EventEmitter {
 		super();
 		this.log = rootDatabase.useLog('local');
 		this.rootStore = rootDatabase;
+		// Break reports are keyed per log name, but every store has its own 'local' log, so they
+		// need a scope. The path, not `databaseName`: one root store can back several logical
+		// databases and they share these logs, and `databaseName` is not set on it yet here.
+		this.corruptFrameScope = rootDatabase.path;
 	}
 
 	/**
@@ -231,9 +238,25 @@ export class RocksTransactionLogStore extends EventEmitter {
 		startByLog?: Map<string, number>;
 		startFromLastFlushed?: boolean;
 		readUncommitted?: boolean;
-	}): Iterable<AuditRecord> {
+	}): TransactionLogIterable {
 		let iterable = new ExtendedIterable<TransactionEntry>();
 		let aggregateIterator: TransactionLogIterator;
+		// Set only for a single-log range, where every entry comes from the same log.
+		let singleLogIterator: TrackedIterator;
+		const corruptFrameStop: CorruptFrameStop = { breaks: 0, truncatedVersions: new Set() };
+		// Each log's iterator carries the version of the last entry it yielded, so a break can be
+		// attributed to the source transaction whose remaining entries it swallowed. On the iterator
+		// itself, not in a map keyed by log: this is written per entry on the replay/broadcast path,
+		// and it stays attached when a removed log is spliced out of the aggregate.
+		const trackCorruptFrames = (log: TransactionLog, queryOptions: typeof options): TrackedIterator => {
+			const report = reportCorruptFrame(`${this.corruptFrameScope}/${log.name}`);
+			const iterator: TrackedIterator = endIteratorOnCorruptFrame(log.query(queryOptions), (error) => {
+				corruptFrameStop.breaks++;
+				if (iterator.lastVersion !== undefined) corruptFrameStop.truncatedVersions.add(iterator.lastVersion);
+				report(error);
+			});
+			return iterator;
+		};
 		if (options.log !== undefined) {
 			let log = typeof options.log === 'number' ? this.nodeLogs?.[options.log] : this.logByName.get(options.log);
 			if (!log) {
@@ -247,15 +270,16 @@ export class RocksTransactionLogStore extends EventEmitter {
 					log = this.rootStore.useLog(options.log);
 				}
 			}
-			const queryIterator = endIteratorOnCorruptFrame(log.query(options), warnCorruptFrame(log.name));
+			const queryIterator = trackCorruptFrames(log, options);
+			singleLogIterator = queryIterator;
 			iterable.iterate = () => queryIterator;
 		} else {
 			const onlyKeys = options.onlyKeys;
 			let logs: TransactionLog[] = [];
 			// holds the queue of next entries from each iterator
 			let nextEntries: any[];
 			let latestUpdates: number;
-			const iterators: IterableIterator<TransactionEntry>[] = [];
+			const iterators: TrackedIterator[] = [];
 			// Iterators that have permanently failed (corrupt entry stuck at the same
 			// position). Tracked by identity so the retry-poll path in next() and
 			// updateIterators() never calls .next() on them again — otherwise every
@@ -298,7 +322,7 @@ export class RocksTransactionLogStore extends EventEmitter {
 								// condition of potentially missing an initial update
 								queryOptions = { ...options, start: options.start ?? 0 };
 							}
-							iterators.push(endIteratorOnCorruptFrame(log.query(queryOptions), warnCorruptFrame(log.name)));
+							iterators.push(trackCorruptFrames(log, queryOptions));
 						}
 					}
 					latestUpdates = this.updates;
@@ -355,7 +379,8 @@ export class RocksTransactionLogStore extends EventEmitter {
 							}
 						}
 						if (earliestIndex >= 0) {
-							// replace the entry with the next one from the iterator we pulled from
+							// before the refill, which is where a break surfaces and needs this entry's version
+							iterators[earliestIndex].lastVersion = earliest.timestamp;
 							nextEntries[earliestIndex] = safeNext(iterators[earliestIndex], logs[earliestIndex]);
 							return {
 								value: onlyKeys ? earliest.timestamp : earliest,
@@ -389,6 +414,10 @@ export class RocksTransactionLogStore extends EventEmitter {
 			iterable.iterate = () => aggregateIterator;
 		}
 		const mappedAggregateIterable = iterable.map(({ timestamp, data, endTxn }: TransactionEntry) => {
+			// A break surfaces on the pull after this entry, so recording it here is in time to attribute
+			// that break to this entry's transaction. The aggregate branch records its own, per source
+			// log, because there this callback cannot tell which log an entry came from.
+			if (singleLogIterator) singleLogIterator.lastVersion = timestamp;
 			// Per-entry try/catch: a corrupt rocks prelude (first 4-16 bytes) would otherwise
 			// throw a raw `RangeError: Offset is outside the bounds of the DataView` out
 			// through `iterable.map`, escape the for-of consumer, and land as an
@@ -442,7 +471,8 @@ export class RocksTransactionLogStore extends EventEmitter {
 			mappedAggregateIterable.addLog = aggregateIterator.addLog;
 			mappedAggregateIterable.removeLog = aggregateIterator.removeLog;
 		}
-		return mappedAggregateIterable;
+		mappedAggregateIterable.corruptFrameStop = corruptFrameStop;
+		return mappedAggregateIterable as TransactionLogIterable;
 	}
 	getKeys(_options?: any) {
 		return []; // TODO: implement this

resources/replayLogs.ts

@@ -78,6 +78,10 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise<void>
 		let transaction: DatabaseTransaction;
 		let lastTimestamp = 0;
 		let writes = 0;
+		// Writes staged on the currently-open transaction, so a discard can be taken back out of `writes`.
+		let stagedWrites = 0;
+		// Records dropped because a corrupt frame truncated the transaction they belong to.
+		let discardedWrites = 0;
 		let skipped = 0;
 		// Track forward progress so a backlog of unwritable entries can't grind the boot thread
 		// forever (harper#1266). `noProgressRun` counts every entry processed without a successful
@@ -92,7 +96,8 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise<void>
 		const configuredReplayTimeout = Number(envGet(CONFIG_PARAMS.REPLICATION_REPLAYTIMEOUT));
 		const replayTimeoutMs = configuredReplayTimeout > 0 ? configuredReplayTimeout : REPLAY_WALL_CLOCK_LIMIT_MS;
 		const txnLog: RocksTransactionLogStore = (rootStore as any).auditStore;
-		for (const auditRecord of txnLog.getRange({ startFromLastFlushed: true, readUncommitted: true }) as any) {
+		const entries = txnLog.getRange({ startFromLastFlushed: true, readUncommitted: true });
+		for (const auditRecord of entries as any) {
 			if (noProgressRun > 0 && shouldAbortStalledReplay(noProgressRun, performance.now() - lastProgressTime)) {
 				logger.fatal(
 					`Aborting transaction-log replay in ${(rootStore as any).databaseName} database: ${noProgressRun} consecutive audit entries with no successful write (${skipped} skipped as unrecoverable, ${writes} replayed so far). This backlog is making no forward progress and was blocking startup (harper#1266) — typically a peer transaction log whose values reference unresolvable shared structures (harper#1163), or a backlog for a dropped table. Continuing boot without replaying the remainder; shed or relocate the oversized/undecodable peer transaction log(s), or re-clone this node, to recover the unreplayed data.`
@@ -166,13 +171,21 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise<void>
 					console.warn('Harper was not properly shutdown, replaying transaction logs to synchronize database');
 				}
 				if (lastTimestamp !== version) {
+					const torn = entries.corruptFrameStop.truncatedVersions.has(lastTimestamp);
 					lastTimestamp = version;
 					try {
-						// commit the last transaction since we are starting a new one
-						transaction?.directCommitSync();
+						// commit the last transaction since we are starting a new one, unless a corrupt
+						// frame swallowed the rest of it — half of a source transaction must never become
+						// durable, so it is dropped whole and stays in the log for a repaired retry
+						if (torn) {
+							writes -= stagedWrites;
+							discardedWrites += stagedWrites;
+							transaction?.abort();
+						} else transaction?.directCommitSync();
 					} catch (error) {
-						logger.error('Error committing replay transaction', error);
+						logger.error(`Error ${torn ? 'discarding a torn' : 'committing'} replay transaction`, error);
 					}
+					stagedWrites = 0;
 					// Abort if replay has exceeded the total wall-clock budget even while making progress
 					// (harper#1316, facet a). shouldAbortStalledReplay resets its counters on every write,
 					// so a slow-but-progressing replay (deep out-of-order audit chain walk per entry) can
@@ -200,6 +213,7 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise<void>
 				context.transaction = transaction;
 				const options = { context, residencyId, nodeId, originatingOperation };
 				writes++;
+				stagedWrites++;
 				switch (type) {
 					case 'put':
 						tableInstance._writeUpdate(recordId, record, true, options);
@@ -283,10 +297,20 @@ export function replayLogs(rootStore: RocksDatabase, tables: any): Promise<void>
 				});
 			}
 		}
+		const finalTorn = entries.corruptFrameStop.truncatedVersions.has(lastTimestamp);
 		try {
-			transaction?.directCommitSync();
+			if (finalTorn) {
+				writes -= stagedWrites;
+				discardedWrites += stagedWrites;
+				transaction?.abort();
+			} else transaction?.directCommitSync();
 		} catch (error) {
-			logger.error('Error committing replay transaction', error);
+			logger.error(`Error ${finalTorn ? 'discarding a torn' : 'committing'} replay transaction`, error);
+		}
+		if (entries.corruptFrameStop.breaks > 0) {
+			logger.error(
+				`Transaction-log replay in ${(rootStore as any).databaseName} database stopped at a corrupt entry after replaying ${writes} records. Every entry after the break is quarantined — neither replayed nor replicated — and ${discardedWrites} record(s) of the transaction the break truncated were discarded rather than applied in part. Repair the transaction log or re-clone this node to recover them.`
+			);
 		}
 		if (writes > 0) logger.warn(`Replayed ${writes} records in ${(rootStore as any).databaseName} database`);
 		if (skipped > 0)