From 8414d168191599dec08588080d0b77506c849424 Mon Sep 17 00:00:00 2001 From: Aaron Queen Date: Wed, 14 Oct 2026 06:26:46 -0600 Subject: [PATCH] perf(server): thread search runs on the SQLite reader worker (#19382) --- .../src/orchestration-v2/ThreadSearch.test.ts | 56 +++++++++++++++++++ .../src/orchestration-v2/ThreadSearch.ts | 32 +++++++---- 2 files changed, 76 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSearch.test.ts b/apps/server/src/orchestration-v2/ThreadSearch.test.ts index 684722111384692aa810bc6d15dd4461d7d25261..f52d146f8084969759f135980614790d93665c23 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.test.ts @@ -7,9 +7,14 @@ import { ThreadId, type OrchestrationV2DomainEvent, } from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -199,3 +204,54 @@ it.layer(layerTest)("ThreadSearch", (it) => { }), ); }); + +it.effect("searches on the reader worker while the writer is busy", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-thread-search-" }); + const layerFile = Layer.mergeAll( + ThreadSearch.layer, + ProjectionStore.layer, + ProjectStore.layer, + ).pipe( + Layer.provideMerge(SqlitePersistence.layerFromPath(path.join(directory, "state.sqlite"))), + Layer.provide(NodeServices.layer), + ); + + yield* Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const search = yield* ThreadSearch.ThreadSearch; + const sql = yield* SqlClient.SqlClient; + const project = ProjectId.make("project:search-reader"); + const threadId = ThreadId.make("thread:reader"); + yield* createProject(project); + yield* Effect.forEach( + [thread(threadId, project), message(threadId, "reader", "user", "needle on the reader")], + projections.apply, + { discard: true }, + ); + + // A write transaction holds the writer connection. A search on the writer would queue + // behind it and never finish; one on the reader worker does not wait. + const writerHeld = yield* Deferred.make(); + const releaseWriter = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + sql.withTransaction( + Deferred.succeed(writerHeld, undefined).pipe( + Effect.andThen(Deferred.await(releaseWriter)), + ), + ), + ); + yield* Deferred.await(writerHeld); + const result = yield* search.search({ query: "needle" }); + yield* Deferred.succeed(releaseWriter, undefined); + yield* Fiber.join(writer); + + assert.deepEqual( + result.matches.map((match) => match.threadId), + [threadId], + ); + }).pipe(Effect.provide(layerFile)); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/orchestration-v2/ThreadSearch.ts b/apps/server/src/orchestration-v2/ThreadSearch.ts index 855556dd3d19eb406fbc07e6fde7528522fb26eb..e36b90a614e89f131d4144c6ff30eb222e626626 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.ts @@ -12,6 +12,7 @@ import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; import * as SqlSchema from "effect/sql/SqlSchema"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; /** Carries no query text: search input is user content. */ export class ThreadSearchError extends Schema.TaggedError()( @@ -138,18 +139,25 @@ export const make = Effect.gen(function* () { const search: ThreadSearch["Service"]["search"] = Effect.fn("ThreadSearch.search")( function* (input) { - const rows = yield* searchRows({ - pattern: `%${escapeLikePattern(input.query)}%`, - limit: input.limit ?? 50, - }).pipe( - Effect.mapError( - (cause) => - new ThreadSearchError({ - operation: Schema.isSchemaError(cause) ? "decode" : "query", - cause, - }), - ), - ); + // The scan reads every message, so run it on the reader worker rather + // than the event loop. + const rows = yield* sql + .withTransaction( + searchRows({ + pattern: `%${escapeLikePattern(input.query)}%`, + limit: input.limit ?? 50, + }), + ) + .pipe( + NodeSqliteClient.readOnly, + Effect.mapError( + (cause) => + new ThreadSearchError({ + operation: Schema.isSchemaError(cause) ? "decode" : "query", + cause, + }), + ), + ); return { matches: rows.map((row) => ({ threadId: row.threadId, -- 2.54.0