diff --git a/packages/fleet/src/store/__tests__/watcher.test.ts b/packages/fleet/src/store/__tests__/watcher.test.ts index 5e4aeaf2..feebbd06 100644 --- a/packages/fleet/src/store/__tests__/watcher.test.ts +++ b/packages/fleet/src/store/__tests__/watcher.test.ts @@ -2,36 +2,17 @@ import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { describe, expect, it, onTestFinished, vi } from 'vitest'; -import { startWatcher, type Watcher } from '../watcher.ts'; +import { startWatcher, type WatchStarter } from '../watcher.ts'; -let dir: string; -let watchers: Watcher[]; - -/** Starts a watcher on a short rescan interval, registering it for cleanup. */ -function startTestWatcher(overrides: { dir?: string } = {}): { - log: ReturnType; - onDirty: ReturnType; -} { - const log = vi.fn(); - const onDirty = vi.fn(); - watchers.push(startWatcher({ debounceMs: 5, dir, log, onDirty, rescanMs: 10, ...overrides })); - return { log, onDirty }; +interface WatcherOverrides { + debounceMs?: number; + dir?: string; + rescanMs?: number; + startWatch?: WatchStarter; } -beforeEach(() => { - dir = mkdtempSync(join(tmpdir(), 'watcher-')); - watchers = []; -}); - -afterEach(() => { - for (const watcher of watchers) { - watcher.stop(); - } - rmSync(dir, { recursive: true, force: true }); -}); - describe('startWatcher', () => { it('logs the recursive-watch mode when the directory is watchable', () => { const { log } = startTestWatcher(); @@ -40,15 +21,40 @@ describe('startWatcher', () => { }); it('fires onDirty on every rescan tick', async () => { - const { onDirty } = startTestWatcher(); + const { onDirty } = startTestWatcher({ startWatch: createFakeWatch().startWatch }); await vi.waitFor(() => { expect(onDirty.mock.calls.length).toBeGreaterThanOrEqual(2); }); }); + it('collapses a burst of watch events into one onDirty', async () => { + const fake = createFakeWatch(); + // The rescan interval fires the same callback, so it is pushed past this test's lifetime to leave the count + // attributable to the burst alone. + const { onDirty } = startTestWatcher({ rescanMs: 60_000, startWatch: fake.startWatch }); + + fake.fireEvent(); + fake.fireEvent(); + fake.fireEvent(); + await vi.waitFor(() => expect(onDirty).toHaveBeenCalled()); + + expect(onDirty).toHaveBeenCalledOnce(); + }); + + it('when the watch emits an error, announces the downgrade and closes the handle', () => { + const fake = createFakeWatch(); + const { log } = startTestWatcher({ startWatch: fake.startWatch }); + log.mockClear(); + + fake.emitError(new Error('EMFILE: too many open files')); + + expect(log).toHaveBeenCalledExactlyOnceWith(expect.stringContaining('downgrading to rescan-only')); + expect(fake.isClosed()).toBe(true); + }); + it('when the directory does not exist, announces rescan-only mode and still rescans', async () => { - const { log, onDirty } = startTestWatcher({ dir: join(dir, 'missing') }); + const { log, onDirty } = startTestWatcher({ dir: join(createTempDir(), 'missing') }); expect(log).toHaveBeenCalledExactlyOnceWith(expect.stringContaining('rescan-only')); await vi.waitFor(() => { @@ -56,18 +62,97 @@ describe('startWatcher', () => { }); }); + it('when the directory does not exist, does not attempt a watch', () => { + const fake = createFakeWatch(); + + startTestWatcher({ dir: join(createTempDir(), 'missing'), startWatch: fake.startWatch }); + + expect(fake.isStarted()).toBe(false); + }); + + it('when the watch fails to start, names the target and the reason', () => { + const dir = createTempDir(); + + const { log } = startTestWatcher({ + dir, + startWatch: () => { + throw new Error('EPERM: operation not permitted'); + }, + }); + + expect(log).toHaveBeenCalledExactlyOnceWith( + expect.stringContaining(`cannot watch ${dir} (EPERM: operation not permitted); rescan-only`), + ); + }); + it('stops firing after stop()', async () => { - const { onDirty } = startTestWatcher(); + const { onDirty, stop } = startTestWatcher({ startWatch: createFakeWatch().startWatch }); await vi.waitFor(() => { expect(onDirty).toHaveBeenCalled(); }); - for (const watcher of watchers.splice(0)) { - watcher.stop(); - } + stop(); onDirty.mockClear(); await new Promise((resolve) => setTimeout(resolve, 50)); expect(onDirty).not.toHaveBeenCalled(); }); }); + +// region | Helpers + +/** A `WatchStarter` that records how it was driven, so tests exercise the watch contract without an OS watch. */ +function createFakeWatch(): { + emitError: (error: Error) => void; + fireEvent: () => void; + isClosed: () => boolean; + isStarted: () => boolean; + startWatch: WatchStarter; +} { + let closed = false; + let onError: ((error: Error) => void) | undefined; + let onEvent: (() => void) | undefined; + let started = false; + + return { + emitError: (error) => onError?.(error), + fireEvent: () => onEvent?.(), + isClosed: () => closed, + isStarted: () => started, + startWatch: (_dir, event) => { + started = true; + onEvent = event; + return { + close: () => { + closed = true; + }, + on: (_event, listener) => { + onError = listener; + }, + }; + }, + }; +} + +/** Creates a temp directory for the current test, removed when the test finishes. */ +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), 'watcher-')); + onTestFinished(() => rmSync(dir, { recursive: true, force: true })); + return dir; +} + +/** Starts a watcher on a short rescan interval, stopped when the current test finishes. */ +function startTestWatcher(overrides: WatcherOverrides = {}): { + log: ReturnType; + onDirty: ReturnType; + stop: () => void; +} { + const dir = overrides.dir ?? createTempDir(); + const log = vi.fn(); + const onDirty = vi.fn(); + const watcher = startWatcher({ debounceMs: 5, dir, log, onDirty, rescanMs: 10, ...overrides }); + onTestFinished(() => watcher.stop()); + return { log, onDirty, stop: () => watcher.stop() }; +} + +// endregion | Helpers diff --git a/packages/fleet/src/store/watcher.ts b/packages/fleet/src/store/watcher.ts index 2dc93b22..bddbf2c5 100644 --- a/packages/fleet/src/store/watcher.ts +++ b/packages/fleet/src/store/watcher.ts @@ -2,17 +2,26 @@ // an always-on rescan interval as the correctness backstop. Watching is best-effort — recursive support varies by // platform, and a watch can fail mid-run — so degradation to rescan-only is announced, never silent, and never fatal. -import { type FSWatcher, watch } from 'node:fs'; +import { statSync, watch } from 'node:fs'; /** A running watcher; `stop` releases the watch handle and every timer. */ export interface Watcher { stop(): void; } +/** The subset of a watch handle this module uses: stopping it, and learning that it failed. */ +export interface WatchHandle { + close(): void; + on(event: 'error', listener: (error: Error) => void): unknown; +} + +/** Starts a recursive watch on `dir`, invoking `onEvent` for every filesystem event. */ +export type WatchStarter = (dir: string, onEvent: () => void) => WatchHandle; + /** * Starts watching `dir`, invoking `onDirty` on debounced watch events and on every rescan tick. `log` receives one * startup line naming the active mode — recursive watch or rescan-only, with the reason — and a line on any later - * downgrade. + * downgrade. `startWatch` is injectable for tests and defaults to {@link startRecursiveWatch}. */ export function startWatcher(input: { debounceMs: number; @@ -20,9 +29,11 @@ export function startWatcher(input: { log: (message: string) => void; onDirty: () => void; rescanMs: number; + startWatch?: WatchStarter; }): Watcher { + const startWatch = input.startWatch ?? startRecursiveWatch; let debounce: ReturnType | undefined; - let watcher: FSWatcher | undefined; + let watcher: WatchHandle | undefined; function handleWatchEvent(): void { if (debounce !== undefined) { @@ -35,7 +46,10 @@ export function startWatcher(input: { } try { - watcher = watch(input.dir, { recursive: true }, handleWatchEvent); + // `fs.watch` reports a missing target inconsistently across platforms, returning an inert watcher on Linux, so its + // not throwing is no evidence that the watch is live. + statSync(input.dir); + watcher = startWatch(input.dir, handleWatchEvent); watcher.on('error', (error) => { // A mid-run watch error — the tree removed, an OS watch limit — must not crash the server; the rescan // interval keeps state current. @@ -45,7 +59,7 @@ export function startWatcher(input: { }); input.log(`recursive watch active on ${input.dir}; rescan backstop every ${input.rescanMs}ms`); } catch (error) { - input.log(`fs.watch unavailable (${readMessage(error)}); rescan-only every ${input.rescanMs}ms`); + input.log(`cannot watch ${input.dir} (${readMessage(error)}); rescan-only every ${input.rescanMs}ms`); } const rescanTimer = setInterval(() => input.onDirty(), input.rescanMs); @@ -68,4 +82,9 @@ function readMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } +/** Starts Node's recursive watch on `dir`. */ +function startRecursiveWatch(dir: string, onEvent: () => void): WatchHandle { + return watch(dir, { recursive: true }, onEvent); +} + // endregion | Helpers