Skip to content

fix(history): stream request-history index ingestion (#1189) #1287

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lidge-jun merged 1 commit into from
Aug 8, 2026
Merged

fix(history): stream request-history index ingestion (#1189) #1287

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 54 additions & 39 deletions src/routing/history/indexer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment on lines +80 to +83

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Disclose oversized records missing from request history

When usage.jsonl contains a complete valid record larger than 1 MiB, the indexer advances past it but permanently omits it from request history and routing analytics. This conflicts with docs-site/src/content/docs/reference/configuration/routing.md, which promises “full history,” and the returned metadata provides no truncation signal, so users can unknowingly make decisions from incomplete results. Document the limit and affected endpoints, or expose an explicit omission indicator.

AGENTS.md reference: src/AGENTS.md:L28-L28

Useful? React with 👍 / 👎.


let db: Database | null = null;
let dbPath = "";
Expand Down Expand Up @@ -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 {
Expand All @@ -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<Array<string | number | null>> = [];
const insert = dbHandle.prepare(ROW_INSERT);
let fd: number | undefined;
try {
const commitBatch = () => {
dbHandle.transaction((rows: Array<Array<string | number | null>>) => {
Expand All @@ -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);
Expand All @@ -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();
}
}

Expand Down
52 changes: 48 additions & 4 deletions tests/request-history-index.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,20 +1,32 @@
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";
import { ManagementRequest } from "./helpers/management-auth";
import {
appendUsageEntry,
resetUsageReadCacheForTests,
usageLogPath,
type PersistedUsageEntry,
} from "../src/usage/log";
import {
closeRequestHistoryIndex,
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";
Expand Down Expand Up @@ -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 () => {
Expand Down
Loading