Skip to content

Commit 200c88e

Browse files
committed
Sweep turn screenshots in every deployment, not only where a culler runs
1 parent e8aa344 commit 200c88e

6 files changed

Lines changed: 213 additions & 49 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,30 @@ It is a view over the trail, not a second record of it. Refusals and stalls are
2323
transactionally by the gateway and the stall guard, so the inbox cannot miss one and nothing new
2424
runs on the action path. The only state it owns is the resolution, held beside the append-only trail
2525
rather than in it. The trail itself still keeps everything; the inbox is only what is open now.
26+
27+
### Turn screenshots are swept in every deployment, not one
28+
29+
A page a Bot opens is photographed and kept in `computer_page_frame`, so a conversation read back
30+
later shows what it was looking at. The reaper for those rows had one caller: the idle-computer
31+
culler, which refuses to run unless each Bot has its own computer and is scheduled only by the Helm
32+
chart's CronJob, which exists only when `computers.mode` is `sandbox`. On Compose, on the all-in-one
33+
image, and on the chart's own default of `shared`, nothing ever called it. One browsing Bot over
34+
ninety days is several hundred megabytes of rows that nothing was ever going to remove.
35+
36+
The sweep now runs on the server, on the same hourly timer that removes old audit rows, and does not
37+
wait for a retention policy to be configured: a month of screenshots is what the store already meant
38+
to keep. It also removes them in batches, because one statement over that much data held its locks
39+
for seventeen seconds.
40+
41+
Deployments using `computers.mode: sandbox` are unaffected in what they keep. The culler no longer
42+
purges frames, because the server does it there too and one owner is better than two.
43+
44+
**On upgrade, the first sweep removes the backlog.** A deployment that has been keeping every
45+
screenshot since it was installed will lose the ones older than a month, about a minute after the
46+
server starts. That is the window the store has always documented and the one sandbox deployments
47+
have been enforcing, but it has never been applied anywhere else, so it is worth knowing before the
48+
upgrade rather than after. It is drained in batches, forty thousand rows an hour, rather than in one
49+
statement.
2650
### A channel a Bot has spoken in unseen shows a dot
2751

2852
The sidebar marks a channel when a Bot has said something since you last had it open: a dot beside

‎server/scripts/cull-idle-computers.ts‎

