authorgravatar for bompus@users.noreply.github.comAaron Queen <bompus@users.noreply.github.com> 2026-10-14 06:26:46-06:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 05:26:46-07:00
log8414d168191599dec08588080d0b77506c849424
tree491239b8644c756fe519b8c7a53a4cb614567c50
parentdd5c7bd09e83bbeb7c3f63aa787e6ffdd3cc6221
signature Signed by PGP key B5690EEEBB952194

perf(server): thread search runs on the SQLite reader worker (#19382)

2 files changed, 76 insertions(+), 12 deletions(-)

apps/server/src/orchestration-v2/ThreadSearch.test.ts+56
...@@ -7,9 +7,14 @@ import {...@@ -7,9 +7,14 @@ import {
7 ThreadId,7 ThreadId,
8 type OrchestrationV2DomainEvent,8 type OrchestrationV2DomainEvent,
9} from "@t3tools/contracts";9} from "@t3tools/contracts";
10import * as NodeServices from "@effect/platform-node/NodeServices";
10import * as DateTime from "effect/DateTime";11import * as DateTime from "effect/DateTime";
12import * as Deferred from "effect/Deferred";
11import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
14import * as FileSystem from "effect/FileSystem";
15import * as Fiber from "effect/Fiber";
12import * as Layer from "effect/Layer";16import * as Layer from "effect/Layer";
17import * as Path from "effect/Path";
13import * as SqlClient from "effect/sql/SqlClient";18import * as SqlClient from "effect/sql/SqlClient";
1419
15import * as SqlitePersistence from "../persistence/Sqlite.ts";20import * as SqlitePersistence from "../persistence/Sqlite.ts";
...@@ -199,3 +204,54 @@ it.layer(layerTest)("ThreadSearch", (it) => {...@@ -199,3 +204,54 @@ it.layer(layerTest)("ThreadSearch", (it) => {
199 }),204 }),
200 );205 );
201});206});
207
208it.effect("searches on the reader worker while the writer is busy", () =>
209 Effect.gen(function* () {
210 const fileSystem = yield* FileSystem.FileSystem;
211 const path = yield* Path.Path;
212 const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-thread-search-" });
213 const layerFile = Layer.mergeAll(
214 ThreadSearch.layer,
215 ProjectionStore.layer,
216 ProjectStore.layer,
217 ).pipe(
218 Layer.provideMerge(SqlitePersistence.layerFromPath(path.join(directory, "state.sqlite"))),
219 Layer.provide(NodeServices.layer),
220 );
221
222 yield* Effect.gen(function* () {
223 const projections = yield* ProjectionStore.ProjectionStoreV2;
224 const search = yield* ThreadSearch.ThreadSearch;
225 const sql = yield* SqlClient.SqlClient;
226 const project = ProjectId.make("project:search-reader");
227 const threadId = ThreadId.make("thread:reader");
228 yield* createProject(project);
229 yield* Effect.forEach(
230 [thread(threadId, project), message(threadId, "reader", "user", "needle on the reader")],
231 projections.apply,
232 { discard: true },
233 );
234
235 // A write transaction holds the writer connection. A search on the writer would queue
236 // behind it and never finish; one on the reader worker does not wait.
237 const writerHeld = yield* Deferred.make<void>();
238 const releaseWriter = yield* Deferred.make<void>();
239 const writer = yield* Effect.forkChild(
240 sql.withTransaction(
241 Deferred.succeed(writerHeld, undefined).pipe(
242 Effect.andThen(Deferred.await(releaseWriter)),
243 ),
244 ),
245 );
246 yield* Deferred.await(writerHeld);
247 const result = yield* search.search({ query: "needle" });
248 yield* Deferred.succeed(releaseWriter, undefined);
249 yield* Fiber.join(writer);
250
251 assert.deepEqual(
252 result.matches.map((match) => match.threadId),
253 [threadId],
254 );
255 }).pipe(Effect.provide(layerFile));
256 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
257);
apps/server/src/orchestration-v2/ThreadSearch.ts+20-12
...@@ -12,6 +12,7 @@ import * as Layer from "effect/Layer";...@@ -12,6 +12,7 @@ import * as Layer from "effect/Layer";
12import * as Schema from "effect/Schema";12import * as Schema from "effect/Schema";
13import * as SqlClient from "effect/sql/SqlClient";13import * as SqlClient from "effect/sql/SqlClient";
14import * as SqlSchema from "effect/sql/SqlSchema";14import * as SqlSchema from "effect/sql/SqlSchema";
15import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
1516
16/** Carries no query text: search input is user content. */17/** Carries no query text: search input is user content. */
17export class ThreadSearchError extends Schema.TaggedError<ThreadSearchError>()(18export class ThreadSearchError extends Schema.TaggedError<ThreadSearchError>()(
...@@ -138,18 +139,25 @@ export const make = Effect.gen(function* () {...@@ -138,18 +139,25 @@ export const make = Effect.gen(function* () {
138139
139 const search: ThreadSearch["Service"]["search"] = Effect.fn("ThreadSearch.search")(140 const search: ThreadSearch["Service"]["search"] = Effect.fn("ThreadSearch.search")(
140 function* (input) {141 function* (input) {
141 const rows = yield* searchRows({142 // The scan reads every message, so run it on the reader worker rather
142 pattern: `%${escapeLikePattern(input.query)}%`,143 // than the event loop.
143 limit: input.limit ?? 50,144 const rows = yield* sql
144 }).pipe(145 .withTransaction(
145 Effect.mapError(146 searchRows({
146 (cause) =>147 pattern: `%${escapeLikePattern(input.query)}%`,
147 new ThreadSearchError({148 limit: input.limit ?? 50,
148 operation: Schema.isSchemaError(cause) ? "decode" : "query",149 }),
149 cause,150 )
150 }),151 .pipe(
151 ),152 NodeSqliteClient.readOnly,
152 );153 Effect.mapError(
154 (cause) =>
155 new ThreadSearchError({
156 operation: Schema.isSchemaError(cause) ? "decode" : "query",
157 cause,
158 }),
159 ),
160 );
153 return {161 return {
154 matches: rows.map((row) => ({162 matches: rows.map((row) => ({
155 threadId: row.threadId,163 threadId: row.threadId,