diff --git a/AGENTS.md b/AGENTS.md index 4ba717abd..2074301fd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -343,8 +343,9 @@ root logger (src/logger.ts) - Data-layer functions (`vault-filesystem`, `vault-patcher`, `note-mover`, `memory-store`, `search-index`) take the logger as a **required** second argument (two-arg pattern: `(params, logger)`) -- Background callers (file-watcher, startup) use the root logger - directly — no request context available +- Background modules (`file-watcher`) take the logger the same way, + through their entry point. Startup wiring in `server.ts` has no + request context, so it passes the root logger **Two-arg `(params, logger)` pattern:** diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 1e3a16c6f..69f526feb 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -492,11 +492,22 @@ When no embedder is configured (`EMBEDDING_ENABLED=false`), no vectors are index ### Indexing -**Indexing flow:** `rebuildFromVault` runs three passes, then returns so the server can start accepting requests: +**Indexing flow:** `rebuildFromVault` runs passes 1 and 2, then returns while pass 3 runs in the background: 1. **Pass 1** — index notes (FTS5 + metadata) 2. **Pass 2** — extract links (with the complete path list for resolution), then index file content (canvas, PDF, text → FTS5) -3. **Pass 3 (background)** — embed notes, then file content. Search works with FTS-only until vectors are ready +3. **Pass 3 (background)** — embed notes, then file content. Until a note's vectors exist, search finds it by keyword only + +**Startup order:** `watchVaultAndBuildIndex` starts the file watcher before the rebuild, so a change that lands while the rebuild reads the vault is neither missed nor replaced by the rebuild's older read: + +1. The watcher starts holding: from then on, it records each path chokidar reports as changed instead of indexing it. Startup waits for chokidar's initial scan to finish. +2. `rebuildFromVault` runs. The watcher is the index's only other writer, because a tool's vault write reaches the index as a watcher event. The watcher is holding, so nothing else writes while the rebuild reads. +3. The watcher goes live. In the same step, it removes each held path whose latest event was a deletion, and marks every held path so that a live change to it takes over from the step 4 batch. +4. One batch checks each held path against the disk, `INDEXING_FILE_READ_CONCURRENCY` (16) at a time. Paths gone from disk are removed first. The rest are read again within the rebuild's limits: other files 16 at a time, then the PDFs whose text is extracted, `PDF_EXTRACTION_CONCURRENCY` (4) at a time. + +The server starts accepting requests once the step 4 batch has finished writing to the index, or after 10 s if a Sync backlog makes the batch longer; the batch then finishes in the background. A held path that a live change takes over is indexed by that change, which can finish just after the server starts. + +A file that is created after step 1 and deleted within chokidar's 2 s write-stability window emits no event, so if the rebuild read it, it stays searchable until the next restart. **Startup cleanup:** before resetting the source tables, the rebuild removes vectors whose parent chunk or memory-entry row is missing and retains vectors with surviving parents. @@ -928,8 +939,9 @@ graph LR both processes read/write the shared `/vault` volume. On startup: bootstraps memory templates if the memory folder doesn't exist, `MEMORY_ENABLED` is not `false`, and the server is not in - `READONLY_MODE`, builds the FTS5 search index including those templates, - then starts the file watcher. + `READONLY_MODE`, starts the file watcher, builds the FTS5 search index + including those templates, then applies the changes the watcher saw while + the index was built (see **Startup order** under [Indexing](#indexing)). `svc-vault-mcp` declares `svc-obsidian-sync` in its `dependencies.d`, so the MCP server starts only after the full init chain has finished and the sync diff --git a/deploy/local/README.md b/deploy/local/README.md index e26a8ad2b..b4fbe69af 100644 --- a/deploy/local/README.md +++ b/deploy/local/README.md @@ -27,7 +27,8 @@ What happens on first start: - The image is pulled (~150MB) - The keyword index builds in seconds — search works right away - The semantic (embedding) index builds in the background — expect a few - minutes on a large vault, with search served keyword-only until it finishes + minutes on a large vault; keyword search works meanwhile, and semantic + matches appear as each note is embedded > **On Windows?** Set `WINDOWS_MODE=true` in your `.env` — then `VAULT_PATH` can point > at a normal Windows path like `C:\Users\you\MyVault`. With the CLI, edit the @@ -192,8 +193,10 @@ The server runs these startup tasks on every boot: 1. Create memory template files if the memory folder doesn't exist (skipped when `MEMORY_ENABLED=false` or `READONLY_MODE=true`). -2. Rebuild the search index, including any new memory template files. -3. Start the file watcher. +2. Start the file watcher, which holds the changes it sees until step 4. +3. Rebuild the search index, including any new memory template files. +4. Apply the changes the watcher held, so search reflects notes added, edited + or deleted while the server started. Restarting the container re-runs this flow (useful when testing bootstrap behavior). The command is the same for both setup methods, diff --git a/deploy/remote/README.md b/deploy/remote/README.md index 2e433e087..cc0214c4f 100644 --- a/deploy/remote/README.md +++ b/deploy/remote/README.md @@ -437,8 +437,10 @@ starts. The server then runs these startup tasks: 1. Create memory template files if the memory folder doesn't exist (skipped when `MEMORY_ENABLED=false` or `READONLY_MODE=true`). -2. Rebuild the search index, including any new memory template files. -3. Start the file watcher. +2. Start the file watcher, which holds the changes it sees until step 4. +3. Rebuild the search index, including any new memory template files. +4. Apply the changes the watcher held, so search reflects notes added, edited + or deleted while the server started. Restarting the container re-runs this flow (useful when testing bootstrap behavior). The command is the same for every setup diff --git a/src/__tests__/integration/server-integration.test.ts b/src/__tests__/integration/server-integration.test.ts index 8b25a4ff7..2932d39e6 100644 --- a/src/__tests__/integration/server-integration.test.ts +++ b/src/__tests__/integration/server-integration.test.ts @@ -77,6 +77,41 @@ it("indexes newly bootstrapped memory templates before accepting requests", asyn ) }) +// Startup holds file-watcher events until its index batch ends, so a server +// that never stopped holding would never index this note. +it("indexes a note written after startup once the file watcher reports it", async () => { + const server = await startServer(await freePort()) + onTestFinished(server.cleanup) + const client = await createTestClient(server.port) + onTestFinished(() => client.close()) + + const writeResult = await callTool({ + client, + name: "vault_write_note", + args: { path: "watcher-smoke.md", body: "watchersmokeopal\n" }, + }) + expect(writeResult.isError).not.toBe(true) + + // Polls without throwing, so a server that never indexes the note fails on + // the exact assertion below rather than on a timeout. + const pollDeadline = DateTime.now().plus({ seconds: 10 }) + const searchUntilFound = async (): Promise => { + const searchResult = await callTool({ + client, + name: "vault_search", + args: { query: "watchersmokeopal" }, + }) + const foundPaths = JSON.parse(textContent(searchResult)).results.map( + (entry: { path: string }) => entry.path, + ) + + if (foundPaths.length > 0 || DateTime.now() > pollDeadline) return foundPaths + await new Promise((resolve) => setTimeout(resolve, 250)) + return searchUntilFound() + } + expect(await searchUntilFound()).toEqual(["watcher-smoke.md"]) +}, 30_000) + /** Extract joined text from a prompt result's messages. */ const promptText = (result: Awaited>): string => { return result.messages diff --git a/src/vault-mcp/search/__tests__/file-watcher.test.ts b/src/vault-mcp/search/__tests__/file-watcher.test.ts index 1e92865df..e83f5d1b3 100644 --- a/src/vault-mcp/search/__tests__/file-watcher.test.ts +++ b/src/vault-mcp/search/__tests__/file-watcher.test.ts @@ -10,17 +10,19 @@ import { symlink, utimes, readFile, + readdir, } from "node:fs/promises" import { join, resolve } from "node:path" import { tmpdir } from "node:os" import { watch, FSWatcher } from "chokidar" import Database from "better-sqlite3" +import { DateTime } from "luxon" import * as sqliteVec from "sqlite-vec" import { createSearchIndex } from "../search-index.js" import { extractPdfText } from "../../obsidian-markdown/pdf.js" import { readdirOrNull, statOrNull } from "../../../utils/fs.js" import type { SearchIndex } from "../search-index.js" -import { startFileWatcher } from "../file-watcher.js" +import { watchVaultAndBuildIndex } from "../file-watcher.js" import { logger } from "../../../logger.js" // Auto-spy chokidar: spy: true keeps the real implementation, so the @@ -38,6 +40,25 @@ vi.mock("../../obsidian-markdown/pdf.js", { spy: true }) let vault: string let index: SearchIndex +type WatcherOptions = NonNullable[0]["watcherOptions"]> + +/** Starts the watcher with the startup rebuild stubbed to find nothing, so the + * only indexing a test sees comes from the watcher's own events and rescans, + * including for files the test wrote before starting it. */ +const startWatcherWithoutRebuild = async ( + target: Readonly<{ vaultPath: string; search: SearchIndex }>, + watcherOptions?: WatcherOptions, +): Promise => { + vi.spyOn(target.search, "rebuildFromVault").mockResolvedValue({ + count: 0, + embedding: Promise.resolve(), + }) + await watchVaultAndBuildIndex( + { ...target, ...(watcherOptions ? { watcherOptions } : {}) }, + logger, + ) +} + /** Poll until a condition is met, with timeout. */ const waitFor = async (check: () => boolean, timeoutMs = 8000, intervalMs = 100): Promise => { const deadline = Date.now() + timeoutMs @@ -75,24 +96,29 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { registerSharedVaultHooks() it("indexes a new .md file", { timeout: 15000 }, async () => { - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "test.md"), "---\ntitle: Test\n---\n\nHello watcher\n", "utf8") await waitFor(() => index.fullTextSearch({ query: "watcher" }, logger).length > 0) const results = index.fullTextSearch({ query: "watcher" }, logger) - expect(results).toHaveLength(1) - expect(results[0]?.path).toBe("test.md") + expect(results.map((result) => result.path)).toEqual(["test.md"]) }) it("re-indexes a modified file", { timeout: 15000 }, async () => { - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "modify.md"), "original content\n", "utf8") await waitFor(() => index.fullTextSearch({ query: "original" }, logger).length > 0) @@ -101,14 +127,17 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { await waitFor(() => index.fullTextSearch({ query: "updated" }, logger).length > 0) const results = index.fullTextSearch({ query: "updated" }, logger) - expect(results).toHaveLength(1) + expect(results.map((result) => result.path)).toEqual(["modify.md"]) }) it("removes a deleted file from index", { timeout: 15000 }, async () => { - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "delete-me.md"), "ephemeral\n", "utf8") await waitFor(() => index.fullTextSearch({ query: "ephemeral" }, logger).length > 0) @@ -121,10 +150,13 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { }) it("ignores non-.md files", { timeout: 15000 }, async () => { - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "data.json"), '{"key": "value"}', "utf8") await writeFile(join(vault, "check.md"), "check file\n", "utf8") @@ -148,10 +180,13 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { vi.mocked(statOrNull).mockRestore() }) - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "vanished.png"), "gone", "utf8") await writeFile(join(vault, "kept.png"), "here", "utf8") @@ -167,16 +202,22 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { // must not upsert (without the guard, reading .size off null would throw // inside the watcher callback). await waitFor(() => upsertNonMdFileSpy.mock.calls.length > 0) - expect(upsertNonMdFileSpy.mock.calls.map(([path]) => path)).toEqual(["kept.png"]) + expect(upsertNonMdFileSpy).toHaveBeenCalledExactlyOnceWith( + "kept.png", + Buffer.byteLength("here"), + ) }) it("ignores hidden directories", { timeout: 15000 }, async () => { await mkdir(join(vault, ".obsidian"), { recursive: true }) - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, ".obsidian/workspace.md"), "hidden content\n", "utf8") await writeFile(join(vault, "visible.md"), "visible content\n", "utf8") @@ -190,10 +231,13 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { it("indexes a symlinked .md file", { timeout: 15000 }, async () => { await writeFile(join(vault, "real.md"), "symlink watcher target\n", "utf8") - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await symlink("real.md", join(vault, "linked.md")) @@ -205,6 +249,8 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { const paths = index .fullTextSearch({ query: "symlink watcher" }, logger) .map((result) => result.path) + // On macOS a late event for real.md, written just before the watcher + // started, can index it too, so only the link's presence is asserted. expect(paths).toContain("linked.md") }) @@ -217,10 +263,13 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { capturedVersion.resolve(sourceVersion) return sourceVersion }) - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile( join(vault, "embed-test.md"), @@ -244,11 +293,14 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { "indexes a note atomically written into a brand-new directory", { timeout: 15000 }, async () => { - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - newDirectoryRescanDelay: 500, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + newDirectoryRescanDelay: 500, + }, + ) // vault_write_note's sequence — mkdir -p, stage a temp file, rename over // the target — is the pattern that races chokidar's new-directory scan. @@ -260,8 +312,7 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { await waitFor(() => index.fullTextSearch({ query: "raced" }, logger).length > 0) const results = index.fullTextSearch({ query: "raced" }, logger) - expect(results).toHaveLength(1) - expect(results[0]?.path).toBe("brand-new/nested/note.md") + expect(results.map((result) => result.path)).toEqual(["brand-new/nested/note.md"]) }, ) @@ -269,18 +320,20 @@ describe("file-watcher", REAL_WATCHER_RETRY, () => { // WSL2 bridge). It works on any filesystem, so this verifies the usePolling // option is wired through and indexing still happens under polling. it("indexes a new .md file when polling", { timeout: 15000 }, async () => { - await startFileWatcher(vault, index, { - stabilityThreshold: 200, - pollInterval: 50, - usePolling: true, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: index }, + { + stabilityThreshold: 200, + pollInterval: 50, + usePolling: true, + }, + ) await writeFile(join(vault, "polled.md"), "polled content\n", "utf8") await waitFor(() => index.fullTextSearch({ query: "polled" }, logger).length > 0) const results = index.fullTextSearch({ query: "polled" }, logger) - expect(results).toHaveLength(1) - expect(results[0]?.path).toBe("polled.md") + expect(results.map((result) => result.path)).toEqual(["polled.md"]) }) }) @@ -293,10 +346,13 @@ describe("file-watcher — file content indexing", REAL_WATCHER_RETRY, () => { }) const upsertSpy = vi.spyOn(fileIndex, "upsertFileContent") - await startFileWatcher(vault, fileIndex, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: fileIndex }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "notes.txt"), "deployment checklist for production release", "utf8") @@ -315,10 +371,13 @@ describe("file-watcher — file content indexing", REAL_WATCHER_RETRY, () => { const upsertSpy = vi.spyOn(fileIndex, "upsertFileContent") const removeSpy = vi.spyOn(fileIndex, "removeFileContent") - await startFileWatcher(vault, fileIndex, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: fileIndex }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "temp.csv"), "id,name\n1,ephemeral-data-row", "utf8") await waitFor(() => upsertSpy.mock.calls.length > 0) @@ -337,10 +396,13 @@ describe("file-watcher — file content indexing", REAL_WATCHER_RETRY, () => { }) const upsertSpy = vi.spyOn(fileIndex, "upsertFileContent") - await startFileWatcher(vault, fileIndex, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: fileIndex }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "doc.pdf"), buildMinimalPdf()) await waitFor(() => upsertSpy.mock.calls.length > 0) @@ -364,10 +426,13 @@ describe("file-watcher — file content indexing", REAL_WATCHER_RETRY, () => { return sourceVersion }) - await startFileWatcher(vault, fileIndex, { - stabilityThreshold: 200, - pollInterval: 50, - }) + await startWatcherWithoutRebuild( + { vaultPath: vault, search: fileIndex }, + { + stabilityThreshold: 200, + pollInterval: 50, + }, + ) await writeFile(join(vault, "data.csv"), "id,name,value\n1,deploy,active\n", "utf8") @@ -380,7 +445,7 @@ describe("file-watcher — file content indexing", REAL_WATCHER_RETRY, () => { }) }) -describe("startFileWatcher — obsolete events and embedding queues", () => { +describe("file watcher — obsolete events and embedding queues", () => { const createControlledWatcher = async (embedder?: Parameters[1]) => { const testVault = await mkdtemp(join(tmpdir(), "watcher-events-")) onTestFinished(() => rm(testVault, { recursive: true })) @@ -397,7 +462,7 @@ describe("startFileWatcher — obsolete events and embedding queues", () => { watchMock.mockRestore() await watcher.close() }) - const starting = startFileWatcher(testVault, search) + const starting = startWatcherWithoutRebuild({ vaultPath: testVault, search }) watcher.emit("ready") await starting @@ -1382,158 +1447,1583 @@ describe("startFileWatcher — obsolete events and embedding queues", () => { }) }) -describe("startFileWatcher — chokidar watch options", () => { - registerSharedVaultHooks() +/** Pauses the first readFile of one path. "before-disk" waits before reading, + * so the read sees the disk as it is on release; "after-disk" reads first and + * waits before returning what it read. */ +const gateFirstRead = async (filePath: string, placement: "before-disk" | "after-disk") => { + const actualFs = await vi.importActual("node:fs/promises") + const reached = Promise.withResolvers() + const release = Promise.withResolvers() + const gatedPaths = new Set() + vi.mocked(readFile).mockImplementation(async (requestedPath, options) => { + const isGatedRead = requestedPath === filePath && !gatedPaths.has(filePath) + + if (!isGatedRead) return actualFs.readFile(requestedPath, options) + gatedPaths.add(filePath) + if (placement === "before-disk") { + reached.resolve(undefined) + await release.promise + return actualFs.readFile(requestedPath, options) + } + const content = await actualFs.readFile(requestedPath, options) + reached.resolve(undefined) + await release.promise + return content + }) + onTestFinished(() => vi.mocked(readFile).mockRestore()) + return { reached: reached.promise, release: () => release.resolve(undefined) } +} - type FakeWatcher = { - on: (event: string, handler: (...args: unknown[]) => void) => FakeWatcher - } +/** Pauses the first readdir of the vault root after it has read the folder, + * so the startup batch's presence checks use names from before whatever the + * test does while the listing waits. */ +const gateRootListing = async (testVault: string) => { + const actualFs = await vi.importActual("node:fs/promises") + const reached = Promise.withResolvers() + const release = Promise.withResolvers() + const gateState = { used: false } + vi.mocked(readdir).mockImplementation(async (requestedPath, options) => { + const isGatedListing = requestedPath === testVault && !gateState.used + + if (!isGatedListing) return actualFs.readdir(requestedPath, options) + gateState.used = true + const names = await actualFs.readdir(requestedPath, options) + reached.resolve(undefined) + await release.promise + return names + }) + onTestFinished(() => vi.mocked(readdir).mockRestore()) + return { reached: reached.promise, release: () => release.resolve(undefined) } +} - /** Chainable watcher stub that fires "ready" so startFileWatcher resolves. */ - const createFakeWatcher = (): FakeWatcher => { - const watcher: FakeWatcher = { - on(event, handler) { - if (event === "ready") queueMicrotask(() => handler()) - return watcher - }, +describe("watchVaultAndBuildIndex — changes during startup", () => { + /** A controlled watcher whose startup rebuild waits for finishRebuild, unless + * realRebuild is set. The database lives outside the vault, so its own files + * never appear among the held changes. */ + const createControlledStartup = async (setup?: { + embedder?: Parameters[1] + heldChangesDeadline?: number + newDirectoryRescanDelay?: number + rebuild?: "stubbed" | "real" + seedNotes?: Readonly> + /** Runs against the empty vault before the watcher starts, e.g. to write + * files the real rebuild reads or to gate one of those reads. */ + beforeStart?: (testVault: string) => Promise + }) => { + const testVault = await mkdtemp(join(tmpdir(), "watcher-startup-vault-")) + const dataDirectory = await mkdtemp(join(tmpdir(), "watcher-startup-data-")) + onTestFinished(async () => { + await rm(testVault, { recursive: true }) + await rm(dataDirectory, { recursive: true }) + }) + const databasePath = join(dataDirectory, "index.db") + const search = createSearchIndex(databasePath, setup?.embedder, undefined, { + fileToolsEnabled: true, + }) + const database = new Database(databasePath, { readonly: true }) + sqliteVec.load(database) + onTestFinished(() => { + database.close() + }) + for (const [fileName, content] of Object.entries(setup?.seedNotes ?? {})) { + search.upsertNote( + { filePath: fileName, rawContent: content, fileStat: { mtimeMs: 0, size: content.length } }, + logger, + ) + } + const rebuildGate = Promise.withResolvers() + + if (setup?.rebuild !== "real") { + vi.spyOn(search, "rebuildFromVault").mockImplementation(async () => { + await rebuildGate.promise + return { count: 0, embedding: Promise.resolve() } + }) + } + await setup?.beforeStart?.(testVault) + const watcher = new FSWatcher() + const watchMock = vi.mocked(watch).mockReturnValue(watcher) + onTestFinished(async () => { + watchMock.mockRestore() + await watcher.close() + }) + const heldChangesDeadline = setup?.heldChangesDeadline + const newDirectoryRescanDelay = setup?.newDirectoryRescanDelay + const watcherOptions = { + ...(heldChangesDeadline ? { heldChangesDeadline } : {}), + ...(newDirectoryRescanDelay ? { newDirectoryRescanDelay } : {}), + } + const startup = watchVaultAndBuildIndex( + { vaultPath: testVault, search, watcherOptions }, + logger, + ) + watcher.emit("ready") + + const fire = async ( + event: "add" | "change" | "unlink" | "addDir", + fileName: string, + ): Promise => { + const handler = watcher.listeners(event)[0] + + if (!handler) throw new Error(`${event} handler was not registered`) + await handler(join(testVault, fileName)) + } + const notePaths = (): unknown[] => { + return database.prepare("SELECT path FROM notes ORDER BY path").all() + } + return { + testVault, + search, + database, + watcher, + fire, + notePaths, + startup, + finishRebuild: () => rebuildGate.resolve(undefined), + failRebuild: (error: Error) => rebuildGate.reject(error), } - return watcher } - /** Starts the watcher with the given FileWatcherOptions (chokidar.watch stubbed - * for this one test) and returns the options object chokidar.watch was called - * with. mockRestore (via onTestFinished) hands the real, spied watch back to - * the integration tests. */ - const chokidarOptionsFrom = async ( - watcherOptions?: Parameters[2], - ): Promise> => { - const watchMock = vi.mocked(watch) - watchMock.mockReset() - watchMock.mockImplementation(() => createFakeWatcher() as unknown as ReturnType) - onTestFinished(() => watchMock.mockRestore()) - await startFileWatcher("/vault", index, watcherOptions) - expect(watchMock).toHaveBeenCalledTimes(1) - return watchMock.mock.calls[0]?.[1] as Record + const spyOnLogger = (level: "debug" | "info" | "warn" | "error") => { + const levelSpy = vi.spyOn(logger, level) + onTestFinished(() => levelSpy.mockRestore()) + return levelSpy } - it("defaults to native events (usePolling false) with no interval when unset", async () => { - const chokidarOptions = await chokidarOptionsFrom() - expect(chokidarOptions.usePolling).toBe(false) - expect("interval" in chokidarOptions).toBe(false) - }) + it("indexes a change held during the rebuild once the rebuild returns", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup() + const debugSpy = spyOnLogger("debug") + const infoSpy = spyOnLogger("info") + await writeFile(join(testVault, "note.md"), "heldamber text\n") - it("omits interval when usePolling is false", async () => { - const chokidarOptions = await chokidarOptionsFrom({ usePolling: false }) - expect(chokidarOptions.usePolling).toBe(false) - expect("interval" in chokidarOptions).toBe(false) + await fire("add", "note.md") + + expect(debugSpy).toHaveBeenCalledWith("held a change until startup indexing ends", { + path: "note.md", + }) + expect(search.fullTextSearch({ query: "heldamber" }, logger)).toEqual([]) + finishRebuild() + await startup + expect(search.fullTextSearch({ query: "heldamber" }, logger).map((hit) => hit.path)).toEqual([ + "note.md", + ]) + expect(infoSpy).toHaveBeenCalledWith("applied the changes held during startup indexing", { + heldPaths: 1, + }) }) - it("passes interval (300ms) only when usePolling is true", async () => { - const chokidarOptions = await chokidarOptionsFrom({ usePolling: true }) - expect(chokidarOptions.usePolling).toBe(true) - expect(chokidarOptions.interval).toBe(300) + it("removes a held deletion's rows even when a change for the path arrives after it", async () => { + const { search, database, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "gone.md": "ghostquartz text links [[other]]\n\n- [ ] ghost task\n" }, + }) + const debugSpy = spyOnLogger("debug") + const linkRows = () => database.prepare("SELECT source, target FROM links").all() + const taskRows = () => { + return database.prepare("SELECT note_path FROM tasks").all() + } + expect(search.fullTextSearch({ query: "ghostquartz" }, logger).map((hit) => hit.path)).toEqual([ + "gone.md", + ]) + expect(linkRows()).toEqual([{ source: "gone.md", target: "other" }]) + expect(taskRows()).toEqual([{ note_path: "gone.md" }]) + + await fire("unlink", "gone.md") + await fire("change", "gone.md") + + expect(debugSpy).toHaveBeenCalledWith("held a deletion until startup indexing ends", { + path: "gone.md", + }) + expect(search.fullTextSearch({ query: "ghostquartz" }, logger).map((hit) => hit.path)).toEqual([ + "gone.md", + ]) + finishRebuild() + await startup + expect(search.fullTextSearch({ query: "ghostquartz" }, logger)).toEqual([]) + expect(database.prepare("SELECT path FROM notes").all()).toEqual([]) + expect(linkRows()).toEqual([]) + expect(taskRows()).toEqual([]) }) - it("defaults awaitWriteFinish to 2000ms stability and 100ms poll interval", async () => { - const chokidarOptions = await chokidarOptionsFrom() - expect(chokidarOptions.awaitWriteFinish).toEqual({ - stabilityThreshold: 2000, - pollInterval: 100, + it("reads a held path whose last event was an unlink but which exists again", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "back.md": "oldtext\n" }, }) + await writeFile(join(testVault, "back.md"), "returnedopal\n") + expect(search.fullTextSearch({ query: "oldtext" }, logger).map((hit) => hit.path)).toEqual([ + "back.md", + ]) + + await fire("change", "back.md") + await fire("unlink", "back.md") + finishRebuild() + await startup + + expect(search.fullTextSearch({ query: "returnedopal" }, logger).map((hit) => hit.path)).toEqual( + ["back.md"], + ) + expect(search.fullTextSearch({ query: "oldtext" }, logger)).toEqual([]) }) -}) -// The chokidar race the rescan closes (a file landing between the -// new-directory scan and its watch registration) can't be forced from the -// outside, so these tests drive the reconciliation directly: a fake watcher -// captures the addDir handler, reports test-controlled tracking via -// getWatched(), and records add() calls — against a real temp vault and index. -describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => { - registerSharedVaultHooks() + // chokidar sends an unlink for every file it tracks, so the batch leaves a + // path that vanished by its read to that live event. + it("removes a held path deleted between its check and its read when its unlink arrives during the read", async () => { + const { testVault, notePaths, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "late.md": "oldtext\n" }, + }) + const notePath = join(testVault, "late.md") + await writeFile(notePath, "newer text\n") + const readGate = await gateFirstRead(notePath, "before-disk") + expect(notePaths()).toEqual([{ path: "late.md" }]) + + await fire("change", "late.md") + finishRebuild() + await readGate.reached + await unlink(notePath) + await fire("unlink", "late.md") + readGate.release() + await startup + + expect(notePaths()).toEqual([]) + }) - const RESCAN_TEST_OPTIONS = { - stabilityThreshold: 200, - pollInterval: 50, - newDirectoryRescanDelay: 50, - } + it("leaves a held path deleted between its check and its read to its unlink after startup", async () => { + const { testVault, notePaths, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "late.md": "oldtext\n" }, + }) + const notePath = join(testVault, "late.md") + await writeFile(notePath, "newer text\n") + const readGate = await gateFirstRead(notePath, "before-disk") + const infoSpy = spyOnLogger("info") + expect(notePaths()).toEqual([{ path: "late.md" }]) + + await fire("change", "late.md") + finishRebuild() + await readGate.reached + await unlink(notePath) + readGate.release() + await startup + // The batch's read found the file gone and changed nothing, and the batch + // still completed. + expect(notePaths()).toEqual([{ path: "late.md" }]) + expect(infoSpy).toHaveBeenCalledWith("applied the changes held during startup indexing", { + heldPaths: 1, + }) + await fire("unlink", "late.md") - type RescanFakeWatcher = { - fireAddDir: (dirPath: string) => void - addedPaths: string[] - } + expect(notePaths()).toEqual([]) + }) - /** Stubs chokidar.watch with a fake exposing what the rescan consumes, then - * starts the file watcher against it. mockRestore (via onTestFinished) - * hands the real, spied watch back to the integration tests. */ - const startWatcherWithFakeChokidar = async ( - watchedChildren: Record, - watcherOptions: Parameters[2], - ): Promise => { - const handlers = new Map void>() - const addedPaths: string[] = [] - const fakeWatcher = { - on(event: string, handler: (path: string) => void) { - if (event === "ready") queueMicrotask(() => handler("")) - handlers.set(event, handler) - return fakeWatcher - }, - getWatched: () => watchedChildren, - add: (path: string) => { - addedPaths.push(path) - }, + it("removes a renamed note before reading its new path, so incoming links follow", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "B.md": "old home\n", "source.md": "points at [[B]]\n" }, + }) + await mkdir(join(testVault, "folder")) + await writeFile(join(testVault, "folder", "B.md"), "new home\n") + await writeFile(join(testVault, "source.md"), "points at [[B]]\n") + const linkFromSource = () => { + const outgoingLinks = search.getOutgoingLinks({ path: "source.md" }, logger) + return outgoingLinks.map((link) => ({ path: link.path, exists: link.exists })) } - const watchMock = vi.mocked(watch) - watchMock.mockReset() - watchMock.mockImplementation(() => fakeWatcher as unknown as ReturnType) - onTestFinished(() => watchMock.mockRestore()) - await startFileWatcher(vault, index, watcherOptions) + expect(linkFromSource()).toEqual([{ path: "B.md", exists: true }]) - const fireAddDir = (dirPath: string): void => { - const addDirHandler = handlers.get("addDir") + await fire("add", "folder/B.md") + await fire("unlink", "B.md") + finishRebuild() + await startup - if (addDirHandler === undefined) { - throw new Error("addDir handler was not registered") - } - addDirHandler(dirPath) + expect(linkFromSource()).toEqual([{ path: "folder/B.md", exists: true }]) + }) + + it("removes a renamed note's old path before reading its new one when the old path's latest held event is a change", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "B.md": "old home\n", "source.md": "points at [[B]]\n" }, + }) + await mkdir(join(testVault, "folder")) + await writeFile(join(testVault, "folder", "B.md"), "new home\n") + await writeFile(join(testVault, "source.md"), "points at [[B]]\n") + const linkFromSource = () => { + const outgoingLinks = search.getOutgoingLinks({ path: "source.md" }, logger) + return outgoingLinks.map((link) => ({ path: link.path, exists: link.exists })) } - return { fireAddDir, addedPaths } - } + expect(linkFromSource()).toEqual([{ path: "B.md", exists: true }]) - /** Backdates a file's mtime (10 minutes) so the rescan's "still being - * written" guard sees it as settled. */ - const backdateMtime = async (filePath: string): Promise => { - const backdated = new Date(Date.now() - 600_000) - await utimes(filePath, backdated, backdated) - } + // The held change makes B.md's latest event a change, so going live does + // not remove it; the batch's removal step does, before the reads. + await fire("unlink", "B.md") + await fire("change", "B.md") + await fire("add", "folder/B.md") + finishRebuild() + await startup - it("indexes a file on disk that chokidar does not track", { timeout: 15000 }, async () => { - const newDirectory = join(vault, "new-folder") - await mkdir(newDirectory) - const missedPath = join(newDirectory, "missed.md") - await writeFile(missedPath, "missed by the scan\n", "utf8") - await backdateMtime(missedPath) + expect(linkFromSource()).toEqual([{ path: "folder/B.md", exists: true }]) + }) - const { fireAddDir, addedPaths } = await startWatcherWithFakeChokidar({}, RESCAN_TEST_OPTIONS) - fireAddDir(newDirectory) + // A path's latest held event decides what going live does, so a change held + // before B.md's deletion must not keep B.md indexed. + it.each([ + { label: "its deletion", heldEvents: ["unlink"] }, + { label: "a change then its deletion", heldEvents: ["change", "unlink"] }, + ] as const)( + "points incoming links at a renamed note whose new path arrives live during the batch, after $label was held", + async ({ heldEvents }) => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "B.md": "old home\n", "source.md": "points at [[B]]\n" }, + }) + await writeFile(join(testVault, "source.md"), "points at [[B]]\n") + const listingGate = await gateRootListing(testVault) + const linkFromSource = () => { + const outgoingLinks = search.getOutgoingLinks({ path: "source.md" }, logger) + return outgoingLinks.map((link) => ({ path: link.path, exists: link.exists })) + } + expect(linkFromSource()).toEqual([{ path: "B.md", exists: true }]) - await waitFor(() => index.fullTextSearch({ query: "missed" }, logger).length > 0) - const results = index.fullTextSearch({ query: "missed" }, logger) - expect(results).toHaveLength(1) - expect(results[0]?.path).toBe("new-folder/missed.md") - // The file must also be registered with chokidar — indexed but - // untracked, its later deletion would emit no unlink (ghost entry). - expect(addedPaths).toEqual([missedPath]) - }) + for (const heldEvent of heldEvents) { + await fire(heldEvent, "B.md") + } + finishRebuild() + await listingGate.reached + await mkdir(join(testVault, "folder")) + await writeFile(join(testVault, "folder", "B.md"), "new home\n") + await fire("add", "folder/B.md") + listingGate.release() + await startup + + expect(linkFromSource()).toEqual([{ path: "folder/B.md", exists: true }]) + }, + ) - it("does not re-index a file chokidar already tracks", { timeout: 15000 }, async () => { - const newDirectory = join(vault, "new-folder") - await mkdir(newDirectory) - const trackedPath = join(newDirectory, "tracked.md") - const missedPath = join(newDirectory, "missed.md") - await writeFile(trackedPath, "already tracked note\n", "utf8") - await writeFile(missedPath, "missed sibling note\n", "utf8") - await backdateMtime(trackedPath) - await backdateMtime(missedPath) + it("keeps a note that a live add indexed while the batch checked its held deletion", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "P.md": "oldtext\n" }, + }) + const listingGate = await gateRootListing(testVault) + const pathsFor = (query: string) => { + return search.fullTextSearch({ query }, logger).map((hit) => hit.path) + } + expect(pathsFor("oldtext")).toEqual(["P.md"]) + + await fire("unlink", "P.md") + finishRebuild() + // The root listing was read before P.md came back, so P.md's check finds it missing. + await listingGate.reached + await writeFile(join(testVault, "P.md"), "liveopal\n") + await fire("add", "P.md") + expect(pathsFor("liveopal")).toEqual(["P.md"]) + listingGate.release() + await startup + + expect(pathsFor("liveopal")).toEqual(["P.md"]) + }) + + it("keeps a live add for a held path whose check waits behind 16 others", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "P.md": "oldtext\n" }, + }) + const blockerNames = Array.from( + { length: 16 }, + (_unused, blockerIndex) => `blocker-${blockerIndex}.md`, + ) + for (const blockerName of blockerNames) { + await writeFile(join(testVault, blockerName), "blockertext\n") + } + // The 16 blocker checks fill the batch's first group and wait at their stat, + // so P.md's check starts only after its live add. The root listing those + // checks shared was read before P.md came back. + const actualFsUtils = + await vi.importActual("../../../utils/fs.js") + const statGate = Promise.withResolvers() + const blockerStats = { started: 0 } + const blockerPathPrefix = join(testVault, "blocker-") + vi.mocked(statOrNull).mockImplementation(async (requestedPath) => { + const isBlockerCheck = + requestedPath.startsWith(blockerPathPrefix) && blockerStats.started < 16 + + if (isBlockerCheck) { + blockerStats.started += 1 + await statGate.promise + } + return actualFsUtils.statOrNull(requestedPath) + }) + onTestFinished(() => vi.mocked(statOrNull).mockRestore()) + for (const blockerName of blockerNames) { + await fire("change", blockerName) + } + await fire("unlink", "P.md") + + finishRebuild() + await vi.waitFor(() => { + expect(blockerStats.started).toBe(16) + }) + await writeFile(join(testVault, "P.md"), "liveopal\n") + await fire("add", "P.md") + statGate.resolve(undefined) + await startup + + expect(search.fullTextSearch({ query: "liveopal" }, logger).map((hit) => hit.path)).toEqual([ + "P.md", + ]) + }) + + it("removes a held deletion of a re-created note when going live, then reads it again", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "back.md": "oldtext\n" }, + }) + const debugSpy = spyOnLogger("debug") + await writeFile(join(testVault, "back.md"), "returnedopal\n") + + await fire("unlink", "back.md") + finishRebuild() + await startup + + expect(debugSpy).toHaveBeenCalledWith("removed from index", { path: "back.md" }) + expect(search.fullTextSearch({ query: "returnedopal" }, logger).map((hit) => hit.path)).toEqual( + ["back.md"], + ) + expect(search.fullTextSearch({ query: "oldtext" }, logger)).toEqual([]) + }) + + it("applies live events from during the batch, so the latest state on disk wins", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "P.md": "oldamber\n", "Q.md": "quartzrow\n" }, + }) + const infoSpy = spyOnLogger("info") + const pPath = join(testVault, "P.md") + await writeFile(pPath, "oldamber\n") + await writeFile(join(testVault, "Q.md"), "quartzrow\n") + const readGate = await gateFirstRead(pPath, "after-disk") + const pathsFor = (query: string) => { + return search.fullTextSearch({ query }, logger).map((hit) => hit.path) + } + expect(pathsFor("oldamber")).toEqual(["P.md"]) + expect(pathsFor("quartzrow")).toEqual(["Q.md"]) + + await fire("change", "P.md") + finishRebuild() + await readGate.reached + await writeFile(pPath, "newopal\n") + await fire("change", "P.md") + await unlink(join(testVault, "Q.md")) + await fire("unlink", "Q.md") + await fire("change", "Q.md") + await writeFile(join(testVault, "R.md"), "replayjade\n") + await fire("change", "R.md") + readGate.release() + await startup + + expect(infoSpy).toHaveBeenCalledWith("applied the changes held during startup indexing", { + heldPaths: 1, + }) + expect(pathsFor("newopal")).toEqual(["P.md"]) + expect(pathsFor("oldamber")).toEqual([]) + expect(pathsFor("quartzrow")).toEqual([]) + expect(pathsFor("replayjade")).toEqual(["R.md"]) + }) + + it("contains a failed read and keeps applying the other held paths", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup() + const errorSpy = spyOnLogger("error") + const failingPath = join(testVault, "locked.md") + await writeFile(failingPath, "lockedtext\n") + await writeFile(join(testVault, "fine.md"), "finetext\n") + const accessFailure = Object.assign(new Error("controlled access failure"), { code: "EACCES" }) + const actualFs = await vi.importActual("node:fs/promises") + vi.mocked(readFile).mockImplementation(async (requestedPath, options) => { + if (requestedPath === failingPath) throw accessFailure + return actualFs.readFile(requestedPath, options) + }) + onTestFinished(() => vi.mocked(readFile).mockRestore()) + + await fire("change", "locked.md") + await fire("change", "fine.md") + finishRebuild() + await expect(startup).resolves.toEqual({ count: 0, embedding: expect.any(Promise) }) + + expect(errorSpy).toHaveBeenCalledExactlyOnceWith("failed to process file change", { + path: "locked.md", + error: "[Error]: controlled access failure", + }) + expect(search.fullTextSearch({ query: "finetext" }, logger).map((hit) => hit.path)).toEqual([ + "fine.md", + ]) + }) + + it("leaves a path whose check fails to its read, which reports the failure", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup() + const errorSpy = spyOnLogger("error") + const failingPath = join(testVault, "locked.md") + await writeFile(failingPath, "lockedtext\n") + await writeFile(join(testVault, "fine.md"), "finetext\n") + const accessFailure = Object.assign(new Error("controlled stat failure"), { code: "EACCES" }) + const actualFs = await vi.importActual("node:fs/promises") + vi.mocked(stat).mockImplementation(async (requestedPath, options) => { + if (requestedPath === failingPath) throw accessFailure + return actualFs.stat(requestedPath, options) + }) + onTestFinished(() => vi.mocked(stat).mockRestore()) + + await fire("change", "locked.md") + await fire("change", "fine.md") + finishRebuild() + await startup + + expect(errorSpy).toHaveBeenCalledExactlyOnceWith("failed to process file change", { + path: "locked.md", + error: "[Error]: controlled stat failure", + }) + expect(search.fullTextSearch({ query: "finetext" }, logger).map((hit) => hit.path)).toEqual([ + "fine.md", + ]) + }) + + it("goes live and resolves startup when the batch throws unexpectedly", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "gone.md": "ghostquartz text\n" }, + }) + const errorSpy = spyOnLogger("error") + // No batch step catches its own debug line, so a throw here escapes the batch. + spyOnLogger("debug").mockImplementation((message) => { + if (message === "removing a held path that is gone from disk") { + throw new Error("controlled batch failure") + } + }) + + await fire("unlink", "gone.md") + finishRebuild() + await expect(startup).resolves.toEqual({ count: 0, embedding: expect.any(Promise) }) + expect(errorSpy).toHaveBeenCalledExactlyOnceWith( + "failed to apply the changes held during startup", + { error: "[Error]: controlled batch failure" }, + ) + + await writeFile(join(testVault, "live.md"), "livetopaz\n") + await fire("add", "live.md") + expect(search.fullTextSearch({ query: "livetopaz" }, logger).map((hit) => hit.path)).toEqual([ + "live.md", + ]) + }) + + it("reports a note embedding that fails during the batch once, with no unhandled rejection", async () => { + const vector = new Float32Array(384).fill(0.1) + const embedder = { + embedText: vi.fn(async (text: string) => { + if (text === "failing\n\nfailingtext") throw new Error("controlled model failure") + return vector + }), + embedBatch: vi.fn(async (texts: readonly string[]) => texts.map(() => vector)), + } + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + embedder, + }) + const errorSpy = spyOnLogger("error") + const slowPath = join(testVault, "slow.md") + await writeFile(join(testVault, "failing.md"), "failingtext\n") + await writeFile(slowPath, "slowtext\n") + const readGate = await gateFirstRead(slowPath, "before-disk") + + await fire("change", "failing.md") + await fire("change", "slow.md") + finishRebuild() + await readGate.reached + await vi.waitFor(() => { + expect(embedder.embedText).toHaveBeenCalledWith("failing\n\nfailingtext") + }) + // One macrotask lets Node report the rejection as unhandled if nothing + // caught it while the other path's read is still held open. + await new Promise((resolve) => setImmediate(resolve)) + readGate.release() + await startup + + expect(errorSpy).toHaveBeenCalledExactlyOnceWith("failed to process file change", { + path: "failing.md", + error: "[Error]: controlled model failure", + }) + expect(search.fullTextSearch({ query: "slowtext" }, logger).map((hit) => hit.path)).toEqual([ + "slow.md", + ]) + }) + + it("lets startup finish while held paths' embeddings are still running", async () => { + const vector = new Float32Array(384).fill(0.1) + const modelReady = Promise.withResolvers() + const embedder = { + embedText: vi.fn(async () => { + await modelReady.promise + return vector + }), + embedBatch: vi.fn(async (texts: readonly string[]) => { + await modelReady.promise + return texts.map(() => vector) + }), + } + const { testVault, search, database, fire, startup, finishRebuild } = + await createControlledStartup({ embedder }) + await writeFile(join(testVault, "note.md"), "pendingopal\n") + await writeFile(join(testVault, "content.txt"), "pendingjade") + + await fire("change", "note.md") + await fire("change", "content.txt") + finishRebuild() + const waitForStartup = async (): Promise => { + await startup + return "resolved" + } + const startupOutcome = await Promise.race([ + waitForStartup(), + new Promise((resolve) => setTimeout(() => resolve("still waiting"), 1000)), + ]) + + expect(startupOutcome).toBe("resolved") + expect(search.fullTextSearch({ query: "pendingopal" }, logger).map((hit) => hit.path)).toEqual([ + "note.md", + ]) + expect(database.prepare("SELECT path FROM file_content").all()).toEqual([ + { path: "content.txt" }, + ]) + expect(database.prepare("SELECT note_path FROM note_chunks").all()).toEqual([]) + modelReady.resolve(undefined) + await vi.waitFor(() => { + expect(database.prepare("SELECT note_path, chunk_text FROM note_chunks").all()).toEqual([ + { note_path: "note.md", chunk_text: "note\n\npendingopal" }, + ]) + }) + }) + + it("reads at most 16 held files at once", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup() + const fileNames = Array.from({ length: 40 }, (_unused, noteIndex) => `bulk-${noteIndex}.md`) + for (const fileName of fileNames) { + await writeFile(join(testVault, fileName), "bulktopaz\n") + } + const actualFs = await vi.importActual("node:fs/promises") + const readCounts = { inFlight: 0, peak: 0 } + vi.mocked(readFile).mockImplementation(async (requestedPath, options) => { + readCounts.inFlight += 1 + readCounts.peak = Math.max(readCounts.peak, readCounts.inFlight) + await new Promise((resolve) => setTimeout(resolve, 5)) + try { + return await actualFs.readFile(requestedPath, options) + } finally { + readCounts.inFlight -= 1 + } + }) + onTestFinished(() => vi.mocked(readFile).mockRestore()) + + for (const fileName of fileNames) { + await fire("add", fileName) + } + finishRebuild() + await startup + + expect(readCounts.peak).toBe(16) + const indexedPaths = search + .fullTextSearch({ query: "bulktopaz", limit: 50 }, logger) + .map((hit) => hit.path) + expect(indexedPaths.toSorted()).toEqual(fileNames.toSorted()) + }) + + it("indexes a change that follows a deletion of the same path during the batch", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "note.md": "oldtext\n" }, + }) + const batchPath = join(testVault, "batch.md") + await writeFile(batchPath, "batchonyx\n") + const readGate = await gateFirstRead(batchPath, "before-disk") + const pathsFor = (query: string) => { + return search.fullTextSearch({ query }, logger).map((hit) => hit.path) + } + + await fire("add", "batch.md") + finishRebuild() + await readGate.reached + await writeFile(join(testVault, "note.md"), "newopal\n") + await fire("unlink", "note.md") + await fire("change", "note.md") + readGate.release() + await startup + + expect(pathsFor("newopal")).toEqual(["note.md"]) + expect(pathsFor("oldtext")).toEqual([]) + }) + + it("drops a queued read of the old spelling when a case-only rename lands during the batch", async () => { + const { testVault, notePaths, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "note.md": "caseopal\n" }, + }) + const oldPath = join(testVault, "note.md") + const renamedPath = join(testVault, "Note.md") + await writeFile(oldPath, "caseopal\n") + const blockerNames = Array.from( + { length: 16 }, + (_unused, blockerIndex) => `blocker-${blockerIndex}.md`, + ) + for (const blockerName of blockerNames) { + await writeFile(join(testVault, blockerName), "blockertext\n") + } + // The 16 blocker reads fill the batch's first group and wait for + // blockerGate, so note.md's read stays queued after its check saw it on + // disk. Once renamed, a case-insensitive filesystem still answers for the + // old spelling, so reading it would index note.md again. + const actualFs = await vi.importActual("node:fs/promises") + const blockerGate = Promise.withResolvers() + const blockerReads = { started: 0 } + const renameState = { renamed: false } + const blockerPathPrefix = join(testVault, "blocker-") + const asFoundOnDisk = (requestedPath: RequestedPath): RequestedPath | string => { + return renameState.renamed && requestedPath === oldPath ? renamedPath : requestedPath + } + vi.mocked(stat).mockImplementation(async (requestedPath, options) => { + return actualFs.stat(asFoundOnDisk(requestedPath), options) + }) + vi.mocked(readFile).mockImplementation(async (requestedPath, options) => { + const isBlockerRead = + typeof requestedPath === "string" && requestedPath.startsWith(blockerPathPrefix) + + if (isBlockerRead) { + blockerReads.started += 1 + await blockerGate.promise + } + return actualFs.readFile(asFoundOnDisk(requestedPath), options) + }) + onTestFinished(() => { + vi.mocked(stat).mockRestore() + vi.mocked(readFile).mockRestore() + }) + for (const blockerName of blockerNames) { + await fire("change", blockerName) + } + await fire("change", "note.md") + + finishRebuild() + await vi.waitFor(() => { + expect(blockerReads.started).toBe(16) + }) + await rename(oldPath, renamedPath) + renameState.renamed = true + await fire("unlink", "note.md") + await fire("add", "Note.md") + blockerGate.resolve(undefined) + await startup + + const expectedPaths = ["Note.md", ...blockerNames].toSorted() + expect(notePaths()).toEqual(expectedPaths.map((path) => ({ path }))) + }) + + it("skips a folder rescan that was scheduled before a failed rebuild closed the watcher", async () => { + const { testVault, watcher, fire, startup, failRebuild } = await createControlledStartup({ + newDirectoryRescanDelay: 50, + }) + const debugSpy = spyOnLogger("debug") + const freshFolder = join(testVault, "fresh") + await mkdir(freshFolder) + await writeFile(join(freshFolder, "note.md"), "freshtext\n") + const addSpy = vi.spyOn(watcher, "add") + + await fire("addDir", "fresh") + failRebuild(new Error("controlled rebuild failure")) + await expect(startup).rejects.toThrow(new Error("controlled rebuild failure")) + await vi.waitFor(() => { + expect(debugSpy).toHaveBeenCalledWith("rescan skipped, watcher closed", { path: "fresh" }) + }) + + expect(vi.mocked(readdirOrNull)).not.toHaveBeenCalledWith(freshFolder) + expect(addSpy).not.toHaveBeenCalled() + expect(watcher.closed).toBe(true) + }) + + it("indexes and watches nothing from a rescan that was running when the watcher closed", async () => { + const { testVault, watcher, fire, startup, failRebuild } = await createControlledStartup({ + newDirectoryRescanDelay: 20, + }) + const debugSpy = spyOnLogger("debug") + const freshNote = join(testVault, "fresh", "note.md") + await mkdir(join(testVault, "fresh")) + await writeFile(freshNote, "freshtext\n") + // Older than the stability window, so the rescan watches and indexes it at once. + const settledSeconds = DateTime.now().minus({ minutes: 10 }).toUnixInteger() + await utimes(freshNote, settledSeconds, settledSeconds) + const actualFs = await vi.importActual("node:fs/promises") + const statReached = Promise.withResolvers() + const statRelease = Promise.withResolvers() + // Holds the rescan's stat of the note until the watcher has closed. + vi.mocked(stat).mockImplementation(async (requestedPath, options) => { + if (requestedPath === freshNote) { + statReached.resolve(undefined) + await statRelease.promise + } + return actualFs.stat(requestedPath, options) + }) + onTestFinished(() => vi.mocked(stat).mockRestore()) + const addSpy = vi.spyOn(watcher, "add") + + await fire("addDir", "fresh") + await statReached.promise + failRebuild(new Error("controlled rebuild failure")) + await expect(startup).rejects.toThrow(new Error("controlled rebuild failure")) + statRelease.resolve(undefined) + await vi.waitFor(() => { + expect(debugSpy).toHaveBeenCalledWith("rescan indexing missed file", { + path: join("fresh", "note.md"), + }) + }) + + // The rescan logs just before handleChange, which would read the note + // before its first await. + expect(vi.mocked(readFile)).not.toHaveBeenCalledWith(freshNote, "utf8") + expect(addSpy).not.toHaveBeenCalled() + expect(watcher.closed).toBe(true) + }) + + it("returns 10 seconds after the rebuild by default while held changes are still applied", async () => { + const { testVault, fire, startup, finishRebuild } = await createControlledStartup() + const infoSpy = spyOnLogger("info") + const slowPath = join(testVault, "slow.md") + await writeFile(slowPath, "slowtext\n") + const readGate = await gateFirstRead(slowPath, "before-disk") + onTestFinished(() => readGate.release()) + // Fake only setTimeout, so the batch's real filesystem work still runs. + vi.useFakeTimers({ toFake: ["setTimeout"] }) + onTestFinished(() => { + vi.useRealTimers() + }) + + await fire("change", "slow.md") + finishRebuild() + await readGate.reached + await vi.advanceTimersByTimeAsync(9_999) + + expect(infoSpy).not.toHaveBeenCalledWith( + "startup continues while the changes held during it are still applied", + { heldPaths: 1, heldChangesDeadline: 10_000 }, + ) + await vi.advanceTimersByTimeAsync(1) + expect(infoSpy).toHaveBeenCalledWith( + "startup continues while the changes held during it are still applied", + { heldPaths: 1, heldChangesDeadline: 10_000 }, + ) + await startup + }) + + it("extracts at most 4 held PDFs at once", async () => { + const { testVault, database, fire, startup, finishRebuild } = await createControlledStartup() + const fileNames = Array.from({ length: 10 }, (_unused, pdfIndex) => `doc-${pdfIndex}.pdf`) + // Extraction is stubbed, so the bytes only need to exist. + for (const fileName of fileNames) { + await writeFile(join(testVault, fileName), "controlled PDF bytes") + } + const extractionCounts = { inFlight: 0, peak: 0 } + const extractionSpy = vi.mocked(extractPdfText).mockImplementation(async () => { + extractionCounts.inFlight += 1 + extractionCounts.peak = Math.max(extractionCounts.peak, extractionCounts.inFlight) + await new Promise((resolve) => setTimeout(resolve, 5)) + extractionCounts.inFlight -= 1 + return { text: "pdftopaz", totalPages: 1 } + }) + onTestFinished(() => extractionSpy.mockRestore()) + + for (const fileName of fileNames) { + await fire("add", fileName) + } + finishRebuild() + await startup + + expect(extractionCounts.peak).toBe(4) + expect(database.prepare("SELECT path FROM file_content ORDER BY path").all()).toEqual( + fileNames.toSorted().map((path) => ({ path })), + ) + }) + + it("returns at the deadline and finishes the held changes in the background", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup({ + heldChangesDeadline: 50, + }) + const infoSpy = spyOnLogger("info") + const debugSpy = spyOnLogger("debug") + const slowPath = join(testVault, "slow.md") + await writeFile(slowPath, "slowtext\n") + const readGate = await gateFirstRead(slowPath, "before-disk") + const pathsFor = (query: string) => { + return search.fullTextSearch({ query }, logger).map((hit) => hit.path) + } + + await fire("change", "slow.md") + finishRebuild() + const waitForStartup = async (): Promise => { + await startup + return "resolved" + } + const startupOutcome = await Promise.race([ + waitForStartup(), + new Promise((resolve) => setTimeout(() => resolve("still waiting"), 1000)), + ]) + + expect(startupOutcome).toBe("resolved") + expect(infoSpy).toHaveBeenCalledWith( + "startup continues while the changes held during it are still applied", + { heldPaths: 1, heldChangesDeadline: 50 }, + ) + expect(pathsFor("slowtext")).toEqual([]) + // The watcher went live before the batch started, so a change while the + // batch's read is still gated is indexed at once. + await writeFile(join(testVault, "S.md"), "aftertext\n") + await fire("change", "S.md") + expect(pathsFor("aftertext")).toEqual(["S.md"]) + expect(debugSpy).not.toHaveBeenCalledWith("held a change until startup indexing ends", { + path: "S.md", + }) + readGate.release() + await vi.waitFor(() => { + expect(pathsFor("slowtext")).toEqual(["slow.md"]) + }) + }) + + it("does not index the old name again after a case-only rename", async () => { + const { testVault, notePaths, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "note.md": "caseopal\n" }, + }) + const renamedPath = join(testVault, "Note.md") + const oldPath = join(testVault, "note.md") + await writeFile(renamedPath, "caseopal\n") + const actualFs = await vi.importActual("node:fs/promises") + // A case-insensitive filesystem answers for the old spelling too; the + // folder listing still holds only the new one. + const asFoundOnDisk = (requestedPath: RequestedPath): RequestedPath | string => { + return requestedPath === oldPath ? renamedPath : requestedPath + } + vi.mocked(stat).mockImplementation(async (requestedPath, options) => { + return actualFs.stat(asFoundOnDisk(requestedPath), options) + }) + vi.mocked(readFile).mockImplementation(async (requestedPath, options) => { + return actualFs.readFile(asFoundOnDisk(requestedPath), options) + }) + onTestFinished(() => { + vi.mocked(stat).mockRestore() + vi.mocked(readFile).mockRestore() + }) + + await fire("unlink", "note.md") + await fire("add", "Note.md") + finishRebuild() + await startup + + expect(notePaths()).toEqual([{ path: "Note.md" }]) + }) + + it("does not index the old path again after a case-only rename of its folder", async () => { + const { testVault, notePaths, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "notes/x.md": "folderjade\n" }, + }) + const oldFolder = join(testVault, "notes") + const renamedFolder = join(testVault, "Notes") + await mkdir(renamedFolder) + await writeFile(join(renamedFolder, "x.md"), "folderjade\n") + const actualFs = await vi.importActual("node:fs/promises") + // A case-insensitive filesystem answers for the old folder spelling too; + // only the vault root's listing holds the new one. + const asFoundOnDisk = (requestedPath: RequestedPath): RequestedPath | string => { + if (typeof requestedPath !== "string") return requestedPath + if (requestedPath !== oldFolder && !requestedPath.startsWith(`${oldFolder}/`)) { + return requestedPath + } + return `${renamedFolder}${requestedPath.slice(oldFolder.length)}` + } + vi.mocked(readdir).mockImplementation(async (requestedPath, options) => { + return actualFs.readdir(asFoundOnDisk(requestedPath), options) + }) + vi.mocked(stat).mockImplementation(async (requestedPath, options) => { + return actualFs.stat(asFoundOnDisk(requestedPath), options) + }) + vi.mocked(readFile).mockImplementation(async (requestedPath, options) => { + return actualFs.readFile(asFoundOnDisk(requestedPath), options) + }) + onTestFinished(() => { + vi.mocked(readdir).mockRestore() + vi.mocked(stat).mockRestore() + vi.mocked(readFile).mockRestore() + }) + expect(notePaths()).toEqual([{ path: "notes/x.md" }]) + + // On macOS, chokidar reports a case-only folder rename as a change to the + // old path, never an unlink. + await fire("change", "notes/x.md") + await fire("add", "Notes/x.md") + finishRebuild() + await startup + + expect(notePaths()).toEqual([{ path: "Notes/x.md" }]) + }) + + it("removes a held note whose path is now a folder", async () => { + const { testVault, notePaths, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "shape.md": "shapeopal\n" }, + }) + const errorSpy = spyOnLogger("error") + await mkdir(join(testVault, "shape.md")) + expect(notePaths()).toEqual([{ path: "shape.md" }]) + + await fire("change", "shape.md") + finishRebuild() + await startup + + expect(notePaths()).toEqual([]) + expect(errorSpy).not.toHaveBeenCalled() + }) + + it("removes a held note whose folder is gone, with no listing warning", async () => { + const { notePaths, fire, startup, finishRebuild } = await createControlledStartup({ + seedNotes: { "gone-folder/note.md": "folderopal\n" }, + }) + const warnSpy = spyOnLogger("warn") + expect(notePaths()).toEqual([{ path: "gone-folder/note.md" }]) + + await fire("unlink", "gone-folder/note.md") + finishRebuild() + await startup + + expect(notePaths()).toEqual([]) + expect(warnSpy).not.toHaveBeenCalled() + }) + + it("reads a held note whose folder cannot be listed", async () => { + const { testVault, search, fire, startup, finishRebuild } = await createControlledStartup() + const warnSpy = spyOnLogger("warn") + const lockedFolder = join(testVault, "locked") + await mkdir(lockedFolder) + await writeFile(join(lockedFolder, "note.md"), "listedopal\n") + const actualFs = await vi.importActual("node:fs/promises") + const listingFailure = Object.assign(new Error("controlled listing failure"), { + code: "EACCES", + }) + vi.mocked(readdir).mockImplementation(async (requestedPath, options) => { + if (requestedPath === lockedFolder) throw listingFailure + return actualFs.readdir(requestedPath, options) + }) + onTestFinished(() => vi.mocked(readdir).mockRestore()) + + await fire("change", "locked/note.md") + finishRebuild() + await startup + + expect(warnSpy).toHaveBeenCalledExactlyOnceWith( + "could not list a folder of held changes, so stat checks its files", + { path: "locked", error: "[Error]: controlled listing failure" }, + ) + expect(search.fullTextSearch({ query: "listedopal" }, logger).map((hit) => hit.path)).toEqual([ + "locked/note.md", + ]) + }) + + it("keeps a held file's rows when its stat fails during the batch", async () => { + const { testVault, search, database, fire, startup, finishRebuild } = + await createControlledStartup() + const errorSpy = spyOnLogger("error") + const imagePath = join(testVault, "image.png") + await writeFile(imagePath, "image data") + await writeFile(join(testVault, "fine.md"), "finetext\n") + search.upsertNonMdFile("image.png", 10) + const assetPaths = () => database.prepare("SELECT path FROM non_md_files").all() + expect(assetPaths()).toEqual([{ path: "image.png" }]) + const actualFsUtils = + await vi.importActual("../../../utils/fs.js") + const statFailure = Object.assign(new Error("controlled stat failure"), { code: "EACCES" }) + vi.mocked(statOrNull).mockImplementation(async (requestedPath) => { + if (requestedPath === imagePath) throw statFailure + return actualFsUtils.statOrNull(requestedPath) + }) + onTestFinished(() => vi.mocked(statOrNull).mockRestore()) + + await fire("change", "image.png") + await fire("change", "fine.md") + finishRebuild() + await startup + + expect(errorSpy).toHaveBeenCalledExactlyOnceWith("failed to stat non-md file", { + path: "image.png", + error: "[Error]: controlled stat failure", + }) + expect(assetPaths()).toEqual([{ path: "image.png" }]) + expect(search.fullTextSearch({ query: "finetext" }, logger).map((hit) => hit.path)).toEqual([ + "fine.md", + ]) + }) + + it("removes a held file's rows when it is deleted between its check and its read and its unlink arrives", async () => { + const { testVault, search, database, fire, startup, finishRebuild } = + await createControlledStartup() + const contentPath = join(testVault, "content.txt") + await writeFile(contentPath, "vanishjade") + search.upsertNonMdFile("content.txt", 10) + search.upsertFileContent( + { filePath: "content.txt", rawContent: "vanishjade", fileStat: { mtimeMs: 0, size: 10 } }, + logger, + ) + const fileRows = () => ({ + assets: database.prepare("SELECT path FROM non_md_files").all(), + contents: database.prepare("SELECT path FROM file_content").all(), + }) + expect(fileRows()).toEqual({ + assets: [{ path: "content.txt" }], + contents: [{ path: "content.txt" }], + }) + // The batch's check sees the file, which is deleted before the check + // returns, so only the read that follows can find it gone. + const actualFsUtils = + await vi.importActual("../../../utils/fs.js") + const checkedPaths = new Set() + vi.mocked(statOrNull).mockImplementation(async (requestedPath) => { + const fileStat = await actualFsUtils.statOrNull(requestedPath) + + if (requestedPath === contentPath && !checkedPaths.has(contentPath)) { + checkedPaths.add(contentPath) + await unlink(contentPath) + } + return fileStat + }) + onTestFinished(() => vi.mocked(statOrNull).mockRestore()) + const infoSpy = spyOnLogger("info") + + await fire("change", "content.txt") + finishRebuild() + await startup + // The batch's read finds the file gone and leaves it to the unlink + // chokidar sends for it, and the batch still completes. + expect(checkedPaths).toEqual(new Set([contentPath])) + expect(fileRows()).toEqual({ + assets: [{ path: "content.txt" }], + contents: [{ path: "content.txt" }], + }) + expect(infoSpy).toHaveBeenCalledWith("applied the changes held during startup indexing", { + heldPaths: 1, + }) + await fire("unlink", "content.txt") + + expect(fileRows()).toEqual({ assets: [], contents: [] }) + }) + + it("closes the watcher and rejects with the rebuild's error when the rebuild fails", async () => { + const { watcher, startup, failRebuild } = await createControlledStartup() + expect(watcher.closed).toBe(false) + + failRebuild(new Error("controlled rebuild failure")) + + await expect(startup).rejects.toThrow(new Error("controlled rebuild failure")) + expect(watcher.closed).toBe(true) + }) + + it("keeps the rebuild's error when closing the watcher also fails", async () => { + const { watcher, startup, failRebuild } = await createControlledStartup() + const errorSpy = spyOnLogger("error") + vi.spyOn(watcher, "close").mockRejectedValueOnce(new Error("controlled close failure")) + + failRebuild(new Error("controlled rebuild failure")) + + await expect(startup).rejects.toThrow(new Error("controlled rebuild failure")) + expect(errorSpy).toHaveBeenCalledWith("failed to close the file watcher", { + error: "[Error]: controlled close failure", + }) + }) + + it("lets a held change win over the rebuild's older read of the same note", async () => { + const readGates: Array>> = [] + const { testVault, search, fire, startup } = await createControlledStartup({ + rebuild: "real", + beforeStart: async (emptyVault) => { + await writeFile(join(emptyVault, "note.md"), "oldquartz\n") + readGates.push(await gateFirstRead(join(emptyVault, "note.md"), "after-disk")) + }, + }) + const [readGate] = readGates + const notePath = join(testVault, "note.md") + const pathsFor = (query: string) => { + return search.fullTextSearch({ query }, logger).map((hit) => hit.path) + } + + if (!readGate) throw new Error("the rebuild's read gate was not installed") + await readGate.reached + await writeFile(notePath, "newopal\n") + await fire("change", "note.md") + readGate.release() + await startup + + expect(pathsFor("newopal")).toEqual(["note.md"]) + expect(pathsFor("oldquartz")).toEqual([]) + }) +}) + +describe("watchVaultAndBuildIndex — startup through real chokidar", REAL_WATCHER_RETRY, () => { + it( + "indexes edits, creations and deletions that land while the rebuild reads the vault", + { timeout: 20000 }, + async () => { + const testVault = await mkdtemp(join(tmpdir(), "watcher-startup-real-")) + const dataDirectory = await mkdtemp(join(tmpdir(), "watcher-startup-real-data-")) + onTestFinished(async () => { + await rm(testVault, { recursive: true }) + await rm(dataDirectory, { recursive: true }) + }) + const databasePath = join(dataDirectory, "index.db") + const search = createSearchIndex(databasePath, undefined, undefined, { + fileToolsEnabled: true, + }) + const database = new Database(databasePath, { readonly: true }) + onTestFinished(() => { + database.close() + }) + const modifiedPath = join(testVault, "modified.md") + const deletedPath = join(testVault, "deleted.md") + const deletedTextPath = join(testVault, "deleted.txt") + await writeFile(deletedPath, "ghostquartz links [[modified]]\n") + await writeFile(modifiedPath, "oldtext\n") + await writeFile(deletedTextPath, "ghostquartz") + const { mtime: modifiedTimeTheRebuildSaw } = await stat(modifiedPath) + + // The rebuild's read of modified.md is held after it read "oldtext"; + // deleted.md's read is only observed, so the test deletes the file + // after the rebuild has read it. + const actualFs = await vi.importActual("node:fs/promises") + const modifiedReadReached = Promise.withResolvers() + const releaseModifiedRead = Promise.withResolvers() + const deletedNoteRead = Promise.withResolvers() + const gatedPaths = new Set() + vi.mocked(readFile).mockImplementation(async (requestedPath, options) => { + const content = await actualFs.readFile(requestedPath, options) + + if (requestedPath === deletedPath) deletedNoteRead.resolve(undefined) + if (requestedPath === modifiedPath && !gatedPaths.has(modifiedPath)) { + gatedPaths.add(modifiedPath) + modifiedReadReached.resolve(undefined) + await releaseModifiedRead.promise + } + return content + }) + onTestFinished(() => vi.mocked(readFile).mockRestore()) + + // A path counts as delivered once the watcher either holds its event + // (the intended path) or writes it to the index at once (a broken hold), + // so the state assertions below decide which happened. + const deliveredPaths = new Set() + const debugSpy = vi.spyOn(logger, "debug").mockImplementation((message, fields) => { + const isHeldLine = message.startsWith("held a ") + const heldPath = fields?.path + + if (isHeldLine && typeof heldPath === "string") deliveredPaths.add(heldPath) + }) + const infoSpy = vi.spyOn(logger, "info") + onTestFinished(() => { + debugSpy.mockRestore() + infoSpy.mockRestore() + }) + const recordDeliveredPath = ( + operation: (...operationArguments: Arguments) => Result, + pathOf: (...operationArguments: Arguments) => string, + ) => { + return (...operationArguments: Arguments): Result => { + deliveredPaths.add(pathOf(...operationArguments)) + return operation(...operationArguments) + } + } + // The real functions are captured before spying, because inside + // mockImplementation's argument the property already holds the spy, + // and wrapping the spy would recurse. + const { upsertNote, upsertNonMdFile, removeNote, removeNonMdFile } = search + vi.spyOn(search, "upsertNote").mockImplementation( + recordDeliveredPath(upsertNote, (params) => params.filePath), + ) + vi.spyOn(search, "upsertNonMdFile").mockImplementation( + recordDeliveredPath(upsertNonMdFile, (filePath) => filePath), + ) + vi.spyOn(search, "removeNote").mockImplementation( + recordDeliveredPath(removeNote, (filePath) => filePath), + ) + vi.spyOn(search, "removeNonMdFile").mockImplementation( + recordDeliveredPath(removeNonMdFile, (filePath) => filePath), + ) + + const startup = watchVaultAndBuildIndex( + { + vaultPath: testVault, + search, + watcherOptions: { stabilityThreshold: 100, pollInterval: 25 }, + }, + logger, + ) + await modifiedReadReached.promise + await deletedNoteRead.promise + await writeFile(modifiedPath, "newopal is longer than the old text\n") + // Sync writes with equal access and modification times; the old + // modification time shows the outcome does not depend on it changing. + await utimes(modifiedPath, modifiedTimeTheRebuildSaw, modifiedTimeTheRebuildSaw) + await unlink(deletedPath) + await unlink(deletedTextPath) + await writeFile(join(testVault, "created.md"), "newjade\n") + await vi.waitFor( + () => { + expect([...deliveredPaths].toSorted()).toEqual([ + "created.md", + "deleted.md", + "deleted.txt", + "modified.md", + ]) + }, + { timeout: 8000 }, + ) + releaseModifiedRead.resolve(undefined) + await startup + + const pathsFor = (query: string) => { + return search.fullTextSearch({ query }, logger).map((hit) => hit.path) + } + expect(pathsFor("newopal")).toEqual(["modified.md"]) + expect(pathsFor("oldtext")).toEqual([]) + expect(pathsFor("newjade")).toEqual(["created.md"]) + expect(pathsFor("ghostquartz")).toEqual([]) + expect( + database.prepare("SELECT source FROM links WHERE source = 'deleted.md'").all(), + ).toEqual([]) + expect(database.prepare("SELECT path FROM non_md_files").all()).toEqual([]) + expect(database.prepare("SELECT path FROM file_content").all()).toEqual([]) + expect(infoSpy).toHaveBeenCalledWith("applied the changes held during startup indexing", { + heldPaths: 4, + }) + }, + ) +}) + +describe("file watcher — chokidar watch options", () => { + registerSharedVaultHooks() + + type FakeWatcher = { + on: (event: string, handler: (...args: unknown[]) => void) => FakeWatcher + } + + /** Chainable watcher stub that fires "ready" so the watcher becomes ready. */ + const createFakeWatcher = (): FakeWatcher => { + const watcher: FakeWatcher = { + on(event, handler) { + if (event === "ready") queueMicrotask(() => handler()) + return watcher + }, + } + return watcher + } + + /** Starts the watcher with the given FileWatcherOptions, with chokidar.watch + * stubbed for this one test. mockRestore (via onTestFinished) hands the + * real, spied watch back to the integration tests. */ + const startWatcherWithStubbedChokidar = async ( + watcherOptions?: WatcherOptions, + ): Promise => { + const watchMock = vi.mocked(watch) + watchMock.mockReset() + watchMock.mockImplementation(() => createFakeWatcher() as unknown as ReturnType) + onTestFinished(() => watchMock.mockRestore()) + await startWatcherWithoutRebuild({ vaultPath: "/vault", search: index }, watcherOptions) + } + + // The options every watcher passes chokidar. The whole object is asserted, + // so an interval key passed without polling fails the match. + const sharedChokidarOptions = { + ignored: expect.any(Function), + persistent: true, + ignoreInitial: true, + } + const defaultWriteFinish = { stabilityThreshold: 2000, pollInterval: 100 } + + it("defaults to native events (usePolling false) with no interval when unset", async () => { + await startWatcherWithStubbedChokidar() + + expect(vi.mocked(watch)).toHaveBeenCalledExactlyOnceWith("/vault", { + ...sharedChokidarOptions, + usePolling: false, + awaitWriteFinish: defaultWriteFinish, + }) + }) + + it("omits interval when usePolling is false", async () => { + await startWatcherWithStubbedChokidar({ usePolling: false }) + + expect(vi.mocked(watch)).toHaveBeenCalledExactlyOnceWith("/vault", { + ...sharedChokidarOptions, + usePolling: false, + awaitWriteFinish: defaultWriteFinish, + }) + }) + + it("passes interval (300ms) only when usePolling is true", async () => { + await startWatcherWithStubbedChokidar({ usePolling: true }) + + expect(vi.mocked(watch)).toHaveBeenCalledExactlyOnceWith("/vault", { + ...sharedChokidarOptions, + usePolling: true, + interval: 300, + awaitWriteFinish: defaultWriteFinish, + }) + }) + + it("defaults awaitWriteFinish to 2000ms stability and 100ms poll interval", async () => { + await startWatcherWithStubbedChokidar() + + expect(vi.mocked(watch)).toHaveBeenCalledExactlyOnceWith("/vault", { + ...sharedChokidarOptions, + usePolling: false, + awaitWriteFinish: { stabilityThreshold: 2000, pollInterval: 100 }, + }) + }) +}) + +// The chokidar race the rescan closes (a file landing between the +// new-directory scan and its watch registration) can't be forced from the +// outside, so these tests drive the reconciliation directly: a fake watcher +// captures the addDir handler, reports test-controlled tracking via +// getWatched(), and records add() calls — against a real temp vault and index. +describe("file watcher — new-directory rescan", REAL_WATCHER_RETRY, () => { + registerSharedVaultHooks() + + const RESCAN_TEST_OPTIONS = { + stabilityThreshold: 200, + pollInterval: 50, + newDirectoryRescanDelay: 50, + } + + type RescanFakeWatcher = { + fireAddDir: (dirPath: string) => void + addedPaths: string[] + } + + /** Stubs chokidar.watch with a fake exposing what the rescan consumes. + * mockRestore (via onTestFinished) hands the real, spied watch back to the + * integration tests. */ + const installFakeChokidar = (watchedChildren: Record): RescanFakeWatcher => { + const handlers = new Map void>() + const addedPaths: string[] = [] + const fakeWatcher = { + on(event: string, handler: (path: string) => void) { + if (event === "ready") queueMicrotask(() => handler("")) + handlers.set(event, handler) + return fakeWatcher + }, + getWatched: () => watchedChildren, + add: (path: string) => { + addedPaths.push(path) + }, + } + const watchMock = vi.mocked(watch) + watchMock.mockReset() + watchMock.mockImplementation(() => fakeWatcher as unknown as ReturnType) + onTestFinished(() => watchMock.mockRestore()) + + const fireAddDir = (dirPath: string): void => { + const addDirHandler = handlers.get("addDir") + + if (addDirHandler === undefined) { + throw new Error("addDir handler was not registered") + } + addDirHandler(dirPath) + } + return { fireAddDir, addedPaths } + } + + /** Starts the file watcher against a fake chokidar (see installFakeChokidar). */ + const startWatcherWithFakeChokidar = async ( + watchedChildren: Record, + watcherOptions: WatcherOptions, + ): Promise => { + const fakeChokidar = installFakeChokidar(watchedChildren) + await startWatcherWithoutRebuild({ vaultPath: vault, search: index }, watcherOptions) + return fakeChokidar + } + + /** Backdates a file's mtime (10 minutes) so the rescan's "still being + * written" guard sees it as settled. */ + const backdateMtime = async (filePath: string): Promise => { + const backdated = new Date(Date.now() - 600_000) + await utimes(filePath, backdated, backdated) + } + + it("indexes a file on disk that chokidar does not track", { timeout: 15000 }, async () => { + const newDirectory = join(vault, "new-folder") + await mkdir(newDirectory) + const missedPath = join(newDirectory, "missed.md") + await writeFile(missedPath, "missed by the scan\n", "utf8") + await backdateMtime(missedPath) + + const { fireAddDir, addedPaths } = await startWatcherWithFakeChokidar({}, RESCAN_TEST_OPTIONS) + fireAddDir(newDirectory) + + await waitFor(() => index.fullTextSearch({ query: "missed" }, logger).length > 0) + const results = index.fullTextSearch({ query: "missed" }, logger) + expect(results.map((result) => result.path)).toEqual(["new-folder/missed.md"]) + // The file must also be registered with chokidar — indexed but + // untracked, its later deletion would emit no unlink (ghost entry). + expect(addedPaths).toEqual([missedPath]) + }) + + it( + "holds a rescan's indexing until the startup rebuild returns", + { timeout: 15000 }, + async () => { + const newDirectory = join(vault, "new-folder") + await mkdir(newDirectory) + const missedPath = join(newDirectory, "missed.md") + await writeFile(missedPath, "missed during startup\n", "utf8") + await backdateMtime(missedPath) + const rebuildGate = Promise.withResolvers() + vi.spyOn(index, "rebuildFromVault").mockImplementation(async () => { + await rebuildGate.promise + return { count: 0, embedding: Promise.resolve() } + }) + // Resolves when the rescan reaches the missed file, whether it holds the + // change (the intended path) or indexes it at once (a broken hold), so the + // state assertion below decides which happened. + const missedFileReached = Promise.withResolvers() + const debugSpy = vi.spyOn(logger, "debug").mockImplementation((message, fields) => { + const isMissedFileHeld = + message === "held a change until startup indexing ends" && + fields?.path === "new-folder/missed.md" + + if (isMissedFileHeld) missedFileReached.resolve(undefined) + }) + const realUpsertNote = index.upsertNote + vi.spyOn(index, "upsertNote").mockImplementation((params, requestLogger) => { + missedFileReached.resolve(undefined) + return realUpsertNote(params, requestLogger) + }) + const infoSpy = vi.spyOn(logger, "info") + onTestFinished(() => { + debugSpy.mockRestore() + infoSpy.mockRestore() + }) + const { fireAddDir } = installFakeChokidar({}) + const startup = watchVaultAndBuildIndex( + { vaultPath: vault, search: index, watcherOptions: RESCAN_TEST_OPTIONS }, + logger, + ) + await vi.waitFor(() => { + expect(index.rebuildFromVault).toHaveBeenCalledTimes(1) + }) + + fireAddDir(newDirectory) + await missedFileReached.promise + + expect(index.fullTextSearch({ query: "missed" }, logger)).toEqual([]) + rebuildGate.resolve(undefined) + await startup + expect(index.fullTextSearch({ query: "missed" }, logger).map((hit) => hit.path)).toEqual([ + "new-folder/missed.md", + ]) + expect(infoSpy).toHaveBeenCalledWith("applied the changes held during startup indexing", { + heldPaths: 1, + }) + }, + ) + + it("does not re-index a file chokidar already tracks", { timeout: 15000 }, async () => { + const newDirectory = join(vault, "new-folder") + await mkdir(newDirectory) + const trackedPath = join(newDirectory, "tracked.md") + const missedPath = join(newDirectory, "missed.md") + await writeFile(trackedPath, "already tracked note\n", "utf8") + await writeFile(missedPath, "missed sibling note\n", "utf8") + await backdateMtime(trackedPath) + await backdateMtime(missedPath) const upsertNoteSpy = vi.spyOn(index, "upsertNote") const { fireAddDir, addedPaths } = await startWatcherWithFakeChokidar( @@ -1545,8 +3035,14 @@ describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => // The untracked sibling getting indexed proves the rescan ran — the // tracked file being skipped can't be a silent no-op. await waitFor(() => index.fullTextSearch({ query: "sibling" }, logger).length > 0) - expect(upsertNoteSpy).toHaveBeenCalledTimes(1) - expect(upsertNoteSpy.mock.calls[0]?.[0]?.filePath).toBe("new-folder/missed.md") + expect(upsertNoteSpy).toHaveBeenCalledExactlyOnceWith( + { + filePath: "new-folder/missed.md", + rawContent: "missed sibling note\n", + fileStat: { mtimeMs: expect.any(Number), size: 20 }, + }, + logger, + ) // Only the missed file is registered — the tracked one is not re-added. expect(addedPaths).toEqual([missedPath]) }) @@ -1595,8 +3091,7 @@ describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => await waitFor(() => index.fullTextSearch({ query: "nested" }, logger).length > 0) const results = index.fullTextSearch({ query: "nested" }, logger) - expect(results).toHaveLength(1) - expect(results[0]?.path).toBe("new-folder/missed-subdir/nested.md") + expect(results.map((result) => result.path)).toEqual(["new-folder/missed-subdir/nested.md"]) // The subdirectory and the missed file are both handed back to chokidar // so they gain watches and future events (unlink included) fire. expect(addedPaths).toEqual([missedSubdirectory, nestedPath]) @@ -1642,9 +3137,8 @@ describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => await waitFor(() => index.fullTextSearch({ query: "symlinked" }, logger).length > 0) const results = index.fullTextSearch({ query: "symlinked" }, logger) - expect(results).toHaveLength(1) // Indexed under the link path, matching chokidar's followSymlinks view. - expect(results[0]?.path).toBe("new-folder/linked-dir/inside.md") + expect(results.map((result) => result.path)).toEqual(["new-folder/linked-dir/inside.md"]) // The inner file registers during the recursion, then the link itself. expect(addedPaths).toEqual([join(linkedDirectory, "inside.md"), linkedDirectory]) }) @@ -1680,8 +3174,7 @@ describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => await waitFor(() => index.fullTextSearch({ query: "symlinked" }, logger).length > 0) const results = index.fullTextSearch({ query: "symlinked" }, logger) - expect(results).toHaveLength(1) - expect(results[0]?.path).toBe("new-folder/linked-dir/inside.md") + expect(results.map((result) => result.path)).toEqual(["new-folder/linked-dir/inside.md"]) // Registration order proves the retry ran: the symlinked directory // registered during the rescan; the file only after its retry settled. expect(addedPaths).toEqual([linkedDirectory, join(linkedDirectory, "inside.md")]) @@ -1711,8 +3204,14 @@ describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => // Give a (broken) unbounded recursion time to produce extra upserts // before asserting the exact count. await new Promise((finished) => setTimeout(finished, 300)) - expect(upsertNoteSpy).toHaveBeenCalledTimes(1) - expect(upsertNoteSpy.mock.calls[0]?.[0]?.filePath).toBe("new-folder/settled.md") + expect(upsertNoteSpy).toHaveBeenCalledExactlyOnceWith( + { + filePath: "new-folder/settled.md", + rawContent: "settled beside a cycle\n", + fileStat: { mtimeMs: expect.any(Number), size: 23 }, + }, + logger, + ) // readdir order isn't guaranteed, so compare sorted ("loop" < "settled.md"). expect([...addedPaths].sort()).toEqual([loopLink, settledPath]) }, @@ -1741,10 +3240,16 @@ describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => // swallowed — not merely absent from the listing. expect(debugSpy).toHaveBeenCalledWith("rescan skipped unreadable entry", { path: "new-folder/dangling.md", - error: expect.stringContaining("ENOENT"), + error: `[Error]: ENOENT: no such file or directory, stat '${danglingLink}'`, }) - expect(upsertNoteSpy).toHaveBeenCalledTimes(1) - expect(upsertNoteSpy.mock.calls[0]?.[0]?.filePath).toBe("new-folder/settled.md") + expect(upsertNoteSpy).toHaveBeenCalledExactlyOnceWith( + { + filePath: "new-folder/settled.md", + rawContent: "settled beside a broken link\n", + fileStat: { mtimeMs: expect.any(Number), size: 29 }, + }, + logger, + ) expect(addedPaths).toEqual([settledPath]) }) @@ -1869,8 +3374,7 @@ describe("startFileWatcher — new-directory rescan", REAL_WATCHER_RETRY, () => await waitFor(() => index.fullTextSearch({ query: "skew" }, logger).length > 0) const results = index.fullTextSearch({ query: "skew" }, logger) - expect(results).toHaveLength(1) - expect(results[0]?.path).toBe("new-folder/skewed.md") + expect(results.map((result) => result.path)).toEqual(["new-folder/skewed.md"]) expect(addedPaths).toEqual([skewedPath]) }, ) diff --git a/src/vault-mcp/search/file-watcher.ts b/src/vault-mcp/search/file-watcher.ts index a7ec2b182..3a660f339 100644 --- a/src/vault-mcp/search/file-watcher.ts +++ b/src/vault-mcp/search/file-watcher.ts @@ -1,20 +1,26 @@ -/** File watcher — keeps the SQLite FTS5 index and vector embeddings current - * via chokidar events, plus a rescan safety net for the files chokidar's - * new-directory handling loses (see rescanNewDirectory). */ +/** File watcher — builds the search index at startup and keeps it and the + * vector embeddings current via chokidar events, plus a rescan safety net for + * the files chokidar's new-directory handling loses (see rescanNewDirectory). + * Startup order is watchVaultAndBuildIndex's job. */ import { watch } from "chokidar" import { DateTime } from "luxon" -import { readFile, stat } from "node:fs/promises" +import { readFile, readdir, stat } from "node:fs/promises" import type { Stats } from "node:fs" -import { extname, join, relative, resolve as resolvePath } from "node:path" -import { INDEXABLE_TEXT_EXTENSIONS } from "./search-index.js" +import { extname, join, relative, resolve as resolvePath, sep } from "node:path" +import { + INDEXABLE_TEXT_EXTENSIONS, + INDEXING_FILE_READ_CONCURRENCY, + PDF_EXTRACTION_CONCURRENCY, +} from "./search-index.js" import type { SearchIndex } from "./search-index.js" import { extractPdfText } from "../obsidian-markdown/pdf.js" -import { logger } from "../../logger.js" +import type { Logger } from "../../logger.js" import { describeError, describeErrorForLog } from "../../utils/describe-error.js" import { isErrnoException } from "../../utils/is-errno-exception.js" -import { readdirOrNull, realpathOrNull, statOrNull } from "../../utils/fs.js" +import { isMissingPathError, readdirOrNull, realpathOrNull, statOrNull } from "../../utils/fs.js" import { hasHiddenPathSegment } from "../../utils/has-hidden-path-segment.js" +import { mapWithConcurrency } from "../../utils/map-with-concurrency.js" /** ms between filesystem polls when usePolling is on. chokidar's raw default is * 100ms, which stat()s the whole tree 10×/sec; 300ms meaningfully cuts CPU, and @@ -25,6 +31,20 @@ const POLLING_INTERVAL_MS = 300 /** Default for FileWatcherOptions.stabilityThreshold (see its doc). */ const DEFAULT_STABILITY_THRESHOLD_MS = 2000 +/** Default for FileWatcherOptions.pollInterval. */ +const DEFAULT_POLL_INTERVAL_MS = 100 + +/** + * Default for FileWatcherOptions.heldChangesDeadline. The server starts + * listening only after watchVaultAndBuildIndex resolves, so the deadline caps + * how long a Sync backlog arriving during startup delays listening. + * - The local compose health check's 20 s start period sets the value: 10 s + * leaves the server time to listen inside it. + * - Without a deadline, the backlog could keep the server from listening past + * the CLI's 120 s health wait. + */ +const DEFAULT_HELD_CHANGES_DEADLINE_MS = 10_000 + type FileWatcherOptions = Readonly<{ /** ms a file's size must stay unchanged before we index it (default 2000). * Prevents reading partial writes from Obsidian Sync. */ @@ -40,222 +60,401 @@ type FileWatcherOptions = Readonly<{ * past the awaitWriteFinish window lets in-flight writes settle before the * rescan reads them. */ newDirectoryRescanDelay?: number + /** ms startup waits after the rebuild for the changes chokidar reported + * meanwhile to be indexed, before returning anyway (default 10000); the + * rest are applied in the background. */ + heldChangesDeadline?: number +}> + +type WatchVaultParams = Readonly<{ + vaultPath: string + search: SearchIndex + watcherOptions?: FileWatcherOptions +}> + +/** The kind of the latest event chokidar reported for a held path. `add` and + * `change` both record "change", since either means "read this path again". + * The startup batch checks every held path against the disk; going live uses + * the kind only to remove a deleted path early. */ +type HeldEventKind = "change" | "unlink" + +/** While holding, the paths chokidar reports are recorded instead of applied; + * once live, each event is applied as it arrives; once closed, they are + * ignored. */ +type WatcherState = + { mode: "holding"; heldPaths: Map } | { mode: "live" } | { mode: "closed" } + +/** A held path and the event token reserved for it when the watcher went live. + * An event token is a symbol stored per path in currentEvents, which each + * newer event for the path replaces. The startup batch removes the path, or + * indexes what it read, only while the reservation is still the path's token. */ +type ReservedPath = Readonly<{ filePath: string; reservation: symbol }> + +/** What indexing one changed file did. A note's embedding is still running when + * this returns, so whoever receives `note-indexed` reports its failure. */ +type ChangeIndexResult = + | { status: "note-indexed"; embedding: Promise } + | { status: "file-indexed" } + // The file was gone when read. + | { status: "vanished" } + // A newer event or a removal for the same path replaced this read's token + // before the read finished, so its content was dropped. + | { status: "superseded" } + // A note's read, stat or index write failed, or a non-md file's stat or + // asset-table write did; already logged. A non-md file whose content alone + // could not be read, extracted or indexed is still "file-indexed". + | { status: "failed" } + +/** Whether a held path names a file on disk. "unknown" leaves the path to its + * read, which reports why it cannot be checked. */ +type HeldPathPresence = "present" | "missing" | "unknown" + +type FolderListing = + { status: "listed"; names: ReadonlySet } | { status: "missing" } | { status: "failed" } + +type HoldingWatcher = Readonly<{ + /** Resolves once chokidar's initial scan is done and every folder is watched. */ + ready: Promise + /** Goes live and starts applying what chokidar reported while holding, and + * returns how many distinct paths that is. `applied` resolves once the + * batch's index writes commit, and never rejects. */ + applyHeldChanges: () => { heldPathCount: number; applied: Promise } + /** Stops watching; a rescan or retry still scheduled fires later and does + * nothing. Used only when startup fails before the held changes apply. + * Never rejects, so the startup failure stays the error reported. */ + close: () => Promise }> /** - * Watches the vault and mirrors every file add/change/delete into the search - * index, so search stays current without rebuilds. Resolves once the watcher - * is ready (initial scan done); the watcher then runs for the process - * lifetime — there is no shutdown path. + * Starts a vault watcher in holding mode: from the moment chokidar starts, it + * records each changed path instead of indexing it. applyHeldChanges switches + * it live, and from then on it mirrors every file add/change/delete into the + * search index for the process lifetime. */ -export const startFileWatcher = ( - vaultPath: string, - search: SearchIndex, - options?: FileWatcherOptions, -): Promise => { - // Serializing per path limits model concurrency; source versions reject - // obsolete results independently of the order jobs finish. +const startHoldingWatcher = (params: WatchVaultParams, logger: Logger): HoldingWatcher => { + const { vaultPath, search, watcherOptions } = params + + // Each path's latest embedding job. A new job for the path waits for this one, + // so a note edited many times runs one model job at a time. A job that still + // finishes after a newer edit writes nothing: the index checks the source + // version the job carries (the token upsertNote or upsertFileContent returned + // for the content it embeds) against the path's current one. const pendingEmbeds = new Map>() - /** Event tokens reject superseded filesystem reads until their handler finishes. - * Source versions belong to the index and protect model work after the handler returns. */ + /** + * Each path's event token, which guards the path's read; source versions + * guard the embedding that runs after it. + * - indexChangedFile stores a fresh token before it reads the file, and drops + * what it read if a newer event or a removal replaced the token meanwhile. + * - applyHeldChanges reserves one for each held path as the watcher goes + * live. The startup batch reads the path under it, so after a later + * deletion or a newer event for the path, the batch skips its removal or + * drops what it read. + */ const currentEvents = new Map() - /** Indexes an added or modified file: non-md files land in the asset table; - * notes are read from disk, upserted into the FTS index, and re-embedded - * (embeds serialized per path via pendingEmbeds). */ - const handleChange = async (filePath: string): Promise => { - const relativePath = relative(vaultPath, filePath) - const eventToken = Symbol() - currentEvents.set(relativePath, eventToken) + // Reassigned, so `let`: going live ends holding, and a failed startup closes + // the watcher. + let watcherState: WatcherState = { mode: "holding", heldPaths: new Map() } - try { - if (!filePath.endsWith(".md")) { - const readNonMdFileStat = async (): Promise => { - try { - return await statOrNull(filePath) - } catch (error) { - logger.error("failed to stat non-md file", { - path: relativePath, - error: describeError(error), - }) - return null - } - } - const fileStat = await readNonMdFileStat() + /** Indexes an added or modified non-md file: its asset-table row, plus its + * content when it is a canvas or an indexable text or PDF file. */ + const indexChangedNonMdFile = async (params: { + filePath: string + relativePath: string + eventToken: symbol + }): Promise => { + const { filePath, relativePath, eventToken } = params - if (currentEvents.get(relativePath) !== eventToken) return - // Vanished between the watcher event and the stat — the unlink event - // that follows will remove any existing row. - if (!fileStat) return - try { - search.upsertNonMdFile(relativePath, fileStat.size) - } catch (error) { - logger.error("failed to index non-md file metadata", { - path: relativePath, - error: describeError(error), - }) - return - } + const readNonMdFileStat = async (): Promise< + { status: "found"; fileStat: Stats } | { status: "missing" } | { status: "failed" } + > => { + try { + const fileStat = await statOrNull(filePath) + return fileStat ? { status: "found", fileStat } : { status: "missing" } + } catch (error) { + logger.error("failed to stat non-md file", { + path: relativePath, + error: describeError(error), + }) + return { status: "failed" } + } + } + const nonMdStat = await readNonMdFileStat() - // Canvas files are always read — link extraction is unconditional. - // PDF and text files are only read when file content FTS is enabled. - const extension = extname(filePath) - const isCanvas = extension === ".canvas" - const isIndexableNonCanvas = - search.fileContentIndexingEnabled && INDEXABLE_TEXT_EXTENSIONS.has(extension) - - if (isCanvas || isIndexableNonCanvas) { - try { - const readContentToIndex = async (): Promise => { - if (extension !== ".pdf") return await readFile(filePath, "utf8") - - const buffer = await readFile(filePath) - const pdfData = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength) - const pdfResult = await extractPdfText(pdfData) - return pdfResult.text - } - const contentToIndex = await readContentToIndex() - - if (currentEvents.get(relativePath) !== eventToken) return - - const sourceVersion = search.upsertFileContent( - { - filePath: relativePath, - rawContent: contentToIndex, - fileStat: { mtimeMs: fileStat.mtimeMs, size: fileStat.size }, - }, - logger, - ) - - // Embed file content vectors — serialized per path via the same - // pendingEmbeds map (note paths end in .md, file paths don't). - // Reads the processed content from the file_content table. - const previousEmbed = pendingEmbeds.get(relativePath) ?? Promise.resolve() - const currentEmbed = previousEmbed - .catch((previousError) => { - logger.debug("previous file embed failed, proceeding with current", { - path: relativePath, - error: describeError(previousError), - }) - }) - .then(() => { - return search.embedFileContent({ filePath: relativePath, sourceVersion }, logger) - }) - pendingEmbeds.set(relativePath, currentEmbed) - - /** This detached job owns its error reporting and queue cleanup after the - * file handler returns; an older job must leave a newer queued job registered. */ - currentEmbed - .catch((embedError) => { - logger.error("file content embedding failed", { - path: relativePath, - error: describeError(embedError), - }) - }) - .finally(() => { - if (pendingEmbeds.get(relativePath) === currentEmbed) { - pendingEmbeds.delete(relativePath) - } - }) - } catch (error) { - // A PDF reader that failed to start arrives wrapped; the cause holds why - logger.warn("file content indexing failed", { - path: relativePath, - ...describeErrorForLog(error), - }) - } - } + if (currentEvents.get(relativePath) !== eventToken) return { status: "superseded" } + // Vanished between the event and the stat: the unlink event that follows + // removes any existing row. + if (nonMdStat.status === "missing") return { status: "vanished" } + if (nonMdStat.status === "failed") return { status: "failed" } - logger.debug("indexed non-md file", { path: relativePath }) - return - } + const { fileStat } = nonMdStat + + // Nothing is awaited between the token check above and this write, so it + // cannot be superseded; only the content read below needs a second check. + try { + search.upsertNonMdFile(relativePath, fileStat.size) + } catch (error) { + logger.error("failed to index non-md file metadata", { + path: relativePath, + error: describeError(error), + }) + return { status: "failed" } + } + // Canvas files are always read — link extraction is unconditional. + // PDF and text files (INDEXABLE_TEXT_EXTENSIONS, which includes .pdf) are + // only read when file content FTS is enabled. + const extension = extname(filePath) + const isCanvas = extension === ".canvas" + const isIndexableNonCanvas = + search.fileContentIndexingEnabled && INDEXABLE_TEXT_EXTENSIONS.has(extension) + + if (isCanvas || isIndexableNonCanvas) { try { - const readNoteSource = async (): Promise<{ content: string; fileStat: Stats } | null> => { - try { - const [content, fileStat] = await Promise.all([ - readFile(filePath, "utf8"), - stat(filePath), - ]) - return { content, fileStat } - } catch (error) { - if (!isErrnoException(error, "ENOENT")) throw error - - logger.debug("change event skipped, file vanished", { path: relativePath }) - return null - } - } - const noteSource = await readNoteSource() + const readContentToIndex = async (): Promise => { + if (extension !== ".pdf") return await readFile(filePath, "utf8") - if (currentEvents.get(relativePath) !== eventToken) return - if (!noteSource) return + const buffer = await readFile(filePath) + // pdfjs throws on a Node Buffer, so it gets a plain Uint8Array view + // over the same bytes. + const pdfData = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength) + const pdfResult = await extractPdfText(pdfData) + return pdfResult.text + } + const contentToIndex = await readContentToIndex() - const { content, fileStat } = noteSource + if (currentEvents.get(relativePath) !== eventToken) return { status: "superseded" } - const sourceVersion = search.upsertNote( + const sourceVersion = search.upsertFileContent( { filePath: relativePath, - rawContent: content, + rawContent: contentToIndex, fileStat: { mtimeMs: fileStat.mtimeMs, size: fileStat.size }, }, logger, ) - // Each path queues model work; recovering a rejected predecessor - // allows later jobs to run after a transient failure. + + // Embed file content vectors — serialized per path via the same + // pendingEmbeds map. A note path ends in .md and a file path never + // does, so their keys never collide. Reads the processed content from + // the file_content table. const previousEmbed = pendingEmbeds.get(relativePath) ?? Promise.resolve() const currentEmbed = previousEmbed .catch((previousError) => { - logger.debug("previous embed failed, proceeding with current", { + logger.debug("previous file embed failed, proceeding with current", { path: relativePath, error: describeError(previousError), }) }) .then(() => { - return search.embedNote( - { notePath: relativePath, rawContent: content, sourceVersion }, - logger, - ) + return search.embedFileContent({ filePath: relativePath, sourceVersion }, logger) }) pendingEmbeds.set(relativePath, currentEmbed) - /** This awaited job reports failures through the handler's catch. Await the - * cleanup promise too, so its rejection is handled and newer queued jobs remain registered. */ - await currentEmbed.finally(() => { - if (pendingEmbeds.get(relativePath) === currentEmbed) { - pendingEmbeds.delete(relativePath) - } - }) - } catch (err) { - logger.error("failed to process file change", { + // Unlike a note's, this embedding is not returned: neither the live + // handler nor the startup batch waits on file-content model work. So + // the job reports its own failure and removes its own queue entry, + // leaving a newer queued job registered. + currentEmbed + .catch((embedError) => { + logger.error("file content embedding failed", { + path: relativePath, + error: describeError(embedError), + }) + }) + .finally(() => { + if (pendingEmbeds.get(relativePath) === currentEmbed) { + pendingEmbeds.delete(relativePath) + } + }) + } catch (error) { + // Warn, as the rebuild does for an unreadable file: the asset-table row + // above is indexed, and content search keeps what it had for the file. + // describeErrorForLog keeps the cause, because a PDF reader that + // failed to start arrives wrapped. + logger.warn("file content indexing failed", { path: relativePath, - error: describeError(err), + ...describeErrorForLog(error), }) } - } finally { - // A newer handler may already have finished and removed its entry. - // Only the current event owns cleanup; absence still invalidates old reads. - if (currentEvents.get(relativePath) === eventToken) { - currentEvents.delete(relativePath) + } + + logger.debug("indexed non-md file", { path: relativePath }) + return { status: "file-indexed" } + } + + /** Indexes an added or modified note and queues its re-embedding. The + * embedding is returned unawaited, so the caller decides whether to wait. */ + const indexChangedNote = async (params: { + filePath: string + relativePath: string + eventToken: symbol + }): Promise => { + const { filePath, relativePath, eventToken } = params + + try { + const readNoteSource = async (): Promise<{ content: string; fileStat: Stats } | null> => { + try { + const [content, fileStat] = await Promise.all([ + readFile(filePath, "utf8"), + stat(filePath), + ]) + return { content, fileStat } + } catch (error) { + if (!isErrnoException(error, "ENOENT")) throw error + + logger.debug("change event skipped, file vanished", { path: relativePath }) + return null + } } + const noteSource = await readNoteSource() + + if (currentEvents.get(relativePath) !== eventToken) return { status: "superseded" } + if (!noteSource) return { status: "vanished" } + + const { content, fileStat } = noteSource + + const sourceVersion = search.upsertNote( + { + filePath: relativePath, + rawContent: content, + fileStat: { mtimeMs: fileStat.mtimeMs, size: fileStat.size }, + }, + logger, + ) + + // Each path queues model work; recovering a rejected predecessor + // allows later jobs to run after a transient failure. + const previousEmbed = pendingEmbeds.get(relativePath) ?? Promise.resolve() + const currentEmbed = previousEmbed + .catch((previousError) => { + logger.debug("previous embed failed, proceeding with current", { + path: relativePath, + error: describeError(previousError), + }) + }) + .then(() => { + return search.embedNote( + { notePath: relativePath, rawContent: content, sourceVersion }, + logger, + ) + }) + pendingEmbeds.set(relativePath, currentEmbed) + + // The caller gets the cleanup promise, not currentEmbed, because a + // promise nobody holds rejects unhandled: `.finally` passes currentEmbed's + // rejection through to it. The cleanup leaves a newer queued job registered. + const embedding = currentEmbed.finally(() => { + if (pendingEmbeds.get(relativePath) === currentEmbed) { + pendingEmbeds.delete(relativePath) + } + }) + return { status: "note-indexed", embedding } + } catch (err) { + logger.error("failed to process file change", { + path: relativePath, + error: describeError(err), + }) + return { status: "failed" } + } + } + + /** Indexes an added or modified file under the token its caller stored for + * the path, and reports what it did. */ + const indexFileUnderToken = (params: { + filePath: string + relativePath: string + eventToken: symbol + }): Promise => { + if (params.filePath.endsWith(".md")) return indexChangedNote(params) + return indexChangedNonMdFile(params) + } + + /** Releases a path's event token unless a newer event replaced it: deleting + * a newer event's token would make that event's read look superseded. An + * older read that finds no token still fails its check. */ + const releaseEventToken = (relativePath: string, eventToken: symbol): void => { + if (currentEvents.get(relativePath) === eventToken) currentEvents.delete(relativePath) + } + + /** Indexes an added or modified file and reports what it did. */ + const indexChangedFile = async (filePath: string): Promise => { + const relativePath = relative(vaultPath, filePath) + + // Taken before the first await, so events take their tokens in the order + // their handlers start, and the read that starts last is the one indexed. + const eventToken = Symbol() + currentEvents.set(relativePath, eventToken) + + // Awaited inside the try, so the token is released only once indexing ends. + try { + return await indexFileUnderToken({ filePath, relativePath, eventToken }) + } finally { + releaseEventToken(relativePath, eventToken) + } + } + + /** Logs a note embedding's failure under the same message as a failed read, + * so each change reports one error whichever step failed. */ + const reportEmbeddingFailure = (filePath: string, error: unknown): void => { + logger.error("failed to process file change", { + path: relative(vaultPath, filePath), + error: describeError(error), + }) + } + + /** chokidar's add and change listener. While the watcher holds, it records + * the path for applyHeldChanges instead of indexing it. chokidar ignores + * the returned promise, so each step logs its own failure and the listener + * never rejects. */ + const handleChange = async (filePath: string): Promise => { + if (watcherState.mode === "closed") return + if (watcherState.mode === "holding") { + watcherState.heldPaths.set(filePath, "change") + logger.debug("held a change until startup indexing ends", { + path: relative(vaultPath, filePath), + }) + return + } + + const indexResult = await indexChangedFile(filePath) + + if (indexResult.status !== "note-indexed") return + + try { + await indexResult.embedding + } catch (error) { + reportEmbeddingFailure(filePath, error) } } /** Removes a deleted file from the index — the asset table for non-md * files, the note tables (FTS, links, tasks, vectors) for notes. */ - const handleDelete = (filePath: string): void => { + const removeFromIndex = (filePath: string): void => { const relativePath = relative(vaultPath, filePath) + + // Dropping the path's token makes a read still in flight for it return + // "superseded", so that read cannot re-add the file removed here. currentEvents.delete(relativePath) if (!filePath.endsWith(".md")) { - try { - search.removeNonMdFile(relativePath) - } catch (error) { - logger.error("failed to remove non-md file metadata", { - path: relativePath, - error: describeError(error), - }) + const removeAssetRow = (): boolean => { + try { + search.removeNonMdFile(relativePath) + return true + } catch (error) { + logger.error("failed to remove non-md file metadata", { + path: relativePath, + error: describeError(error), + }) + return false + } } + const assetRowRemoved = removeAssetRow() + + // The content rows live in their own tables, so they are removed even + // when the asset-table removal above failed. const deletedExtension = extname(filePath) const isDeletedCanvas = deletedExtension === ".canvas" const isDeletedIndexable = @@ -272,7 +471,8 @@ export const startFileWatcher = ( return } } - logger.debug("processed non-md file removal", { path: relativePath }) + // Reported only when every removal succeeded; a failure logged its own error. + if (assetRowRemoved) logger.debug("processed non-md file removal", { path: relativePath }) return } @@ -288,8 +488,22 @@ export const startFileWatcher = ( logger.debug("removed from index", { path: relativePath }) } - const stabilityThreshold = options?.stabilityThreshold ?? DEFAULT_STABILITY_THRESHOLD_MS - const newDirectoryRescanDelay = options?.newDirectoryRescanDelay ?? 2 * stabilityThreshold + /** chokidar's unlink listener. While the watcher holds, it records the path + * for applyHeldChanges instead of removing it. */ + const handleDelete = (filePath: string): void => { + if (watcherState.mode === "closed") return + if (watcherState.mode === "holding") { + watcherState.heldPaths.set(filePath, "unlink") + logger.debug("held a deletion until startup indexing ends", { + path: relative(vaultPath, filePath), + }) + return + } + removeFromIndex(filePath) + } + + const stabilityThreshold = watcherOptions?.stabilityThreshold ?? DEFAULT_STABILITY_THRESHOLD_MS + const newDirectoryRescanDelay = watcherOptions?.newDirectoryRescanDelay ?? 2 * stabilityThreshold /** True when the file's last write falls inside the stability window — * plausibly still being written. A negative age (mtime in the future — @@ -312,19 +526,26 @@ export const startFileWatcher = ( persistent: true, ignoreInitial: true, // Poll across the Docker Desktop ↔ WSL2 bridge, where inotify is dropped. - usePolling: options?.usePolling ?? false, + usePolling: watcherOptions?.usePolling ?? false, // interval only drives chokidar's polling backend. Pass it only when // polling, so we never depend on how chokidar treats an interval set // without usePolling (today it's ignored — a future release needn't be). - ...(options?.usePolling ? { interval: POLLING_INTERVAL_MS } : {}), + ...(watcherOptions?.usePolling ? { interval: POLLING_INTERVAL_MS } : {}), // Obsidian Sync writes files in chunks — wait for write stability before // indexing to avoid reading partial content awaitWriteFinish: { stabilityThreshold, - pollInterval: options?.pollInterval ?? 100, + pollInterval: watcherOptions?.pollInterval ?? DEFAULT_POLL_INTERVAL_MS, }, }) + /** Adds a path to chokidar unless the watcher is closed. A rescan or retry + * timer can fire after close, and chokidar's add reopens a closed watcher. */ + const watchPathUnlessClosed = (path: string): void => { + if (watcherState.mode === "closed") return + watcher.add(path) + } + /** * Polls a mid-write file until its writes settle, then indexes it. Serves * the new-directory rescan (rescanNewDirectory, below) for one case: a file @@ -338,18 +559,26 @@ export const startFileWatcher = ( const scheduleUnstableFileRetry = (filePath: string): void => { const retryTimer = setTimeout(() => { const indexIfSettled = async (): Promise => { + // Ends the retries, which would otherwise keep polling a closed + // watcher's file until it settles. + if (watcherState.mode === "closed") return + const fileStat = await stat(filePath) + // A file still being written is checked again after another window. + // The retries have no limit, since a file that keeps changing is still + // arriving; a deleted file ends them when the stat above rejects. if (isWithinStabilityWindow(fileStat.mtimeMs)) { scheduleUnstableFileRetry(filePath) return } - watcher.add(filePath) + watchPathUnlessClosed(filePath) logger.debug("rescan indexing settled file after retry", { path: relative(vaultPath, filePath), }) await handleChange(filePath) } + // A timer callback cannot await, so the retry runs detached and logs its own failure. indexIfSettled().catch((err) => { logger.debug("rescan retry skipped unreadable file", { path: relative(vaultPath, filePath), @@ -377,7 +606,17 @@ export const startFileWatcher = ( dirPath: string, visitedRealPaths: Set, ): Promise => { - /** The wrapper lists descendants recursively; symlinked directories are followed below. */ + // A rescan scheduled before close still fires, but a closed watcher has + // nothing to reconcile, and close empties getWatched, so every entry would + // look missed. + if (watcherState.mode === "closed") { + logger.debug("rescan skipped, watcher closed", { path: relative(vaultPath, dirPath) }) + return + } + + // readdirOrNull lists every descendant recursively, or returns null when the + // directory is gone. It does not enter symlinked directories; the loop + // below rescans those. const entries = await readdirOrNull(dirPath) if (!entries) { @@ -422,7 +661,7 @@ export const startFileWatcher = ( // A subdirectory chokidar never saw would stay unwatched forever — // watcher.add() registers its watches. Its contents are already covered // by this recursive listing, so the add's suppressed events don't matter. - watcher.add(fullPath) + watchPathUnlessClosed(fullPath) logger.debug("rescan registered missed directory", { path: relativePath, }) @@ -441,7 +680,7 @@ export const startFileWatcher = ( // emitting events, which would make a later pass skip them as // already tracked. await rescanNewDirectory(fullPath, visitedRealPaths) - watcher.add(fullPath) + watchPathUnlessClosed(fullPath) logger.debug("rescan registered missed symlinked directory", { path: relativePath, }) @@ -464,7 +703,7 @@ export const startFileWatcher = ( // Register the file with chokidar too — indexed but untracked, its // later deletion would emit no unlink, leaving a ghost index entry. - watcher.add(fullPath) + watchPathUnlessClosed(fullPath) logger.debug("rescan indexing missed file", { path: relativePath }) await handleChange(fullPath) } catch (err) { @@ -481,6 +720,7 @@ export const startFileWatcher = ( * reads them. Wired to chokidar's addDir event below. */ const scheduleNewDirectoryRescan = (dirPath: string): void => { const rescanTimer = setTimeout(() => { + // A timer callback cannot await, so the rescan runs detached and logs its own failure. rescanNewDirectory(dirPath, new Set()).catch((err) => { logger.error("failed to rescan new directory", { path: relative(vaultPath, dirPath), @@ -492,6 +732,242 @@ export const startFileWatcher = ( rescanTimer.unref() } + /** Lists a folder's entry names for the startup batch's presence checks. */ + const listFolderNames = async (folderPath: string): Promise => { + try { + // A plain readdir, not readdirOrNull: the check needs only this folder's + // own names, and readdirOrNull lists every descendant. + const names = await readdir(folderPath) + return { status: "listed", names: new Set(names) } + } catch (error) { + if (isMissingPathError(error)) return { status: "missing" } + + logger.warn("could not list a folder of held changes, so stat checks its files", { + path: relative(vaultPath, folderPath), + error: describeError(error), + }) + return { status: "failed" } + } + } + + /** + * Decides whether a held path still names a file. + * - Each folder from the vault root down must list the next name in the path + * exactly. On a case-insensitive filesystem, `stat` still succeeds for the + * old path after a case-only rename of the file (`note.md` to `Note.md`) or + * of a folder above it (`notes/` to `Notes/`), and reading it would index + * the old path again. + * - `listFolderOnce` returns the batch's shared listing of a folder, so each + * folder is read once per batch. + */ + const checkHeldPathOnDisk = async ( + filePath: string, + listFolderOnce: (folderPath: string) => Promise, + ): Promise => { + const pathSegments = relative(vaultPath, filePath).split(sep) + + // Walks down from the vault root: the vault root must list the first + // segment, that folder the second, and so on to the file's own name. + // Sequential, so the first folder that lacks the next name ends the check. + for (const [segmentIndex, segmentName] of pathSegments.entries()) { + const folderPath = join(vaultPath, ...pathSegments.slice(0, segmentIndex)) + const folderListing = await listFolderOnce(folderPath) + const missingFromListing = + folderListing.status === "listed" && !folderListing.names.has(segmentName) + + // A folder that could not be listed proves nothing either way, so the + // walk goes on and the stat below decides. + if (folderListing.status === "missing" || missingFromListing) return "missing" + } + + try { + const fileStat = await statOrNull(filePath) + + // A folder now at the path also counts as missing. Removing it is + // harmless: it clears only rows for a file that is no longer there. + return fileStat?.isFile() ? "present" : "missing" + } catch (error) { + // The path's read runs next and reports the failure at error level. + logger.debug("could not stat a held path, so its read reports why", { + path: relative(vaultPath, filePath), + error: describeError(error), + }) + return "unknown" + } + } + + /** Removes a held path the startup batch found missing on disk. A held + * deletion that applyHeldChanges already removed comes here again while its + * path is still gone; the second removal finds no rows and changes nothing. */ + const removeHeldPath = (filePath: string): void => { + logger.debug("removing a held path that is gone from disk", { + path: relative(vaultPath, filePath), + }) + removeFromIndex(filePath) + } + + /** + * Reads one held path again for the startup batch, under its reservation, + * without waiting for its embedding. + * - If a live event for the path replaced the reservation after going live, + * the read drops what it read, as an older live read does. + * - A path that vanished by its read is left to the `unlink` chokidar sends + * for every file it tracks, as a live read leaves it. + */ + const indexHeldPath = async ({ filePath, reservation }: ReservedPath): Promise => { + const relativePath = relative(vaultPath, filePath) + + // Awaited inside the try, so the reservation is released only once indexing ends. + try { + const indexResult = await indexFileUnderToken({ + filePath, + relativePath, + eventToken: reservation, + }) + + if (indexResult.status !== "note-indexed") return + + // The batch does not wait for embeddings, so the failure handler is + // attached now. Attached after the batch, an embedding that rejected + // first would be reported as an unhandled rejection. + indexResult.embedding.catch((error) => { + reportEmbeddingFailure(filePath, error) + }) + } finally { + releaseEventToken(relativePath, reservation) + } + } + + /** + * Reads held paths again within the concurrency limits the rebuild also uses. + * - Other files go first, INDEXING_FILE_READ_CONCURRENCY at a time, so slow + * PDF extraction does not hold them back. + * - PDFs whose text is extracted go after, PDF_EXTRACTION_CONCURRENCY at a + * time. The groups run one after the other, so the two limits never add up. + */ + const indexHeldPathsWithinBounds = async ( + reservedPaths: ReadonlyArray, + ): Promise => { + // indexChangedNonMdFile extracts a PDF's text only when file content + // indexing is on; otherwise a PDF is only statted and keeps the general bound. + const needsPdfExtraction = ({ filePath }: ReservedPath): boolean => { + return search.fileContentIndexingEnabled && extname(filePath) === ".pdf" + } + const pdfReads = reservedPaths.filter(needsPdfExtraction) + const otherReads = reservedPaths.filter((reservedPath) => !needsPdfExtraction(reservedPath)) + + await mapWithConcurrency({ + items: otherReads, + concurrency: INDEXING_FILE_READ_CONCURRENCY, + mapper: indexHeldPath, + }) + await mapWithConcurrency({ + items: pdfReads, + concurrency: PDF_EXTRACTION_CONCURRENCY, + mapper: indexHeldPath, + }) + } + + /** Checks each held path against the disk, removes the missing ones, then + * reads the rest again. The watcher is live meanwhile, so for a path whose + * reservation a live event replaced, the removal is skipped and the read's + * result is dropped: that event is newer. */ + const reconcileHeldPaths = async (reservedPaths: ReadonlyArray): Promise => { + const folderListings = new Map>() + + /** Lists each folder once per batch, however many held paths sit in it. */ + const listFolderOnce = (folderPath: string): Promise => { + const knownListing = folderListings.get(folderPath) + + if (knownListing) return knownListing + const folderListing = listFolderNames(folderPath) + folderListings.set(folderPath, folderListing) + return folderListing + } + + const presenceChecks = await mapWithConcurrency({ + items: reservedPaths, + concurrency: INDEXING_FILE_READ_CONCURRENCY, + mapper: async (reservedPath) => { + const presence = await checkHeldPathOnDisk(reservedPath.filePath, listFolderOnce) + return { reservedPath, presence } + }, + }) + const missingPaths = presenceChecks + .filter((check) => check.presence === "missing") + .map((check) => check.reservedPath) + const pathsToRead = presenceChecks + .filter((check) => check.presence !== "missing") + .map((check) => check.reservedPath) + + // Removals run before reads. upsertNote re-resolves only links whose target + // is not indexed, so removing a renamed note's old path first lets the new + // path's read claim the links that pointed at the old path. Going live + // already removed an old path whose latest held event is a deletion; this + // loop removes one whose latest held event is a change. + for (const { filePath, reservation } of missingPaths) { + // Skips a path a live event has taken since going live, so a live `add` + // that indexed it is not removed. No await sits between this check and + // the removal, so no live event can take the path in between. + const reservationIsCurrent = currentEvents.get(relative(vaultPath, filePath)) === reservation + + if (!reservationIsCurrent) continue + removeHeldPath(filePath) + } + + await indexHeldPathsWithinBounds(pathsToRead) + } + + /** + * Goes live and starts the startup batch over the distinct held paths. The + * first three steps run with no await between them, before any live event + * can start: + * + * 1. The watcher goes live. + * 2. Each path whose latest held event is a deletion is removed, so a renamed + * note's new path, held or arriving as a live event, claims the links that + * pointed at the old one. reconcileHeldPaths removes an old path whose + * latest held event is a change, before the batch reads any path. A file + * that came back is read again by the batch. + * 3. Each held path gets a reserved event token. It comes after step 2, + * because removeFromIndex deletes the path's token. + * + * `applied` never rejects: one catch logs a failure of the batch. + */ + const applyHeldChanges = (): { heldPathCount: number; applied: Promise } => { + // Unreachable today, because applyHeldChanges runs once per startup and the + // watcher closes only when the rebuild fails, before that call. + if (watcherState.mode !== "holding") return { heldPathCount: 0, applied: Promise.resolve() } + + const { heldPaths } = watcherState + watcherState = { mode: "live" } + + for (const [filePath, latestEventKind] of heldPaths) { + if (latestEventKind === "unlink") removeFromIndex(filePath) + } + + const reservedPaths: ReservedPath[] = [] + for (const filePath of heldPaths.keys()) { + const reservation = Symbol() + currentEvents.set(relative(vaultPath, filePath), reservation) + reservedPaths.push({ filePath, reservation }) + } + + const applyBatch = async (): Promise => { + try { + await reconcileHeldPaths(reservedPaths) + logger.info("applied the changes held during startup indexing", { + heldPaths: reservedPaths.length, + }) + } catch (error) { + logger.error("failed to apply the changes held during startup", { + error: describeError(error), + }) + } + } + return { heldPathCount: reservedPaths.length, applied: applyBatch() } + } + watcher .on("add", handleChange) .on("change", handleChange) @@ -503,10 +979,74 @@ export const startFileWatcher = ( }) }) - return new Promise((resolve) => { + const ready = new Promise((resolve) => { watcher.on("ready", () => { logger.info("file watcher started", { vaultPath }) resolve() }) }) + + const close = async (): Promise => { + watcherState = { mode: "closed" } + try { + await watcher.close() + } catch (error) { + logger.error("failed to close the file watcher", { error: describeError(error) }) + } + } + + return { ready, applyHeldChanges, close } +} + +/** + * Starts the watcher, builds the search index from disk, then applies the + * changes the watcher saw meanwhile. The watcher is ready before the rebuild + * reads anything and holds every change until the rebuild returns, so a change + * landing mid-rebuild is applied after it instead of being missed or replaced + * by the rebuild's older read. Resolves with the rebuild's result (the note + * count and its background embedding promise) once the batch over the held + * changes commits its index writes, or at heldChangesDeadline if it takes + * longer; the rest then apply in the background. + */ +export const watchVaultAndBuildIndex = async ( + params: WatchVaultParams, + logger: Logger, +): Promise<{ count: number; embedding: Promise }> => { + const { vaultPath, search, watcherOptions } = params + const holdingWatcher = startHoldingWatcher(params, logger) + await holdingWatcher.ready + + // A failed rebuild closes the watcher rather than leave it holding every + // later change; the caller sees the rebuild's error. + const rebuildIndexOrCloseWatcher = async (): Promise<{ + count: number + embedding: Promise + }> => { + try { + return await search.rebuildFromVault({ vaultPath }, logger) + } catch (error) { + await holdingWatcher.close() + throw error + } + } + const rebuildResult = await rebuildIndexOrCloseWatcher() + + // `applied` never rejects, so past the deadline it can finish unawaited. + const { heldPathCount, applied } = holdingWatcher.applyHeldChanges() + const heldChangesDeadline = + watcherOptions?.heldChangesDeadline ?? DEFAULT_HELD_CHANGES_DEADLINE_MS + const { promise: deadlinePassed, resolve: passDeadline } = Promise.withResolvers<"deadline">() + const deadlineTimer = setTimeout(() => passDeadline("deadline"), heldChangesDeadline) + + // `applied` resolves with nothing, so only the deadline makes this "deadline". + const firstSettled = await Promise.race([applied, deadlinePassed]) + clearTimeout(deadlineTimer) + + if (firstSettled === "deadline") { + logger.info("startup continues while the changes held during it are still applied", { + heldPaths: heldPathCount, + heldChangesDeadline, + }) + } + return rebuildResult } diff --git a/src/vault-mcp/search/search-index.ts b/src/vault-mcp/search/search-index.ts index 5f1ed3977..5bf55d728 100644 --- a/src/vault-mcp/search/search-index.ts +++ b/src/vault-mcp/search/search-index.ts @@ -39,6 +39,15 @@ import { hybridSearch } from "./hybrid-search.js" // ── File content indexing constants ───────────────────────────── +/** Files read at once while indexing from disk, by the startup rebuild and by + * the file watcher's startup batch, so a large vault cannot open every file + * at once. */ +export const INDEXING_FILE_READ_CONCURRENCY = 16 + +/** PDFs whose text is extracted at once while indexing. Extraction is + * CPU-heavy, so fewer run together than other file reads. */ +export const PDF_EXTRACTION_CONCURRENCY = 4 + /** Extensions whose content is indexed into file_content_fts for vault_search. * Excludes .canvas (own path with link extraction) and .md (notes_fts). */ export const INDEXABLE_TEXT_EXTENSIONS = new Set([ @@ -2077,12 +2086,23 @@ export const createSearchIndex = ( } /** Rebuilds note, task, file and graph indexes; reconciles retained memory rows and vectors. - * Returns the note count and a background embedding promise. */ + * Returns the note count and a background embedding promise. + * - No source write may run while it does. A source write is an upsert or remove call + * that changes the tables this clears; one that lands during its reads is replaced by + * the older content it read. + * - The file watcher is the only caller that makes source writes at runtime, and + * watchVaultAndBuildIndex holds it until this returns; a new runtime caller must go + * through that hold. + * - The returned embedding promise rejects if the background pass fails. This function + * already logs that failure, so a caller may leave the promise unawaited. */ const rebuildFromVault = async ( params: { vaultPath: string }, logger: Logger, ): Promise<{ count: number; embedding: Promise }> => { const { vaultPath } = params + // Embedding work that outlived its source can leave vectors whose chunk or + // memory entry is gone. They would take nearest-neighbour slots from live + // vectors, so they are dropped before anything else. const orphanNoteVectorsRemoved = deleteOrphanNoteVectorsStmt?.run().changes ?? 0 const orphanFileVectorsRemoved = deleteOrphanFileVectorsStmt?.run().changes ?? 0 const orphanMemoryVectorsRemoved = deleteOrphanMemoryVectorsStmt?.run().changes ?? 0 @@ -2177,11 +2197,9 @@ export const createSearchIndex = ( return null } } - /** Bound filesystem work so a large vault cannot open every file at once. */ - const REBUILD_IO_CONCURRENCY = 16 const nonMarkdownFileSizeResults = await mapWithConcurrency({ items: allNonMdFiles, - concurrency: REBUILD_IO_CONCURRENCY, + concurrency: INDEXING_FILE_READ_CONCURRENCY, mapper: readFileSize, }) const nonMarkdownFileSizes = nonMarkdownFileSizeResults.filter((entry) => entry !== null) @@ -2198,7 +2216,6 @@ export const createSearchIndex = ( }) : [] - // Read canvas files for content indexing + link extraction. const readRebuildFileContent = async (params: { file: RebuildFilePaths sourceKind: "canvas file" | "text file" | "note" @@ -2224,23 +2241,18 @@ export const createSearchIndex = ( return null } } + // Read canvas files for content indexing + link extraction. const canvasContentResults = await mapWithConcurrency({ items: canvasFiles, - concurrency: REBUILD_IO_CONCURRENCY, + concurrency: INDEXING_FILE_READ_CONCURRENCY, mapper: (file) => readRebuildFileContent({ file, sourceKind: "canvas file" }), }) const canvasContents = canvasContentResults.filter((entry) => entry !== null) // Extract PDF text with bounded concurrency (CPU-intensive pdfjs work). - const extractPdfContent = async (file: { - absolutePath: string - relativePath: string - }): Promise<{ - relativePath: string - content: string - modifiedAtMs: number - sizeBytes: number - } | null> => { + const extractPdfContent = async ( + file: RebuildFilePaths, + ): Promise => { try { const [buffer, fileStat] = await Promise.all([ readFile(file.absolutePath), @@ -2265,7 +2277,7 @@ export const createSearchIndex = ( } const pdfResults = await mapWithConcurrency({ items: pdfFiles, - concurrency: 4, + concurrency: PDF_EXTRACTION_CONCURRENCY, mapper: extractPdfContent, }) const pdfContents = pdfResults.filter((entry) => entry !== null) @@ -2273,14 +2285,14 @@ export const createSearchIndex = ( // Read text files for content indexing (raw UTF-8). const textFileContentResults = await mapWithConcurrency({ items: textFiles, - concurrency: REBUILD_IO_CONCURRENCY, + concurrency: INDEXING_FILE_READ_CONCURRENCY, mapper: (file) => readRebuildFileContent({ file, sourceKind: "text file" }), }) const textFileContents = textFileContentResults.filter((entry) => entry !== null) const noteContentResults = await mapWithConcurrency({ items: markdownFiles, - concurrency: REBUILD_IO_CONCURRENCY, + concurrency: INDEXING_FILE_READ_CONCURRENCY, mapper: (file) => readRebuildFileContent({ file, sourceKind: "note" }), }) const noteContents = noteContentResults.filter((entry) => entry !== null) @@ -2371,6 +2383,9 @@ export const createSearchIndex = ( const pathList = allPaths.map((row) => row.path) cachedCanvasNotePaths = pathList + // No source write may run during a rebuild, so this second clear + // normally finds nothing. It stays so that a link written during the + // file reads above, against that rule, cannot outlive the rebuild. db.exec("DELETE FROM links") for (const note of noteContents) { if (skippedNotePaths.has(note.relativePath)) continue @@ -2465,9 +2480,8 @@ export const createSearchIndex = ( // Every note read from disk, including notes that failed to index, unlike indexedNotes const readableNotePaths = new Set(noteContents.map((note) => note.relativePath)) - // Pass 3 runs in the background — the server can start accepting requests - // immediately after FTS indexing (Passes 1+2) finishes. Embedding is a - // progressive enhancement: search works with FTS-only until vectors are ready. + // Pass 3 runs in the background, so startup does not wait for it. Until a + // note's vectors exist, search finds that note by keyword only. const embeddingPromise = embedder && selectAllNoteChunkPathsStmt ? (async () => { @@ -2497,13 +2511,12 @@ export const createSearchIndex = ( let entriesEmbedded = 0 let embedErrors = 0 for (const note of notesForEmbedding) { - if ( - !isCurrentSourceVersion( - { sourcePath: note.relativePath, sourceVersion: note.sourceVersion }, - logger, - ) + const noteVersionIsCurrent = isCurrentSourceVersion( + { sourcePath: note.relativePath, sourceVersion: note.sourceVersion }, + logger, ) - continue + + if (!noteVersionIsCurrent) continue try { chunksEmbedded += await embedAndStoreChunks( @@ -2516,6 +2529,8 @@ export const createSearchIndex = ( ) const memoryFile = memoryFileNameFromPath(note.relativePath) + // Checked again after the await above: a newer write to the + // note during the chunk embedding makes this entry pass obsolete. if ( memoryFile && isCurrentSourceVersion( @@ -2580,13 +2595,12 @@ export const createSearchIndex = ( let fileChunksEmbedded = 0 let fileEmbedErrors = 0 for (const file of filesForEmbedding) { - if ( - !isCurrentSourceVersion( - { sourcePath: file.path, sourceVersion: file.sourceVersion }, - logger, - ) + const fileVersionIsCurrent = isCurrentSourceVersion( + { sourcePath: file.path, sourceVersion: file.sourceVersion }, + logger, ) - continue + + if (!fileVersionIsCurrent) continue try { fileChunksEmbedded += await embedAndStoreFileChunks( diff --git a/src/vault-mcp/server.ts b/src/vault-mcp/server.ts index 4735eb9cd..9ff501b4f 100644 --- a/src/vault-mcp/server.ts +++ b/src/vault-mcp/server.ts @@ -10,7 +10,7 @@ import { createReranker } from "./search/reranker.js" import { createMemoryStore } from "./vault-operations/memory-store.js" import { readTaskFormatConfig } from "./vault-operations/task-format-config.js" import { trashSweeper } from "./vault-operations/trash-sweeper.js" -import { startFileWatcher } from "./search/file-watcher.js" +import { watchVaultAndBuildIndex } from "./search/file-watcher.js" import { createOAuthProvider } from "./oauth/oauth-provider.js" import { createOAuthRoutes } from "./oauth/oauth-routes.js" import { createMcpRouter } from "./mcp-core/mcp-router.js" @@ -117,6 +117,7 @@ const startServer = async (): Promise => { // /data is the Docker volume where the container persists indexes and logs const dataDir = indexDbPath ? dirname(indexDbPath) : "/data" const searchDbPath = indexDbPath ?? `${dataDir}/search.db` + // oauth.db sits beside the search index, so INDEX_DB_PATH moves both. const oauthDbPath = `${dataDir}/oauth.db` const port = env.get("PORT").default("8000").asPortNumber() const host = env.get("HOST").default("0.0.0.0").asString() @@ -146,13 +147,18 @@ const startServer = async (): Promise => { fileToolsEnabled: config.fileToolsEnabled, statusRegistry, }) + + // Before the watcher starts, so its template writes are not held and re-read. await bootstrapMemoryIfEnabled(config, vaultPath) - const { count } = await search.rebuildFromVault({ vaultPath }, logger) - logger.info("initial index built", { count }) - await startFileWatcher(vaultPath, search, { - usePolling: config.windowsBindMount, - }) + // watchVaultAndBuildIndex also returns the rebuild's embedding promise. Startup + // does not wait for it, so the server takes requests while the embedding pass + // is still running, and rebuildFromVault logs an embedding failure itself. + const { count } = await watchVaultAndBuildIndex( + { vaultPath, search, watcherOptions: { usePolling: config.windowsBindMount } }, + logger, + ) + logger.info("initial index built", { notesIndexedByRebuild: count }) const oauthProvider = createOAuthProvider({ authToken,