Skip to content

Commit 6ed47b5

Browse files
Yash-Singh1Yash-Singh1
authored
fix(server): route SQLite reads around writer contention (#23902)
1 parent a8dba9c commit 6ed47b5

21 files changed

Lines changed: 305 additions & 69 deletions

‎apps/server/src/mcpApps/McpAppModelContext.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import type { ThreadId, TurnItemId } from "@t3tools/contracts";
22
import * as Context from "effect/Context";
3+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
34
import * as DateTime from "effect/DateTime";
45
import * as Effect from "effect/Effect";
56
import * as Layer from "effect/Layer";
@@ -83,7 +84,7 @@ export const layer = Layer.effect(
8384
OR run.thread_id <> context.thread_id
8485
)
8586
ORDER BY context.updated_at, context.item_id
86-
`.pipe(Effect.mapError(fail));
87+
`.pipe(NodeSqliteClient.readOnly, Effect.mapError(fail));
8788
});
8889

8990
return McpAppModelContext.of({ set, forThread });

‎apps/server/src/orchestration-v2/EffectOutbox.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import {
1515
ThreadId,
1616
} from "@t3tools/contracts";
1717
import * as Context from "effect/Context";
18+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
1819
import * as DateTime from "effect/DateTime";
1920
import * as Deferred from "effect/Deferred";
2021
import * as Duration from "effect/Duration";
@@ -420,6 +421,7 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
420421
WHERE effect_id = ${effectId}
421422
LIMIT 1
422423
`.pipe(
424+
NodeSqliteClient.readOnly,
423425
Effect.flatMap((rows) => {
424426
const row = rows[0];
425427
return row === undefined
@@ -437,6 +439,7 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
437439
WHERE command_id = ${commandId}
438440
ORDER BY created_at ASC, effect_id ASC
439441
`.pipe(
442+
NodeSqliteClient.readOnly,
440443
Effect.flatMap((rows) => decodeRows("list", rows)),
441444
Effect.mapError((cause) =>
442445
isEffectOutboxError(cause)
@@ -571,7 +574,7 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
571574
SELECT MIN(candidate.available_at) AS available_at
572575
FROM orchestration_v2_effect_outbox AS candidate
573576
WHERE ${claimableCandidatePredicate()}
574-
`.pipe(Effect.withTracerEnabled(false));
577+
`.pipe(NodeSqliteClient.readOnly, Effect.withTracerEnabled(false));
575578
const availableAt = rows[0]?.available_at;
576579
if (availableAt === undefined || availableAt === null) return Option.none();
577580
const parsed = DateTime.make(availableAt);

‎apps/server/src/orchestration-v2/FoundationPersistence.test.ts‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3218,3 +3218,83 @@ it.effect.each(["events", "receipts"] as const)(
32183218
}).pipe(Effect.provide(maintenanceLayer));
32193219
}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
32203220
);
3221+
3222+
it.effect("routes projection node lookups to the reader connection", () =>
3223+
Effect.gen(function* () {
3224+
const fs = yield* FileSystem.FileSystem;
3225+
const path = yield* Path.Path;
3226+
const directory = yield* fs.makeTempDirectoryScoped({ prefix: "projection-reader-" });
3227+
const database = SqlitePersistence.layerFromPath(path.join(directory, "statev2.sqlite"));
3228+
const projectionLayer = ProjectionStore.layer.pipe(Layer.provideMerge(database));
3229+
yield* Effect.gen(function* () {
3230+
const projections = yield* ProjectionStore.ProjectionStoreV2;
3231+
const now = yield* DateTime.now;
3232+
const nodeId = NodeId.make("node:reader-routing");
3233+
const node = {
3234+
id: nodeId,
3235+
threadId: ThreadId.make("thread:reader-routing"),
3236+
runId: null,
3237+
parentNodeId: null,
3238+
rootNodeId: nodeId,
3239+
kind: "subagent" as const,
3240+
status: "running" as const,
3241+
countsForRun: false,
3242+
providerThreadId: null,
3243+
providerTurnId: null,
3244+
nativeItemRef: null,
3245+
runtimeRequestId: null,
3246+
checkpointScopeId: null,
3247+
startedAt: now,
3248+
completedAt: null,
3249+
};
3250+
const spans: Array<Tracer.Span> = [];
3251+
const observe: Statement.Transformer = (statement, _sql, _fiber, span) => {
3252+
const [query] = statement.compile();
3253+
if (/orchestration_v2_projection_nodes WHERE node_id/.test(query)) {
3254+
spans.push(span);
3255+
}
3256+
return Effect.succeed(statement);
3257+
};
3258+
yield* Effect.gen(function* () {
3259+
assert.isFalse(yield* projections.isNodeStored(node));
3260+
}).pipe(
3261+
Effect.provideService(Statement.CurrentTransformer, observe),
3262+
Effect.provideService(Statement.SpanPropagationEnabled, true),
3263+
);
3264+
assert.equal(spans.length, 1);
3265+
assert.isTrue(
3266+
spans.every((span) => span.attributes.get("db.connection") === "reader"),
3267+
JSON.stringify(spans.map((span) => span.attributes.get("db.connection"))),
3268+
);
3269+
}).pipe(Effect.provide(projectionLayer));
3270+
}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
3271+
);
3272+
3273+
it.effect("loads a shell snapshot while the writer connection is occupied", () =>
3274+
Effect.gen(function* () {
3275+
const fs = yield* FileSystem.FileSystem;
3276+
const path = yield* Path.Path;
3277+
const directory = yield* fs.makeTempDirectoryScoped({ prefix: "shell-reader-" });
3278+
const database = SqlitePersistence.layerFromPath(path.join(directory, "statev2.sqlite"));
3279+
const projectionLayer = ProjectionStore.layer.pipe(Layer.provideMerge(database));
3280+
yield* Effect.gen(function* () {
3281+
const sql = yield* SqlClient.SqlClient;
3282+
const projections = yield* ProjectionStore.ProjectionStoreV2;
3283+
const held = yield* Deferred.make<void>();
3284+
const release = yield* Deferred.make<void>();
3285+
const writer = yield* sql
3286+
.withTransaction(
3287+
sql`SELECT 1`.pipe(
3288+
Effect.andThen(Deferred.succeed(held, undefined)),
3289+
Effect.andThen(Deferred.await(release)),
3290+
),
3291+
)
3292+
.pipe(Effect.forkChild);
3293+
yield* Deferred.await(held);
3294+
const snapshot = yield* projections.getShellSnapshot();
3295+
yield* Deferred.succeed(release, undefined);
3296+
yield* Fiber.join(writer);
3297+
assert.deepEqual(snapshot.threads, []);
3298+
}).pipe(Effect.provide(projectionLayer));
3299+
}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
3300+
);

‎apps/server/src/orchestration-v2/ProjectStore.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import {
99
ThreadEnvMode,
1010
} from "@t3tools/contracts";
1111
import * as Context from "effect/Context";
12+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
1213
import * as Effect from "effect/Effect";
1314
import * as Layer from "effect/Layer";
1415
import * as Option from "effect/Option";
@@ -191,6 +192,7 @@ export const make = Effect.gen(function* () {
191192

192193
const get: ProjectStoreV2["Service"]["get"] = (projectId, options) =>
193194
selectRows({ projectId, includeDeleted: options?.includeDeleted === true }).pipe(
195+
NodeSqliteClient.readOnly,
194196
Effect.map((rows) => Option.fromUndefinedOr(rows[0])),
195197
mapError("get"),
196198
);
@@ -199,12 +201,13 @@ export const make = Effect.gen(function* () {
199201
selectRows({
200202
...(options?.projectIds === undefined ? {} : { projectIds: options.projectIds }),
201203
includeDeleted: options?.includeDeleted === true,
202-
}).pipe(mapError("list"));
204+
}).pipe(NodeSqliteClient.readOnly, mapError("list"));
203205

204206
const findActiveByWorkspaceRoot: ProjectStoreV2["Service"]["findActiveByWorkspaceRoot"] = (
205207
workspaceRoot,
206208
) =>
207209
selectRows({ workspaceRoot, includeDeleted: false }).pipe(
210+
NodeSqliteClient.readOnly,
208211
Effect.map((rows) => Option.fromUndefinedOr(rows[0])),
209212
mapError("findActiveByWorkspaceRoot"),
210213
);

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

Lines changed: 46 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests";
22
import { projectTurnItemForWire } from "./WireProjection.ts";
33
import * as Stream from "effect/Stream";
4+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
45
import { makeThreadFind, findProjectedThreadItems } from "./ThreadFind.ts";
56
import type {
67
OrchestrationV2SearchThreadInput,
@@ -6511,47 +6512,55 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
65116512
)
65126513
.pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })));
65136514

