diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index a7b035629dda8e321d3c6384adfc7ff117278662..db5ee4a85708bf344c0551654125ef4382f432b7 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -16,13 +16,17 @@ import { import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; +import { forkParked } from "../serverActivation.ts"; + export const OrchestrationEffectRequestV2 = Schema.Union([ Schema.Struct({ type: Schema.Literal("provider-runtime.continue"), @@ -138,6 +142,21 @@ export const OrchestrationEffectStatusV2 = Schema.Literals([ ]); export type OrchestrationEffectStatusV2 = typeof OrchestrationEffectStatusV2.Type; +/** + * How long a succeeded or cancelled effect row is kept after it completes. + * Claiming, recovery, the orchestrator and storage cleanup only ask whether a + * row is pending, running or failed, so a missing succeeded or cancelled row + * reads the same as a present one. Its one other use is deduplication: `enqueue` + * ignores an id that already exists, and ids are deterministic per command or + * run. Command retries are deduplicated by their receipts first, and a run's + * effects are re-enqueued only around one server restart, so a week is far + * longer than either and leaves recent history for debugging. + * Failed rows are kept: storage cleanup keeps a deleted thread's worktree while + * any of its effects failed. + */ +export const SETTLED_EFFECT_RETENTION = Duration.days(7); +const PRUNE_BATCH_SIZE = 500; + export interface OrchestrationEffectV2 { readonly id: string; readonly commandId: CommandId; @@ -223,6 +242,8 @@ export interface EffectOutboxV2Shape { readonly workerId: string; readonly error: string; }) => Effect.Effect; + /** Deletes succeeded and cancelled rows older than `SETTLED_EFFECT_RETENTION`. Returns the count. */ + readonly pruneSettled: Effect.Effect; } export class EffectOutboxV2 extends Context.Service()( @@ -635,8 +656,48 @@ export const layer: Layer.Layer = La }).pipe( Effect.mapError((cause) => new EffectOutboxError({ operation: "fail", effectId, cause })), ), + pruneSettled: Effect.gen(function* () { + const cutoff = DateTime.formatIso( + DateTime.subtractDuration(yield* DateTime.now, SETTLED_EFFECT_RETENTION), + ); + // Each batch is its own statement, so a large backlog never holds the + // write lock for long and writers can commit between batches. + const deleteBatch = sql<{ readonly effect_id: string }>` + DELETE FROM orchestration_v2_effect_outbox + WHERE rowid IN ( + SELECT rowid + FROM orchestration_v2_effect_outbox + WHERE status IN ('succeeded', 'cancelled') + AND completed_at < ${cutoff} + LIMIT ${PRUNE_BATCH_SIZE} + ) + RETURNING effect_id + `; + let pruned = 0; + while (true) { + const deleted = (yield* deleteBatch).length; + pruned += deleted; + if (deleted < PRUNE_BATCH_SIZE) return pruned; + yield* Effect.yieldNow; + } + }).pipe( + Effect.mapError((cause) => new EffectOutboxError({ operation: "prune-settled", cause })), + ), }; return service; }), ); + +/** Prunes settled effect rows once the server is active, then every hour. */ +export const pruneWorkerLive = Layer.effectDiscard( + Effect.gen(function* () { + const outbox = yield* EffectOutboxV2; + yield* forkParked( + outbox.pruneSettled.pipe( + Effect.catch((cause) => Effect.logWarning("Failed to prune settled effects", { cause })), + Effect.repeat(Schedule.spaced(Duration.hours(1))), + ), + ); + }), +); diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 03f4886ce1713e9ef807ffac101a54b7382a114d..221d0e6f7d78a636c99c72575ef0f11d88c4c918 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -27,10 +27,12 @@ import { } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Scheduler from "effect/Scheduler"; import * as Schema from "effect/Schema"; @@ -65,6 +67,8 @@ const storesProvided = Layer.mergeAll(databaseLayer, eventStoreProvided, project const eventSinkProvided = EventSink.layer.pipe(Layer.provide(storesProvided)); const effectOutboxProvided = EffectOutbox.layer.pipe(Layer.provide(databaseLayer)); const commandReceiptStoreProvided = CommandReceiptStore.layer.pipe(Layer.provide(databaseLayer)); +// Its own database, for tests that act on every row in the outbox table. +const isolatedOutboxLayer = Layer.fresh(EffectOutbox.layer.pipe(Layer.provideMerge(databaseLayer))); const projectionMaintenanceProvided = ProjectionMaintenance.layer.pipe( Layer.provide(storesProvided), ); @@ -2223,6 +2227,95 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { }), ); + it.effect("prunes settled effects past retention in batches and keeps the rest", () => + Effect.gen(function* () { + const outbox = yield* EffectOutbox.EffectOutboxV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const insert = ( + prefix: string, + count: number, + status: EffectOutbox.OrchestrationEffectStatusV2, + completedAgo: Duration.Duration | null, + ) => { + const completedAt = + completedAgo === null + ? null + : DateTime.formatIso(DateTime.subtractDuration(now, completedAgo)); + const createdAt = DateTime.formatIso(now); + return sql` + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${count}) + INSERT INTO orchestration_v2_effect_outbox ( + effect_id, command_id, thread_id, effect_type, payload_json, status, + available_at, created_at, updated_at, completed_at + ) + SELECT ${prefix} || i, 'command:prune', 'thread:prune', 'terminal.cleanup', + '{"type":"terminal.cleanup"}', ${status}, ${createdAt}, ${createdAt}, ${createdAt}, + ${completedAt} + FROM n + `; + }; + const old = Duration.sum(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1)); + const recent = Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1)); + // More expired rows than one delete batch. + yield* insert("succeeded-old:", 1_201, "succeeded", old); + yield* insert("cancelled-old:", 2, "cancelled", old); + yield* insert("succeeded-recent:", 1, "succeeded", recent); + yield* insert("failed-old:", 1, "failed", old); + yield* insert("pending:", 1, "pending", null); + yield* insert("running:", 1, "running", null); + + assert.equal(yield* outbox.pruneSettled, 1_203); + + const remaining = yield* sql<{ readonly effect_id: string }>` + SELECT effect_id FROM orchestration_v2_effect_outbox ORDER BY effect_id + `; + assert.deepEqual( + remaining.map((row) => row.effect_id), + ["failed-old:1", "pending:1", "running:1", "succeeded-recent:1"], + ); + assert.equal(yield* outbox.pruneSettled, 0); + }).pipe(Effect.provide(isolatedOutboxLayer)), + ); + + it.effect("prunes settled effects hourly from the layer-owned worker", () => + Effect.gen(function* () { + const outbox = yield* EffectOutbox.EffectOutboxV2; + const commandId = CommandId.make("command:foundation-prune-worker"); + const threadId = ThreadId.make("thread:foundation-prune-worker"); + const workerId = "prune-worker"; + const request = { type: "terminal.cleanup" } as const; + yield* outbox.enqueue([{ id: "effect:prune-worker:done", commandId, threadId, request }]); + yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 }); + assert.isTrue(yield* outbox.succeed({ effectId: "effect:prune-worker:done", workerId })); + yield* outbox.enqueue([{ id: "effect:prune-worker:pending", commandId, threadId, request }]); + // Observe each run so the test waits for it instead of racing the clock. + const runs = yield* Queue.unbounded(); + const observed = EffectOutbox.EffectOutboxV2.of({ + ...outbox, + pruneSettled: outbox.pruneSettled.pipe(Effect.tap((pruned) => Queue.offer(runs, pruned))), + }); + const ids = Effect.map(outbox.listByCommandId(commandId), (rows) => + rows.map((row) => row.id).toSorted(), + ); + + yield* TestClock.adjust( + Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(30)), + ); + yield* Layer.build( + EffectOutbox.pruneWorkerLive.pipe( + Layer.provide(Layer.succeed(EffectOutbox.EffectOutboxV2, observed)), + ), + ); + assert.equal(yield* Queue.take(runs), 0); + assert.deepEqual(yield* ids, ["effect:prune-worker:done", "effect:prune-worker:pending"]); + + yield* TestClock.adjust("1 hour"); + assert.equal(yield* Queue.take(runs), 1); + assert.deepEqual(yield* ids, ["effect:prune-worker:pending"]); + }).pipe(Effect.scoped, Effect.provide(isolatedOutboxLayer)), + ); + it.effect("does not emit a SQL span for an empty safety claim", () => Effect.gen(function* () { const outbox = yield* EffectOutbox.EffectOutboxV2; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index b4e262d08a7052466a91b07ee081497a7edd10f4..69be6f5e136c870a1913ee18f2a6bc7c650def73 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -17,7 +17,10 @@ import { layer as checkpointRollbackServiceLayer } from "./CheckpointRollbackSer import { layer as commandPolicyLayer } from "./CommandPolicy.ts"; import { layerFromApplicationReceipts as commandReceiptStoreLayer } from "./CommandReceiptStore.ts"; import { layer as contextHandoffServiceLayer } from "./ContextHandoffService.ts"; -import { layer as effectOutboxLayer } from "./EffectOutbox.ts"; +import { + layer as effectOutboxLayer, + pruneWorkerLive as effectOutboxPruneWorkerLive, +} from "./EffectOutbox.ts"; import { executorLayer as effectExecutorLayer, layer as effectWorkerLayer, @@ -324,6 +327,7 @@ export const OrchestrationV2ProductionLayerLive = Layer.mergeAll( ), providerContinuationWorkerProvided, agentSessionImporterProvided, + effectOutboxPruneWorkerLive.pipe(Layer.provide(effectOutboxLayer)), ).pipe( Layer.provide(Scheduler.layer), Layer.provideMerge(OrchestrationEventInfrastructureLayerLive), diff --git a/apps/server/src/pullRequest/PullRequestReadCache.test.ts b/apps/server/src/pullRequest/PullRequestReadCache.test.ts index c4df383e4b1a5d28853f33bb98b80d6fd3ae5cc6..a6a14dbae9cabd7d6b8d1ddff5e98058d9edd3bc 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.test.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.test.ts @@ -1,18 +1,107 @@ import { assert, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { PullRequestOperationError } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as TestClock from "effect/testing/TestClock"; import * as KeyValueStore from "effect/persistence/KeyValueStore"; +import * as ServerConfig from "../config.ts"; import * as PullRequestReadCache from "./PullRequestReadCache.ts"; const cacheLayer = (directory: string) => PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory))); +/** Sets every file's mtime to `age` before the current test clock time. */ +const ageFiles = (directory: string, names: ReadonlyArray, age: Duration.Duration) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const mtime = DateTime.toDateUtc(DateTime.subtractDuration(yield* DateTime.now, age)); + for (const name of names) yield* fs.utimes(path.join(directory, name), mtime, mtime); + }); + it.layer(NodeServices.layer)("PR filesystem cache", (it) => { + it.effect("prunes entry files written before the max age and keeps fresh ones", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" }); + yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z"))); + const cache = yield* cacheLayer(directory); + yield* cache.get("old", Effect.succeed("old"), ["pr"]); + yield* cache.invalidate("pr"); + const [stale, revisions] = (yield* fs.readDirectory(directory)).toSorted(); + assert.strictEqual(revisions, "revisions"); + yield* cache.get("fresh", Effect.succeed("fresh")); + const fresh = (yield* fs.readDirectory(directory)).find( + (name) => name !== stale && name !== revisions, + ); + const unrelated = "unrelated.json"; + yield* fs.writeFileString(`${directory}/${unrelated}`, "{}"); + const justExpired = Duration.sum( + PullRequestReadCache.ENTRY_FILE_MAX_AGE, + Duration.seconds(1), + ); + yield* ageFiles(directory, [stale!, revisions!, unrelated], justExpired); + yield* ageFiles(directory, [fresh!], PullRequestReadCache.ENTRY_FILE_MAX_AGE); + + yield* PullRequestReadCache.pruneExpiredEntryFiles(directory); + + assert.deepStrictEqual( + (yield* fs.readDirectory(directory)).toSorted(), + [fresh!, revisions!, unrelated].toSorted(), + ); + }), + ); + + it.effect("sweeps the cache directory every hour once the layer is built", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const { providerStatusCacheDir } = yield* ServerConfig.ServerConfig; + const directory = `${providerStatusCacheDir}/pull-requests`; + yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z"))); + yield* (yield* cacheLayer(directory)).get("summary", Effect.succeed("cached")); + const [entry] = yield* fs.readDirectory(directory); + // Within the max age at the first two sweeps, past it at the third. + yield* ageFiles( + directory, + [entry!], + Duration.subtract(PullRequestReadCache.ENTRY_FILE_MAX_AGE, Duration.minutes(90)), + ); + // A sweep's last file operation is the entry's stat when it keeps the + // entry, and its removal otherwise. Waiting for it means the sweep has + // finished before the clock moves. + const operations = yield* Queue.unbounded(); + const observedFs = FileSystem.FileSystem.of({ + ...fs, + stat: (path) => fs.stat(path).pipe(Effect.tap(() => Queue.offer(operations, "stat"))), + remove: (path, options) => + fs.remove(path, options).pipe(Effect.tap(() => Queue.offer(operations, "remove"))), + }); + yield* Layer.build( + PullRequestReadCache.layer.pipe( + Layer.provide(Layer.succeed(FileSystem.FileSystem, observedFs)), + ), + ); + assert.strictEqual(yield* Queue.take(operations), "stat"); + yield* TestClock.adjust("1 hour"); + assert.strictEqual(yield* Queue.take(operations), "stat"); + assert.deepStrictEqual(yield* fs.readDirectory(directory), [entry]); + yield* TestClock.adjust("1 hour"); + assert.deepStrictEqual(yield* Queue.takeN(operations, 2), ["stat", "remove"]); + assert.deepStrictEqual(yield* fs.readDirectory(directory), []); + }).pipe( + Effect.scoped, + Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-pr-cache-layer-" })), + ), + ); + it.effect("reuses files after restart and respects the original expiry", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; diff --git a/apps/server/src/pullRequest/PullRequestReadCache.ts b/apps/server/src/pullRequest/PullRequestReadCache.ts index b7188d514f82118715f7571d6d5cc87de8d39f83..a691898769be43101636062e00e5212f66da03de 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.ts @@ -9,8 +9,11 @@ import * as Context from "effect/Context"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as KeyValueStore from "effect/persistence/KeyValueStore"; @@ -18,8 +21,18 @@ import * as Persistable from "effect/persistence/Persistable"; import * as PersistedCache from "effect/persistence/PersistedCache"; import * as Persistence from "effect/persistence/Persistence"; import * as ServerConfig from "../config.ts"; +import { forkParked } from "../serverActivation.ts"; const CONCURRENT_READS = 512; +// Persistence prefixes every entry key with the store id, so each entry file +// name starts with it. The scope `revisions` file does not. +const STORE_ID = "pr-v2"; +/** + * Entries expire a minute after they are written, and a write sets the file's + * mtime. A file untouched for a day is long expired; the day only leaves slack + * for clock changes. Pruning a live entry would cost one refetch. + */ +export const ENTRY_FILE_MAX_AGE = Duration.days(1); type ReadError = PullRequestOperationError | PullRequestUnavailableError; const revisionCodec = Schema.fromJsonString( Schema.Record( @@ -97,7 +110,7 @@ export const make = Effect.gen(function* () { })), ), { - storeId: "pr-v2", + storeId: STORE_ID, timeToLive, inMemoryTTL: timeToLive, inMemoryCapacity: CONCURRENT_READS, @@ -162,15 +175,60 @@ export const make = Effect.gen(function* () { }); }); +/** + * Deletes entry files in `directory` not written within `ENTRY_FILE_MAX_AGE`. + * The persisted cache drops an expired entry only when it is read again, so + * files for PRs nobody reopens would otherwise stay forever. + */ +export const pruneExpiredEntryFiles = Effect.fn("PullRequestReadCache.pruneExpiredEntryFiles")( + function* (directory: string) { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cutoff = (yield* Clock.currentTimeMillis) - Duration.toMillis(ENTRY_FILE_MAX_AGE); + const entries = (yield* fileSystem.readDirectory(directory)).filter((name) => + name.startsWith(STORE_ID), + ); + // One file at a time, and `partition` visits every file, so one locked file + // does not stop the sweep. + const [, failures] = yield* Effect.partition(entries, (name) => { + const entryPath = path.join(directory, name); + return fileSystem.stat(entryPath).pipe( + Effect.flatMap((info) => + Option.exists(info.mtime, (mtime) => mtime.getTime() < cutoff) + ? fileSystem.remove(entryPath) + : Effect.void, + ), + Effect.catchReason("PlatformError", "NotFound", () => Effect.void), + ); + }); + if (failures.length > 0) { + yield* Effect.logWarning("Failed to prune some PR cache files", { + failed: failures.length, + cause: failures[0], + }); + } + }, +); + export const layer = Layer.unwrap( Effect.gen(function* () { const config = yield* ServerConfig.ServerConfig; const path = yield* Path.Path; + const directory = path.join(config.providerStatusCacheDir, "pull-requests"); return Layer.effect(PullRequestReadCache, make).pipe( Layer.provide( - KeyValueStore.layerFileSystem( - path.join(config.providerStatusCacheDir, "pull-requests"), - ).pipe( + KeyValueStore.layerFileSystem(directory).pipe( + // Prunes once the server is active, then every hour. + Layer.tap(() => + forkParked( + pruneExpiredEntryFiles(directory).pipe( + Effect.catch((cause) => + Effect.logWarning("Failed to prune PR cache files", { cause }), + ), + Effect.repeat(Schedule.spaced(Duration.hours(1))), + ), + ), + ), Layer.catch(() => Layer.effectDiscard( Effect.logWarning("PR cache directory unavailable; using memory cache"),