From 75c439f4ecc5431f4b1ae95ddd620245ff57bb27 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Sat, 10 Oct 2026 20:11:28 -0500 Subject: [PATCH] fix(usage): bound OpenCode history reads to prevent backend OOM (#17961) --- .../src/server/usage.test.ts | 71 +++++++++++++++++++ .../provider-opencode/src/server/usage.ts | 60 +++++++++++----- 2 files changed, 112 insertions(+), 19 deletions(-) diff --git a/packages/provider-opencode/src/server/usage.test.ts b/packages/provider-opencode/src/server/usage.test.ts index 5a517e40c5cb9e72438a895e4f9ad48f936451d0..7575c3659040ef8bbdeda78e8cf6c9b943dd8f16 100644 --- a/packages/provider-opencode/src/server/usage.test.ts +++ b/packages/provider-opencode/src/server/usage.test.ts @@ -7,6 +7,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; +import { vi } from "vite-plus/test"; import { readOpenCodeUsage } from "./usage.ts"; @@ -17,6 +18,75 @@ const openDatabase = (filename: string) => ); describe("readOpenCodeUsage", () => { + it.effect("bounds raw history allocations while counting every eligible message once", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "usage-reader-large-test-" }); + const db = yield* openDatabase(path.join(dir, "opencode.db")); + yield* db.unsafe( + "CREATE TABLE message (id TEXT, session_id TEXT, data TEXT, time_created INTEGER)", + ); + const sinceMs = 1780000000000; + const message = { + role: "assistant", + modelID: "claude-sonnet-4-5", + time: { created: sinceMs }, + tokens: { input: 100, output: 20 }, + // Transcript bodies are irrelevant to usage, but can be very large. + body: "x".repeat(64 * 1024), + }; + const data = JSON.stringify(message); + const insert = (rowId: number, id: string, payload: string, created = sinceMs) => + db.unsafe( + "INSERT INTO message (rowid, id, session_id, data, time_created) VALUES (?, ?, ?, ?, ?)", + [rowId, id, "session-1", payload, created], + ); + // Sparse IDs, including negative rowids, must not skip records at page boundaries. + for (let index = 0; index < 130; index++) { + yield* insert(index * 2 - 10, `msg-${index}`, data); + } + yield* insert(300, "old", data, sinceMs - 1); + yield* insert(301, "user", JSON.stringify({ ...message, role: "user" })); + yield* insert(302, "invalid", "{invalid"); + yield* insert(303, "msg-0", data); + + let largestRawResult = 0; + const originalAll = NodeSqlite.StatementSync.prototype.all; + yield* Effect.acquireRelease( + Effect.sync(() => + vi.spyOn(NodeSqlite.StatementSync.prototype, "all").mockImplementation(function ( + this: NodeSqlite.StatementSync, + ...params + ) { + const rows = originalAll.apply(this, params); + const rawSize = rows.reduce( + (size, row) => size + (typeof row.data === "string" ? row.data.length : 0), + 0, + ); + largestRawResult = Math.max(largestRawResult, rawSize); + return rows; + }), + ), + (spy) => Effect.sync(() => spy.mockRestore()), + ); + const result = yield* readOpenCodeUsage(dir, sinceMs); + assert.isFalse(result.error); + const records = result.files.flatMap((file) => file.records); + assert.deepStrictEqual( + records.map((record) => record.dedupeKey), + Array.from({ length: 130 }, (_, index) => `opencode:msg-${index}`), + ); + assert.strictEqual( + records.reduce((total, record) => total + record.totals.uncachedInputTokens, 0), + 13000, + ); + // The fixture exceeds 8 MiB. No query may retain its complete bodies. + assert.isAbove(largestRawResult, 0); + assert.isBelow(largestRawResult, 6 * 1024 * 1024); + }).pipe(Effect.provide(NodeServices.layer)), + ); + it.effect("counts migrated OpenCode messages once and sees subsequent WAL writes", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -68,3 +138,4 @@ describe("readOpenCodeUsage", () => { }).pipe(Effect.provide(NodeServices.layer)), ); }); +import * as NodeSqlite from "node:sqlite"; diff --git a/packages/provider-opencode/src/server/usage.ts b/packages/provider-opencode/src/server/usage.ts index 9d09266118eb702cbe9ee01c03c947b031ddeaee..677d502a72b0a7303bd81e4f2648f5bfff43dbda 100644 --- a/packages/provider-opencode/src/server/usage.ts +++ b/packages/provider-opencode/src/server/usage.ts @@ -91,6 +91,7 @@ export interface OpenCodeUsageReadResult { } const isNotFound = (cause: PlatformError.PlatformError) => cause.reason._tag === "NotFound"; +const MESSAGE_PAGE_SIZE = 64; /** A regular file or directory, never a symlink to one. */ const entryType = Effect.fn("entryType")(function* (path: string) { @@ -172,26 +173,47 @@ export const readOpenCodeUsage = Effect.fn("readOpenCodeUsage")(function* ( const timestamp = columns.has("time_created") ? "time_created" : "NULL"; const predicates = table === "session_message" ? ["type = 'assistant'"] : []; if (timestamp !== "NULL") predicates.push("time_created >= ?"); - const where = predicates.length > 0 ? ` WHERE ${predicates.join(" AND ")}` : ""; - const rows = yield* sql.unsafe<{ - readonly id: unknown; - readonly session_id: unknown; - readonly data: unknown; - readonly created: unknown; - }>( - `SELECT id, session_id, data, ${timestamp} AS created FROM ${table}${where}`, - timestamp === "NULL" ? [] : [sinceMs], - ); - for (const [index, row] of rows.entries()) { - append( - file.records, - parseOpenCodeMessage(text(row.data), { - id: text(row.id), - sessionId: text(row.session_id), - ...(typeof row.created === "number" ? { timestampMs: row.created } : {}), - }), + const lastRowId = (yield* sql.unsafe<{ readonly last_row_id: number | null }>( + `SELECT MAX(rowid) AS last_row_id FROM ${table}`, + ))[0]?.last_row_id; + if (lastRowId === null || lastRowId === undefined) continue; + let afterRowId: number | undefined; + // SqlClient materializes each result with StatementSync.all(). Keep + // message bodies in small pages, and exclude rows appended mid-scan. + while (true) { + const pagePredicates = [ + ...predicates, + "rowid <= ?", + ...(afterRowId === undefined ? [] : ["rowid > ?"]), + ]; + const rows = yield* sql.unsafe<{ + readonly row_id: number; + readonly id: unknown; + readonly session_id: unknown; + readonly data: unknown; + readonly created: unknown; + }>( + `SELECT rowid AS row_id, id, session_id, data, ${timestamp} AS created + FROM ${table} WHERE ${pagePredicates.join(" AND ")} ORDER BY rowid LIMIT ${MESSAGE_PAGE_SIZE}`, + [ + ...(timestamp === "NULL" ? [] : [sinceMs]), + lastRowId, + ...(afterRowId === undefined ? [] : [afterRowId]), + ], ); - if (index % 256 === 255) yield* Effect.yieldNow; + for (const row of rows) { + append( + file.records, + parseOpenCodeMessage(text(row.data), { + id: text(row.id), + sessionId: text(row.session_id), + ...(typeof row.created === "number" ? { timestampMs: row.created } : {}), + }), + ); + } + if (rows.length < MESSAGE_PAGE_SIZE) break; + afterRowId = rows[rows.length - 1]!.row_id; + yield* Effect.yieldNow; } } }).pipe( -- 2.54.0