authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-05 16:20:37-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 16:20:37-07:00
logca952677321abf9f3762ae4bb0129454e5a63955
tree03773563a0b29bc6888381ad31162cc3a648f1c0
parent27ab45593fdcaffe718c1fa54dcbb2df4bd769a2
signature Signed by PGP key B5690EEEBB952194

fix(server): finished outbox rows and old PR cache files are pruned (#16247)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

5 files changed, 310 insertions(+), 5 deletions(-)

apps/server/src/orchestration-v2/EffectOutbox.ts+61
......@@ -16,13 +16,17 @@ import {
1616import * as Context from "effect/Context";
1717import * as DateTime from "effect/DateTime";
1818import * as Deferred from "effect/Deferred";
19import * as Duration from "effect/Duration";
1920import * as Effect from "effect/Effect";
2021import * as Layer from "effect/Layer";
2122import * as Option from "effect/Option";
2223import * as Queue from "effect/Queue";
24import * as Schedule from "effect/Schedule";
2325import * as Schema from "effect/Schema";
2426import * as SqlClient from "effect/sql/SqlClient";
2527
28import { forkParked } from "../serverActivation.ts";
29
2630export const OrchestrationEffectRequestV2 = Schema.Union([
2731 Schema.Struct({
2832 type: Schema.Literal("provider-runtime.continue"),
......@@ -138,6 +142,21 @@ export const OrchestrationEffectStatusV2 = Schema.Literals([
138142]);
139143export type OrchestrationEffectStatusV2 = typeof OrchestrationEffectStatusV2.Type;
140144
145/**
146 * How long a succeeded or cancelled effect row is kept after it completes.
147 * Claiming, recovery, the orchestrator and storage cleanup only ask whether a
148 * row is pending, running or failed, so a missing succeeded or cancelled row
149 * reads the same as a present one. Its one other use is deduplication: `enqueue`
150 * ignores an id that already exists, and ids are deterministic per command or
151 * run. Command retries are deduplicated by their receipts first, and a run's
152 * effects are re-enqueued only around one server restart, so a week is far
153 * longer than either and leaves recent history for debugging.
154 * Failed rows are kept: storage cleanup keeps a deleted thread's worktree while
155 * any of its effects failed.
156 */
157export const SETTLED_EFFECT_RETENTION = Duration.days(7);
158const PRUNE_BATCH_SIZE = 500;
159
141160export interface OrchestrationEffectV2 {
142161 readonly id: string;
143162 readonly commandId: CommandId;
......@@ -223,6 +242,8 @@ export interface EffectOutboxV2Shape {
223242 readonly workerId: string;
224243 readonly error: string;
225244 }) => Effect.Effect<boolean, EffectOutboxError>;
245 /** Deletes succeeded and cancelled rows older than `SETTLED_EFFECT_RETENTION`. Returns the count. */
246 readonly pruneSettled: Effect.Effect<number, EffectOutboxError>;
226247}
227248
228249export class EffectOutboxV2 extends Context.Service<EffectOutboxV2, EffectOutboxV2Shape>()(
......@@ -635,8 +656,48 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
635656 }).pipe(
636657 Effect.mapError((cause) => new EffectOutboxError({ operation: "fail", effectId, cause })),
637658 ),
659 pruneSettled: Effect.gen(function* () {
660 const cutoff = DateTime.formatIso(
661 DateTime.subtractDuration(yield* DateTime.now, SETTLED_EFFECT_RETENTION),
662 );
663 // Each batch is its own statement, so a large backlog never holds the
664 // write lock for long and writers can commit between batches.
665 const deleteBatch = sql<{ readonly effect_id: string }>`
666 DELETE FROM orchestration_v2_effect_outbox
667 WHERE rowid IN (
668 SELECT rowid
669 FROM orchestration_v2_effect_outbox
670 WHERE status IN ('succeeded', 'cancelled')
671 AND completed_at < ${cutoff}
672 LIMIT ${PRUNE_BATCH_SIZE}
673 )
674 RETURNING effect_id
675 `;
676 let pruned = 0;
677 while (true) {
678 const deleted = (yield* deleteBatch).length;
679 pruned += deleted;
680 if (deleted < PRUNE_BATCH_SIZE) return pruned;
681 yield* Effect.yieldNow;
682 }
683 }).pipe(
684 Effect.mapError((cause) => new EffectOutboxError({ operation: "prune-settled", cause })),
685 ),
638686 };
639687
640688 return service;
641689 }),
642690);
691
692/** Prunes settled effect rows once the server is active, then every hour. */
693export const pruneWorkerLive = Layer.effectDiscard(
694 Effect.gen(function* () {
695 const outbox = yield* EffectOutboxV2;
696 yield* forkParked(
697 outbox.pruneSettled.pipe(
698 Effect.catch((cause) => Effect.logWarning("Failed to prune settled effects", { cause })),
699 Effect.repeat(Schedule.spaced(Duration.hours(1))),
700 ),
701 );
702 }),
703);
apps/server/src/orchestration-v2/FoundationPersistence.test.ts+93
......@@ -27,10 +27,12 @@ import {
2727} from "@t3tools/contracts";
2828import * as DateTime from "effect/DateTime";
2929import * as Deferred from "effect/Deferred";
30import * as Duration from "effect/Duration";
3031import * as Effect from "effect/Effect";
3132import * as Fiber from "effect/Fiber";
3233import * as Layer from "effect/Layer";
3334import * as Option from "effect/Option";
35import * as Queue from "effect/Queue";
3436import * as Ref from "effect/Ref";
3537import * as Scheduler from "effect/Scheduler";
3638import * as Schema from "effect/Schema";
......@@ -65,6 +67,8 @@ const storesProvided = Layer.mergeAll(databaseLayer, eventStoreProvided, project
6567const eventSinkProvided = EventSink.layer.pipe(Layer.provide(storesProvided));
6668const effectOutboxProvided = EffectOutbox.layer.pipe(Layer.provide(databaseLayer));
6769const commandReceiptStoreProvided = CommandReceiptStore.layer.pipe(Layer.provide(databaseLayer));
70// Its own database, for tests that act on every row in the outbox table.
71const isolatedOutboxLayer = Layer.fresh(EffectOutbox.layer.pipe(Layer.provideMerge(databaseLayer)));
6872const projectionMaintenanceProvided = ProjectionMaintenance.layer.pipe(
6973 Layer.provide(storesProvided),
7074);
......@@ -2223,6 +2227,95 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => {
22232227 }),
22242228 );
22252229
2230 it.effect("prunes settled effects past retention in batches and keeps the rest", () =>
2231 Effect.gen(function* () {
2232 const outbox = yield* EffectOutbox.EffectOutboxV2;
2233 const sql = yield* SqlClient.SqlClient;
2234 const now = yield* DateTime.now;
2235 const insert = (
2236 prefix: string,
2237 count: number,
2238 status: EffectOutbox.OrchestrationEffectStatusV2,
2239 completedAgo: Duration.Duration | null,
2240 ) => {
2241 const completedAt =
2242 completedAgo === null
2243 ? null
2244 : DateTime.formatIso(DateTime.subtractDuration(now, completedAgo));
2245 const createdAt = DateTime.formatIso(now);
2246 return sql`
2247 WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${count})
2248 INSERT INTO orchestration_v2_effect_outbox (
2249 effect_id, command_id, thread_id, effect_type, payload_json, status,
2250 available_at, created_at, updated_at, completed_at
2251 )
2252 SELECT ${prefix} || i, 'command:prune', 'thread:prune', 'terminal.cleanup',
2253 '{"type":"terminal.cleanup"}', ${status}, ${createdAt}, ${createdAt}, ${createdAt},
2254 ${completedAt}
2255 FROM n
2256 `;
2257 };
2258 const old = Duration.sum(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1));
2259 const recent = Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1));
2260 // More expired rows than one delete batch.
2261 yield* insert("succeeded-old:", 1_201, "succeeded", old);
2262 yield* insert("cancelled-old:", 2, "cancelled", old);
2263 yield* insert("succeeded-recent:", 1, "succeeded", recent);
2264 yield* insert("failed-old:", 1, "failed", old);
2265 yield* insert("pending:", 1, "pending", null);
2266 yield* insert("running:", 1, "running", null);
2267
2268 assert.equal(yield* outbox.pruneSettled, 1_203);
2269
2270 const remaining = yield* sql<{ readonly effect_id: string }>`
2271 SELECT effect_id FROM orchestration_v2_effect_outbox ORDER BY effect_id
2272 `;
2273 assert.deepEqual(
2274 remaining.map((row) => row.effect_id),
2275 ["failed-old:1", "pending:1", "running:1", "succeeded-recent:1"],
2276 );
2277 assert.equal(yield* outbox.pruneSettled, 0);
2278 }).pipe(Effect.provide(isolatedOutboxLayer)),
2279 );
2280
2281 it.effect("prunes settled effects hourly from the layer-owned worker", () =>
2282 Effect.gen(function* () {
2283 const outbox = yield* EffectOutbox.EffectOutboxV2;
2284 const commandId = CommandId.make("command:foundation-prune-worker");
2285 const threadId = ThreadId.make("thread:foundation-prune-worker");
2286 const workerId = "prune-worker";
2287 const request = { type: "terminal.cleanup" } as const;
2288 yield* outbox.enqueue([{ id: "effect:prune-worker:done", commandId, threadId, request }]);
2289 yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2290 assert.isTrue(yield* outbox.succeed({ effectId: "effect:prune-worker:done", workerId }));
2291 yield* outbox.enqueue([{ id: "effect:prune-worker:pending", commandId, threadId, request }]);
2292 // Observe each run so the test waits for it instead of racing the clock.
2293 const runs = yield* Queue.unbounded<number>();
2294 const observed = EffectOutbox.EffectOutboxV2.of({
2295 ...outbox,
2296 pruneSettled: outbox.pruneSettled.pipe(Effect.tap((pruned) => Queue.offer(runs, pruned))),
2297 });
2298 const ids = Effect.map(outbox.listByCommandId(commandId), (rows) =>
2299 rows.map((row) => row.id).toSorted(),
2300 );
2301
2302 yield* TestClock.adjust(
2303 Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(30)),
2304 );
2305 yield* Layer.build(
2306 EffectOutbox.pruneWorkerLive.pipe(
2307 Layer.provide(Layer.succeed(EffectOutbox.EffectOutboxV2, observed)),
2308 ),
2309 );
2310 assert.equal(yield* Queue.take(runs), 0);
2311 assert.deepEqual(yield* ids, ["effect:prune-worker:done", "effect:prune-worker:pending"]);
2312
2313 yield* TestClock.adjust("1 hour");
2314 assert.equal(yield* Queue.take(runs), 1);
2315 assert.deepEqual(yield* ids, ["effect:prune-worker:pending"]);
2316 }).pipe(Effect.scoped, Effect.provide(isolatedOutboxLayer)),
2317 );
2318
22262319 it.effect("does not emit a SQL span for an empty safety claim", () =>
22272320 Effect.gen(function* () {
22282321 const outbox = yield* EffectOutbox.EffectOutboxV2;
apps/server/src/orchestration-v2/runtimeLayer.ts+5-1
......@@ -17,7 +17,10 @@ import { layer as checkpointRollbackServiceLayer } from "./CheckpointRollbackSer
1717import { layer as commandPolicyLayer } from "./CommandPolicy.ts";
1818import { layerFromApplicationReceipts as commandReceiptStoreLayer } from "./CommandReceiptStore.ts";
1919import { layer as contextHandoffServiceLayer } from "./ContextHandoffService.ts";
20import { layer as effectOutboxLayer } from "./EffectOutbox.ts";
20import {
21 layer as effectOutboxLayer,
22 pruneWorkerLive as effectOutboxPruneWorkerLive,
23} from "./EffectOutbox.ts";
2124import {
2225 executorLayer as effectExecutorLayer,
2326 layer as effectWorkerLayer,
......@@ -324,6 +327,7 @@ export const OrchestrationV2ProductionLayerLive = Layer.mergeAll(
324327 ),
325328 providerContinuationWorkerProvided,
326329 agentSessionImporterProvided,
330 effectOutboxPruneWorkerLive.pipe(Layer.provide(effectOutboxLayer)),
327331).pipe(
328332 Layer.provide(Scheduler.layer),
329333 Layer.provideMerge(OrchestrationEventInfrastructureLayerLive),
apps/server/src/pullRequest/PullRequestReadCache.test.ts+89
......@@ -1,18 +1,107 @@
11import { assert, it } from "@effect/vitest";
22import * as NodeServices from "@effect/platform-node/NodeServices";
33import { PullRequestOperationError } from "@t3tools/contracts";
4import * as DateTime from "effect/DateTime";
45import * as Deferred from "effect/Deferred";
6import * as Duration from "effect/Duration";
57import * as Effect from "effect/Effect";
68import * as Fiber from "effect/Fiber";
79import * as FileSystem from "effect/FileSystem";
10import * as Layer from "effect/Layer";
11import * as Path from "effect/Path";
12import * as Queue from "effect/Queue";
813import * as TestClock from "effect/testing/TestClock";
914import * as KeyValueStore from "effect/persistence/KeyValueStore";
15import * as ServerConfig from "../config.ts";
1016import * as PullRequestReadCache from "./PullRequestReadCache.ts";
1117
1218const cacheLayer = (directory: string) =>
1319 PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory)));
1420
21/** Sets every file's mtime to `age` before the current test clock time. */
22const ageFiles = (directory: string, names: ReadonlyArray<string>, age: Duration.Duration) =>
23 Effect.gen(function* () {
24 const fs = yield* FileSystem.FileSystem;
25 const path = yield* Path.Path;
26 const mtime = DateTime.toDateUtc(DateTime.subtractDuration(yield* DateTime.now, age));
27 for (const name of names) yield* fs.utimes(path.join(directory, name), mtime, mtime);
28 });
29
1530it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
31 it.effect("prunes entry files written before the max age and keeps fresh ones", () =>
32 Effect.gen(function* () {
33 const fs = yield* FileSystem.FileSystem;
34 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
35 yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z")));
36 const cache = yield* cacheLayer(directory);
37 yield* cache.get("old", Effect.succeed("old"), ["pr"]);
38 yield* cache.invalidate("pr");
39 const [stale, revisions] = (yield* fs.readDirectory(directory)).toSorted();
40 assert.strictEqual(revisions, "revisions");
41 yield* cache.get("fresh", Effect.succeed("fresh"));
42 const fresh = (yield* fs.readDirectory(directory)).find(
43 (name) => name !== stale && name !== revisions,
44 );
45 const unrelated = "unrelated.json";
46 yield* fs.writeFileString(`${directory}/${unrelated}`, "{}");
47 const justExpired = Duration.sum(
48 PullRequestReadCache.ENTRY_FILE_MAX_AGE,
49 Duration.seconds(1),
50 );
51 yield* ageFiles(directory, [stale!, revisions!, unrelated], justExpired);
52 yield* ageFiles(directory, [fresh!], PullRequestReadCache.ENTRY_FILE_MAX_AGE);
53
54 yield* PullRequestReadCache.pruneExpiredEntryFiles(directory);
55
56 assert.deepStrictEqual(
57 (yield* fs.readDirectory(directory)).toSorted(),
58 [fresh!, revisions!, unrelated].toSorted(),
59 );
60 }),
61 );
62
63 it.effect("sweeps the cache directory every hour once the layer is built", () =>
64 Effect.gen(function* () {
65 const fs = yield* FileSystem.FileSystem;
66 const { providerStatusCacheDir } = yield* ServerConfig.ServerConfig;
67 const directory = `${providerStatusCacheDir}/pull-requests`;
68 yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z")));
69 yield* (yield* cacheLayer(directory)).get("summary", Effect.succeed("cached"));
70 const [entry] = yield* fs.readDirectory(directory);
71 // Within the max age at the first two sweeps, past it at the third.
72 yield* ageFiles(
73 directory,
74 [entry!],
75 Duration.subtract(PullRequestReadCache.ENTRY_FILE_MAX_AGE, Duration.minutes(90)),
76 );
77 // A sweep's last file operation is the entry's stat when it keeps the
78 // entry, and its removal otherwise. Waiting for it means the sweep has
79 // finished before the clock moves.
80 const operations = yield* Queue.unbounded<string>();
81 const observedFs = FileSystem.FileSystem.of({
82 ...fs,
83 stat: (path) => fs.stat(path).pipe(Effect.tap(() => Queue.offer(operations, "stat"))),
84 remove: (path, options) =>
85 fs.remove(path, options).pipe(Effect.tap(() => Queue.offer(operations, "remove"))),
86 });
87 yield* Layer.build(
88 PullRequestReadCache.layer.pipe(
89 Layer.provide(Layer.succeed(FileSystem.FileSystem, observedFs)),
90 ),
91 );
92 assert.strictEqual(yield* Queue.take(operations), "stat");
93 yield* TestClock.adjust("1 hour");
94 assert.strictEqual(yield* Queue.take(operations), "stat");
95 assert.deepStrictEqual(yield* fs.readDirectory(directory), [entry]);
96 yield* TestClock.adjust("1 hour");
97 assert.deepStrictEqual(yield* Queue.takeN(operations, 2), ["stat", "remove"]);
98 assert.deepStrictEqual(yield* fs.readDirectory(directory), []);
99 }).pipe(
100 Effect.scoped,
101 Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-pr-cache-layer-" })),
102 ),
103 );
104
16105 it.effect("reuses files after restart and respects the original expiry", () =>
17106 Effect.gen(function* () {
18107 const fs = yield* FileSystem.FileSystem;
apps/server/src/pullRequest/PullRequestReadCache.ts+62-4
......@@ -9,8 +9,11 @@ import * as Context from "effect/Context";
99import * as Duration from "effect/Duration";
1010import * as Effect from "effect/Effect";
1111import * as Exit from "effect/Exit";
12import * as FileSystem from "effect/FileSystem";
1213import * as Layer from "effect/Layer";
14import * as Option from "effect/Option";
1315import * as Path from "effect/Path";
16import * as Schedule from "effect/Schedule";
1417import * as Schema from "effect/Schema";
1518import * as Semaphore from "effect/Semaphore";
1619import * as KeyValueStore from "effect/persistence/KeyValueStore";
......@@ -18,8 +21,18 @@ import * as Persistable from "effect/persistence/Persistable";
1821import * as PersistedCache from "effect/persistence/PersistedCache";
1922import * as Persistence from "effect/persistence/Persistence";
2023import * as ServerConfig from "../config.ts";
24import { forkParked } from "../serverActivation.ts";
2125
2226const CONCURRENT_READS = 512;
27// Persistence prefixes every entry key with the store id, so each entry file
28// name starts with it. The scope `revisions` file does not.
29const STORE_ID = "pr-v2";
30/**
31 * Entries expire a minute after they are written, and a write sets the file's
32 * mtime. A file untouched for a day is long expired; the day only leaves slack
33 * for clock changes. Pruning a live entry would cost one refetch.
34 */
35export const ENTRY_FILE_MAX_AGE = Duration.days(1);
2336type ReadError = PullRequestOperationError | PullRequestUnavailableError;
2437const revisionCodec = Schema.fromJsonString(
2538 Schema.Record(
......@@ -97,7 +110,7 @@ export const make = Effect.gen(function* () {
97110 })),
98111 ),
99112 {
100 storeId: "pr-v2",
113 storeId: STORE_ID,
101114 timeToLive,
102115 inMemoryTTL: timeToLive,
103116 inMemoryCapacity: CONCURRENT_READS,
......@@ -162,15 +175,60 @@ export const make = Effect.gen(function* () {
162175 });
163176});
164177
178/**
179 * Deletes entry files in `directory` not written within `ENTRY_FILE_MAX_AGE`.
180 * The persisted cache drops an expired entry only when it is read again, so
181 * files for PRs nobody reopens would otherwise stay forever.
182 */
183export const pruneExpiredEntryFiles = Effect.fn("PullRequestReadCache.pruneExpiredEntryFiles")(
184 function* (directory: string) {
185 const fileSystem = yield* FileSystem.FileSystem;
186 const path = yield* Path.Path;
187 const cutoff = (yield* Clock.currentTimeMillis) - Duration.toMillis(ENTRY_FILE_MAX_AGE);
188 const entries = (yield* fileSystem.readDirectory(directory)).filter((name) =>
189 name.startsWith(STORE_ID),
190 );
191 // One file at a time, and `partition` visits every file, so one locked file
192 // does not stop the sweep.
193 const [, failures] = yield* Effect.partition(entries, (name) => {
194 const entryPath = path.join(directory, name);
195 return fileSystem.stat(entryPath).pipe(
196 Effect.flatMap((info) =>
197 Option.exists(info.mtime, (mtime) => mtime.getTime() < cutoff)
198 ? fileSystem.remove(entryPath)
199 : Effect.void,
200 ),
201 Effect.catchReason("PlatformError", "NotFound", () => Effect.void),
202 );
203 });
204 if (failures.length > 0) {
205 yield* Effect.logWarning("Failed to prune some PR cache files", {
206 failed: failures.length,
207 cause: failures[0],
208 });
209 }
210 },
211);
212
165213export const layer = Layer.unwrap(
166214 Effect.gen(function* () {
167215 const config = yield* ServerConfig.ServerConfig;
168216 const path = yield* Path.Path;
217 const directory = path.join(config.providerStatusCacheDir, "pull-requests");
169218 return Layer.effect(PullRequestReadCache, make).pipe(
170219 Layer.provide(
171 KeyValueStore.layerFileSystem(
172 path.join(config.providerStatusCacheDir, "pull-requests"),
173 ).pipe(
220 KeyValueStore.layerFileSystem(directory).pipe(
221 // Prunes once the server is active, then every hour.
222 Layer.tap(() =>
223 forkParked(
224 pruneExpiredEntryFiles(directory).pipe(
225 Effect.catch((cause) =>
226 Effect.logWarning("Failed to prune PR cache files", { cause }),
227 ),
228 Effect.repeat(Schedule.spaced(Duration.hours(1))),
229 ),
230 ),
231 ),
174232 Layer.catch(() =>
175233 Layer.effectDiscard(
176234 Effect.logWarning("PR cache directory unavailable; using memory cache"),