From a523a42f2f2b5a35678b6453b80b57d60e5829a0 Mon Sep 17 00:00:00 2001 From: Guitaraholic Date: Wed, 2 Sep 2026 20:38:02 +0100 Subject: [PATCH] Move thread search SQLite reads off the serving loop. List and sidebar already used the readonly WAL worker. GET /threads/search still called searchThreadsWithPendingInteractionState on the serving connection, so a palette keystroke could stall streams. Same worker, new request name; :memory: stays sync. --- apps/server/src/routes/threads/base.ts | 8 +++---- packages/db/src/index.ts | 1 + packages/db/src/sqlite-read-queue.ts | 28 +++++++++++++++++----- packages/db/src/sqlite-read-worker.ts | 11 ++++++++- packages/db/test/sqlite-read-queue.test.ts | 17 +++++++++++++ 5 files changed, 54 insertions(+), 11 deletions(-) diff --git a/apps/server/src/routes/threads/base.ts b/apps/server/src/routes/threads/base.ts index aaba545452..194b48208b 100644 --- a/apps/server/src/routes/threads/base.ts +++ b/apps/server/src/routes/threads/base.ts @@ -8,7 +8,7 @@ import { listThreadMentionRowsByIds, listThreadsWithPendingInteractionStateOffThread, markThreadDeleted, - searchThreadsWithPendingInteractionState, + searchThreadsWithPendingInteractionStateOffThread, updateThread, type ThreadSearchResultGroup as DbThreadSearchResultGroup, type UpdateThreadInput, @@ -309,7 +309,7 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void { ); }); - get(routes.search, (context, query) => { + get(routes.search, async (context, query) => { const searchQuery = query.query.trim(); if (countNonWhitespaceChars(searchQuery) < 2) { throw new ApiError( @@ -321,10 +321,10 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void { const limitPerGroup = parseSearchLimitPerGroup(query.limitPerGroup); return context.json( buildThreadSearchResponse(deps, { - ...searchThreadsWithPendingInteractionState(deps.db, { + ...(await searchThreadsWithPendingInteractionStateOffThread(deps.db, { query: searchQuery, limitPerGroup, - }), + })), }) satisfies ThreadSearchResponse, ); }); diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 93cc7927a9..e542375acf 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -2,6 +2,7 @@ export { createConnection } from "./connection.js"; export { listThreadsWithPendingInteractionStateForProjectsOffThread, listThreadsWithPendingInteractionStateOffThread, + searchThreadsWithPendingInteractionStateOffThread, startSqliteReadWorker, stopSqliteReadWorker, } from "./sqlite-read-queue.js"; diff --git a/packages/db/src/sqlite-read-queue.ts b/packages/db/src/sqlite-read-queue.ts index 91bd68a271..386f4bee77 100644 --- a/packages/db/src/sqlite-read-queue.ts +++ b/packages/db/src/sqlite-read-queue.ts @@ -5,8 +5,11 @@ import type { DbConnection } from "./connection.js"; import { listThreadsWithPendingInteractionState, listThreadsWithPendingInteractionStateForProjects, + searchThreadsWithPendingInteractionState, type ListThreadsForProjectsOptions, type ListThreadsOptions, + type SearchThreadsWithPendingInteractionStateArgs, + type ThreadSearchResults, type ThreadWithPendingInteractionState, } from "./data/threads.js"; import type { @@ -16,7 +19,7 @@ import type { type PendingRead = { reject: (error: Error) => void; - resolve: (result: ThreadWithPendingInteractionState[]) => void; + resolve: (result: unknown) => void; }; let worker: Worker | null = null; @@ -74,7 +77,7 @@ export function startSqliteReadWorker(args: { } pending.delete(response.id); if (response.ok) { - request.resolve(response.result as ThreadWithPendingInteractionState[]); + request.resolve(response.result); return; } request.reject(new Error(response.error)); @@ -102,9 +105,9 @@ export async function stopSqliteReadWorker(): Promise { await current.terminate(); } -async function request( +async function request( message: Omit, -): Promise { +): Promise { const current = worker; if (current === null) { throw new Error("sqlite read worker is not running"); @@ -124,7 +127,7 @@ export async function listThreadsWithPendingInteractionStateOffThread( if (worker === null) { return listThreadsWithPendingInteractionState(db, options); } - return request({ + return request({ name: "listThreadsWithPendingInteractionState", args: options, }); @@ -137,8 +140,21 @@ export async function listThreadsWithPendingInteractionStateForProjectsOffThread if (worker === null) { return listThreadsWithPendingInteractionStateForProjects(db, options); } - return request({ + return request({ name: "listThreadsWithPendingInteractionStateForProjects", args: options, }); } + +export async function searchThreadsWithPendingInteractionStateOffThread( + db: DbConnection, + args: SearchThreadsWithPendingInteractionStateArgs, +): Promise { + if (worker === null) { + return searchThreadsWithPendingInteractionState(db, args); + } + return request({ + name: "searchThreadsWithPendingInteractionState", + args, + }); +} diff --git a/packages/db/src/sqlite-read-worker.ts b/packages/db/src/sqlite-read-worker.ts index f2b8e47a07..e35531642e 100644 --- a/packages/db/src/sqlite-read-worker.ts +++ b/packages/db/src/sqlite-read-worker.ts @@ -3,8 +3,10 @@ import { createConnection } from "./connection.js"; import { listThreadsWithPendingInteractionState, listThreadsWithPendingInteractionStateForProjects, + searchThreadsWithPendingInteractionState, type ListThreadsForProjectsOptions, type ListThreadsOptions, + type SearchThreadsWithPendingInteractionStateArgs, } from "./data/threads.js"; export type SqliteReadRequest = @@ -17,6 +19,11 @@ export type SqliteReadRequest = id: number; name: "listThreadsWithPendingInteractionStateForProjects"; args: ListThreadsForProjectsOptions; + } + | { + id: number; + name: "searchThreadsWithPendingInteractionState"; + args: SearchThreadsWithPendingInteractionStateArgs; }; export type SqliteReadResponse = @@ -36,7 +43,9 @@ port.on("message", (request: SqliteReadRequest) => { const result = request.name === "listThreadsWithPendingInteractionState" ? listThreadsWithPendingInteractionState(db, request.args) - : listThreadsWithPendingInteractionStateForProjects(db, request.args); + : request.name === "listThreadsWithPendingInteractionStateForProjects" + ? listThreadsWithPendingInteractionStateForProjects(db, request.args) + : searchThreadsWithPendingInteractionState(db, request.args); const response: SqliteReadResponse = { id: request.id, ok: true, result }; port.postMessage(response); } catch (error) { diff --git a/packages/db/test/sqlite-read-queue.test.ts b/packages/db/test/sqlite-read-queue.test.ts index 781835f418..82926c0191 100644 --- a/packages/db/test/sqlite-read-queue.test.ts +++ b/packages/db/test/sqlite-read-queue.test.ts @@ -11,6 +11,8 @@ import { createThread, listThreadsWithPendingInteractionState, listThreadsWithPendingInteractionStateForProjects, + searchThreadsWithPendingInteractionState, + updateThread, } from "../src/data/threads.js"; import { migrate } from "../src/migrate.js"; import { noopNotifier } from "../src/notifier.js"; @@ -18,6 +20,7 @@ import { isSqliteReadWorkerActive, listThreadsWithPendingInteractionStateForProjectsOffThread, listThreadsWithPendingInteractionStateOffThread, + searchThreadsWithPendingInteractionStateOffThread, startSqliteReadWorker, stopSqliteReadWorker, } from "../src/sqlite-read-queue.js"; @@ -101,4 +104,18 @@ describe("sqlite read queue", () => { expect(fromWorker).toEqual(fromServing); expect(fromProjectsWorker).toEqual(fromProjectsServing); }); + + it("returns the same thread search from a file-backed worker as the serving connection", async () => { + const { db, source, thread } = createFileDatabase(); + updateThread(db, noopNotifier, thread.id, { + title: "alpha search target", + }); + startSqliteReadWorker({ source }); + expect(isSqliteReadWorkerActive()).toBe(true); + + const args = { query: "alpha", limitPerGroup: 10 }; + await expect( + searchThreadsWithPendingInteractionStateOffThread(db, args), + ).resolves.toEqual(searchThreadsWithPendingInteractionState(db, args)); + }); });