perf(server): port #13704 to V2 — pull request sync reads only threads with linked pull requests

Main stopped the per-minute pull request sync from reading the whole
shell snapshot. V2's sync had the same hot path, and a heavier one: it
called OrchestratorV2.getShellSnapshot(), which builds every thread's
shell (run, item, provider-thread and message subqueries) including
archived threads, then dropped all but the linked ones. It runs every
minute and again whenever a new link needs its first snapshot.

ProjectionStoreV2 gains getThreadsWithPullRequests: one SELECT over
orchestration_v2_projection_threads filtered to active rows whose
payload has a non-empty pullRequests array, returning only the fields
the sync reads. The in-memory store mirrors it. The main-side query on
projection_thread_pull_requests does not apply: V2 stores links in the
thread payload and never writes that V1 table.

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

Author: juliusmarminge

apps/server/src/orchestration-v2/ProjectionSettlement.test.ts

@@ -298,6 +298,52 @@ for (const [name, testLayer] of [
   );
 }
 
+const pullRequestLink = (number: number) => ({
+  host: "github.com",
+  repository: "owner/repository",
+  number,
+  url: `https://github.com/owner/repository/pull/${number}`,
+  source: "manual" as const,
+  linkedAt: DateTime.formatIso(old),
+  snapshot: null,
+  stack: null,
+});
+
+for (const [name, testLayer] of [
+  ["sql", SqlLayer],
+  ["memory", layerMemory],
+] as const) {
+  it.effect(`${name}: lists only active threads with pull request links, oldest first`, () =>
+    Effect.gen(function* () {
+      const store = yield* ProjectionStoreV2;
+      yield* createThread("no-links");
+      yield* createThread("empty-links", { pullRequests: [] });
+      yield* createThread("archived-link", {
+        archivedAt: old,
+        pullRequests: [pullRequestLink(1)],
+      });
+      const settled = yield* createThread("settled-link", {
+        settledOverride: "settled",
+        settledAt: old,
+        updatedAt: DateTime.subtract(now, { days: 12 }),
+        pullRequests: [pullRequestLink(2)],
+      });
+      const open = yield* createThread("open-link", {
+        pullRequests: [pullRequestLink(3), pullRequestLink(4)],
+      });
+
+      const threads = yield* store.getThreadsWithPullRequests();
+      assert.deepEqual(
+        threads.map((thread) => [thread.id, thread.settledOverride, thread.pullRequests?.length]),
+        [
+          [settled, "settled", 1],
+          [open, null, 2],
+        ],
+      );
+    }).pipe(Effect.provide(testLayer)),
+  );
+}
+
 it.effect(
   "reads settlement candidates and thread metadata without loading historical or archived payloads",
   () =>

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -149,6 +149,12 @@ export type ProjectionLimitRecoveryCandidate = Pick<
   | "snoozedUntil"
 >;
 
+/** The thread fields pull request sync reads, for a thread with at least one link. */
+export type ProjectionThreadPullRequests = Pick<
+  OrchestrationV2AppThread,
+  "id" | "projectId" | "settledOverride" | "settledAt" | "pullRequests"
+>;
+
 /** Thread activity needed by settlement, without transcript or fork history. */
 export type ProjectionSettlementCandidate = Pick<
   OrchestrationV2ThreadShell,
@@ -331,6 +337,14 @@ export interface ProjectionStoreV2Shape {
     ReadonlyArray<ProjectionSettlementCandidate>,
     ProjectionStoreV2Error
   >;
+  /**
+   * Active (not deleted, not archived) threads with at least one pull request
+   * link, in shell snapshot order. Skips run, message and item reads.
+   */
+  readonly getThreadsWithPullRequests: () => Effect.Effect<
+    ReadonlyArray<ProjectionThreadPullRequests>,
+    ProjectionStoreV2Error
+  >;
   readonly getTurnStartContext: (
     threadId: ThreadId,
     runId: RunId,
@@ -5029,6 +5043,29 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         )
         .pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })));
 
