diff --git a/CHANGELOG.md b/CHANGELOG.md index bd28b9a3..07e864c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,30 @@ than accepting one it cannot attribute. `scripts/start.sh` runs the worker local turns it on with `routines.enabled` and takes the secret as `secrets.workerSharedSecret`. No new port is opened for any of this — the worker only ever calls out to the server it already trusts. +### Turn screenshots are swept in every deployment, not one + +A page a Bot opens is photographed and kept in `computer_page_frame`, so a conversation read back +later shows what it was looking at. The reaper for those rows had one caller: the idle-computer +culler, which refuses to run unless each Bot has its own computer and is scheduled only by the Helm +chart's CronJob, which exists only when `computers.mode` is `sandbox`. On Compose, on the all-in-one +image, and on the chart's own default of `shared`, nothing ever called it. One browsing Bot over +ninety days is several hundred megabytes of rows that nothing was ever going to remove. + +The sweep now runs on the server, on the same hourly timer that removes old audit rows, and does not +wait for a retention policy to be configured: a month of screenshots is what the store already meant +to keep. It also removes them in batches, because one statement over that much data held its locks +for seventeen seconds. + +Deployments using `computers.mode: sandbox` are unaffected in what they keep. The culler no longer +purges frames, because the server does it there too and one owner is better than two. + +**On upgrade, the first sweep removes the backlog.** A deployment that has been keeping every +screenshot since it was installed will lose the ones older than a month, about a minute after the +server starts. That is the window the store has always documented and the one sandbox deployments +have been enforcing, but it has never been applied anywhere else, so it is worth knowing before the +upgrade rather than after. It is drained in batches, forty thousand rows an hour, rather than in one +statement. + ### A channel a Bot has spoken in unseen shows a dot The sidebar marks a channel when a Bot has said something since you last had it open: a dot beside diff --git a/server/scripts/cull-idle-computers.ts b/server/scripts/cull-idle-computers.ts index 43183ac3..43074a48 100644 --- a/server/scripts/cull-idle-computers.ts +++ b/server/scripts/cull-idle-computers.ts @@ -11,7 +11,6 @@ * losing anything, and a failing CronJob that pages somebody at 3am should mean something worse. */ import { randomUUID } from "node:crypto"; -import { createPageFrameStore } from "../src/computer/page-frames"; import { createComputerProvider } from "../src/computer/provider"; import { loadConfig } from "../src/config"; import { createDatabase } from "../src/db/client"; @@ -36,21 +35,11 @@ if (config.computer.provider !== "sandbox") { const database = createDatabase(config.databaseUrl); const queue = createWorkQueue(database); -const pageFrames = createPageFrameStore(database); const provider = createComputerProvider(config.computer); // A name for the lease, so a stuck claim can be traced back to the pod that took it. const owner = `culler/${process.env.HOSTNAME ?? randomUUID().slice(0, 8)}`; -/** - * How long a turn's screenshot is kept. - * - * A month, because reading back a conversation is the thing these exist for and people do that long - * after the run. Past that the transcript names the page it opened instead, which is the same - * sentence with less in it rather than a broken one. - */ -const FRAME_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; - try { const options = { database, @@ -86,19 +75,6 @@ try { */ finishedOlderThanMs: config.computer.idleAfterMs, }); - /* - * And the screenshots, which had a reaper and nothing calling it. - * - * A page is a row and a Bot that browses makes them for as long as it runs, so this table only - * ever grew: written on every navigation, taken out by a profile wipe and by nothing else. Kept - * long enough that reading back a conversation from last month still shows what it opened, and not - * for ever, because these are the largest thing this deployment stores and the least useful once - * nobody is reading that conversation any more. - * - * Here rather than in the API server because this is already the sweep that runs on a schedule - * with a claim under it, and a second timer would be a second thing to get wrong. - */ - const framesPurged = await pageFrames.purge(FRAME_RETENTION_MS); console.info( JSON.stringify({ type: "computer-cull", @@ -106,7 +82,6 @@ try { suspended: report.suspended, skipped: report.skipped, purged, - framesPurged, }), ); } finally { diff --git a/server/src/audit-retention.ts b/server/src/audit-retention.ts index 5654969d..57c6cb08 100644 --- a/server/src/audit-retention.ts +++ b/server/src/audit-retention.ts @@ -21,6 +21,10 @@ * every action writes to. */ import postgres from "postgres"; +import { + FRAME_RETENTION_MS, + type PageFrameStore, +} from "./computer/page-frames"; /** * The advisory lock this takes, as an arbitrary but fixed number. @@ -117,18 +121,39 @@ export type RetentionSweeper = { stop: () => void }; * Not immediately at boot: a deployment rolling several servers would have all of them contend for * the lock in the same second, and the one that wins would compete with start-up for the database. */ -export function startAuditRetention( +export function startRetentionSweeps( databaseUrl: string, retentionDays: number | undefined, + // Swept whatever the audit policy is: their only other caller runs in `computers.mode: sandbox` alone. + pageFrames?: PageFrameStore, options: { intervalMs?: number; firstRunMs?: number } = {}, ): RetentionSweeper { - if (!retentionDays || retentionDays < 1) return { stop: () => undefined }; + const sweepsAudit = Boolean(retentionDays && retentionDays >= 1); + if (!sweepsAudit && !pageFrames) return { stop: () => undefined }; const intervalMs = options.intervalMs ?? 60 * 60_000; const firstRunMs = options.firstRunMs ?? 60_000; const timers: ReturnType[] = []; const run = () => { + if (pageFrames) { + void pageFrames + .purge(FRAME_RETENTION_MS) + .then((removed) => { + if (removed === 0) return; + console.info(JSON.stringify({ type: "page-frames-swept", removed })); + }) + .catch((error) => { + console.error( + JSON.stringify({ + type: "page-frames-sweep-failed", + note: "Old turn screenshots were not removed. Nothing is broken; the table is larger than it should be.", + error: String(error), + }), + ); + }); + } + if (!sweepsAudit || !retentionDays) return; void sweepAuditTrail(databaseUrl, retentionDays) .then(({ deleted }) => { if (deleted === null || deleted === 0) return; diff --git a/server/src/computer/page-frames.ts b/server/src/computer/page-frames.ts index 727dbd0d..55177592 100644 --- a/server/src/computer/page-frames.ts +++ b/server/src/computer/page-frames.ts @@ -4,7 +4,7 @@ * Written where the navigation happens, which is the one moment the screen is certainly showing the * page that was asked for, and read back when somebody reopens the conversation that asked for it. */ -import { and, eq, lt, sql } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import type { Database } from "../db/client"; import { computerPageFrame } from "../db/schema"; @@ -17,6 +17,19 @@ import { computerPageFrame } from "../db/schema"; */ const MAX_FRAME_BYTES = 4 * 1024 * 1024; +/** + * How long a turn's screenshot is kept. + * + * A month, because reading back a conversation is the thing these exist for and people do that long + * after the run. Past that the transcript names the page it opened instead, which is the same + * sentence with less in it rather than a broken one. + */ +export const FRAME_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; + +// Small batches: a frame is hundreds of kilobytes, so the audit sweep's five thousand would be a gigabyte a statement. +const PURGE_BATCH = 200; +const MAX_PURGE_BATCHES = 200; + /** * How big a base64 string actually is, in bytes. * @@ -124,16 +137,21 @@ export function createPageFrameStore(database: Database): PageFrameStore { }, async purge(olderThanMs) { - const gone = await database - .delete(computerPageFrame) - .where( - lt( - computerPageFrame.capturedAt, - sql`now() - make_interval(secs => ${olderThanMs / 1000})`, - ), - ) - .returning({ url: computerPageFrame.url }); - return gone.length; + // Batched like the audit sweep: one statement over ninety days of one Bot held its locks for 17s. + let removed = 0; + for (let batch = 0; batch < MAX_PURGE_BATCHES; batch += 1) { + const result = (await database.execute(sql` + delete from ${computerPageFrame} where ctid in ( + select ctid from ${computerPageFrame} + where ${computerPageFrame.capturedAt} < now() - make_interval(secs => ${olderThanMs / 1000}) + limit ${PURGE_BATCH} + ) + `)) as unknown as { count?: number }; + const count = result?.count ?? 0; + removed += count; + if (count < PURGE_BATCH) break; + } + return removed; }, }; } diff --git a/server/src/index.ts b/server/src/index.ts index 15104de5..d2164ab8 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -11,7 +11,7 @@ import type { AgentActor } from "./agents/profile-types"; import { createRuntimeAgentLoader } from "./agents/runtime-agents"; import { createApp } from "./app"; import { createAuditReader, createAuditStore, recordAuditEvent } from "./audit"; -import { startAuditRetention } from "./audit-retention"; +import { startRetentionSweeps } from "./audit-retention"; import { createAuth } from "./auth"; import { DEV_ACTOR, initializeDevActorUser } from "./auth/dev-actor"; import { createRoleRepository } from "./auth/guards"; @@ -248,15 +248,13 @@ const policyListener = await startPolicyListener( * unavailable, and the row is a note for a reader rather than something the server depends on. */ const bootAuditStore = createAuditStore(database); -/* - * Old audit rows removed on a schedule, when a deployment has asked for that. - * - * One server sweeps rather than all of them, decided by an advisory lock. Off unless - * `AUDIT_RETENTION_DAYS` is set. See audit-retention.ts. - */ -const auditRetention = startAuditRetention( +// One store: the gateway writes through it, a route reads it, and the sweep below takes the old ones out. +const pageFrameStore = createPageFrameStore(database); +// Housekeeping on a schedule: audit rows when asked for, screenshots always, one timer. See audit-retention.ts. +const retentionSweeps = startRetentionSweeps( config.databaseUrl, config.auditRetentionDays, + pageFrameStore, ); const computerGateway = computerProvider ? createComputerGateway({ @@ -269,7 +267,7 @@ const computerGateway = computerProvider snapshots: createSnapshotStore(database), // So wiping a profile takes the pictures of its signed-in pages with it, which is what the // sentence on that button already promised. - pageFrames: createPageFrameStore(database), + pageFrames: pageFrameStore, allowPrivateHosts: config.computer?.allowPrivateHosts, token: config.computer?.token, }) @@ -711,7 +709,7 @@ const app = createApp( // Chooses the coworker for an untagged message, on the deployment's own model and key. intentRouter, // What a browsing turn's screen looked like when it finished, so the transcript can show it later. - createPageFrameStore(database), + pageFrameStore, // What a due routine actually does: a turn, run as its owner, into the thread they will open. routineRunner, // A person's own standing instructions: the list, and a switch to stop one. @@ -880,7 +878,7 @@ for (const signal of ["SIGINT", "SIGTERM"] as const) { void Promise.allSettled([ channelActivityListener.stop(), policyListener.stop(), - Promise.resolve(auditRetention.stop()), + Promise.resolve(retentionSweeps.stop()), ]).finally(() => process.exit(0)); }); } diff --git a/server/tests/page-frame-retention.integration.test.ts b/server/tests/page-frame-retention.integration.test.ts new file mode 100644 index 00000000..6f74c851 --- /dev/null +++ b/server/tests/page-frame-retention.integration.test.ts @@ -0,0 +1,124 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { eq, sql } from "drizzle-orm"; +import { startRetentionSweeps } from "../src/audit-retention"; +import { + createPageFrameStore, + FRAME_RETENTION_MS, +} from "../src/computer/page-frames"; +import { createDatabase } from "../src/db/client"; +import { computerPageFrame } from "../src/db/schema"; +import { TEST_POOL } from "./support/database"; + +/** + * The screenshots have to be able to stop growing, in every deployment rather than in one. + * + * Their reaper had a single caller, `scripts/cull-idle-computers.ts`, which refuses to run unless the + * provider is `sandbox` and is scheduled only by the chart's culler CronJob, which renders only in + * that mode. Compose, the all-in-one image and the chart's own default of `computers.mode: shared` + * therefore wrote a row per navigation and removed none, ever. + * + * Against a real database because the interval arithmetic and the batching are both SQL. + */ + +const databaseUrl = + process.env.DATABASE_URL ?? + "postgres://openbot:openbot@localhost:5432/openbot"; +const database = createDatabase(databaseUrl, TEST_POOL); +const store = createPageFrameStore(database); + +const COMPUTER = `frame-retention-${crypto.randomUUID().slice(0, 8)}`; + +async function frame(daysAgo: number, index: number): Promise { + await store.save({ + computerId: COMPUTER, + toolCallId: `turn-${index}`, + url: `https://example.com/${index}`, + title: `page ${index}`, + frame: "iVBORw0KGgo=", + }); + await database + .update(computerPageFrame) + .set({ capturedAt: sql`now() - make_interval(days => ${daysAgo})` }) + .where(eq(computerPageFrame.toolCallId, `turn-${index}`)); +} + +const kept = () => + database + .select({ toolCallId: computerPageFrame.toolCallId }) + .from(computerPageFrame) + .where(eq(computerPageFrame.computerId, COMPUTER)); + +afterEach(async () => { + await database + .delete(computerPageFrame) + .where(eq(computerPageFrame.computerId, COMPUTER)); +}); + +describe("page frame retention", () => { + test("a deployment that has not configured audit retention still sweeps frames", async () => { + await frame(90, 1); + await frame(31, 2); + await frame(1, 3); + + /* + * `undefined` is the whole point of this case. It is what a deployment that never set + * `AUDIT_RETENTION_DAYS` has, and the sweeper used to return before starting a timer at all, + * which is how the frames went unswept everywhere the culler does not run. + */ + const sweeps = startRetentionSweeps(databaseUrl, undefined, store, { + firstRunMs: 10, + intervalMs: 3_600_000, + }); + try { + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((await kept()).length === 1) break; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + } finally { + sweeps.stop(); + } + + expect((await kept()).map((row) => row.toolCallId)).toEqual(["turn-3"]); + }); + + test("purge removes everything past the window and nothing inside it", async () => { + await frame(40, 1); + await frame(29, 2); + + await store.purge(FRAME_RETENTION_MS); + expect((await kept()).map((row) => row.toolCallId)).toEqual(["turn-2"]); + }); + + test("purge past its batch size removes every eligible row", async () => { + for (let index = 1; index <= 205; index += 1) await frame(45, index); + + await store.purge(FRAME_RETENTION_MS); + expect(await kept()).toHaveLength(0); + }); + + test("purge leaves a frame inside the window alone", async () => { + await frame(1, 1); + + await store.purge(FRAME_RETENTION_MS); + expect(await kept()).toHaveLength(1); + }); + + /* + * Asserted on what survives for this computer rather than on what `purge` returns: the sweep is + * deployment-wide by design, so its count moves with whatever else is in the table. + * + * Two replicas sweeping at once, which is the ordinary case: this half takes no advisory lock, + * because a delete keyed on age run twice removes the same rows once. The risk it has to be shown + * not to have is double-counting or deadlocking on the same `ctid`s. + */ + test("two sweeps at once remove each row once and neither stalls", async () => { + for (let index = 1; index <= 400; index += 1) await frame(45, index); + + await Promise.all([ + store.purge(FRAME_RETENTION_MS), + store.purge(FRAME_RETENTION_MS), + ]); + + expect(await kept()).toHaveLength(0); + }); +});