# Files changed · PR \#23867 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/23867/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[fix\(server\): route projection reads through the reader connection](https://git.cafe/pingdotgg/t3code/pull/23867.md)

Version: 3

Head commit: 7ca41e9da546c518d1b48d7190dca0998bdcf434

Base commit: 2c8bdf6177159d8fcd12a897d7c9311f6eb93237

Comparison base: 2c8bdf6177159d8fcd12a897d7c9311f6eb93237

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

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

Status: modified; +64; −0

```diff
@@ -26,12 +26,16 @@
   TurnItemId,
   type ThreadPullRequestLink,
 } from "@t3tools/contracts";
+import * as NodeServices from "@effect/platform-node/NodeServices";
 import * as DateTime from "effect/DateTime";
 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 * as Schema from "effect/Schema";
 import * as Tracer from "effect/Tracer";
 import * as SqlClient from "effect/sql/SqlClient";
+import * as Statement from "effect/sql/Statement";
 import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";
 
 import * as SqlitePersistence from "../persistence/Sqlite.ts";
@@ -5644,3 +5648,63 @@
     }),
   );
 });
+
+// regression: #23687 pure reads must not queue behind the write lock
+const readerReads: ReadonlyArray<{
+  readonly name: string;
+  readonly read: (
+    store: ProjectionStore.ProjectionStoreV2Shape,
+    threadId: ThreadId,
+  ) => Effect.Effect<unknown, ProjectionStore.ProjectionStoreV2Error>;
+}> = [
+  {
+    name: "getThreadRecords",
+    read: (store, threadId) => store.getThreadRecords(threadId, ["runs"]),
+  },
+  { name: "getShellSnapshot", read: (store) => store.getShellSnapshot() },
+  { name: "getThreadShell", read: (store, threadId) => store.getThreadShell(threadId) },
+  { name: "getRecoveryThreadIds", read: (store) => store.getRecoveryThreadIds("runtime") },
+  {
+    name: "getSettlementCandidates",
+    read: (store, threadId) => store.getSettlementCandidates(threadId),
+  },
+];
+it.effect.each(readerReads)("$name runs on the reader connection", ({ read }) =>
+  Effect.gen(function* () {
+    const fileSystem = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    // The reader worker is a second connection, so the database must be a file.
+    const directory = yield* fileSystem.makeTempDirectoryScoped({
+      prefix: "t3-projection-reader-",
+    });
+    const layerFile = ProjectionStore.layer.pipe(
+      Layer.provideMerge(SqlitePersistence.layerFromPath(path.join(directory, "statev2.sqlite"))),
+    );
+
+    yield* Effect.gen(function* () {
+      const store = yield* ProjectionStore.ProjectionStoreV2;
+      const threadId = yield* addRolledBackRecoveryCandidate("reader");
+      const connections: Array<unknown> = [];
+      // nodeSqliteClient stamps db.connection only on the span that acquired the connection.
+      const tracer = Tracer.make({
+        span(options) {
+          const span = new Tracer.NativeSpan(options);
+          const end = span.end.bind(span);
+          span.end = (endTime, exit) => {
+            end(endTime, exit);
+            const connection = span.attributes.get("db.connection");
+            if (connection !== undefined) connections.push(connection);
+          };
+          return span;
+        },
+      });
+
+      yield* read(store, threadId).pipe(
+        Effect.provideService(Statement.SpanPropagationEnabled, true),
+        Effect.withTracer(tracer),
+      );
+
+      assert.deepEqual(connections, ["reader"]);
+    }).pipe(Effect.provide(layerFile));
+  }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+);
```

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

Status: modified; +29; −6

```diff
@@ -83,6 +83,7 @@
 import * as Layer from "effect/Layer";
 import * as Ref from "effect/Ref";
 import * as Schema from "effect/Schema";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
 import * as SqlClient from "effect/sql/SqlClient";
 import type * as Statement from "effect/sql/Statement";
 
@@ -4014,7 +4015,11 @@
             AND thread_id IN (${candidates})
             ${kind === "queued-runs" ? sql`AND archived_at IS NULL` : sql``}
           ORDER BY updated_at ASC, thread_id ASC
+        `.pipe(
+          // A pure read, on the reader connection: startup recovery must not
+          // take the write lock.
+          NodeSqliteClient.readOnly,
-        `;
+        );
         return rows.map((row) => ThreadId.make(row.thread_id));
       },
       Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })),
@@ -5245,6 +5250,10 @@
       filter,
     ) =>
       sql.withTransaction(readCanonicalProjection(threadId, undefined, fields, filter)).pipe(
+        // The MCP read path runs this for every thread read. It is a pure
+        // read, so it belongs on the reader connection: on the writer it
+        // contends with the writes that the same load is producing.
+        NodeSqliteClient.readOnly,
         Effect.map(
           (projection) =>
             Object.fromEntries([
@@ -6079,7 +6088,12 @@
             );
           }),
         )
+        .pipe(
+          // A pure read, on the reader connection: the settlement sweep must
+          // not take the write lock.
-        .pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })));
+          NodeSqliteClient.readOnly,
+          Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })),
+        );
 
     const getThreadsWithPullRequests: ProjectionStoreV2Shape["getThreadsWithPullRequests"] =
       Effect.fn("ProjectionStore.getThreadsWithPullRequests")(
@@ -6404,9 +6418,13 @@
       );
 
     const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = (options) =>
+      sql.withTransaction(readShellSnapshotRows(options)).pipe(
+        // A pure read: the shell list is read far more often than it is
+        // written, and on the writer it queues behind the writes.
+        NodeSqliteClient.readOnly,
+        Effect.mapError(shellSnapshotReadError),
-      sql
-        .withTransaction(readShellSnapshotRows(options))
+        Effect.flatMap(decodeShellSnapshot),
+      );
-        .pipe(Effect.mapError(shellSnapshotReadError), Effect.flatMap(decodeShellSnapshot));
 
     const getThreadShell: ProjectionStoreV2Shape["getThreadShell"] = (threadId) =>
       sql
@@ -6480,7 +6498,12 @@
             });
           }),
         )
-        .pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })));
+        .pipe(
+          // A pure read: the shell row for one thread is read on every
+          // thread read, and on the writer it queues behind the writes.
+          NodeSqliteClient.readOnly,
+          Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })),
+        );
 
     return {
       apply,
```