+    const getThreadsWithPullRequests: ProjectionStoreV2Shape["getThreadsWithPullRequests"] = () =>
+      Effect.gen(function* () {
+        const rows = yield* sql<PayloadRow>`
+          SELECT payload_json
+          FROM orchestration_v2_projection_threads
+          WHERE deleted_at IS NULL
+            AND json_extract(payload_json, '$.archivedAt') IS NULL
+            AND json_array_length(payload_json, '$.pullRequests') > 0
+          ORDER BY updated_at ASC, thread_id ASC
+        `;
+        return yield* Effect.forEach(rows, (row) =>
+          decodeThreadPayload(row.payload_json).pipe(
+            Effect.map((thread): ProjectionThreadPullRequests => ({
+              id: thread.id,
+              projectId: thread.projectId,
+              settledOverride: thread.settledOverride,
+              settledAt: thread.settledAt,
+              pullRequests: thread.pullRequests ?? [],
+            })),
+          ),
+        );
+      }).pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })));
+
     const shellThreadStateFromRow = (input: {
       readonly row: ShellThreadRow;
       readonly runOrdinalsByThreadId: ReadonlyMap<ThreadId, Map<RunId, number>>;
@@ -5292,6 +5329,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       getThreadShell,
       getThread,
       getSettlementCandidates,
+      getThreadsWithPullRequests,
       getThreadProjection,
       getTurnStartContext,
       getTurnStartHistory,
@@ -5413,6 +5451,31 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
                 left.id.localeCompare(right.id),
             );
         }),
+      getThreadsWithPullRequests: () =>
+        Ref.get(replayState).pipe(
+          Effect.map((state) =>
+            [...state.projections.values()]
+              .map(({ thread }) => thread)
+              .filter(
+                (thread) =>
+                  thread.deletedAt === null &&
+                  thread.archivedAt === null &&
+                  (thread.pullRequests ?? []).length > 0,
+              )
+              .toSorted(
+                (left, right) =>
+                  DateTime.toEpochMillis(left.updatedAt) -
+                    DateTime.toEpochMillis(right.updatedAt) || left.id.localeCompare(right.id),
+              )
+              .map((thread): ProjectionThreadPullRequests => ({
+                id: thread.id,
+                projectId: thread.projectId,
+                settledOverride: thread.settledOverride,
+                settledAt: thread.settledAt,
+                pullRequests: thread.pullRequests ?? [],
+              })),
+          ),
+        ),
       getLimitRecoveryCandidates: (options) =>
         Ref.get(replayState).pipe(
           Effect.map((state) =>

apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts

@@ -219,6 +219,7 @@ it.effect(
           getThreadShell: () => Effect.die("unused getThreadShell"),
           getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
           getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),
+          getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"),
           getThreadProjection: () => Effect.die("control effects must not load transcript"),
           getTurnStartContext: () => Effect.die("unused"),
           getTurnStartHistory: () => Effect.die("unused"),

apps/server/src/orchestration/PullRequestSyncReactor.test.ts

@@ -16,6 +16,7 @@ import {
 } from "@t3tools/contracts";
 import { assert, describe, it } from "@effect/vitest";
 import * as Crypto from "effect/Crypto";
+import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
@@ -26,7 +27,7 @@ import { TestClock } from "effect/testing";
 import { PullRequestService } from "../pullRequest/PullRequestService.ts";
 import { ServerActivation } from "../serverActivation.ts";
 import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts";
-import { v2PullRequestThread } from "../orchestration-v2/testkit/pullRequestFixtures.ts";
+import { ProjectionStoreV2 } from "../orchestration-v2/ProjectionStore.ts";
 import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";
 import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts";
 
@@ -164,6 +165,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
   const activation = yield* Deferred.make<void>();
   const snapshots = yield* Ref.make(options.snapshot);
   const snapshotReads = yield* Queue.unbounded<void>();
+  const shellSnapshotReads = yield* Ref.make(0);
   const syncCommands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);
   const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);
   const summaryCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
@@ -204,16 +206,28 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
       stack,
       invalidate: options.invalidate ?? (() => Effect.void),
     }),
-    Layer.mock(OrchestratorV2)({
-      getShellSnapshot: () =>
+    Layer.mock(ProjectionStoreV2)({
+      // Mirrors the store's filter: active threads that have at least one link.
+      getThreadsWithPullRequests: () =>
         Queue.offer(snapshotReads, undefined).pipe(
           Effect.andThen(Ref.get(snapshots)),
-          Effect.map((snapshot) => ({
-            schemaVersion: 2,
-            snapshotSequence: snapshot.snapshotSequence,
-            threads: snapshot.threads.map(v2PullRequestThread),
-            archivedThreads: [],
-          })),
+          Effect.map((snapshot) =>
+            snapshot.threads
+              .filter((thread) => thread.archivedAt === null && thread.pullRequests.length > 0)
+              .map((thread) => ({
+                id: thread.id,
+                projectId: thread.projectId,
+                settledOverride: thread.settledOverride,
+                settledAt: thread.settledAt === null ? null : DateTime.makeUnsafe(thread.settledAt),
+                pullRequests: thread.pullRequests,
+              })),
+          ),
+        ),
+    }),
+    Layer.mock(OrchestratorV2)({
+      getShellSnapshot: () =>
+        Ref.update(shellSnapshotReads, (count) => count + 1).pipe(
+          Effect.andThen(Effect.die(new Error("pull request sync must not read the shell"))),
         ),
       dispatch,
       streamDomainEvents: Stream.empty,
@@ -226,6 +240,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
     activation,
     snapshots,
     snapshotReads,
+    shellSnapshotReads,
     syncCommands,
     linkCommands,
     summaryCalls,
@@ -386,6 +401,8 @@ describe("PullRequestSyncReactor", () => {
             ],
           );
           assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1);
+          // Reads only linked threads, never the full shell snapshot of every thread.
+          assert.strictEqual(yield* Ref.get(fixture.shellSnapshotReads), 0);
         }).pipe(Effect.provide(fixture.layer));
       }),
     ),

