diff --git a/package.json b/package.json index dfff4b6..a5f16bd 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,7 @@ }, "scripts": { "build": "tsc", - "test": "npm run build && node --test tests/**/*.test.mjs", + "test": "npm run build && node --test \"tests/**/*.test.mjs\"", "prepublishOnly": "npm run build", "start": "node dist/index.js", "format": "prettier --write \"src/**/*.ts\"", diff --git a/src/history/store.ts b/src/history/store.ts index a4d0f2f..b937b1c 100644 --- a/src/history/store.ts +++ b/src/history/store.ts @@ -12,6 +12,59 @@ 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; + +/** Split a buffer into complete newline-delimited lines and an incomplete remainder. */ +function splitCompleteLines(combined: Buffer): { lines: Buffer[]; remainder: Buffer } { + const firstLineEnd = combined.indexOf(0x0a); + if (firstLineEnd === -1) return { lines: [], remainder: combined }; + + 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)); + return { lines, remainder: combined.subarray(0, firstLineEnd) }; +} + +/** Parse a single line buffer, pushing valid entries or recording corrupted line numbers. */ +function processHistoryLine( + lineBytes: Buffer, + lineIndexFromEnd: number, + entries: CommitEntry[], + corruptedLineIndexesFromEnd: number[], +): void { + const line = lineBytes.toString('utf-8'); + if (line.trim().length === 0) return; + + try { + entries.push(JSON.parse(line) as CommitEntry); + } catch { + corruptedLineIndexesFromEnd.push(lineIndexFromEnd); + } +} + +/** 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; @@ -205,52 +258,108 @@ export async function appendEntry(entry: CommitEntry): Promise { } } +/** State accumulated while scanning a history file backward. */ +interface HistoryScanState { + entries: CommitEntry[]; + processedLineCount: number; + totalLineCount: number | undefined; + remainderChunks: Buffer[]; + corruptedLineIndexesFromEnd: number[]; +} + +/** Process one backward-read chunk, returning updated scan state. */ +function processChunk( + chunk: Buffer, + state: HistoryScanState, + position: number, + limit: number, +): void { + if (!chunk.includes(0x0a)) { + state.remainderChunks.unshift(chunk); + if (position === 0) { + state.totalLineCount = state.processedLineCount + 1; + } + return; + } + + const combined = state.remainderChunks.length === 0 ? chunk : Buffer.concat([chunk, ...state.remainderChunks]); + const { lines, remainder: newRemainder } = splitCompleteLines(combined); + state.remainderChunks = [newRemainder]; + + if (position === 0) { + state.totalLineCount = state.processedLineCount + lines.length + 1; + } + + for (let index = lines.length - 1; index >= 0 && state.entries.length < limit; index--) { + processHistoryLine(lines[index]!, state.processedLineCount++, state.entries, state.corruptedLineIndexesFromEnd); + } +} + +/** Process the remaining partial-line bytes after the main scan loop. */ +function processRemainder(state: HistoryScanState, limit: number): void { + const remainder = state.remainderChunks.length === 1 ? state.remainderChunks[0]! : Buffer.concat(state.remainderChunks); + processHistoryLine(remainder, state.processedLineCount++, state.entries, state.corruptedLineIndexesFromEnd); +} + /** Load recent valid history entries while warning once about corrupted JSONL rows. */ 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 corruptedLineNumbers: Array = []; + const state: HistoryScanState = { + entries: [], + processedLineCount: 0, + totalLineCount: undefined, + remainderChunks: [], + corruptedLineIndexesFromEnd: [], + }; - const entries: CommitEntry[] = []; - const corruptedLineNumbers: number[] = []; + const history = await open(historyPath, 'r'); + try { + const { size } = await history.stat(); + let position = size; + + while (position > 0 && state.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); + processChunk(buffer.subarray(0, bytesRead), state, position, limit); + } - for (const { line, lineNumber } of lines) { - try { - const entry = JSON.parse(line) as CommitEntry; - if (entries.length < limit) { - entries.push(entry); - } - } catch { - corruptedLineNumbers.push(lineNumber); + if (position === 0 && state.entries.length < limit) { + processRemainder(state, limit); + } + + for (const lineIndexFromEnd of state.corruptedLineIndexesFromEnd) { + corruptedLineNumbers.push(state.totalLineCount === undefined ? undefined : state.totalLineCount - lineIndexFromEnd); } + } finally { + await history.close(); } warnCorruptedHistory(historyPath, corruptedLineNumbers); - return entries; + return state.entries; } /** 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..a17ff92 100644 --- a/tests/history-corruption.test.mjs +++ b/tests/history-corruption.test.mjs @@ -3,9 +3,8 @@ import test from 'node:test'; import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; 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 +30,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 +165,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 +213,64 @@ 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); +}); + +test('loadEntries reports line 1 for a single corrupted row without trailing newline', async () => { + const originalHome = process.env.HOME; + const originalAppData = process.env.APPDATA; + const originalXdgConfigHome = process.env.XDG_CONFIG_HOME; + const tempHome = mkdtempSync(join(tmpdir(), 'commit-echo-history-')); + const warnings = []; + const originalWarn = console.warn; + + try { + process.env.HOME = tempHome; + process.env.APPDATA = join(tempHome, 'AppData', 'Roaming'); + process.env.XDG_CONFIG_HOME = join(tempHome, '.config'); + console.warn = (message) => warnings.push(String(message)); + + const configDir = getConfigDir(); + mkdirSync(configDir, { recursive: true }); + writeFileSync(join(configDir, 'history.jsonl'), '{not valid json', 'utf-8'); + + const entries = await loadEntries(10); + + assert.deepEqual(entries, []); + assert.equal(warnings.length, 1); + assert.match(warnings[0], /ignored 1 corrupted commit history entry/); + assert.match(warnings[0], /line 1/); + } finally { + console.warn = originalWarn; + restoreEnv('HOME', originalHome); + restoreEnv('APPDATA', originalAppData); + restoreEnv('XDG_CONFIG_HOME', originalXdgConfigHome); + rmSync(tempHome, { recursive: true, force: true }); + } +});