Lines changed: 0 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111
* losing anything, and a failing CronJob that pages somebody at 3am should mean something worse.
1212
*/
1313
import { randomUUID } from "node:crypto";
14-
import { createPageFrameStore } from "../src/computer/page-frames";
1514
import { createComputerProvider } from "../src/computer/provider";
1615
import { loadConfig } from "../src/config";
1716
import { createDatabase } from "../src/db/client";
@@ -36,21 +35,11 @@ if (config.computer.provider !== "sandbox") {
3635

3736
const database = createDatabase(config.databaseUrl);
3837
const queue = createWorkQueue(database);
39-
const pageFrames = createPageFrameStore(database);
4038
const provider = createComputerProvider(config.computer);
4139

4240
// A name for the lease, so a stuck claim can be traced back to the pod that took it.
4341
const owner = `culler/${process.env.HOSTNAME ?? randomUUID().slice(0, 8)}`;
4442

45-
/**
46-
* How long a turn's screenshot is kept.
47-
*
48-
* A month, because reading back a conversation is the thing these exist for and people do that long
49-
* after the run. Past that the transcript names the page it opened instead, which is the same
50-
* sentence with less in it rather than a broken one.
51-
*/
52-
const FRAME_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
53-
5443
try {
5544
const options = {
5645
database,
@@ -86,27 +75,13 @@ try {
8675
*/
8776
finishedOlderThanMs: config.computer.idleAfterMs,
8877
});
89-
/*
90-
* And the screenshots, which had a reaper and nothing calling it.
91-
*
92-
* A page is a row and a Bot that browses makes them for as long as it runs, so this table only
93-
* ever grew: written on every navigation, taken out by a profile wipe and by nothing else. Kept
94-
* long enough that reading back a conversation from last month still shows what it opened, and not
95-
* for ever, because these are the largest thing this deployment stores and the least useful once
96-
* nobody is reading that conversation any more.
97-
*
98-
* Here rather than in the API server because this is already the sweep that runs on a schedule
99-
* with a claim under it, and a second timer would be a second thing to get wrong.
100-
*/
101-
const framesPurged = await pageFrames.purge(FRAME_RETENTION_MS);
10278
console.info(
10379
JSON.stringify({
10480
type: "computer-cull",
10581
offered,
10682
suspended: report.suspended,
10783
skipped: report.skipped,
10884
purged,
109-
framesPurged,
11085
}),
11186
);
11287
} finally {

‎server/src/audit-retention.ts‎

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@
2121
* every action writes to.
2222
*/
2323
import postgres from "postgres";
24+
import {
25+
FRAME_RETENTION_MS,
26+
type PageFrameStore,
27+
} from "./computer/page-frames";
2428

2529
/**
2630
* The advisory lock this takes, as an arbitrary but fixed number.
@@ -117,18 +121,39 @@ export type RetentionSweeper = { stop: () => void };
117121
* Not immediately at boot: a deployment rolling several servers would have all of them contend for
118122
* the lock in the same second, and the one that wins would compete with start-up for the database.
119123
*/
120-
export function startAuditRetention(
124+
export function startRetentionSweeps(
121125
databaseUrl: string,
122126
retentionDays: number | undefined,
127+
// Swept whatever the audit policy is: their only other caller runs in `computers.mode: sandbox` alone.
128+
pageFrames?: PageFrameStore,
123129
options: { intervalMs?: number; firstRunMs?: number } = {},
124130
): RetentionSweeper {
125-
if (!retentionDays || retentionDays < 1) return { stop: () => undefined };
131+
const sweepsAudit = Boolean(retentionDays && retentionDays >= 1);
132+
if (!sweepsAudit && !pageFrames) return { stop: () => undefined };
126133

127134
const intervalMs = options.intervalMs ?? 60 * 60_000;
128135
const firstRunMs = options.firstRunMs ?? 60_000;
129136
const timers: ReturnType<typeof setInterval>[] = [];
130137

131138
const run = () => {
139+
if (pageFrames) {
140+
void pageFrames
141+
.purge(FRAME_RETENTION_MS)
142+
.then((removed) => {
143+
if (removed === 0) return;
144+
console.info(JSON.stringify({ type: "page-frames-swept", removed }));
145+
})
146+
.catch((error) => {
147+
console.error(
148+
JSON.stringify({
149+
type: "page-frames-sweep-failed",
150+
note: "Old turn screenshots were not removed. Nothing is broken; the table is larger than it should be.",
151+
error: String(error),
152+
}),
153+
);
154+
});
155+
}
156+
if (!sweepsAudit || !retentionDays) return;
132157
void sweepAuditTrail(databaseUrl, retentionDays)
133158
.then(({ deleted }) => {
134159
if (deleted === null || deleted === 0) return;

‎server/src/computer/page-frames.ts‎

Lines changed: 29 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
* Written where the navigation happens, which is the one moment the screen is certainly showing the
55
* page that was asked for, and read back when somebody reopens the conversation that asked for it.
66
*/
7-
import { and, eq, lt, sql } from "drizzle-orm";
7+
import { and, eq, sql } from "drizzle-orm";
88
import type { Database } from "../db/client";
99
import { computerPageFrame } from "../db/schema";
1010

@@ -17,6 +17,19 @@ import { computerPageFrame } from "../db/schema";
1717
*/
1818
const MAX_FRAME_BYTES = 4 * 1024 * 1024;
1919

20+
/**
21+
* How long a turn's screenshot is kept.
22+
*
23+
* A month, because reading back a conversation is the thing these exist for and people do that long
24+
* after the run. Past that the transcript names the page it opened instead, which is the same
25+
* sentence with less in it rather than a broken one.
26+
*/
27+
export const FRAME_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
28+
29+
// Small batches: a frame is hundreds of kilobytes, so the audit sweep's five thousand would be a gigabyte a statement.
30+
const PURGE_BATCH = 200;
31+
const MAX_PURGE_BATCHES = 200;
32+
2033
/**
2134
* How big a base64 string actually is, in bytes.
2235
*
@@ -124,16 +137,21 @@ export function createPageFrameStore(database: Database): PageFrameStore {
124137
},
125138

126139
async purge(olderThanMs) {
127-
const gone = await database
128-
.delete(computerPageFrame)
129-
.where(
130-
lt(
131-
computerPageFrame.capturedAt,
132-
sql`now() - make_interval(secs => ${olderThanMs / 1000})`,
133-
),
134-
)
135-
.returning({ url: computerPageFrame.url });
136-
return gone.length;
140+
// Batched like the audit sweep: one statement over ninety days of one Bot held its locks for 17s.
141+
let removed = 0;
142+
for (let batch = 0; batch < MAX_PURGE_BATCHES; batch += 1) {
143+
const result = (await database.execute(sql`
144+
delete from ${computerPageFrame} where ctid in (
145+
select ctid from ${computerPageFrame}
146+
where ${computerPageFrame.capturedAt} < now() - make_interval(secs => ${olderThanMs / 1000})
147+
limit ${PURGE_BATCH}
148+
)
149+
`)) as unknown as { count?: number };
150+
const count = result?.count ?? 0;
151+
removed += count;
152+
if (count < PURGE_BATCH) break;
153+
}
154+
return removed;
137155
},
138156
};
139157
}

‎server/src/index.ts‎

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import { createAgentProfileStore } from "./agents/profile-store";
55
import { createRuntimeAgentLoader } from "./agents/runtime-agents";
66
import { createApp } from "./app";
77
import { createAuditReader, createAuditStore, recordAuditEvent } from "./audit";
8-
import { startAuditRetention } from "./audit-retention";
8+
import { startRetentionSweeps } from "./audit-retention";
99
import { createAuth } from "./auth";
1010
import { DEV_ACTOR, initializeDevActorUser } from "./auth/dev-actor";
1111
import { createRoleRepository } from "./auth/guards";
@@ -237,15 +237,13 @@ const policyListener = await startPolicyListener(
237237
* unavailable, and the row is a note for a reader rather than something the server depends on.
238238
*/
239239
const bootAuditStore = createAuditStore(database);
240-
/*
241-
* Old audit rows removed on a schedule, when a deployment has asked for that.
242-
*
243-
* One server sweeps rather than all of them, decided by an advisory lock. Off unless
244-
* `AUDIT_RETENTION_DAYS` is set. See audit-retention.ts.
245-
*/
246-
const auditRetention = startAuditRetention(
240+
// One store: the gateway writes through it, a route reads it, and the sweep below takes the old ones out.
241+
const pageFrameStore = createPageFrameStore(database);
242+
// Housekeeping on a schedule: audit rows when asked for, screenshots always, one timer. See audit-retention.ts.
243+
const retentionSweeps = startRetentionSweeps(
247244
config.databaseUrl,
248245
config.auditRetentionDays,
246+
pageFrameStore,
249247
);
250248
const computerGateway = computerProvider
251249
? createComputerGateway({
@@ -258,7 +256,7 @@ const computerGateway = computerProvider
258256
snapshots: createSnapshotStore(database),
259257
// So wiping a profile takes the pictures of its signed-in pages with it, which is what the
260258
// sentence on that button already promised.
261-
pageFrames: createPageFrameStore(database),
259+
pageFrames: pageFrameStore,
262260
allowPrivateHosts: config.computer?.allowPrivateHosts,
263261
token: config.computer?.token,
264262
})
@@ -559,7 +557,7 @@ const app = createApp(
559557
// Resolutions for the attention inbox: which trail rows a person has marked handled.
560558
createAttentionStore(database),
561559
// What a browsing turn's screen looked like when it finished, so the transcript can show it later.
562-
createPageFrameStore(database),
560+
pageFrameStore,
563561
);
564562

565563
/**
@@ -724,7 +722,7 @@ for (const signal of ["SIGINT", "SIGTERM"] as const) {
724722
void Promise.allSettled([
725723
channelActivityListener.stop(),
726724
policyListener.stop(),
727-
Promise.resolve(auditRetention.stop()),
725+
Promise.resolve(retentionSweeps.stop()),
728726
]).finally(() => process.exit(0));
729727
});
730728
}
Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
import { afterEach, describe, expect, test } from "bun:test";
2+
import { eq, sql } from "drizzle-orm";
3+
import { startRetentionSweeps } from "../src/audit-retention";
4+
import {
5+
createPageFrameStore,
6+
FRAME_RETENTION_MS,
7+
} from "../src/computer/page-frames";
8+
import { createDatabase } from "../src/db/client";
9+
import { computerPageFrame } from "../src/db/schema";
10+
import { TEST_POOL } from "./support/database";
11+
12+
/**
13+
* The screenshots have to be able to stop growing, in every deployment rather than in one.
14+
*
15+
* Their reaper had a single caller, `scripts/cull-idle-computers.ts`, which refuses to run unless the
16+
* provider is `sandbox` and is scheduled only by the chart's culler CronJob, which renders only in
17+
* that mode. Compose, the all-in-one image and the chart's own default of `computers.mode: shared`
18+
* therefore wrote a row per navigation and removed none, ever.
19+
*
20+
* Against a real database because the interval arithmetic and the batching are both SQL.
21+
*/
22+
23+
const databaseUrl =
24+
process.env.DATABASE_URL ??
25+
"postgres://openbot:openbot@localhost:5432/openbot";
26+
const database = createDatabase(databaseUrl, TEST_POOL);
27+
const store = createPageFrameStore(database);
28+
29+
const COMPUTER = `frame-retention-${crypto.randomUUID().slice(0, 8)}`;
30+
31+
async function frame(daysAgo: number, index: number): Promise<void> {
32+
await store.save({
33+
computerId: COMPUTER,
34+
toolCallId: `turn-${index}`,
35+
url: `https://example.com/${index}`,
36+
title: `page ${index}`,
37+
frame: "iVBORw0KGgo=",
38+
});
39+
await database
40+
.update(computerPageFrame)
41+
.set({ capturedAt: sql`now() - make_interval(days => ${daysAgo})` })
42+
.where(eq(computerPageFrame.toolCallId, `turn-${index}`));
43+
}
44+
45+
const kept = () =>
46+
database
47+
.select({ toolCallId: computerPageFrame.toolCallId })
48+
.from(computerPageFrame)
49+
.where(eq(computerPageFrame.computerId, COMPUTER));
50+
51+
afterEach(async () => {
52+
await database
53+
.delete(computerPageFrame)
54+
.where(eq(computerPageFrame.computerId, COMPUTER));
55+
});
56+
57+
describe("page frame retention", () => {
58+
test("a deployment that has not configured audit retention still sweeps frames", async () => {
59+
await frame(90, 1);
60+
await frame(31, 2);
61+
await frame(1, 3);
62+
63+
/*
64+
* `undefined` is the whole point of this case. It is what a deployment that never set
65+
* `AUDIT_RETENTION_DAYS` has, and the sweeper used to return before starting a timer at all,
66+
* which is how the frames went unswept everywhere the culler does not run.
67+
*/
68+
const sweeps = startRetentionSweeps(databaseUrl, undefined, store, {
69+
firstRunMs: 10,
70+
intervalMs: 3_600_000,
71+
});
72+
try {
73+
for (let attempt = 0; attempt < 100; attempt += 1) {
74+
if ((await kept()).length === 1) break;
75+
await new Promise((resolve) => setTimeout(resolve, 50));
76+
}
77+
} finally {
78+
sweeps.stop();
79+
}
80+
81+
expect((await kept()).map((row) => row.toolCallId)).toEqual(["turn-3"]);
82+
});
83+
84+
test("purge removes everything past the window and nothing inside it", async () => {
85+
await frame(40, 1);
86+
await frame(29, 2);
87+
88+
await store.purge(FRAME_RETENTION_MS);
89+
expect((await kept()).map((row) => row.toolCallId)).toEqual(["turn-2"]);
90+
});
91+
92+
test("purge past its batch size removes every eligible row", async () => {
93+
for (let index = 1; index <= 205; index += 1) await frame(45, index);
94+
95+
await store.purge(FRAME_RETENTION_MS);
96+
expect(await kept()).toHaveLength(0);
97+
});
98+
99+
test("purge leaves a frame inside the window alone", async () => {
100+
await frame(1, 1);
101+
102+
await store.purge(FRAME_RETENTION_MS);
103+
expect(await kept()).toHaveLength(1);
104+
});
105+
106+
/*
107+
* Asserted on what survives for this computer rather than on what `purge` returns: the sweep is
108+
* deployment-wide by design, so its count moves with whatever else is in the table.
109+
*
110+
* Two replicas sweeping at once, which is the ordinary case: this half takes no advisory lock,
111+
* because a delete keyed on age run twice removes the same rows once. The risk it has to be shown
112+
* not to have is double-counting or deadlocking on the same `ctid`s.
113+
*/
114+
test("two sweeps at once remove each row once and neither stalls", async () => {
115+
for (let index = 1; index <= 400; index += 1) await frame(45, index);
116+
117+
await Promise.all([
118+
store.purge(FRAME_RETENTION_MS),
119+
store.purge(FRAME_RETENTION_MS),
120+
]);
121+
122+
expect(await kept()).toHaveLength(0);
123+
});
124+
});

0 commit comments

Comments
 (0)