From 4f09a9407a6b4e48a90113bc60b22e23e78730af Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 09:20:12 +0000 Subject: [PATCH 1/2] fix(server): a slow file read in the search watcher no longer freezes the server fff's watcher holds its index lock while it reads each new file, and WorkspaceSearchIndex called fff synchronously on the event loop, so one slow read stopped every request until it finished. Each fff finder now lives in its own worker thread and every native call (create, search, grep, scan, isScanning, destroy) is an awaited message. A slow read delays only the search waiting on it. The main thread still loads fff-node first: ffi-rs crashes the process when the first thread that loaded it exits. Fixes pingdotgg/t3code#11306 AI-assisted; verified with the tests and tools listed above. --- .../src/workspace/WorkspaceEntries.test.ts | 71 ++++++----- .../src/workspace/WorkspaceFinderThread.ts | 116 ++++++++++++++++++ .../workspace/WorkspaceSearchIndex.test.ts | 29 +++-- .../src/workspace/WorkspaceSearchIndex.ts | 57 +++++---- 4 files changed, 204 insertions(+), 69 deletions(-) create mode 100644 apps/server/src/workspace/WorkspaceFinderThread.ts diff --git a/apps/server/src/workspace/WorkspaceEntries.test.ts b/apps/server/src/workspace/WorkspaceEntries.test.ts index 62d6ab0a63b33..15f320a700837 100644 --- a/apps/server/src/workspace/WorkspaceEntries.test.ts +++ b/apps/server/src/workspace/WorkspaceEntries.test.ts @@ -1,7 +1,6 @@ // @effect-diagnostics nodeBuiltinImport:off import * as NodeFSP from "node:fs/promises"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { FileFinder } from "@ff-labs/fff-node"; import { it, afterEach, describe, expect } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -13,6 +12,7 @@ import { vi } from "vite-plus/test"; import * as ServerConfig from "../config.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as VcsProcess from "../vcs/VcsProcess.ts"; +import { WorkspaceFinderOpener } from "./WorkspaceFinderThread.ts"; import * as WorkspaceEntries from "./WorkspaceEntries.ts"; import * as WorkspacePaths from "./WorkspacePaths.ts"; @@ -793,39 +793,42 @@ it.layer(layerTest, { excludeTestServices: true })("WorkspaceEntries", (it) => { let scansOutlastWait = false; let finders = 0; let rescans = 0; - const createFinder = FileFinder.create.bind(FileFinder); - vi.spyOn(FileFinder, "create").mockImplementation((options) => { - const created = createFinder(options); - if (created.ok) { - const finder = created.value; - finders += 1; - const scanFiles = finder.scanFiles.bind(finder); - finder.scanFiles = () => { - rescans += 1; - return scanFiles(); - }; - const waitForIndexReady = finder.waitForIndexReady.bind(finder); - finder.waitForIndexReady = (timeoutMs) => - scansOutlastWait - ? Promise.resolve({ ok: true, value: false }) - : waitForIndexReady(timeoutMs); - } - return created; - }); - const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; - const searchNeedle = workspaceEntries.searchContents({ - cwd, - query: "needle", - limit: 100, - caseSensitive: true, - wholeWord: false, - useRegex: false, - }); - yield* searchNeedle; - - scansOutlastWait = true; - yield* workspaceEntries.refresh(cwd); - const result = yield* searchNeedle; + const openFinder = yield* WorkspaceFinderOpener; + const countingOpener: typeof openFinder = async (options) => { + const opened = await openFinder(options); + if (!opened.ok) return opened; + finders += 1; + const call: typeof opened.value.call = (method, ...args) => { + if (method === "scanFiles") rescans += 1; + if (method === "waitForIndexReady" && scansOutlastWait) { + return Promise.resolve({ ok: true, value: false } as never); + } + return opened.value.call(method, ...args); + }; + return { ok: true, value: { ...opened.value, call } }; + }; + const result = yield* Effect.gen(function* () { + const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; + const searchNeedle = workspaceEntries.searchContents({ + cwd, + query: "needle", + limit: 100, + caseSensitive: true, + wholeWord: false, + useRegex: false, + }); + yield* searchNeedle; + + scansOutlastWait = true; + yield* workspaceEntries.refresh(cwd); + return yield* searchNeedle; + }).pipe( + Effect.provide( + Layer.fresh(WorkspaceEntries.layer).pipe( + Layer.provide(Layer.succeed(WorkspaceFinderOpener, countingOpener)), + ), + ), + ); expect(result.matches.map((match) => match.path)).toEqual(["src/notes.ts"]); // The second search ran on the rescanned live finder, not a rebuild. diff --git a/apps/server/src/workspace/WorkspaceFinderThread.ts b/apps/server/src/workspace/WorkspaceFinderThread.ts new file mode 100644 index 0000000000000..051efd3af1a2f --- /dev/null +++ b/apps/server/src/workspace/WorkspaceFinderThread.ts @@ -0,0 +1,116 @@ +// @effect-diagnostics nodeBuiltinImport:off -- fff's native calls block their thread, so each finder owns a Node worker. +import * as NodeModule from "node:module"; +import * as NodeWorkerThreads from "node:worker_threads"; + +import type { FileFinder, InitOptions, Result } from "@ff-labs/fff-node"; +import * as Context from "effect/Context"; + +type FinderMethod = + | "destroy" + | "directorySearch" + | "fileSearch" + | "grep" + | "isScanning" + | "mixedSearch" + | "scanFiles" + | "waitForIndexReady"; + +export interface ThreadedFinder { + readonly call: ( + method: M, + ...args: Parameters + ) => Promise>>; + /** Stops the thread once any native call in flight returns; call `destroy` first. */ + readonly close: () => Promise; +} + +export type OpenFinder = (options: InitOptions) => Promise>; + +// An inline worker needs no extra build entry, so src, dist, the single executable and Electron all load it. +// Eval workers define a global `__dirname` of ".", which fff-node would use to look for its native library. +const FINDER_WORKER_SOURCE = ` +delete globalThis.__dirname; +delete globalThis.__filename; +const { parentPort, workerData } = require("node:worker_threads"); +const { FileFinder } = require(workerData.fffEntryPath); +const created = FileFinder.create(workerData.options); +parentPort.postMessage(created.ok ? { ok: true } : created); +if (created.ok) { + const finder = created.value; + parentPort.on("message", async ({ id, method, args }) => { + try { + parentPort.postMessage({ id, value: await finder[method](...args) }); + } catch (cause) { + parentPort.postMessage({ id, failed: true, cause }); + } + }); +} +`; + +type WorkerReply = + | { readonly id: number; readonly value: unknown } + | { readonly id: number; readonly failed: true; readonly cause: unknown }; + +// fff-node's native library stays external; single-executables require disk packages instead of importing them. +const requireForFff = NodeModule.createRequire(import.meta.url); +const fffEntryPath = requireForFff.resolve("@ff-labs/fff-node"); +// ffi-rs crashes the process when the first thread that loaded it exits, so the main thread loads it first. +requireForFff(fffEntryPath); + +const openWorkerFinder: OpenFinder = (options) => + new Promise((resolve, reject) => { + const worker = new NodeWorkerThreads.Worker(FINDER_WORKER_SOURCE, { + eval: true, + workerData: { fffEntryPath, options }, + }); + worker.unref(); + const pending = new Map< + number, + { readonly resolve: (value: never) => void; readonly reject: (cause: unknown) => void } + >(); + let nextId = 0; + let failure: unknown; + const fail = (cause: unknown) => { + failure ??= cause; + reject(failure); + for (const request of pending.values()) request.reject(failure); + pending.clear(); + }; + worker.on("error", fail); + worker.on("exit", (code) => fail(new Error(`The fff worker exited with code ${code}.`))); + + const call: ThreadedFinder["call"] = (method, ...args) => + new Promise((resolveCall, rejectCall) => { + if (failure !== undefined) { + rejectCall(failure); + return; + } + const id = nextId++; + pending.set(id, { resolve: resolveCall, reject: rejectCall }); + worker.postMessage({ id, method, args }); + }); + + worker.once("message", (created: Result) => { + if (!created.ok) { + resolve(created); + return; + } + worker.on("message", (reply: WorkerReply) => { + const request = pending.get(reply.id); + if (!request) return; + pending.delete(reply.id); + if ("failed" in reply) request.reject(reply.cause); + else request.resolve(reply.value as never); + }); + resolve({ + ok: true, + value: { call, close: () => worker.terminate().then(() => undefined) }, + }); + }); + }); + +/** Opens fff finders off the server's event loop, where a native call can block for as long as a file read. */ +export const WorkspaceFinderOpener = Context.Reference( + "t3/workspace/WorkspaceFinderThread/WorkspaceFinderOpener", + { defaultValue: () => openWorkerFinder }, +); diff --git a/apps/server/src/workspace/WorkspaceSearchIndex.test.ts b/apps/server/src/workspace/WorkspaceSearchIndex.test.ts index 2f9ac05bf9e3c..c667e4bf26bcf 100644 --- a/apps/server/src/workspace/WorkspaceSearchIndex.test.ts +++ b/apps/server/src/workspace/WorkspaceSearchIndex.test.ts @@ -1,10 +1,4 @@ -import { - FileFinder, - type FileItem, - type GrepCursor, - type GrepOptions, - type GrepResult, -} from "@ff-labs/fff-node"; +import type { FileFinder, FileItem, GrepCursor, GrepOptions, GrepResult } from "@ff-labs/fff-node"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { afterEach, expect, it } from "@effect/vitest"; import * as HostProcess from "@t3tools/shared/HostProcess"; @@ -13,12 +7,23 @@ import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import { vi } from "vite-plus/test"; +import { WorkspaceFinderOpener } from "./WorkspaceFinderThread.ts"; import * as WorkspaceSearchIndex from "./WorkspaceSearchIndex.ts"; afterEach(() => { vi.restoreAllMocks(); }); +const withInProcessFinder = (finder: FileFinder) => + Effect.provideService(WorkspaceFinderOpener, async () => ({ + ok: true as const, + value: { + call: async (method, ...args) => + (finder[method] as (...input: typeof args) => never)(...args), + close: async () => undefined, + }, + })); + function fileItem(relativePath: string): FileItem { return { relativePath, @@ -53,9 +58,10 @@ it.effect("filters image searches before applying the result limit", () => waitForIndexReady: vi.fn(async () => ({ ok: true as const, value: true })), fileSearch, } as unknown as FileFinder; - vi.spyOn(FileFinder, "create").mockReturnValueOnce({ ok: true, value: finder }); - const searchIndex = yield* WorkspaceSearchIndex.make("/workspace/project"); + const searchIndex = yield* WorkspaceSearchIndex.make("/workspace/project").pipe( + withInProcessFinder(finder), + ); const resultWithoutKind = yield* searchIndex.search("", 200, undefined, true); const resultWithDirectoryKind = yield* searchIndex.search("", 200, "directory", true); @@ -116,9 +122,10 @@ it.effect("continues whole-word searches after a filtered grep page", () => waitForIndexReady: vi.fn(async () => ({ ok: true as const, value: true })), grep, } as unknown as FileFinder; - vi.spyOn(FileFinder, "create").mockReturnValueOnce({ ok: true, value: finder }); - const searchIndex = yield* WorkspaceSearchIndex.make("/workspace/project", "content"); + const searchIndex = yield* WorkspaceSearchIndex.make("/workspace/project", "content").pipe( + withInProcessFinder(finder), + ); const result = yield* searchIndex.searchContents({ query: "needle", limit: 1, diff --git a/apps/server/src/workspace/WorkspaceSearchIndex.ts b/apps/server/src/workspace/WorkspaceSearchIndex.ts index 11a19e1839861..4cd7967196067 100644 --- a/apps/server/src/workspace/WorkspaceSearchIndex.ts +++ b/apps/server/src/workspace/WorkspaceSearchIndex.ts @@ -1,10 +1,7 @@ -import * as NodeModule from "node:module"; - import type { DirItem, DirSearchResult, FileItem, - FileFinder as FileFinderType, GrepCursor, MixedItem, MixedSearchResult, @@ -27,9 +24,7 @@ import type { } from "@t3tools/contracts"; import { isProjectFaviconPath } from "@t3tools/shared/projectFavicon"; -// fff-node's native library stays external; single-executables require disk packages instead of importing them. -const requireForFff = NodeModule.createRequire(import.meta.url); -const { FileFinder } = requireForFff("@ff-labs/fff-node") as typeof import("@ff-labs/fff-node"); +import { type ThreadedFinder, WorkspaceFinderOpener } from "./WorkspaceFinderThread.ts"; const WORKSPACE_INDEX_MAX_ENTRIES = 25_000; const WORKSPACE_INDEX_PAGE_SIZE = WORKSPACE_INDEX_MAX_ENTRIES + 2; @@ -288,9 +283,10 @@ const createFinder = Effect.fn("WorkspaceSearchIndex.createFinder")(function* ( cwd: string, variant: WorkspaceSearchIndexVariant, ) { - const result = yield* Effect.try({ + const openFinder = yield* WorkspaceFinderOpener; + const result = yield* Effect.tryPromise({ try: () => - FileFinder.create({ + openFinder({ basePath: cwd, disableMmapCache: true, // Content indexing costs CPU/memory; path-only consumers use the lightweight index. @@ -313,11 +309,11 @@ const createFinder = Effect.fn("WorkspaceSearchIndex.createFinder")(function* ( const waitForIndexReady = Effect.fn("WorkspaceSearchIndex.waitForIndexReady")(function* ( cwd: string, - finder: FileFinderType, + finder: ThreadedFinder, onFailure: (input: { readonly reason: string; readonly cause?: unknown }) => E, ): Effect.fn.Return { const result = yield* Effect.tryPromise({ - try: () => finder.waitForIndexReady(WORKSPACE_INDEX_SCAN_TIMEOUT_MS), + try: () => finder.call("waitForIndexReady", WORKSPACE_INDEX_SCAN_TIMEOUT_MS), catch: (cause) => onFailure({ reason: "FileFinder.waitForIndexReady rejected unexpectedly.", @@ -340,8 +336,8 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( variant: WorkspaceSearchIndexVariant = "paths", ) { const finder = yield* Effect.acquireRelease(createFinder(cwd, variant), (finder) => - Effect.try({ - try: () => finder.destroy(), + Effect.tryPromise({ + try: () => finder.call("destroy").finally(finder.close), catch: (cause) => new WorkspaceSearchIndexDestroyFailed({ cwd, cause }), }).pipe(Effect.orDie), ); @@ -366,15 +362,28 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( }), ); - const hasIncompleteInitialScan = () => initialScanTimedOut && finder.isScanning(); + const hasIncompleteInitialScan = () => + initialScanTimedOut + ? Effect.tryPromise({ + try: () => finder.call("isScanning"), + catch: (cause) => + new WorkspaceSearchIndexSearchFailed({ + cwd, + queryLength: 0, + pageSize: 0, + reason: "FileFinder.isScanning threw unexpectedly.", + cause, + }), + }) + : Effect.succeed(false); const runSearch = Effect.fn("WorkspaceSearchIndex.runSearch")(function* ( query: string, pageSize: number, operation: "directorySearch" | "fileSearch" | "grep" | "mixedSearch", - execute: () => Result, + execute: () => Promise>, ): Effect.fn.Return { - const result = yield* Effect.try({ + const result = yield* Effect.tryPromise({ try: execute, catch: (cause) => new WorkspaceSearchIndexSearchFailed({ @@ -399,8 +408,8 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( const refresh: WorkspaceSearchIndex["Service"]["refresh"] = Effect.fn( "WorkspaceSearchIndex.refresh", )(function* () { - const result = yield* Effect.try({ - try: () => finder.scanFiles(), + const result = yield* Effect.tryPromise({ + try: () => finder.call("scanFiles"), catch: (cause) => new WorkspaceSearchIndexRefreshFailed({ cwd, @@ -430,9 +439,9 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( const list: WorkspaceSearchIndex["Service"]["list"] = Effect.fn("WorkspaceSearchIndex.list")( function* () { - const incompleteBeforeQuery = hasIncompleteInitialScan(); + const incompleteBeforeQuery = yield* hasIncompleteInitialScan(); const result = yield* runSearch("", WORKSPACE_INDEX_PAGE_SIZE, "mixedSearch", () => - finder.mixedSearch("", { pageSize: WORKSPACE_INDEX_PAGE_SIZE }), + finder.call("mixedSearch", "", { pageSize: WORKSPACE_INDEX_PAGE_SIZE }), ); const mapped = mapMixedSearchResult(result, WORKSPACE_INDEX_MAX_ENTRIES); const sortedEntries = withDirectoryAncestors(mapped.entries).toSorted((left, right) => @@ -454,23 +463,23 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( // matching is subsequence-based, so dropping "[" still matches "[id]" folders. const query = rawQuery.replaceAll("[", ""); const pageSize = imageOnly ? WORKSPACE_INDEX_PAGE_SIZE : Math.max(1, limit + 1); - const incompleteBeforeQuery = hasIncompleteInitialScan(); + const incompleteBeforeQuery = yield* hasIncompleteInitialScan(); if (kind === "file" || imageOnly) { const result = yield* runSearch(query, pageSize, "fileSearch", () => - finder.fileSearch(query, { pageSize }), + finder.call("fileSearch", query, { pageSize }), ); const mapped = mapFileSearchResult(result, limit, imageOnly); return { ...mapped, truncated: incompleteBeforeQuery || mapped.truncated }; } if (kind === "directory") { const result = yield* runSearch(query, pageSize, "directorySearch", () => - finder.directorySearch(query, { pageSize }), + finder.call("directorySearch", query, { pageSize }), ); const mapped = mapDirectorySearchResult(result, limit); return { ...mapped, truncated: incompleteBeforeQuery || mapped.truncated }; } const result = yield* runSearch(query, pageSize, "mixedSearch", () => - finder.mixedSearch(query, { pageSize }), + finder.call("mixedSearch", query, { pageSize }), ); const mapped = mapMixedSearchResult(result, limit); return { ...mapped, truncated: incompleteBeforeQuery || mapped.truncated }; @@ -493,7 +502,7 @@ export const make = Effect.fn("WorkspaceSearchIndex.make")(function* ( do { const remainingTimeBudgetMs = Math.max(1, Math.ceil(deadline - performance.now())); const result = yield* runSearch(input.query, input.limit, "grep", () => - finder.grep(searchQuery, { + finder.call("grep", searchQuery, { mode: regexMode ? "regex" : "plain", smartCase: !input.caseSensitive && !regexMode, maxMatchesPerFile: Math.min(CONTENT_SEARCH_MAX_MATCHES_PER_FILE, rawPageSize), From 732eeb5a901e007900f81a814fdf0976e521fcca Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 10:05:42 +0000 Subject: [PATCH 2/2] fix(server): fff worker ignores Node watch-mode messages Under node --watch (how the dev server runs) Node posts its own watch:require messages on a worker's parentPort. The finder opener read the first one as fff's create result, so every search index failed to open with a schema error. Replies are now tagged and other messages are ignored. AI-assisted; verified with the tests and tools listed above. --- .../src/workspace/WorkspaceFinderThread.ts | 32 +++++++++---------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/apps/server/src/workspace/WorkspaceFinderThread.ts b/apps/server/src/workspace/WorkspaceFinderThread.ts index 051efd3af1a2f..ec15b59661943 100644 --- a/apps/server/src/workspace/WorkspaceFinderThread.ts +++ b/apps/server/src/workspace/WorkspaceFinderThread.ts @@ -34,7 +34,7 @@ delete globalThis.__filename; const { parentPort, workerData } = require("node:worker_threads"); const { FileFinder } = require(workerData.fffEntryPath); const created = FileFinder.create(workerData.options); -parentPort.postMessage(created.ok ? { ok: true } : created); +parentPort.postMessage({ created: created.ok ? { ok: true } : created }); if (created.ok) { const finder = created.value; parentPort.on("message", async ({ id, method, args }) => { @@ -47,7 +47,9 @@ if (created.ok) { } `; -type WorkerReply = +// Node's watch mode posts its own `watch:require` messages on the same port, so every reply is tagged. +type WorkerMessage = + | { readonly created: Result } | { readonly id: number; readonly value: unknown } | { readonly id: number; readonly failed: true; readonly cause: unknown }; @@ -90,22 +92,20 @@ const openWorkerFinder: OpenFinder = (options) => worker.postMessage({ id, method, args }); }); - worker.once("message", (created: Result) => { - if (!created.ok) { - resolve(created); + worker.on("message", (message: WorkerMessage) => { + if ("created" in message) { + resolve( + message.created.ok + ? { ok: true, value: { call, close: () => worker.terminate().then(() => undefined) } } + : message.created, + ); return; } - worker.on("message", (reply: WorkerReply) => { - const request = pending.get(reply.id); - if (!request) return; - pending.delete(reply.id); - if ("failed" in reply) request.reject(reply.cause); - else request.resolve(reply.value as never); - }); - resolve({ - ok: true, - value: { call, close: () => worker.terminate().then(() => undefined) }, - }); + const request = pending.get(message.id); + if (!request) return; + pending.delete(message.id); + if ("failed" in message) request.reject(message.cause); + else request.resolve(message.value as never); }); });