diff --git a/packages/server/src/server/watcher-liveness-canary.test.ts b/packages/server/src/server/watcher-liveness-canary.test.ts index 4da1481408..7e719da9f0 100644 --- a/packages/server/src/server/watcher-liveness-canary.test.ts +++ b/packages/server/src/server/watcher-liveness-canary.test.ts @@ -1,4 +1,5 @@ -import { mkdtemp, rm } from "node:fs/promises"; +import { watchFile, unwatchFile } from "node:fs"; +import { mkdtemp, readFile, rm, stat } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, expect, test } from "vitest"; @@ -31,3 +32,39 @@ test("rejects when the watcher never reports the canary", async () => { await expect(canary.verify()).rejects.toThrow("did not report its liveness canary"); }); + +test("verifies a watcher that starts delivering after the initial canary write", async () => { + const watchRoot = await mkdtemp(join(tmpdir(), "paseo-watcher-canary-")); + cleanupPaths.push(watchRoot); + const canary = createWatcherLivenessCanary(watchRoot); + const verification = canary.verify(); + const verified = expect(verification).resolves.toBeUndefined(); + + // The initial write happens before the observer starts delivering changes. + await expect.poll(() => readFile(canary.path, "utf8")).toBe("paseo watcher liveness canary\n"); + watchFile(canary.path, { interval: 10 }, () => { + canary.filterEvents([{ path: canary.path, type: "update" }]); + }); + try { + await verified; + await expect(readFile(canary.path)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { + unwatchFile(canary.path); + } +}); + +test("aborting verification stops pulses before removing the canary", async () => { + const watchRoot = await mkdtemp(join(tmpdir(), "paseo-watcher-canary-")); + cleanupPaths.push(watchRoot); + const canary = createWatcherLivenessCanary(watchRoot); + const controller = new AbortController(); + const verification = canary.verify(controller.signal); + const cancelled = expect(verification).rejects.toThrow("stop watching"); + await expect.poll(() => readFile(canary.path, "utf8")).toBe("paseo watcher liveness canary\n"); + const initialWrite = (await stat(canary.path)).mtimeMs; + await expect.poll(async () => (await stat(canary.path)).mtimeMs).toBeGreaterThan(initialWrite); + + controller.abort(new Error("stop watching")); + await cancelled; + await expect(readFile(canary.path)).rejects.toMatchObject({ code: "ENOENT" }); +}); diff --git a/packages/server/src/server/watcher-liveness-canary.ts b/packages/server/src/server/watcher-liveness-canary.ts index a366f0229a..2fc5c5dbdf 100644 --- a/packages/server/src/server/watcher-liveness-canary.ts +++ b/packages/server/src/server/watcher-liveness-canary.ts @@ -1,9 +1,11 @@ import { randomUUID } from "node:crypto"; import { rm, writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { setInterval } from "node:timers/promises"; import type { FileChange } from "./file-observer/index.js"; export const WATCHER_LIVENESS_CANARY_TIMEOUT_MS = 10_000; +const CANARY_CONTENTS = "paseo watcher liveness canary\n"; export interface WatcherLivenessCanary { readonly path: string; @@ -32,7 +34,11 @@ export function createWatcherLivenessCanary( return filtered; }, async verify(signal) { - await writeFile(canaryPath, "paseo watcher liveness canary\n", { flag: "wx" }); + await writeFile(canaryPath, CANARY_CONTENTS, { flag: "wx" }); + // Native registration can return before macOS starts its FSEvents stream. + // Keep probing until an event round-trips instead of relying on the first write. + const pulses = new AbortController(); + const pulsing = pulseCanary(canaryPath, pulses.signal); let timeout: NodeJS.Timeout | null = null; let removeAbortListener = () => {}; try { @@ -55,12 +61,24 @@ export function createWatcherLivenessCanary( signal.addEventListener("abort", rejectForAbort, { once: true }); removeAbortListener = () => signal.removeEventListener("abort", rejectForAbort); }); - await Promise.race([reported, timeoutPromise, abortPromise]); + await Promise.race([reported, timeoutPromise, abortPromise, pulsing]); } finally { if (timeout) clearTimeout(timeout); removeAbortListener(); + pulses.abort(); + await Promise.allSettled([pulsing]); await rm(canaryPath, { force: true }); } }, }; } + +async function pulseCanary(path: string, signal: AbortSignal): Promise { + try { + for await (const _ of setInterval(100, undefined, { signal })) { + await writeFile(path, CANARY_CONTENTS); + } + } catch (error) { + if (!signal.aborted) throw error; + } +} diff --git a/packages/server/src/server/workspace-git-service.observation.integration.test.ts b/packages/server/src/server/workspace-git-service.observation.integration.test.ts index 9f53dec33a..a2cb27739d 100644 --- a/packages/server/src/server/workspace-git-service.observation.integration.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.integration.test.ts @@ -12,12 +12,7 @@ import { WorkspaceGitServiceImpl } from "./workspace-git-service.js"; import type { FileChange, FileObserver, SubscribeToFileChanges } from "./file-observer/index.js"; function createLogger(): pino.Logger { - const logger = { - child: () => logger, - debug: vi.fn(), - warn: vi.fn(), - }; - return logger as unknown as pino.Logger; + return pino({ level: "warn" }); } function createFacts(cwd: string): CheckoutSnapshotFacts { @@ -232,7 +227,14 @@ test("recursive observation updates tracked state and prunes ignored storms", as await vi.waitFor( () => { - expect(activeWatcherCount).toBe(2); + expect( + activeWatcherCount, + JSON.stringify({ + observer: observer.getDiagnostics(), + service: service.getMetrics(), + deliveredEvents, + }), + ).toBe(2); expect(workspaceRuntimeStore.get(repoDir)).not.toBeNull(); expect(service.getMetrics()).toMatchObject({ workspaceObservationSetupInFlightCount: 0,