perf(server): stop decoding unrelated events during startup (#12846)

Co-authored-by: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: 3 people

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -0,0 +1,167 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { assert, it } from "@effect/vitest";
+import {
+  CommandId,
+  EventId,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  type OrchestrationEvent,
+} from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
+import * as Layer from "effect/Layer";
+import * as Path from "effect/Path";
+
+import { ServerConfig } from "../../config.ts";
+import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
+import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
+import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
+import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";
+import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
+import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
+
+const TestLayer = OrchestrationProjectionPipelineLive.pipe(
+  Layer.provideMerge(OrchestrationEventStoreLive),
+  Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-projection-cleanup-" })),
+  Layer.provideMerge(SqlitePersistenceMemory),
+  Layer.provideMerge(NodeServices.layer),
+);
+
+const CLEANUP_PROJECTOR = "projection.attachment-cleanup";
+
+const exists = (filePath: string) =>
+  Effect.gen(function* () {
+    const fileSystem = yield* FileSystem.FileSystem;
+    return (yield* Effect.result(fileSystem.stat(filePath)))._tag === "Success";
+  });
+
+it.layer(TestLayer)("OrchestrationProjectionPipeline attachment cleanup bootstrap", (it) => {
+  it.effect(
+    "advances the cleanup cursor to the latest event, dedupes reverts, and holds on failure",
+    () =>
+      Effect.gen(function* () {
+        const projectionPipeline = yield* OrchestrationProjectionPipeline;
+        const eventStore = yield* OrchestrationEventStore;
+        const projectionState = yield* ProjectionStateRepository;
+        const fileSystem = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const { attachmentsDir } = yield* ServerConfig;
+        const now = "2026-01-01T00:00:00.000Z";
+        const projectId = ProjectId.make("project-cleanup");
+        const revertedThreadId = ThreadId.make("thread-cleanup-revert");
+        const deletedThreadId = ThreadId.make("thread-cleanup-delete");
+        const revertedAttachmentPath = path.join(
+          attachmentsDir,
+          "thread-cleanup-revert-00000000-0000-4000-8000-000000000001.png",
+        );
+        const deletedAttachmentPath = path.join(
+          attachmentsDir,
+          "thread-cleanup-delete-00000000-0000-4000-8000-000000000001.png",
+        );
+        const cleanupCursor = projectionState
+          .listAll()
+          .pipe(
+            Effect.map(
+              (states) =>
+                states.find((state) => state.projector === CLEANUP_PROJECTOR)?.lastAppliedSequence,
+            ),
+          );
+
+        let eventOrdinal = 0;
+        const append = (
+          event: Pick<OrchestrationEvent, "type" | "aggregateKind" | "aggregateId" | "payload">,
+        ) => {
+          eventOrdinal += 1;
+          return eventStore.append({
+            ...event,
+            eventId: EventId.make(`evt-cleanup-${eventOrdinal}`),
+            occurredAt: now,
+            commandId: CommandId.make(`cmd-cleanup-${eventOrdinal}`),
+            causationEventId: null,
+            correlationId: CommandId.make(`cmd-cleanup-${eventOrdinal}`),
+            metadata: {},
+          } as Omit<OrchestrationEvent, "sequence">);
+        };
+        const appendThreadCreated = (threadId: ThreadId) =>
+          append({
+            type: "thread.created",
+            aggregateKind: "thread",
+            aggregateId: threadId,
+            payload: {
+              threadId,
+              projectId,
+              title: "Cleanup thread",
+              modelSelection: {
+                instanceId: ProviderInstanceId.make("codex"),
+                model: "gpt-5-codex",
+              },
+              runtimeMode: "full-access",
+              branch: null,
+              worktreePath: null,
+              createdAt: now,
+              updatedAt: now,
+            },
+          });
+        const appendReverted = (turnCount: number) =>
+          append({
+            type: "thread.reverted",
+            aggregateKind: "thread",
+            aggregateId: revertedThreadId,
+            payload: { threadId: revertedThreadId, turnCount },
+          });
+
+        yield* append({
+          type: "project.created",
+          aggregateKind: "project",
+          aggregateId: projectId,
+          payload: {
+            projectId,
+            title: "Cleanup project",
+            workspaceRoot: "/tmp/project-cleanup",
+            defaultModelSelection: null,
+            scripts: [],
+            createdAt: now,
+            updatedAt: now,
+          },
+        });
+        yield* appendThreadCreated(revertedThreadId);
+        yield* appendThreadCreated(deletedThreadId);
+        yield* fileSystem.makeDirectory(attachmentsDir, { recursive: true });
+        yield* fileSystem.writeFileString(revertedAttachmentPath, "stale");
+        // Two reverts for one thread dedupe into a single prune.
+        yield* appendReverted(2);
+        yield* appendReverted(1);
+        // The latest event is neither a revert nor a delete, yet the cursor must land on it.
+        const latest = yield* append({
+          type: "thread.archived",
+          aggregateKind: "thread",
+          aggregateId: deletedThreadId,
+          payload: { threadId: deletedThreadId, archivedAt: now, updatedAt: now },
+        });
+
+        yield* projectionPipeline.bootstrap;
+        assert.isFalse(yield* exists(revertedAttachmentPath));
+        assert.equal(yield* cleanupCursor, latest.sequence);
+
+        // A cleanup that cannot remove its files leaves the cursor behind for the next bootstrap.
+        yield* fileSystem.makeDirectory(deletedAttachmentPath);
+        yield* fileSystem.writeFileString(path.join(deletedAttachmentPath, "keep.txt"), "keep");
+        yield* append({
+          type: "thread.deleted",
+          aggregateKind: "thread",
+          aggregateId: deletedThreadId,
+          payload: { threadId: deletedThreadId, deletedAt: now },
+        });
+        yield* projectionPipeline.bootstrap;
+        assert.isTrue(yield* exists(deletedAttachmentPath));
+        assert.equal(yield* cleanupCursor, latest.sequence);
+
+        yield* fileSystem.remove(deletedAttachmentPath, { recursive: true });
+        yield* fileSystem.writeFileString(deletedAttachmentPath, "retry");
+        yield* projectionPipeline.bootstrap;
+        assert.isFalse(yield* exists(deletedAttachmentPath));
+        assert.equal(yield* cleanupCursor, latest.sequence + 1);
+      }),
+  );
+});

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -1,6 +1,8 @@
 import {
   ApprovalRequestId,
+  IsoDateTime,
   isImportedAgentSessionMessageId,
+  NonNegativeInt,
   UserInputAttachmentAnswerPayload,
   type ChatAttachment,
   ThreadId,
@@ -18,12 +20,17 @@ import * as Path from "effect/Path";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
+import * as SqlSchema from "effect/unstable/sql/SqlSchema";
 import {
   legacyThreadPullRequestKey,
   threadPullRequestKeysEqual,
 } from "@t3tools/shared/threadPullRequests";
 
-import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts";
+import {
+  toPersistenceDecodeError,
+  toPersistenceSqlError,
+  type ProjectionRepositoryError,
+} from "../../persistence/Errors.ts";
 import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
 import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts";
 import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts";
@@ -117,6 +124,13 @@ interface AttachmentSideEffects {
   readonly prunedThreadRelativePaths: Map<string, Set<string>>;
 }
 
+const AttachmentCleanupReplayRow = Schema.Struct({
+  sequence: NonNegativeInt,
+  occurredAt: IsoDateTime,
+  cleanupType: Schema.NullOr(Schema.Literals(["thread.reverted", "thread.deleted"])),
+  threadId: Schema.NullOr(ThreadId),
+});
+
 const materializeAttachmentsForProjection = Effect.fn("materializeAttachmentsForProjection")(
   (input: { readonly attachments: ReadonlyArray<ChatAttachment> }) =>
     Effect.succeed(input.attachments.length === 0 ? [] : input.attachments),
@@ -498,6 +512,49 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
     const path = yield* Path.Path;
     const serverConfig = yield* ServerConfig;
 
+    // Attachment cleanup has a separate retry cursor because its filesystem work
+    // runs after projection transactions commit. Normal runtime events therefore
+    // accumulate behind that cursor. Bootstrap only needs the latest event for
+    // the new cursor plus revert/delete metadata; reading full event payloads here
+    // made restart cost grow with every event since the previous restart.
+    const listAttachmentCleanupReplayRows = SqlSchema.findAll({
+      Request: Schema.Struct({ sequenceExclusive: NonNegativeInt }),
+      Result: AttachmentCleanupReplayRow,
+      execute: ({ sequenceExclusive }) => sql`
+        SELECT
+          sequence,
+          occurred_at AS "occurredAt",
+          CASE
+            WHEN aggregate_kind = 'thread'
+              AND event_type IN ('thread.reverted', 'thread.deleted')
+            THEN event_type
+            ELSE NULL
+          END AS "cleanupType",
+          CASE
+            WHEN aggregate_kind = 'thread'
+              AND event_type IN ('thread.reverted', 'thread.deleted')
+            THEN stream_id
+            ELSE NULL
+          END AS "threadId"
+        -- Force the sequence rowid range. With this OR predicate SQLite can
+        -- otherwise choose the aggregate index and scan every thread event.
+        FROM orchestration_events NOT INDEXED
+        WHERE sequence > ${sequenceExclusive}
+          AND (
+            sequence = (
+              SELECT MAX(sequence)
+              FROM orchestration_events
+              WHERE sequence > ${sequenceExclusive}
+            )
+            OR (
+              aggregate_kind = 'thread'
+              AND event_type IN ('thread.reverted', 'thread.deleted')
+            )
+          )
+        ORDER BY sequence ASC
+      `,
+    });
+
     const applyProjectsProjection: ProjectorDefinition["apply"] = Effect.fn(
       "applyProjectsProjection",
     )(function* (event, _attachmentSideEffects) {
@@ -1962,7 +2019,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
     ];
 
     const applyAttachmentSideEffects = Effect.fn("applyAttachmentSideEffects")(
-      function* (event: OrchestrationEvent, sideEffects: AttachmentSideEffects) {
+      function* (
+        event: Pick<OrchestrationEvent, "sequence" | "type">,
+        sideEffects: AttachmentSideEffects,
+      ) {
         if (
           sideEffects.deletedThreadIds.size === 0 &&
           sideEffects.prunedThreadRelativePaths.size === 0
@@ -2124,27 +2184,37 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
       // Cleanup has its own cursor so retries never have to replay committed text.
       // All message and activity references are current before any files are removed.
-      const pendingCleanup = new Map<string, OrchestrationEvent>();
-      let lastEvent: OrchestrationEvent | undefined;
-      yield* Stream.runForEach(
-        eventStore.readFromSequence(cleanupStart, Number.MAX_SAFE_INTEGER),
-        (event) =>
-          Effect.sync(() => {
-            lastEvent = event;
-            if (event.type === "thread.reverted" || event.type === "thread.deleted") {
-              pendingCleanup.set(`${event.type}:${event.payload.threadId}`, event);
-            }
-          }),
+      const pendingCleanup = new Map<
+        string,
+        Schema.Schema.Type<typeof AttachmentCleanupReplayRow>
+      >();
+      const cleanupReplayRows = yield* listAttachmentCleanupReplayRows({
+        sequenceExclusive: cleanupStart,
+      }).pipe(
+        Effect.mapError((cause) =>
+          Schema.isSchemaError(cause)
+            ? toPersistenceDecodeError("ProjectionPipeline.bootstrap:decodeCleanupRows")(cause)
+            : toPersistenceSqlError("ProjectionPipeline.bootstrap:listCleanupRows")(cause),
+        ),
       );
+      const lastEvent = cleanupReplayRows.at(-1);
+      for (const row of cleanupReplayRows) {
+        if (row.cleanupType !== null && row.threadId !== null) {
+          pendingCleanup.set(`${row.cleanupType}:${row.threadId}`, row);
+        }
+      }
       for (const event of pendingCleanup.values()) {
-        if (event.type !== "thread.reverted" && event.type !== "thread.deleted") continue;
-        const threadId = event.payload.threadId;
-        const cleaned = yield* applyAttachmentSideEffects(event, {
-          deletedThreadIds: new Set(event.type === "thread.deleted" ? [threadId] : []),
-          prunedThreadRelativePaths: new Map(
-            event.type === "thread.reverted" ? [[threadId, new Set<string>()]] : [],
-          ),
-        });
+        if (event.cleanupType === null || event.threadId === null) continue;
+        const threadId = event.threadId;
+        const cleaned = yield* applyAttachmentSideEffects(
+          { sequence: event.sequence, type: event.cleanupType },
+          {
+            deletedThreadIds: new Set(event.cleanupType === "thread.deleted" ? [threadId] : []),
+            prunedThreadRelativePaths: new Map(
+              event.cleanupType === "thread.reverted" ? [[threadId, new Set<string>()]] : [],
+            ),
+          },
+        );
         // Leave the cleanup cursor behind this event so the next bootstrap retries it.
         if (!cleaned) return;
       }