apps/server/src/orchestration/PullRequestSyncReactor.ts

@@ -1,7 +1,6 @@
 import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl";
 import {
   CommandId,
-  type OrchestrationV2ThreadShell,
   type PullRequestSummary,
   type ThreadPullRequestKey,
   type ThreadPullRequestLink,
@@ -29,13 +28,17 @@ import * as Stream from "effect/Stream";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
 import { forkParked } from "../serverActivation.ts";
 import { OrchestratorV2 } from "../orchestration-v2/Orchestrator.ts";
+import {
+  ProjectionStoreV2,
+  type ProjectionThreadPullRequests,
+} from "../orchestration-v2/ProjectionStore.ts";
 
 const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;
 
 type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;
 
 interface LinkEntry {
-  readonly thread: OrchestrationV2ThreadShell;
+  readonly thread: ProjectionThreadPullRequests;
   readonly link: ThreadPullRequestLink;
 }
 
@@ -103,15 +106,15 @@ function stacksEqual(
   );
 }
 
-function isUnsettled(thread: OrchestrationV2ThreadShell): boolean {
+function isUnsettled(thread: ProjectionThreadPullRequests): boolean {
   return thread.settledOverride !== "settled" && thread.settledAt === null;
 }
 
 /**
  * Keeps every thread ↔ pull request link's host snapshot current. One sweep a minute reads
- * the shell snapshot, groups visible links by pull request so the host is asked once per PR
- * no matter how many threads share it, and writes back only what changed. Native stacks the
- * host reports are auto-linked to the thread as `source: "stack"`.
+ * only the active threads that have links, groups visible links by pull request so the host
+ * is asked once per PR no matter how many threads share it, and writes back only what
+ * changed. Native stacks the host reports are auto-linked to the thread as `source: "stack"`.
  */
 export class PullRequestSyncReactor extends Context.Service<
   PullRequestSyncReactor,
@@ -126,6 +129,7 @@ export class PullRequestSyncReactor extends Context.Service<
 /** @public Service construction is part of the canonical Effect module API. */
 export const make = Effect.gen(function* () {
   const engine = yield* OrchestratorV2;
+  const projections = yield* ProjectionStoreV2;
   const pullRequests = yield* PullRequestService.PullRequestService;
   const crypto = yield* Crypto.Crypto;
 
@@ -151,14 +155,13 @@ export const make = Effect.gen(function* () {
       Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields);
 
   const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* (requestedKey?: string) {
-    const snapshot = yield* engine.getShellSnapshot();
+    const threads = yield* projections.getThreadsWithPullRequests();
     const now = yield* DateTime.now;
     const nowMs = DateTime.toEpochMillis(now);
     const nowIso = DateTime.formatIso(now);
 
     const groups = new Map<string, Array<LinkEntry>>();
-    for (const thread of snapshot.threads) {
-      if (thread.archivedAt !== null) continue;
+    for (const thread of threads) {
       for (const link of visibleThreadPullRequests(thread.pullRequests ?? [])) {
         const key = threadPullRequestKeyOf(link);
         const entries = groups.get(key) ?? [];

apps/server/src/server.ts

@@ -513,7 +513,11 @@ const RuntimeCoreDependenciesBaseLive = Layer.mergeAll(
       const service = yield* PullRequestSyncReactor.PullRequestSyncReactor;
       yield* service.start();
     }),
-  ).pipe(Layer.provideMerge(PullRequestSyncReactor.layer), Layer.provide(PullRequestServiceLive)),
+  ).pipe(
+    Layer.provideMerge(PullRequestSyncReactor.layer),
+    Layer.provide(PullRequestServiceLive),
+    Layer.provide(ProjectionStoreV2.layer),
+  ),
   // Subscribes to `account.rate-limits.updated` so usage bars track live
   // telemetry instead of waiting for the next status probe.
   ProviderUsageLimitsIngestionLive,