diff --git a/src/history/store.ts b/src/history/store.ts index a4d0f2f..6f8a1d8 100644 --- a/src/history/store.ts +++ b/src/history/store.ts @@ -12,6 +12,25 @@ const HISTORY_LOCK_STALE_MS = 60_000; const HISTORY_LOCK_TIMEOUT_MS = HISTORY_LOCK_STALE_MS + 10_000; const HISTORY_LOCK_HEARTBEAT_MS = 15_000; const HISTORY_LOCK_TAKEOVER_SUFFIX = '.takeover'; +const HISTORY_READ_CHUNK_SIZE = 64 * 1024; + +/** Read a complete history chunk, retrying when the filesystem returns a short read. */ +export async function readHistoryChunk( + history: Pick, + buffer: Buffer, + position: number, + bytesToRead: number, +): Promise { + let bytesRead = 0; + + while (bytesRead < bytesToRead) { + const result = await history.read(buffer, bytesRead, bytesToRead - bytesRead, position + bytesRead); + if (result.bytesRead === 0) break; + bytesRead += result.bytesRead; + } + + return bytesRead; +} function hasErrorCode(error: unknown, code: string): boolean { return typeof error === 'object' && error !== null && 'code' in error && error.code === code; @@ -210,25 +229,75 @@ export async function loadEntries(limit = 200): Promise { const historyPath = getHistoryPath(); if (!existsSync(historyPath)) return []; - const raw = await readFile(historyPath, 'utf-8'); - const lines = raw - .split('\n') - .map((line, index) => ({ line, lineNumber: index + 1 })) - .filter(({ line }) => line.trim().length > 0) - .reverse(); - const entries: CommitEntry[] = []; - const corruptedLineNumbers: number[] = []; + const corruptedLineNumbers: Array = []; + const history = await open(historyPath, 'r'); - for (const { line, lineNumber } of lines) { - try { - const entry = JSON.parse(line) as CommitEntry; - if (entries.length < limit) { - entries.push(entry); + try { + const { size } = await history.stat(); + let position = size; + let remainderChunks: Buffer[] = []; + let processedLineCount = 0; + let totalLineCount: number | undefined; + const corruptedLineIndexesFromEnd: number[] = []; + + const parseLine = (lineBytes: Buffer) => { + const lineIndexFromEnd = processedLineCount++; + const line = lineBytes.toString('utf-8'); + if (line.trim().length === 0) return; + + try { + entries.push(JSON.parse(line) as CommitEntry); + } catch { + corruptedLineIndexesFromEnd.push(lineIndexFromEnd); } - } catch { - corruptedLineNumbers.push(lineNumber); + }; + + while (position > 0 && entries.length < limit) { + const chunkEnd = position; + position = Math.max(0, chunkEnd - HISTORY_READ_CHUNK_SIZE - 3); + const bytesToRead = chunkEnd - position; + const buffer = Buffer.allocUnsafe(bytesToRead); + const bytesRead = await readHistoryChunk(history, buffer, position, bytesToRead); + const chunk = buffer.subarray(0, bytesRead); + const firstLineEnd = chunk.indexOf(0x0a); + if (firstLineEnd === -1) { + remainderChunks.unshift(chunk); + continue; + } + + const combined = remainderChunks.length === 0 ? chunk : Buffer.concat([chunk, ...remainderChunks]); + remainderChunks = [combined.subarray(0, firstLineEnd)]; + const lines: Buffer[] = []; + let lineStart = firstLineEnd + 1; + for (let index = lineStart; index < combined.length; index++) { + if (combined[index] === 0x0a) { + lines.push(combined.subarray(lineStart, index)); + lineStart = index + 1; + } + } + lines.push(combined.subarray(lineStart)); + const processedBeforeChunk = processedLineCount; + + if (position === 0) { + totalLineCount = processedBeforeChunk + lines.length + 1; + } + + for (let index = lines.length - 1; index >= 0 && entries.length < limit; index--) { + parseLine(lines[index]!); + } + } + + if (position === 0 && entries.length < limit) { + const remainder = remainderChunks.length === 1 ? remainderChunks[0]! : Buffer.concat(remainderChunks); + parseLine(remainder); } + + for (const lineIndexFromEnd of corruptedLineIndexesFromEnd) { + corruptedLineNumbers.push(totalLineCount === undefined ? undefined : totalLineCount - lineIndexFromEnd); + } + } finally { + await history.close(); } warnCorruptedHistory(historyPath, corruptedLineNumbers); @@ -237,20 +306,20 @@ export async function loadEntries(limit = 200): Promise { } /** Emit a compact warning that identifies corrupted history line numbers. */ -function warnCorruptedHistory(historyPath: string, corruptedLineNumbers: number[]): void { +function warnCorruptedHistory(historyPath: string, corruptedLineNumbers: Array): void { if (corruptedLineNumbers.length === 0) return; const count = corruptedLineNumbers.length; const noun = count === 1 ? 'entry' : 'entries'; - const lines = corruptedLineNumbers + const knownLineNumbers = corruptedLineNumbers.filter((lineNumber): lineNumber is number => lineNumber !== undefined); + const lines = knownLineNumbers .slice(0, 5) .sort((a, b) => a - b) .join(', '); const suffix = count > 5 ? `, +${count - 5} more` : ''; + const location = knownLineNumbers.length === count ? `line ${lines}${suffix}` : 'recently scanned rows'; - console.warn( - `Warning: ignored ${count} corrupted commit history ${noun} in ${historyPath} (line ${lines}${suffix}).`, - ); + console.warn(`Warning: ignored ${count} corrupted commit history ${noun} in ${historyPath} (${location}).`); } /** Count raw history rows in the configured history file. */ diff --git a/tests/history-corruption.test.mjs b/tests/history-corruption.test.mjs index 08ab8c7..6d2f234 100644 --- a/tests/history-corruption.test.mjs +++ b/tests/history-corruption.test.mjs @@ -5,7 +5,7 @@ import { join } from 'node:path'; import { tmpdir } from 'node:os'; import { getConfigDir } from '../dist/config/store.js'; -import { countEntries, loadEntries } from '../dist/history/store.js'; +import { countEntries, loadEntries, readHistoryChunk } from '../dist/history/store.js'; function writeHistory(lines) { const configDir = getConfigDir(); @@ -31,6 +31,35 @@ function validEntry(message, timestamp) { }); } +test('readHistoryChunk retries short reads from the unread offset', async () => { + const source = Buffer.from('history entry with a short read'); + const destination = Buffer.alloc(source.length); + const reads = []; + const history = { + async read(buffer, offset, length, position) { + reads.push({ offset, length, position }); + const bytesToCopy = Math.min(5, length); + source.copy(buffer, offset, position, position + bytesToCopy); + return { bytesRead: bytesToCopy, buffer }; + }, + }; + + assert.equal(await readHistoryChunk(history, destination, 0, source.length), source.length); + assert.deepEqual(destination, source); + assert.deepEqual( + reads.map(({ offset, position }) => ({ offset, position })), + [ + { offset: 0, position: 0 }, + { offset: 5, position: 5 }, + { offset: 10, position: 10 }, + { offset: 15, position: 15 }, + { offset: 20, position: 20 }, + { offset: 25, position: 25 }, + { offset: 30, position: 30 }, + ], + ); +}); + async function withIsolatedHistory(lines, assertion) { const originalHome = process.env.HOME; const originalAppData = process.env.APPDATA; @@ -137,6 +166,40 @@ test('loadEntries warns and returns empty entries when all lines are corrupted', }); }); +test('loadEntries stops parsing once it has enough recent valid entries', async () => { + await withIsolatedHistory( + [ + '{invalid old line', + validEntry('fix: keep older valid entry', '2026-06-01T00:00:00Z'), + validEntry('feat: keep newest valid entry', '2026-06-01T00:00:01Z'), + ], + async ({ warnings }) => { + const entries = await loadEntries(1); + + assert.deepEqual(entries.map((entry) => entry.message), ['feat: keep newest valid entry']); + assert.deepEqual(warnings, []); + }, + ); +}); + +test('loadEntries uses a generic location for corrupted rows in a partial scan', async () => { + const olderEntries = Array.from({ length: 700 }, (_, index) => + validEntry(`chore: keep older entry ${index}`, `2026-06-01T00:00:${String(index).padStart(2, '0')}Z`), + ); + + await withIsolatedHistory( + [...olderEntries, validEntry('fix: keep newest valid entry', '2026-06-01T00:01:00Z'), '{invalid newest line'], + async ({ warnings }) => { + const entries = await loadEntries(1); + + assert.deepEqual(entries.map((entry) => entry.message), ['fix: keep newest valid entry']); + assert.equal(warnings.length, 1); + assert.match(warnings[0], /ignored 1 corrupted commit history entry/); + assert.match(warnings[0], /recently scanned rows/); + }, + ); +}); + test('countEntries counts raw non-empty history rows, including malformed JSON lines', async () => { await withIsolatedHistory( [ @@ -151,3 +214,31 @@ test('countEntries counts raw non-empty history rows, including malformed JSON l }, ); }); + +test('loadEntries preserves UTF-8 characters split across backward-read chunks', async () => { + const prefix = '{"timestamp":"2026-06-01T00:00:00Z","message":"a'; + const suffix = '","diff":"","model":"test-model","provider":"local"}'; + const messageTailLength = 2 * 1024 * 1024 - Buffer.byteLength(suffix) - 1; + const row = `${prefix}\u00e9x\u00e9${'b'.repeat(messageTailLength)}${suffix}`; + + const originalConcat = Buffer.concat; + let concatCalls = 0; + Buffer.concat = (...args) => { + concatCalls += 1; + return originalConcat(...args); + }; + + await withIsolatedHistory([row], async () => { + try { + const entries = await loadEntries(1); + + assert.equal(entries.length, 1); + assert.equal(entries[0].message, `a\u00e9x\u00e9${'b'.repeat(messageTailLength)}`); + assert.doesNotMatch(entries[0].message, /\uFFFD/); + } finally { + Buffer.concat = originalConcat; + } + }); + + assert.equal(concatCalls, 1); +});