From 02ec799fe13c8ef836ef9ca0b1fbe44339fd00c8 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sat, 8 Aug 2026 21:48:17 +0900 Subject: [PATCH] fix(history): stream request-history index ingestion (#1189) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the full-tail allocation in the request-history indexer with a 64 KiB streaming reader. The previous `readCompleteTail` allocated `size - indexedOffset` bytes in one shot before parsing, so a large append created a proportional transient allocation even though the SQLite index is a disposable projection. Records are now assembled across chunk boundaries, and a complete record above 1 MiB is omitted from the projection only. `usage.jsonl` stays canonical and is never truncated or rewritten; `indexedRows` still counts successfully projected records, and the indexed offset only advances past a newline so a torn final record is re-read rather than skipped. `insert.finalize()` remains unconditional in `finally` — an unterminated prepared statement keeps the DB file busy on Windows after close. Republished from #1189 by luvs01, whose branch was 300 commits behind dev. Rebased onto f5147cbc8 with no conflicts; authorship preserved below. Co-authored-by: luvs01 <27862058+luvs01@users.noreply.github.com> --- src/routing/history/indexer.ts | 93 +++++++++++++++++------------ tests/request-history-index.test.ts | 52 ++++++++++++++-- 2 files changed, 102 insertions(+), 43 deletions(-) diff --git a/src/routing/history/indexer.ts b/src/routing/history/indexer.ts index cad9ea5c5..186860fa9 100644 --- a/src/routing/history/indexer.ts +++ b/src/routing/history/indexer.ts @@ -76,6 +76,11 @@ export interface RequestHistoryPage { export const REQUEST_HISTORY_MAX_PAGE_SIZE = 100; export const REQUEST_HISTORY_DEFAULT_PAGE_SIZE = 50; export const REQUEST_HISTORY_INSERT_BATCH = 500; +export const REQUEST_HISTORY_READ_CHUNK_BYTES = 64 * 1024; +// The SQLite index is a disposable projection. Complete JSONL records above +// this bound are omitted from the projection; the canonical usage.jsonl is +// never truncated or rewritten by the indexer. +export const REQUEST_HISTORY_MAX_RECORD_BYTES = 1024 * 1024; let db: Database | null = null; let dbPath = ""; @@ -190,24 +195,6 @@ INSERT OR IGNORE INTO requests ( ?, ?, ?, ?, ? )`; -function readCompleteTail(fd: number, fromOffset: number, size: number): { text: string; nextOffset: number } { - if (size <= fromOffset) return { text: "", nextOffset: fromOffset }; - const length = size - fromOffset; - const buf = Buffer.allocUnsafe(length); - let offset = 0; - while (offset < length) { - const read = readSync(fd, buf, offset, length - offset, fromOffset + offset); - if (read === 0) throw new Error("usage log changed while indexing"); - offset += read; - } - const newline = buf.lastIndexOf(0x0a); - if (newline < 0) return { text: "", nextOffset: fromOffset }; - return { - text: buf.subarray(0, newline).toString("utf-8"), - nextOffset: fromOffset + newline + 1, - }; -} - function parsedEntryFromLine(line: string): PersistedUsageEntry | null { if (!line.trim()) return null; try { @@ -227,12 +214,11 @@ function parsedEntryFromLine(line: string): PersistedUsageEntry | null { return null; } -function ingestText(dbHandle: Database, text: string): number { - if (!text) return 0; - const lines = text.split(/\r?\n/); +function ingestSourceTail(dbHandle: Database, path: string, fromOffset: number): number { let inserted = 0; let pending: Array> = []; const insert = dbHandle.prepare(ROW_INSERT); + let fd: number | undefined; try { const commitBatch = () => { dbHandle.transaction((rows: Array>) => { @@ -243,29 +229,55 @@ function ingestText(dbHandle: Database, text: string): number { })(pending); pending = []; }; - for (const line of lines) { - const entry = parsedEntryFromLine(line); - if (!entry) continue; + const ingestLine = (line: Buffer) => { + const entry = parsedEntryFromLine(line.toString("utf-8")); + if (!entry) return; pending.push(extractRow(entry)); if (pending.length >= REQUEST_HISTORY_INSERT_BATCH) commitBatch(); - } - if (pending.length > 0) commitBatch(); - } finally { - // Windows file locks: an unterminated prepared statement keeps the DB - // file busy after close (verified on Bun 1.3.14). Finalize always. - insert.finalize(); - } - return inserted; -} + }; -function ingestSourceTail(dbHandle: Database, path: string, fromOffset: number): number { - let fd: number | undefined; - try { fd = openSync(path, "r"); const stat = fstatSync(fd); - const { text, nextOffset } = readCompleteTail(fd, fromOffset, Number(stat.size)); - if (text.length === 0 && nextOffset === fromOffset) return 0; - const inserted = ingestText(dbHandle, text); + const size = Number(stat.size); + const readBuffer = Buffer.allocUnsafe(REQUEST_HISTORY_READ_CHUNK_BYTES); + let position = fromOffset; + let nextOffset = fromOffset; + let fragments: Buffer[] = []; + let fragmentBytes = 0; + let oversized = false; + while (position < size) { + const requested = Math.min(readBuffer.length, size - position); + const bytesRead = readSync(fd, readBuffer, 0, requested, position); + if (bytesRead === 0) throw new Error("usage log changed while indexing"); + let lineStart = 0; + for (let index = 0; index < bytesRead; index++) { + if (readBuffer[index] !== 0x0a) continue; + const segment = readBuffer.subarray(lineStart, index); + if (!oversized && fragmentBytes + segment.length <= REQUEST_HISTORY_MAX_RECORD_BYTES) { + const line = fragments.length === 0 + ? segment + : Buffer.concat([...fragments, segment], fragmentBytes + segment.length); + ingestLine(line); + } + fragments = []; + fragmentBytes = 0; + oversized = false; + lineStart = index + 1; + nextOffset = position + index + 1; + } + const remainder = readBuffer.subarray(lineStart, bytesRead); + if (!oversized && fragmentBytes + remainder.length <= REQUEST_HISTORY_MAX_RECORD_BYTES) { + // Copy because readBuffer is reused on the next iteration. + fragments.push(Buffer.from(remainder)); + fragmentBytes += remainder.length; + } else if (remainder.length > 0) { + fragments = []; + fragmentBytes = 0; + oversized = true; + } + position += bytesRead; + } + if (pending.length > 0) commitBatch(); const current = readIndexedMeta(dbHandle); setMeta(dbHandle, HISTORY_META_KEYS.indexedOffset, nextOffset); setMeta(dbHandle, HISTORY_META_KEYS.indexedRows, current.indexedRows + inserted); @@ -275,6 +287,9 @@ function ingestSourceTail(dbHandle: Database, path: string, fromOffset: number): return inserted; } finally { if (fd !== undefined) closeSync(fd); + // Windows file locks: an unterminated prepared statement keeps the DB + // file busy after close (verified on Bun 1.3.14). Finalize always. + insert.finalize(); } } diff --git a/tests/request-history-index.test.ts b/tests/request-history-index.test.ts index 97d6a0a30..f67d6dd9e 100644 --- a/tests/request-history-index.test.ts +++ b/tests/request-history-index.test.ts @@ -1,5 +1,14 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { existsSync, mkdtempSync, rmSync, truncateSync, writeFileSync } from "node:fs"; +import { + appendFileSync, + existsSync, + mkdtempSync, + readFileSync, + rmSync, + statSync, + truncateSync, + writeFileSync, +} from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { handleManagementAPI } from "../src/server/management-api"; @@ -7,6 +16,7 @@ import { ManagementRequest } from "./helpers/management-auth"; import { appendUsageEntry, resetUsageReadCacheForTests, + usageLogPath, type PersistedUsageEntry, } from "../src/usage/log"; import { @@ -14,7 +24,9 @@ import { queryRequestHistory, rebuildRequestHistoryIndex, requestHistoryRowById, + REQUEST_HISTORY_MAX_RECORD_BYTES, REQUEST_HISTORY_MAX_PAGE_SIZE, + REQUEST_HISTORY_READ_CHUNK_BYTES, } from "../src/routing/history/indexer"; import { InvalidCursorError } from "../src/routing/history/cursor"; import { HISTORY_DB_FILENAME } from "../src/routing/history/schema"; @@ -194,18 +206,50 @@ describe("request-history index (RI-02)", () => { test("partial final JSONL line is skipped until it completes", async () => { for (const row of seedRows(3)) appendUsageEntry(row); + const completeOffset = statSync(usageLogPath()).size; // Append a partial line without a trailing newline. - const { appendFileSync } = await import("node:fs"); - const { usageLogPath } = await import("../src/usage/log"); appendFileSync(usageLogPath(), '{"requestId":"req-partial","timestamp":', "utf-8"); const page = await queryRequestHistory({}, undefined, 10); expect(page.rows.length).toBe(3); expect(page.meta.indexedRows).toBe(3); + expect(page.meta.indexedOffset).toBe(completeOffset); // Completing the line makes it indexable on the next refresh. appendFileSync(usageLogPath(), '9999,"provider":"a","model":"m1","status":200,"durationMs":1,"usageStatus":"reported"}\n', "utf-8"); const after = await queryRequestHistory({}, undefined, 10); expect(after.rows.length).toBe(4); - expect(after.rows.some(row => row.requestId === "req-partial")).toBe(true); + expect(after.rows.filter(row => row.requestId === "req-partial")).toHaveLength(1); + expect(after.meta.indexedOffset).toBe(statSync(usageLogPath()).size); + }); + + test("streaming refresh indexes a valid record that crosses a read chunk", async () => { + const large = entry("chunk-spanning", 9998, "a", "m1", { + apiKeyId: "x".repeat(REQUEST_HISTORY_READ_CHUNK_BYTES + 1024), + }); + const line = `${JSON.stringify(large)}\n`; + expect(Buffer.byteLength(line)).toBeGreaterThan(REQUEST_HISTORY_READ_CHUNK_BYTES); + expect(Buffer.byteLength(line)).toBeLessThan(REQUEST_HISTORY_MAX_RECORD_BYTES); + appendFileSync(usageLogPath(), line, "utf-8"); + + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.map(row => row.requestId)).toEqual(["chunk-spanning"]); + expect(page.meta.indexedOffset).toBe(statSync(usageLogPath()).size); + }); + + test("streaming refresh skips an oversized record without changing the canonical log", async () => { + const oversized = entry("oversized", 9998, "a", "m1", { + apiKeyId: "x".repeat(REQUEST_HISTORY_MAX_RECORD_BYTES + 1), + }); + const oversizedLine = `${JSON.stringify(oversized)}\n`; + expect(Buffer.byteLength(oversizedLine)).toBeGreaterThan(REQUEST_HISTORY_MAX_RECORD_BYTES); + appendFileSync(usageLogPath(), oversizedLine, "utf-8"); + appendUsageEntry(entry("after-oversized", 9999)); + const canonical = readFileSync(usageLogPath()); + + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.map(row => row.requestId)).toEqual(["after-oversized"]); + expect(page.meta.indexedRows).toBe(1); + expect(page.meta.indexedOffset).toBe(canonical.byteLength); + expect(readFileSync(usageLogPath())).toEqual(canonical); }); test("duplicate replay is ignored", async () => {