fix(usage): bound OpenCode history reads to prevent backend OOM

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";

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(