Move thread-list SQLite reads off the serving loop

apps/server/package.json

@@ -4,7 +4,7 @@
   "type": "module",
   "private": true,
   "scripts": {
-    "build": "node ../../scripts/build-node-entry.mjs src/index.ts dist/index.js --clean-dist --external ./start-server.js --copy-dir ../../packages/db/drizzle dist/drizzle --copy-dir src/assets dist/assets && node --import tsx scripts/copy-builtin-skills.ts && node ../../scripts/build-node-entry.mjs src/start-server.ts dist/start-server.js && node ../../scripts/build-node-entry.mjs ../../packages/plugin-sdk/src/index.ts dist/plugin-sdk-runtime.js",
+    "build": "node ../../scripts/build-node-entry.mjs src/index.ts dist/index.js --clean-dist --external ./start-server.js --copy-dir ../../packages/db/drizzle dist/drizzle --copy-dir src/assets dist/assets && node --import tsx scripts/copy-builtin-skills.ts && node ../../scripts/build-node-entry.mjs src/start-server.ts dist/start-server.js && node ../../scripts/build-node-entry.mjs src/sqlite-read-worker.ts dist/sqlite-read-worker.js && node ../../scripts/build-node-entry.mjs ../../packages/plugin-sdk/src/index.ts dist/plugin-sdk-runtime.js",
     "start": "node dist/index.js",
     "start:prod": "cross-env NODE_ENV=production node dist/index.js",
     "dev": "node --conditions=source --import tsx scripts/dev-supervisor.mjs",

apps/server/src/routes/projects.ts

@@ -13,7 +13,7 @@ import {
   listPublicProjects,
   listProjectSourcesByProjectIds,
   listThreadSections,
-  listThreadsWithPendingInteractionStateForProjects,
+  listThreadsWithPendingInteractionStateForProjectsOffThread,
   reorderProject,
   updateProject,
   updateProjectSource,
@@ -201,26 +201,27 @@ function parseProjectListIncludes(
   return includes;
 }
 
-function buildProjectsWithThreadsResponse(
+async function buildProjectsWithThreadsResponse(
   deps: AppDeps,
   options: ProjectListOptions,
-): ProjectWithThreadsResponse[] {
+): Promise<ProjectWithThreadsResponse[]> {
   return buildProjectsWithThreadsResponseFromRows(
     deps,
     listDiscoverableProjects(deps, options),
   );
 }
 
-function buildProjectsWithThreadsResponseFromRows(
+async function buildProjectsWithThreadsResponseFromRows(
   deps: AppDeps,
   projectRows: ProjectResponseProjectFields[],
-): ProjectWithThreadsResponse[] {
+): Promise<ProjectWithThreadsResponse[]> {
   const projects = buildProjectResponsesFromRows(deps, projectRows);
   const projectIds = projects.map((project) => project.id);
-  const threadRows = listThreadsWithPendingInteractionStateForProjects(
-    deps.db,
-    { archived: false, projectIds },
-  );
+  const threadRows =
+    await listThreadsWithPendingInteractionStateForProjectsOffThread(deps.db, {
+      archived: false,
+      projectIds,
+    });
   const threadResponses = toThreadListEntryResponses(deps, {
     threads: threadRows,
   });
@@ -253,7 +254,7 @@ function buildProjectsWithThreadsResponseFromRows(
   }));
 }
 
-function buildSidebarBootstrapResponse(deps: AppDeps) {
+async function buildSidebarBootstrapResponse(deps: AppDeps) {
   const personalProject = getPersonalProject(deps.db);
   if (!personalProject) {
     throw new ApiError(
@@ -262,9 +263,8 @@ function buildSidebarBootstrapResponse(deps: AppDeps) {
       "Personal project is not initialized",
     );
   }
-  const personalProjectResponse = buildProjectsWithThreadsResponseFromRows(
-    deps,
-    [personalProject],
+  const personalProjectResponse = (
+    await buildProjectsWithThreadsResponseFromRows(deps, [personalProject])
   )[0];
   if (!personalProjectResponse) {
     throw new ApiError(
@@ -275,7 +275,7 @@ function buildSidebarBootstrapResponse(deps: AppDeps) {
   }
   return {
     sections: listThreadSections(deps.db),
-    projects: buildProjectsWithThreadsResponseFromRows(
+    projects: await buildProjectsWithThreadsResponseFromRows(
       deps,
       listPublicProjects(deps.db),
     ),
@@ -338,13 +338,15 @@ export function registerProjectRoutes(app: Hono, deps: AppDeps): void {
   });
   const routes = publicApiRoutes.projects;
 
-  get(routes.list, (context, query) => {
+  get(routes.list, async (context, query) => {
     const includes = parseProjectListIncludes(query);
     const options: ProjectListOptions = {
       includePersonal: query.includePersonal === "true",
     };
     if (includes.has("threads")) {
-      return context.json(buildProjectsWithThreadsResponse(deps, options));
+      return context.json(
+        await buildProjectsWithThreadsResponse(deps, options),
+      );
     }
     return context.json(
       buildProjectResponsesFromRows(
@@ -354,8 +356,8 @@ export function registerProjectRoutes(app: Hono, deps: AppDeps): void {
     );
   });
 
-  get(routes.sidebarBootstrap, (context) =>
-    context.json(buildSidebarBootstrapResponse(deps)),
+  get(routes.sidebarBootstrap, async (context) =>
+    context.json(await buildSidebarBootstrapResponse(deps)),
   );
 
   post(routes.create, async (context, payload) => {

apps/server/src/routes/threads/base.ts

@@ -6,7 +6,7 @@ import {
   getEnvironment,
   getThreadSectionById,
   listThreadMentionRowsByIds,
-  listThreadsWithPendingInteractionState,
+  listThreadsWithPendingInteractionStateOffThread,
   markThreadDeleted,
   searchThreadsWithPendingInteractionState,
   updateThread,
@@ -255,7 +255,7 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void {
     );
   });
 
-  get(routes.list, (context, query) => {
+  get(routes.list, async (context, query) => {
     const limit = parseOptionalInteger(query.limit, "limit");
     if (limit !== undefined && limit <= 0) {
       throw new ApiError(400, "invalid_request", "limit must be positive");
@@ -277,22 +277,33 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void {
     if (query.sectionId) {
       requireThreadSection(deps, query.sectionId);
     }
-    const threads = listThreadsWithPendingInteractionState(deps.db, {
-      ...(query.projectId ? { projectId: query.projectId } : {}),
-      ...(query.parentThreadId ? { parentThreadId: query.parentThreadId } : {}),
-      ...(query.sourceThreadId ? { sourceThreadId: query.sourceThreadId } : {}),
-      ...(query.sectionId ? { sectionId: query.sectionId } : {}),
-      ...(query.unsectioned === "true" ? { unsectioned: true } : {}),
-      ...(query.originKind ? { originKind: query.originKind } : {}),
-      ...(query.originPluginId ? { originPluginId: query.originPluginId } : {}),
-      includeHidden: query.includeHidden === "true",
-      archived:
-        query.archived === undefined ? undefined : query.archived === "true",
-      hasParent:
-        query.hasParent === undefined ? undefined : query.hasParent === "true",
-      ...(limit !== undefined ? { limit } : {}),
-      ...(offset !== undefined ? { offset } : {}),
-    });
+    const threads = await listThreadsWithPendingInteractionStateOffThread(
+      deps.db,
+      {
+        ...(query.projectId ? { projectId: query.projectId } : {}),
+        ...(query.parentThreadId
+          ? { parentThreadId: query.parentThreadId }
+          : {}),
+        ...(query.sourceThreadId
+          ? { sourceThreadId: query.sourceThreadId }
+          : {}),
+        ...(query.sectionId ? { sectionId: query.sectionId } : {}),
+        ...(query.unsectioned === "true" ? { unsectioned: true } : {}),
+        ...(query.originKind ? { originKind: query.originKind } : {}),
+        ...(query.originPluginId
+          ? { originPluginId: query.originPluginId }
+          : {}),
+        includeHidden: query.includeHidden === "true",
+        archived:
+          query.archived === undefined ? undefined : query.archived === "true",
+        hasParent:
+          query.hasParent === undefined
+            ? undefined
+            : query.hasParent === "true",
+        ...(limit !== undefined ? { limit } : {}),
+        ...(offset !== undefined ? { offset } : {}),
+      },
+    );
     return context.json(
       toThreadListEntryResponses(deps, { threads }) satisfies ThreadListEntry[],
     );

apps/server/src/sqlite-read-worker.ts

@@ -0,0 +1 @@
+import "@bb/db/sqlite-read-worker";

apps/server/src/start-server.ts

@@ -6,7 +6,7 @@ import type { ServerConfig } from "@bb/config/server";
 import { isLoopbackHostname } from "@bb/config/loopback";
 import { toOptionalString } from "@bb/config/strings";
 import { createLogger } from "@bb/logger";
-import { getAppSettings } from "@bb/db";
+import { getAppSettings, startSqliteReadWorker } from "@bb/db";
 import { initDb } from "./db.js";
 import { createApp } from "./server.js";
 import { PendingInteractionLifecycle } from "./services/interactions/pending-interactions.js";
@@ -56,6 +56,25 @@ export async function runServer(serverConfig: ServerConfig): Promise<void> {
     dataDir: serverConfig.BB_DATA_DIR,
     logger,
   });
+  if (serverConfig.databasePath !== ":memory:") {
+    try {
+      const jsFilename = fileURLToPath(
+        new URL("./sqlite-read-worker.js", import.meta.url),
+      );
+      const tsFilename = fileURLToPath(
+        new URL("./sqlite-read-worker.ts", import.meta.url),
+      );
+      startSqliteReadWorker({
+        source: serverConfig.databasePath,
+        workerFilename: existsSync(jsFilename) ? jsFilename : tsFilename,
+      });
+    } catch (error) {
+      logger.error(
+        { err: error },
+        "Failed to start sqlite read worker; heavy reads will stay on the serving thread",
+      );
+    }
+  }
   const hub = new NotificationHub();
   const watchInterests = new WatchInterestCoordinator({ db, hub });
   const sharedPorts = new HostSharedPortCoordinator({ db, hub });

packages/db/package.json

@@ -12,6 +12,11 @@
       "source": "./src/internal-environment-lifecycle.ts",
       "types": "./src/internal-environment-lifecycle.ts",
       "default": "./src/internal-environment-lifecycle.ts"
+    },
+    "./sqlite-read-worker": {
+      "source": "./src/sqlite-read-worker.ts",
+      "types": "./src/sqlite-read-worker.ts",
+      "default": "./src/sqlite-read-worker.ts"
     }
   },
   "types": "./src/index.ts",

packages/db/src/connection.ts

@@ -16,6 +16,7 @@ export interface SlowDbQueryLogger {
 }
 
 export interface CreateConnectionOptions {
+  readonly?: boolean;
   slowQueryLogger?: SlowDbQueryLogger;
   slowQueryThresholdMs?: number;
 }
@@ -156,12 +157,16 @@ export function createConnection(
   source: string | Buffer = "bb.db",
   options: CreateConnectionOptions = {},
 ) {
-  const sqlite = new Database(source);
-
-  sqlite.pragma("auto_vacuum = INCREMENTAL");
-  sqlite.pragma("journal_mode = WAL");
+  const sqlite = options.readonly
+    ? new Database(source, { fileMustExist: true, readonly: true })
+    : new Database(source);
+
+  if (!options.readonly) {
+    sqlite.pragma("auto_vacuum = INCREMENTAL");
+    sqlite.pragma("journal_mode = WAL");
+    sqlite.pragma("synchronous = NORMAL");
+  }
   sqlite.pragma("foreign_keys = ON");
-  sqlite.pragma("synchronous = NORMAL");
   sqlite.pragma(`cache_size = -${SQLITE_CACHE_SIZE_KIB}`);
   sqlite.pragma(`mmap_size = ${SQLITE_MMAP_SIZE_BYTES}`);
   sqlite.pragma(`busy_timeout = ${SQLITE_BUSY_TIMEOUT_MS}`);

packages/db/src/index.ts

@@ -1,4 +1,10 @@
 export { createConnection } from "./connection.js";
+export {
+  listThreadsWithPendingInteractionStateForProjectsOffThread,
+  listThreadsWithPendingInteractionStateOffThread,
+  startSqliteReadWorker,
+  stopSqliteReadWorker,
+} from "./sqlite-read-queue.js";
 export type {
   DbConnection,
   DbQueryConnection,