6515+
const readOnReader =
6516+
<Args extends ReadonlyArray<unknown>, A, E, R>(
6517+
method: (...args: Args) => Effect.Effect<A, E, R>,
6518+
) =>
6519+
(...args: Args): Effect.Effect<A, E, R> =>
6520+
method(...args).pipe(NodeSqliteClient.readOnly);
6521+
65146522
return {
65156523
apply,
6516-
getShellSnapshot,
6517-
readShellSnapshot,
6518-
getThreadShell,
6519-
getThread,
6520-
getSettlementCandidates,
6521-
getThreadsWithPullRequests,
6522-
getThreadWorktreePaths,
6524+
getShellSnapshot: readOnReader(getShellSnapshot),
6525+
readShellSnapshot: readOnReader(readShellSnapshot),
6526+
getThreadShell: readOnReader(getThreadShell),
6527+
getThread: readOnReader(getThread),
6528+
getSettlementCandidates: readOnReader(getSettlementCandidates),
6529+
getThreadsWithPullRequests: readOnReader(getThreadsWithPullRequests),
6530+
getThreadWorktreePaths: readOnReader(getThreadWorktreePaths),
65236531
getThreadProjection,
6524-
getTurnStartContext,
6525-
getTurnStartHistory,
6526-
getRuntimeRecoveryProjection,
6527-
getRunningTurnContext,
6528-
getThreadProviderContext,
6529-
getRuntimeResponseContext,
6530-
getCheckpointContext,
6531-
getCheckpointCaptureContext,
6532-
getCheckpointRollbackContext,
6533-
getRunMessage,
6534-
canStartQueuedRun,
6535-
getPendingNativeUserInputs,
6536-
hasUnpairedRunInterruptRequest,
6537-
isNodeStored,
6538-
getMessageCount,
6539-
getNextTurnItemOrdinal,
6540-
getTurnItem,
6541-
getThreadRecords,
6542-
getRuntimeRequest,
6543-
getPlan,
6544-
getProviderControlContext,
6545-
getLimitRecoveryCandidates,
6546-
getRecoveryThreadIds,
6547-
getUnreadableThreadIds,
6548-
getThreadSnapshot,
6549-
getThreadSnapshotWindow,
6550-
getTimelinePage,
6551-
getThreadHistoryPage,
6552-
searchThread,
6532+
getTurnStartContext: readOnReader(getTurnStartContext),
6533+
getTurnStartHistory: readOnReader(getTurnStartHistory),
6534+
getRuntimeRecoveryProjection: readOnReader(getRuntimeRecoveryProjection),
6535+
getRunningTurnContext: readOnReader(getRunningTurnContext),
6536+
getThreadProviderContext: readOnReader(getThreadProviderContext),
6537+
getRuntimeResponseContext: readOnReader(getRuntimeResponseContext),
6538+
getCheckpointContext: readOnReader(getCheckpointContext),
6539+
getCheckpointCaptureContext: readOnReader(getCheckpointCaptureContext),
6540+
getCheckpointRollbackContext: readOnReader(getCheckpointRollbackContext),
6541+
getRunMessage: readOnReader(getRunMessage),
6542+
canStartQueuedRun: readOnReader(canStartQueuedRun),
6543+
getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs),
6544+
hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest),
6545+
isNodeStored: readOnReader(isNodeStored),
6546+
getMessageCount: readOnReader(getMessageCount),
6547+
getNextTurnItemOrdinal: readOnReader(getNextTurnItemOrdinal),
6548+
getTurnItem: readOnReader(getTurnItem),
6549+
getThreadRecords: (threadId, fields, filter) =>
6550+
getThreadRecords(threadId, fields, filter).pipe(NodeSqliteClient.readOnly),
6551+
getRuntimeRequest: readOnReader(getRuntimeRequest),
6552+
getPlan: readOnReader(getPlan),
6553+
getProviderControlContext: readOnReader(getProviderControlContext),
6554+
getLimitRecoveryCandidates: readOnReader(getLimitRecoveryCandidates),
6555+
getRecoveryThreadIds: readOnReader(getRecoveryThreadIds),
6556+
getUnreadableThreadIds: readOnReader(getUnreadableThreadIds),
6557+
getThreadSnapshot: readOnReader(getThreadSnapshot),
6558+
getThreadSnapshotWindow: readOnReader(getThreadSnapshotWindow),
6559+
getTimelinePage: readOnReader(getTimelinePage),
6560+
getThreadHistoryPage: readOnReader(getThreadHistoryPage),
6561+
searchThread: readOnReader(searchThread),
65536562
searchThreadStream,
6554-
getThreadAttachmentIds,
6563+
getThreadAttachmentIds: readOnReader(getThreadAttachmentIds),
65556564
} satisfies ProjectionStoreV2Shape;
65566565
}),
65576566
);

‎apps/server/src/orchestration-v2/ThreadFind.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import * as Data from "effect/Data";
1414
import * as Effect from "effect/Effect";
1515
import * as Schema from "effect/Schema";
1616
import * as Stream from "effect/Stream";
17+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
1718
import * as SqlClient from "effect/sql/SqlClient";
1819
import {
1920
ProjectionStoreReadError,
@@ -345,7 +346,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
345346
}
346347
return { payloads, cwd, sequence, cacheKey, sourceThreadIds };
347348
},
348-
sql.withTransaction,
349+
(effect) => sql.withTransaction(effect).pipe(NodeSqliteClient.readOnly),
349350
(effect, input) =>
350351
effect.pipe(
351352
Effect.mapError((cause) =>

‎apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import {
2222
TurnItemId,
2323
} from "@t3tools/contracts";
2424
import * as KeyedLock from "@t3tools/shared/KeyedLock";
25+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
2526
import * as Cause from "effect/Cause";
2627
import * as Context from "effect/Context";
2728
import * as DateTime from "effect/DateTime";
@@ -359,7 +360,7 @@ const make = Effect.gen(function* () {
359360
WHERE thread_id = ${threadId}
360361
AND role IN ('user', 'assistant')
361362
ORDER BY created_at ASC, message_id ASC
362-
`;
363+
`.pipe(NodeSqliteClient.readOnly);
363364

364365
const listShellMessages = Effect.fn("LegacyV1ThreadImporter.listShellMessages")(function* (
365366
threadId: ThreadId,
@@ -669,6 +670,7 @@ const make = Effect.gen(function* () {
669670
WHERE legacy_import.transcript_imported_at IS NULL
670671
)
671672
`.pipe(
673+
NodeSqliteClient.readOnly,
672674
Effect.map((rows) => rows[0]?.count ?? 0),
673675
Effect.mapError(
674676
(cause) => new LegacyV1ThreadImportError({ operation: "inspect pending", cause }),
@@ -690,7 +692,7 @@ const make = Effect.gen(function* () {
690692
FROM orchestration_v2_legacy_imports
691693
WHERE thread_id = ${threadId}
692694
LIMIT 1
693-
`;
695+
`.pipe(NodeSqliteClient.readOnly);
694696
const imported = imports[0];
695697
if (imported === undefined || imported.transcript_imported_at !== null) {
696698
if (imported !== undefined) {
@@ -708,7 +710,7 @@ const make = Effect.gen(function* () {
708710
AND aggregate_kind = 'thread'
709711
AND stream_id = ${threadId}
710712
AND event_id LIKE ${`${IMPORT_EVENT_PREFIX}:message:%`}
711-
`;
713+
`.pipe(NodeSqliteClient.readOnly);
712714
const existing = new Set(existingRows.map((row) => row.event_id));
713715
const missing = messages.filter(
714716
(message) => !existing.has(`${IMPORT_EVENT_PREFIX}:message:${message.message_id}`),
@@ -772,7 +774,7 @@ const make = Effect.gen(function* () {
772774
FROM orchestration_v2_legacy_imports
773775
WHERE transcript_imported_at IS NULL
774776
ORDER BY shell_imported_at ASC, thread_id ASC
775-
`;
777+
`.pipe(NodeSqliteClient.readOnly);
776778
let importedThreadCount = 0;
777779
let importedMessageCount = 0;
778780
for (const row of rows) {

‎apps/server/src/persistence/AuthPairingLinks.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import * as Context from "effect/Context";
2+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
23
import * as Effect from "effect/Effect";
34
import * as Layer from "effect/Layer";
45
import * as Option from "effect/Option";
@@ -294,6 +295,7 @@ export const make = Effect.gen(function* () {
294295

295296
const listActive: AuthPairingLinkRepository["Service"]["listActive"] = (input) =>
296297
listActivePairingLinkRows(input).pipe(
298+
NodeSqliteClient.readOnly,
297299
Effect.mapError(
298300
toPersistenceSqlOrDecodeError(
299301
"AuthPairingLinkRepository.listActive:query",
@@ -329,6 +331,7 @@ export const make = Effect.gen(function* () {
329331

330332
const getByCredential: AuthPairingLinkRepository["Service"]["getByCredential"] = (input) =>
331333
getPairingLinkRowByCredential(input).pipe(
334+
NodeSqliteClient.readOnly,
332335
Effect.mapError(
333336
toPersistenceSqlOrDecodeError(
334337
"AuthPairingLinkRepository.getByCredential:query",

‎apps/server/src/persistence/AuthSessions.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import * as Context from "effect/Context";
2+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
23
import * as Effect from "effect/Effect";
34
import * as Layer from "effect/Layer";
45
import * as Option from "effect/Option";
@@ -416,6 +417,7 @@ export const make = Effect.gen(function* () {
416417

417418
const getById: AuthSessionRepository["Service"]["getById"] = (input) =>
418419
getSessionRowById(input).pipe(
420+
NodeSqliteClient.readOnly,
419421
Effect.mapError(
420422
toPersistenceSqlOrDecodeError(
421423
"AuthSessionRepository.getById:query",
@@ -443,6 +445,7 @@ export const make = Effect.gen(function* () {
443445

444446
const listActive: AuthSessionRepository["Service"]["listActive"] = (input) =>
445447
listActiveSessionRows(input).pipe(
448+
NodeSqliteClient.readOnly,
446449
Effect.mapError(
447450
toPersistenceSqlOrDecodeError(
448451
"AuthSessionRepository.listActive:query",

‎apps/server/src/persistence/OrchestrationCommandReceipts.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { CommandId, IsoDateTime, NonNegativeInt, ProjectId, ThreadId } from "@t3tools/contracts";
2+
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
23
import * as SqlClient from "effect/sql/SqlClient";
34
import * as SqlSchema from "effect/sql/SqlSchema";
45
import * as Context from "effect/Context";
@@ -149,6 +150,7 @@ const makeOrchestrationCommandReceiptRepository = Effect.gen(function* () {
149150
input,
150151
) =>
151152
findReceiptByCommandId(input).pipe(
153+
NodeSqliteClient.readOnly,
152154
Effect.mapError(
153155
toPersistenceSqlError("OrchestrationCommandReceiptRepository.getByCommandId:query"),
154156
),

0 commit comments

Comments
 (0)