From 48e65b407fd7df4d097df7c264fc168f2665ea4f Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 10 Aug 2026 13:43:48 +0200 Subject: [PATCH] Replace Parcel Watcher with Custom Watcher (#3107) * fix(server): replace wedging file observation Native watcher teardown could block the daemon main thread during workspace archive. Own recursive observation behind an awaited lifecycle boundary, retain an isolated comparison harness, and gate correctness, resources, and sustained CPU across platforms. * test(server): accept root watcher evidence Native recursive watchers may coalesce ignored descendant changes into a root event. Assert that ignored paths stay excluded while a nonignored sentinel proves delivery remains live. * test(server): preserve watcher comparison evidence Record Parcel delivery misses without weakening the production observer gate, and allow the Windows stress test to use its declared polling window. * fix(server): reconcile native watcher omissions Back macOS and Windows recursive watching with a bounded namespace inventory so coalesced create and delete events converge without unbounded rescans. * test(server): compare worktree filesystem identity Windows can return the same worktree through 8.3 and long path spellings. Preserve cardinality and branch checks while comparing the real path. * perf(server): bound native inventory audits Reduce full native inventory cadence while preserving a six-second convergence bound, and gate setup readiness and event-loop p99 against platform-appropriate budgets. * build: refresh Nix npm dependency hash * perf(server): scope native file reconciliation * perf(server): defer native safety audits until quiet * fix(server): move native safety deadline during audits * perf(server): audit native inventories after settled churn * fix(server): reconcile native root evidence promptly * perf(server): deduplicate native audit scheduling * perf(server): coalesce native event clusters * perf(server): suppress fresh native change scans * test(server): separate burst recovery from sustained churn * test(server): stabilize observer performance gates * perf(server): skip fresh native change scans * fix(server): preserve nested native change audits * perf(server): coalesce native repair scans * fix(server): reconcile native directory moves promptly * test(server): bound observer burst repair cost * refactor(server): own file observer lifecycle * build: refresh Nix npm dependency hash * ci: fold observer coverage into server suite * fix(server): await file observer shutdown --- CLAUDE.md | 1 + docs/file-observation.md | 19 + nix/npm-deps.hash | 2 +- package-lock.json | 277 +-------- packages/server/package.json | 5 +- .../server/scripts/measure-file-observer.ts | 403 ++++++++++++ .../server/scripts/measure-git-observation.ts | 25 +- .../reproduce-file-observer-teardown.ts | 169 +++++ .../src/server/file-observer/index.test.ts | 362 +++++++++++ .../server/src/server/file-observer/index.ts | 184 ++++++ .../file-observer/internal/contracts.ts | 37 ++ .../server/file-observer/internal/linux.ts | 270 ++++++++ .../internal/native-recursive.ts | 588 ++++++++++++++++++ .../file-observer/internal/observation.ts | 174 ++++++ .../server/file-observer/internal/paths.ts | 55 ++ ...ate-agent-worktree-autoarchive.e2e.test.ts | 24 +- .../test-utils/workspace-git-service-stub.ts | 24 +- .../server/src/server/websocket-server.ts | 26 +- ...it-service.observation.integration.test.ts | 77 ++- .../workspace-git-service.observation.test.ts | 114 +++- .../workspace-git-service.primitive.test.ts | 32 +- .../src/server/workspace-git-service.test.ts | 22 +- .../src/server/workspace-git-service.ts | 156 +++-- scripts/trace-daemon.mjs | 23 - 24 files changed, 2596 insertions(+), 473 deletions(-) create mode 100644 docs/file-observation.md create mode 100644 packages/server/scripts/measure-file-observer.ts create mode 100644 packages/server/scripts/reproduce-file-observer-teardown.ts create mode 100644 packages/server/src/server/file-observer/index.test.ts create mode 100644 packages/server/src/server/file-observer/index.ts create mode 100644 packages/server/src/server/file-observer/internal/contracts.ts create mode 100644 packages/server/src/server/file-observer/internal/linux.ts create mode 100644 packages/server/src/server/file-observer/internal/native-recursive.ts create mode 100644 packages/server/src/server/file-observer/internal/observation.ts create mode 100644 packages/server/src/server/file-observer/internal/paths.ts diff --git a/CLAUDE.md b/CLAUDE.md index 82776096f..2792cebf0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -46,6 +46,7 @@ At the start of non-trivial work, list `docs/` and skim anything relevant to the | [docs/protocol-compatibility.md](docs/protocol-compatibility.md) | Why app/daemon versions drift, protocol vs feature contract, capability gating, COMPAT tagging | | [docs/protocol-validation.md](docs/protocol-validation.md) | zod-aot generated inbound WebSocket validation, patched compiler regressions, schema-purity rules | | [docs/terminal-performance.md](docs/terminal-performance.md) | Terminal latency pipeline, coalescing/backpressure invariants, benchmark + perf spec usage | +| [docs/file-observation.md](docs/file-observation.md) | Recursive watcher ownership, Linux constraints, teardown invariants, and Parcel comparison | | [docs/testing.md](docs/testing.md) | TDD workflow, determinism, real dependencies over mocks, test organization | | [docs/qa.md](docs/qa.md) | QA evidence bar for pull requests — platform matrix, version drift, performance, UI proof | | [docs/mobile-testing.md](docs/mobile-testing.md) | Maestro and mobile test workflows | diff --git a/docs/file-observation.md b/docs/file-observation.md new file mode 100644 index 000000000..1418422ea --- /dev/null +++ b/docs/file-observation.md @@ -0,0 +1,19 @@ +# File observation + +Use `packages/server/src/server/file-observer` for recursive filesystem observation. Create one observer service at the owning service boundary, subscribe roots through it, and close it during owner shutdown. Keep platform watchers, directory discovery, exclusion reconciliation, event batching, failure cleanup, diagnostics, and teardown inside the module. A consumer supplies a root and excluded subtrees; it must not branch on the platform or manage child watchers. + +Do not replace the Linux implementation with `fs.watch({ recursive: true })`. Node 22 implements Linux recursion in JavaScript by walking the tree and watching every file and directory. Paseo's directory-only Linux watcher already exhausted resources before ignored-tree pruning and a watcher cap were added in [#794](https://github.com/getpaseo/paseo/pull/794). + +Closing an observation must remain safe after its root is renamed or removed. Workspace archive removes owned worktrees while releasing observation references, so teardown cannot assume the watched path still exists. + +`unsubscribe()` is an awaited barrier. Once it resolves, no scan, reconciliation, queued event, native handle, or callback may remain. `updateIgnore()` is the equivalent barrier for excluded paths: events queued below a newly excluded root cannot be delivered after it resolves. Treat path disappearance while scanning, attaching, or receiving a watcher error as rename/archive churn. Surface permission and resource errors so WorkspaceGit can enter bounded polling. + +Linux topology repair is trailing-debounced and scoped to the renamed path. A file rename needs one classification stat; a new or moved directory scans only its subtree. Full-tree scans are reserved for startup and ignore-set replacement. The initial Linux scan conservatively emits existing files discovered before their directory watcher was installed; WorkspaceGit batches these into one refresh. Benchmark this startup cost with pre-populated trees. + +Windows and macOS use one native recursive watcher plus an ignored-pruned, directory-indexed inventory. Native events deliver immediately. Named events trigger serialized shallow scans of their parent directories; known directory events also reconcile that subtree. This recovers concrete create and delete paths omitted by coalescing without walking the whole root for each edit. Startup and exclusion changes use an awaited full inventory barrier. Activity arms a full safety audit 30 seconds after it settles, with a 30-second minimum interval and a five-minute starvation bound; pathless native events also trigger an immediate shallow root scan. Observation fails into polling rather than retaining more than 250,000 file and directory entries per root. + +Read aggregate health from the owning observer service. Runtime metrics include active observations, native handles, tracked native files, pending events and immediate reconciliation work, scoped and full reconciliation activity and latency, and failure counts. Delayed safety audits do not count as pending work. Closing the service releases every subscription and clears its diagnostics. Do not add path lists, watcher handles, or platform-specific controls to this interface. + +Git owns Git-ignore evaluation. The observer accepts absolute excluded roots and applies updates without replacing the observation or exposing its watcher topology. This keeps tracked files inside otherwise ignored directories observable and keeps Git policy out of the filesystem module. + +The real-filesystem contracts and daemon auto-archive lifecycle run in the normal server test suite. Use the scripts only for manual performance and soak work: `npm run measure:file-observer --workspace=@getpaseo/server` measures burst and sustained-create behavior, and `npm run repro:file-observer-teardown --workspace=@getpaseo/server` runs the teardown soak. diff --git a/nix/npm-deps.hash b/nix/npm-deps.hash index a29ac1eea..cb16a408d 100644 --- a/nix/npm-deps.hash +++ b/nix/npm-deps.hash @@ -1 +1 @@ -sha256-hnoCGKnTLB6p2xR+M/q9dT+F8S7ou1aFcx1D+et8obc= +sha256-tB0WZG6F6VFMS5zFqo73giSAn7SqfkJISF5fMjDioJQ= diff --git a/package-lock.json b/package-lock.json index 912320604..2147d093a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9515,280 +9515,6 @@ "node": "^20.19.0 || >=22.12.0" } }, - "node_modules/@parcel/watcher": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher/-/watcher-2.6.0.tgz", - "integrity": "sha512-7FNeNl8NCE7aINx7WXiKQrPYZWC/hvrTsmk6zmxbI7LTXE7hVek/n8AfVgpe2y82zl3w0HvCHN0bVKMBoJcC0w==", - "hasInstallScript": true, - "license": "MIT", - "dependencies": { - "detect-libc": "^2.0.3", - "is-glob": "^4.0.3", - "node-addon-api": "^7.0.0", - "picomatch": "^4.0.4" - }, - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - }, - "optionalDependencies": { - "@parcel/watcher-android-arm64": "2.6.0", - "@parcel/watcher-darwin-arm64": "2.6.0", - "@parcel/watcher-darwin-x64": "2.6.0", - "@parcel/watcher-freebsd-x64": "2.6.0", - "@parcel/watcher-linux-arm-glibc": "2.6.0", - "@parcel/watcher-linux-arm-musl": "2.6.0", - "@parcel/watcher-linux-arm64-glibc": "2.6.0", - "@parcel/watcher-linux-arm64-musl": "2.6.0", - "@parcel/watcher-linux-x64-glibc": "2.6.0", - "@parcel/watcher-linux-x64-musl": "2.6.0", - "@parcel/watcher-win32-arm64": "2.6.0", - "@parcel/watcher-win32-x64": "2.6.0" - } - }, - "node_modules/@parcel/watcher-android-arm64": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-android-arm64/-/watcher-android-arm64-2.6.0.tgz", - "integrity": "sha512-trgpLSCKRC/huFjXX/Smh+0sWe4+YtKfktIToiMl59ghz7z+qkH6kMvNnUbLyRs9N11t8l4svSCs1+5B3rOAhA==", - "cpu": [ - "arm64" - ], - "license": "MIT", - "optional": true, - "os": [ - "android" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-darwin-arm64": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-darwin-arm64/-/watcher-darwin-arm64-2.6.0.tgz", - "integrity": "sha512-Y3QV0gl7Q1zbfueunkWIERICbEojQFCgpyG7YqOGNFLsckXyI1xu9mAIUpKY9QBYzBtSkN8dBPwd3yiAO9ovMw==", - "cpu": [ - "arm64" - ], - "license": "MIT", - "optional": true, - "os": [ - "darwin" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-darwin-x64": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-darwin-x64/-/watcher-darwin-x64-2.6.0.tgz", - "integrity": "sha512-Ohv6OpzhUfKYD7Beb8kDvG0jbIxORCYY1JRdZnaBtnjjkJxgD7ZVL0nw2sCYd0yTMKTvz3nnTnOF3cDifK+kvw==", - "cpu": [ - "x64" - ], - "license": "MIT", - "optional": true, - "os": [ - "darwin" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-freebsd-x64": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-freebsd-x64/-/watcher-freebsd-x64-2.6.0.tgz", - "integrity": "sha512-5HmXvDgs8VK+74jF9y9/2FE3/OnlcKmc56tjmSrEuZjpSZOGL+fvAu+HKJBdPs9uwoP2hE6TlSUpXZ/C5jUFmQ==", - "cpu": [ - "x64" - ], - "license": "MIT", - "optional": true, - "os": [ - "freebsd" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-linux-arm-glibc": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm-glibc/-/watcher-linux-arm-glibc-2.6.0.tgz", - "integrity": "sha512-Ps/hui3A+vMbjdqlqAowK2ZL8+BO8dBjxeWXj6npTBs3jx4wWmbPpaLuqwrQrSqIVMCnpWo238bJ1U37GhQOYg==", - "cpu": [ - "arm" - ], - "license": "MIT", - "optional": true, - "os": [ - "linux" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-linux-arm-musl": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm-musl/-/watcher-linux-arm-musl-2.6.0.tgz", - "integrity": "sha512-9c6AUHgHoG+IY88MRIHupztQiQnrbqHYQjkM2btA+Bf/wQnQMuiD0Wfk1EVv3TlNT3x41uU71rn6E4xh/+zvkw==", - "cpu": [ - "arm" - ], - "license": "MIT", - "optional": true, - "os": [ - "linux" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-linux-arm64-glibc": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm64-glibc/-/watcher-linux-arm64-glibc-2.6.0.tgz", - "integrity": "sha512-yHRqS2owEXe6Hic9z6Mh1ECsCd+ODVOGvZDyciqRd21+v+o+DnXMOrw50DSpIG2sb8GPEaPPmfeCAWKPJdq46g==", - "cpu": [ - "arm64" - ], - "license": "MIT", - "optional": true, - "os": [ - "linux" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-linux-arm64-musl": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-linux-arm64-musl/-/watcher-linux-arm64-musl-2.6.0.tgz", - "integrity": "sha512-WhB2e/V7rqdHHWZusBSPuy5Ei8S6lSz6FE5TKKQz5h3a0O+C+mhY7vxU9b/stqvMb8beLnPY82ZrFTLKs+SrKA==", - "cpu": [ - "arm64" - ], - "license": "MIT", - "optional": true, - "os": [ - "linux" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-linux-x64-glibc": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-linux-x64-glibc/-/watcher-linux-x64-glibc-2.6.0.tgz", - "integrity": "sha512-ulGE6x6Oz6iAwg75T8YQSoguBWasniIbX+QWpaYPcCnDOpdWX3k+4xbEYPZVLxOuoJI+svJJPD3sEj8G7lrQ3A==", - "cpu": [ - "x64" - ], - "license": "MIT", - "optional": true, - "os": [ - "linux" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-linux-x64-musl": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-linux-x64-musl/-/watcher-linux-x64-musl-2.6.0.tgz", - "integrity": "sha512-tkBYKt7YQrjIJWYDnto2YgO8MRkjlMTSNoRHzsXinBqbLdeOM3L32wPZJvIZxqaLMfSlS/4sUjH/6STVP/XDLw==", - "cpu": [ - "x64" - ], - "license": "MIT", - "optional": true, - "os": [ - "linux" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-win32-arm64": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-win32-arm64/-/watcher-win32-arm64-2.6.0.tgz", - "integrity": "sha512-gIZAP23jaHjGWasY/TY6yL7NHFClf0Ga7FN+iINvk+KN94rhm94lYZhFsbYFNcA04/onvGD9kKmiJLJB2HbNwQ==", - "cpu": [ - "arm64" - ], - "license": "MIT", - "optional": true, - "os": [ - "win32" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, - "node_modules/@parcel/watcher-win32-x64": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/@parcel/watcher-win32-x64/-/watcher-win32-x64-2.6.0.tgz", - "integrity": "sha512-cA+/pXV2YkfxlIcXOQ5fSWqAzzPyD78/x5qbK/I0vUkrlYHA8TIz+MXjAbGouguKVSI4bOmkTSJ1/poVSsgt+A==", - "cpu": [ - "x64" - ], - "license": "MIT", - "optional": true, - "os": [ - "win32" - ], - "engines": { - "node": ">= 10.0.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/parcel" - } - }, "node_modules/@pinojs/redact": { "version": "0.4.0", "resolved": "https://registry.npmjs.org/@pinojs/redact/-/redact-0.4.0.tgz", @@ -23278,6 +23004,7 @@ "version": "2.1.1", "resolved": "https://registry.npmjs.org/is-extglob/-/is-extglob-2.1.1.tgz", "integrity": "sha512-SbKbANkN603Vi4jEZv49LeVJMn4yGwsbzZworEoyEiutsN3nJYdbO36zfhGJ6QEDpOZIFkDtnq5JRxmvl3jsoQ==", + "dev": true, "license": "MIT", "engines": { "node": ">=0.10.0" @@ -23331,6 +23058,7 @@ "version": "4.0.3", "resolved": "https://registry.npmjs.org/is-glob/-/is-glob-4.0.3.tgz", "integrity": "sha512-xelSayHH36ZgE7ZWhli7pW34hNbNl8Ojv5KVmkJD4hBdD3th8Tfk9vYasLM+mXWOZhFkgZfxhLSnrwRr4elSSg==", + "dev": true, "license": "MIT", "dependencies": { "is-extglob": "^2.1.1" @@ -38289,7 +38017,6 @@ "@isaacs/ttlcache": "^2.1.4", "@modelcontextprotocol/sdk": "^1.20.1", "@opencode-ai/sdk": "1.14.46", - "@parcel/watcher": "^2.6.0", "@xterm/headless": "^6.0.0", "ai": "5.0.78", "ajv": "^8.20.0", diff --git a/packages/server/package.json b/packages/server/package.json index 4bbd964f2..31ced6203 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -46,12 +46,14 @@ "generate:config-schema": "tsx scripts/generate-config-schema.ts", "measure:agent-tools": "tsx scripts/measure-agent-tools-context.ts", "measure:git-observation": "tsx scripts/measure-git-observation.ts", + "measure:file-observer": "tsx scripts/measure-file-observer.ts", + "repro:file-observer-teardown": "tsx scripts/reproduce-file-observer-teardown.ts", "speech:models": "tsx scripts/list-speech-models.ts", "speech:download": "tsx scripts/download-speech-models.ts", "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts", "test": "npm run test:unit && npm run test:integration", "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"", - "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts", + "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\"", "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts", "test:integration:all": "npm run test:e2e", "test:integration:real": "vitest run real.e2e.test.ts", @@ -76,7 +78,6 @@ "@isaacs/ttlcache": "^2.1.4", "@modelcontextprotocol/sdk": "^1.20.1", "@opencode-ai/sdk": "1.14.46", - "@parcel/watcher": "^2.6.0", "@xterm/headless": "^6.0.0", "ai": "5.0.78", "ajv": "^8.20.0", diff --git a/packages/server/scripts/measure-file-observer.ts b/packages/server/scripts/measure-file-observer.ts new file mode 100644 index 000000000..2b9774ec4 --- /dev/null +++ b/packages/server/scripts/measure-file-observer.ts @@ -0,0 +1,403 @@ +import { execFile } from "node:child_process"; +import { mkdir, mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { monitorEventLoopDelay, performance } from "node:perf_hooks"; +import { promisify } from "node:util"; +import { createFileObserver, type FileChange } from "../src/server/file-observer/index.js"; + +const DIRECTORY_COUNT = readPositiveInteger("PASEO_WATCH_BENCH_DIRS", 500); +const ROOT_COUNT = readPositiveInteger("PASEO_WATCH_BENCH_ROOTS", 1); +const IGNORED_FILE_COUNT = readPositiveInteger("PASEO_WATCH_BENCH_IGNORED_FILES", 2_000); +const PREEXISTING_FILE_COUNT = readPositiveInteger("PASEO_WATCH_BENCH_PREEXISTING_FILES", 1_000); +const EDIT_COUNT = readPositiveInteger("PASEO_WATCH_BENCH_EDITS", 100); +const SUSTAINED_EDIT_COUNT = readPositiveInteger("PASEO_WATCH_BENCH_SUSTAINED_EDITS", 20); +const SUSTAINED_EDIT_INTERVAL_MS = readPositiveInteger( + "PASEO_WATCH_BENCH_SUSTAINED_INTERVAL_MS", + 500, +); +const REPETITIONS = readPositiveInteger("PASEO_WATCH_BENCH_REPETITIONS", 1); +const execFileAsync = promisify(execFile); + +interface Measurement { + backend: string; + run: number; + setupMs: number; + editLatencyMs: number; + editLatencyP50Ms: number; + editLatencyP95Ms: number; + editLatencyP99Ms: number; + burstDurationMs: number; + burstCpuMs: number; + sustainedDurationMs: number; + sustainedCpuMs: number; + sustainedScopedReconciliationCount: number | null; + sustainedFullReconciliationCount: number | null; + sustainedNativeEventCount: number | null; + sustainedNativeChangeEventCount: number | null; + sustainedNativeRenameEventCount: number | null; + sustainedNativePathlessEventCount: number | null; + sustainedNativeClassificationCount: number | null; + sustainedNativeShallowScanCount: number | null; + sustainedMissedPaths: number; + teardownMs: number; + startupEvents: number; + deliveredEvents: number; + ignoredEvents: number; + missedTrackedPaths: number; + cpuUserMs: number; + cpuSystemMs: number; + rssDeltaMiB: number; + eventLoopDelayP99Ms: number; + eventLoopDelayMaxMs: number; + kernelWatchCount: number | null; + scopedReconciliationCount: number | null; + fullReconciliationCount: number | null; +} + +interface FixtureRoot { + root: string; + ignoredRoot: string; + trackedRoot: string; +} + +async function main(): Promise { + const isChild = process.argv.includes("--child"); + + if (isChild) { + process.stdout.write(`${JSON.stringify(await measure(1))}\n`); + return; + } + + const results: Measurement[] = []; + for (let run = 1; run <= REPETITIONS; run += 1) { + results.push(await measureInChild(run)); + } + const evaluation = evaluate(results); + process.stdout.write( + `${JSON.stringify({ fixture: fixtureDescription(), evaluation, results }, null, 2)}\n`, + ); + if (!evaluation.passed) process.exitCode = 1; +} + +async function measureInChild(run: number): Promise { + const { stdout } = await execFileAsync( + process.execPath, + [...process.execArgv, process.argv[1], "--child"], + { env: process.env, maxBuffer: 1024 * 1024 }, + ); + return { ...(JSON.parse(stdout.trim()) as Measurement), run }; +} + +async function measure(run: number): Promise { + const base = await mkdtemp(join(tmpdir(), "paseo-watch-")); + const observer = createFileObserver(); + const roots: FixtureRoot[] = []; + for (let rootIndex = 0; rootIndex < ROOT_COUNT; rootIndex += 1) { + const root = join(base, `worktree-${rootIndex}`); + const ignoredRoot = join(root, "ignored"); + const trackedRoot = join(root, "tracked"); + await Promise.all([ + mkdir(ignoredRoot, { recursive: true }), + mkdir(trackedRoot, { recursive: true }), + ]); + for (let index = 0; index < DIRECTORY_COUNT; index += 1) { + await mkdir(join(trackedRoot, `directory-${index}`)); + } + await Promise.all( + Array.from({ length: PREEXISTING_FILE_COUNT }, (_, index) => + writeFile( + join(trackedRoot, `directory-${index % DIRECTORY_COUNT}`, `existing-${index}.txt`), + `${index}\n`, + ), + ), + ); + roots.push({ root, ignoredRoot, trackedRoot }); + } + const rssBefore = process.memoryUsage().rss; + const cpuBefore = process.cpuUsage(); + const diagnosticsBefore = observer.getDiagnostics(); + const eventLoopDelay = monitorEventLoopDelay({ resolution: 10 }); + eventLoopDelay.enable(); + const delivered: FileChange[] = []; + const firstObservedAt = new Map(); + let error: Error | null = null; + const setupStarted = performance.now(); + const subscriptions = await Promise.all( + roots.map(({ root, ignoredRoot }) => + observer.subscribe( + root, + (nextError, events) => { + error ??= nextError; + delivered.push(...events); + for (const event of events) { + if (!firstObservedAt.has(event.path)) + firstObservedAt.set(event.path, performance.now()); + } + }, + { ignore: [ignoredRoot] }, + ), + ), + ); + const setupMs = performance.now() - setupStarted; + await new Promise((resolve) => setTimeout(resolve, 25)); + const startupEvents = delivered.length; + const kernelWatchCount = await readKernelWatchCount(); + + try { + for (let index = 0; index < IGNORED_FILE_COUNT; index += 1) { + const target = roots[index % roots.length]; + await writeFile(join(target.ignoredRoot, `artifact-${index}.txt`), `${index}\n`); + } + const trackedPaths: string[] = []; + const writtenAt = new Map(); + const burstCpuBefore = process.cpuUsage(); + const editStarted = performance.now(); + for (let index = 0; index < EDIT_COUNT; index += 1) { + const target = roots[index % roots.length]; + const path = join( + target.trackedRoot, + `directory-${Math.floor(index / roots.length) % DIRECTORY_COUNT}`, + `edit-${index}.txt`, + ); + trackedPaths.push(path); + await writeFile(path, `${index}\n`); + writtenAt.set(path, performance.now()); + } + await waitFor( + () => trackedPaths.every((path) => delivered.some((event) => event.path === path)), + 10_000, + ); + const editLatencyMs = performance.now() - editStarted; + const observedLatencies = trackedPaths.map( + (path) => + (firstObservedAt.get(path) ?? performance.now()) - (writtenAt.get(path) ?? editStarted), + ); + if (error) throw error; + + const burstSettled = await waitFor(() => { + const diagnostics = observer.getDiagnostics(); + return ( + diagnostics.pendingEventCount === 0 && + diagnostics.pendingReconciliationWorkCount === 0 && + diagnostics.reconciliationInFlightCount === 0 + ); + }, 30_000); + if (!burstSettled) throw new Error("Timed out waiting for observer reconciliation to settle"); + const burstDurationMs = performance.now() - editStarted; + const burstCpu = process.cpuUsage(burstCpuBefore); + const burstCpuMs = (burstCpu.user + burstCpu.system) / 1_000; + + const sustainedPaths: string[] = []; + const diagnosticsBeforeSustained = observer.getDiagnostics(); + const sustainedCpuBefore = process.cpuUsage(); + const sustainedStarted = performance.now(); + for (let index = 0; index < SUSTAINED_EDIT_COUNT; index += 1) { + const target = roots[index % roots.length]; + const path = join(target.trackedRoot, `sustained-${index}.txt`); + sustainedPaths.push(path); + await writeFile(path, `${index}\n`); + await new Promise((resolve) => setTimeout(resolve, SUSTAINED_EDIT_INTERVAL_MS)); + } + await waitFor( + () => sustainedPaths.every((path) => delivered.some((event) => event.path === path)), + 10_000, + ); + const sustainedSettled = await waitFor(() => { + const diagnostics = observer.getDiagnostics(); + return ( + diagnostics.pendingEventCount === 0 && + diagnostics.pendingReconciliationWorkCount === 0 && + diagnostics.reconciliationInFlightCount === 0 + ); + }, 30_000); + if (!sustainedSettled) { + throw new Error("Timed out waiting for sustained observer reconciliation to settle"); + } + const sustainedDurationMs = performance.now() - sustainedStarted; + const sustainedCpu = process.cpuUsage(sustainedCpuBefore); + const sustainedCpuMs = (sustainedCpu.user + sustainedCpu.system) / 1_000; + const diagnosticsAfterSustained = observer.getDiagnostics(); + + const teardownStarted = performance.now(); + await Promise.all(subscriptions.map((subscription) => subscription.unsubscribe())); + const teardownMs = performance.now() - teardownStarted; + const diagnosticsAfter = observer.getDiagnostics(); + const cpu = process.cpuUsage(cpuBefore); + eventLoopDelay.disable(); + return { + backend: "production", + run, + setupMs, + editLatencyMs, + editLatencyP50Ms: percentile(observedLatencies, 0.5), + editLatencyP95Ms: percentile(observedLatencies, 0.95), + editLatencyP99Ms: percentile(observedLatencies, 0.99), + burstDurationMs, + burstCpuMs, + sustainedDurationMs, + sustainedCpuMs, + sustainedScopedReconciliationCount: + diagnosticsAfterSustained.scopedReconciliationCount - + diagnosticsBeforeSustained.scopedReconciliationCount, + sustainedFullReconciliationCount: + diagnosticsAfterSustained.fullReconciliationCount - + diagnosticsBeforeSustained.fullReconciliationCount, + sustainedNativeEventCount: + diagnosticsAfterSustained.nativeEventCount - diagnosticsBeforeSustained.nativeEventCount, + sustainedNativeChangeEventCount: + diagnosticsAfterSustained.nativeChangeEventCount - + diagnosticsBeforeSustained.nativeChangeEventCount, + sustainedNativeRenameEventCount: + diagnosticsAfterSustained.nativeRenameEventCount - + diagnosticsBeforeSustained.nativeRenameEventCount, + sustainedNativePathlessEventCount: + diagnosticsAfterSustained.nativePathlessEventCount - + diagnosticsBeforeSustained.nativePathlessEventCount, + sustainedNativeClassificationCount: + diagnosticsAfterSustained.nativeClassificationCount - + diagnosticsBeforeSustained.nativeClassificationCount, + sustainedNativeShallowScanCount: + diagnosticsAfterSustained.nativeShallowScanCount - + diagnosticsBeforeSustained.nativeShallowScanCount, + sustainedMissedPaths: sustainedPaths.filter( + (path) => !delivered.some((event) => event.path === path), + ).length, + teardownMs, + startupEvents, + deliveredEvents: delivered.length, + ignoredEvents: delivered.filter((event) => + roots.some(({ ignoredRoot }) => event.path.startsWith(ignoredRoot)), + ).length, + missedTrackedPaths: trackedPaths.filter( + (path) => !delivered.some((event) => event.path === path), + ).length, + cpuUserMs: cpu.user / 1_000, + cpuSystemMs: cpu.system / 1_000, + rssDeltaMiB: (process.memoryUsage().rss - rssBefore) / 1024 / 1024, + eventLoopDelayP99Ms: eventLoopDelay.percentile(99) / 1_000_000, + eventLoopDelayMaxMs: eventLoopDelay.max / 1_000_000, + kernelWatchCount, + scopedReconciliationCount: + diagnosticsAfter.scopedReconciliationCount - diagnosticsBefore.scopedReconciliationCount, + fullReconciliationCount: + diagnosticsAfter.fullReconciliationCount - diagnosticsBefore.fullReconciliationCount, + }; + } finally { + await Promise.all(subscriptions.map((subscription) => subscription.unsubscribe())); + await observer.close(); + await rm(base, { recursive: true, force: true }); + } +} + +async function readKernelWatchCount(): Promise { + if (process.platform !== "linux") return null; + let count = 0; + for (const entry of await readdir("/proc/self/fdinfo")) { + const contents = await readFile(join("/proc/self/fdinfo", entry), "utf8").catch(() => ""); + count += contents.split("\n").filter((line) => line.startsWith("inotify wd:")).length; + } + return count; +} + +function fixtureDescription() { + return { + directoriesPerRoot: DIRECTORY_COUNT, + totalDirectories: DIRECTORY_COUNT * ROOT_COUNT, + roots: ROOT_COUNT, + ignoredFiles: IGNORED_FILE_COUNT, + preexistingFilesPerRoot: PREEXISTING_FILE_COUNT, + trackedEdits: EDIT_COUNT, + sustainedEdits: SUSTAINED_EDIT_COUNT, + sustainedEditIntervalMs: SUSTAINED_EDIT_INTERVAL_MS, + repetitions: REPETITIONS, + platform: process.platform, + arch: process.arch, + node: process.version, + }; +} + +function evaluate(results: Measurement[]): { + passed: boolean; + failures: string[]; +} { + const failures: string[] = []; + for (const result of results) { + evaluateStandaloneResult(result, failures); + } + return { passed: failures.length === 0, failures }; +} + +function evaluateStandaloneResult(result: Measurement, failures: string[]): void { + if (result.missedTrackedPaths !== 0) { + failures.push(`observer missed ${result.missedTrackedPaths} tracked paths`); + } + if (result.ignoredEvents !== 0) { + failures.push(`observer emitted ${result.ignoredEvents} excluded events`); + } + if (result.sustainedMissedPaths !== 0) { + failures.push(`observer missed ${result.sustainedMissedPaths} sustained paths`); + } + if (result.teardownMs >= 1_000) { + failures.push(`observer teardown took ${result.teardownMs.toFixed(1)}ms`); + } + if ((result.sustainedFullReconciliationCount ?? 0) > 0) { + failures.push( + `node ran ${result.sustainedFullReconciliationCount} full reconciliations during sustained writes`, + ); + } + const setupLimitMs = process.platform === "win32" ? 8_000 : 1_000; + if (result.setupMs > setupLimitMs) { + failures.push( + `node setup ${result.setupMs.toFixed(1)}ms exceeded the ${setupLimitMs}ms per-run limit on run ${result.run}`, + ); + } + if (result.burstDurationMs > 60_000) { + failures.push( + `node burst recovery ${result.burstDurationMs.toFixed(1)}ms exceeded the 60000ms per-run limit on run ${result.run}`, + ); + } + // The production observer repairs omitted native siblings with one shallow + // directory diff per touched directory; keep that bounded allowance isolated + // to burst recovery rather than weakening steady-state CPU or correctness. + const burstCpuLimitMs = Math.max(6_000, result.burstDurationMs * 0.5); + if (result.burstCpuMs > burstCpuLimitMs) { + failures.push( + `node burst CPU ${result.burstCpuMs.toFixed(1)}ms exceeded the ${burstCpuLimitMs.toFixed(1)}ms per-run limit on run ${result.run}`, + ); + } + if (result.eventLoopDelayP99Ms > 250) { + failures.push( + `node event-loop p99 ${result.eventLoopDelayP99Ms.toFixed(1)}ms exceeded the 250ms per-run limit on run ${result.run}`, + ); + } + const sustainedCpuLimitMs = result.sustainedDurationMs * 0.25; + if (result.sustainedCpuMs > sustainedCpuLimitMs) { + failures.push( + `node sustained-create CPU ${result.sustainedCpuMs.toFixed(1)}ms exceeded the ${sustainedCpuLimitMs.toFixed(1)}ms per-run limit on run ${result.run}`, + ); + } +} + +function percentile(values: number[], quantile: number): number { + if (values.length === 0) return 0; + const sorted = [...values].sort((left, right) => left - right); + return sorted[Math.min(sorted.length - 1, Math.ceil(sorted.length * quantile) - 1)] ?? 0; +} + +async function waitFor(predicate: () => boolean, timeoutMs: number): Promise { + const deadline = performance.now() + timeoutMs; + while (!predicate()) { + if (performance.now() >= deadline) return false; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return true; +} + +function readPositiveInteger(name: string, fallback: number): number { + const value = Number(process.env[name] ?? fallback); + if (!Number.isInteger(value) || value <= 0) throw new Error(`${name} must be a positive integer`); + return value; +} + +await main(); diff --git a/packages/server/scripts/measure-git-observation.ts b/packages/server/scripts/measure-git-observation.ts index ecd6d07f2..b292c0e06 100644 --- a/packages/server/scripts/measure-git-observation.ts +++ b/packages/server/scripts/measure-git-observation.ts @@ -3,7 +3,6 @@ import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; -import parcelWatcher from "@parcel/watcher"; import type pino from "pino"; import { getCheckoutDiff, @@ -16,11 +15,15 @@ import { type GitCommandMetricsSnapshot, } from "../src/utils/run-git-command.js"; import { CheckoutDiffManager } from "../src/server/checkout-diff-manager.js"; +import { createFileObserver } from "../src/server/file-observer/index.js"; import { - subscribeToWorkspaceFileChanges, WorkspaceGitServiceImpl, type WorkspaceGitRuntimeSnapshot, } from "../src/server/workspace-git-service.js"; +import type { + FileObserverSubscription, + SubscribeToFileChanges, +} from "../src/server/file-observer/index.js"; interface ProducerCounts { structural: number; @@ -90,21 +93,23 @@ function snapshotCounts(counts: ProducerCounts): ProducerCounts { } function createTrackedSubscriber(failWorktreeRoot: string | null) { - const subscriptions = new Set(); - const subscribe = async (...args: Parameters) => { + const observer = createFileObserver(); + const subscriptions = new Set(); + const subscribe: SubscribeToFileChanges = async (...args) => { if (failWorktreeRoot && path.resolve(args[0]) === failWorktreeRoot) { throw new Error("measurement watcher setup failure"); } - const subscription = await subscribeToWorkspaceFileChanges(...args); + const subscription = await observer.subscribe(...args); subscriptions.add(subscription); return { + updateIgnore: (paths) => subscription.updateIgnore(paths), unsubscribe: async () => { await subscription.unsubscribe(); subscriptions.delete(subscription); }, }; }; - return { subscribe, subscriptions }; + return { observer, subscribe, subscriptions }; } function createMeasuredService(input: { @@ -175,10 +180,12 @@ async function measurePhase(input: { async function closeMeasuredService(input: { service: WorkspaceGitServiceImpl; - subscriptions: Set; + observer: ReturnType; + subscriptions: Set; }): Promise { - input.service.dispose(); + await input.service.dispose(); await waitFor(() => input.subscriptions.size === 0, "watchers to close"); + await input.observer.close(); } async function main(): Promise { @@ -396,12 +403,14 @@ async function main(): Promise { if (healthy) { await closeMeasuredService({ service: healthy.service, + observer: healthy.watcher.observer, subscriptions: healthy.watcher.subscriptions, }).catch(() => {}); } if (degraded) { await closeMeasuredService({ service: degraded.service, + observer: degraded.watcher.observer, subscriptions: degraded.watcher.subscriptions, }).catch(() => {}); } diff --git a/packages/server/scripts/reproduce-file-observer-teardown.ts b/packages/server/scripts/reproduce-file-observer-teardown.ts new file mode 100644 index 000000000..180ea46e7 --- /dev/null +++ b/packages/server/scripts/reproduce-file-observer-teardown.ts @@ -0,0 +1,169 @@ +import { execFile } from "node:child_process"; +import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { performance } from "node:perf_hooks"; +import { promisify } from "node:util"; +import { createFileObserver } from "../src/server/file-observer/index.js"; + +const execFileAsync = promisify(execFile); +const ITERATIONS = readPositiveInteger("PASEO_WATCH_REPRO_ITERATIONS", 200); +const CHILD_TIMEOUT_MS = readPositiveInteger( + "PASEO_WATCH_REPRO_TIMEOUT_MS", + Math.max(45_000, ITERATIONS * 150), +); + +interface ChildResult { + backend: string; + iterations: number; + completed: number; + teardownErrors: string[]; + elapsedMs: number; + teardownP99Ms: number; + callbacksAfterClose: number; + fileDescriptorsBefore: number | null; + fileDescriptorsAfter: number | null; + rssGrowthMiB: number; + observerBaselineRestored: boolean; + assertionsPassed: boolean; +} + +async function main(): Promise { + if (process.argv.includes("--child")) { + process.stdout.write(`${JSON.stringify(await runChild())}\n`); + return; + } + + const startedAt = performance.now(); + let result; + try { + const { stdout } = await execFileAsync( + process.execPath, + [...process.execArgv, process.argv[1], "--child"], + { env: process.env, maxBuffer: 1024 * 1024, timeout: CHILD_TIMEOUT_MS }, + ); + result = { ...JSON.parse(stdout.trim()), wedged: false }; + } catch (error) { + const processError = error as Error & { killed?: boolean; signal?: string }; + result = { + backend: "production", + iterations: ITERATIONS, + completed: 0, + teardownErrors: [processError.message], + elapsedMs: performance.now() - startedAt, + wedged: processError.killed === true || processError.signal === "SIGTERM", + }; + } + process.stdout.write( + `${JSON.stringify({ platform: process.platform, node: process.version, result }, null, 2)}\n`, + ); + const productionPassed = + "assertionsPassed" in result && result.assertionsPassed && result.wedged === false; + if (!productionPassed) process.exitCode = 1; +} + +async function runChild(): Promise { + const base = await mkdtemp(join(tmpdir(), "paseo-watch-repro-")); + const observer = createFileObserver(); + const teardownErrors: string[] = []; + const teardownDurations: number[] = []; + const startedAt = performance.now(); + const warmupRoot = join(base, "warmup"); + await mkdir(warmupRoot); + const warmupSubscription = await observer.subscribe(warmupRoot, () => undefined); + await warmupSubscription.unsubscribe(); + await rm(warmupRoot, { recursive: true, force: true }); + const rssBefore = process.memoryUsage().rss; + const fileDescriptorsBefore = await readFileDescriptorCount(); + const diagnosticsBefore = observer.getDiagnostics(); + let completed = 0; + let callbacksAfterClose = 0; + const sharedSubscription = await observer.subscribe(base, () => undefined); + try { + for (let index = 0; index < ITERATIONS; index += 1) { + const root = join(base, `worktree-${index}`); + await mkdir(join(root, "nested"), { recursive: true }); + let closed = false; + const subscription = await observer.subscribe(root, () => { + if (closed) callbacksAfterClose += 1; + }); + + const command = process.platform === "win32" ? "cmd.exe" : "true"; + const commandArgs = process.platform === "win32" ? ["/c", "exit", "0"] : []; + await Promise.all([ + writeFile(join(root, "nested", `edit-${index}.txt`), `${index}\n`), + execFileAsync(command, commandArgs), + ]); + await writeFile(join(root, "archive-trigger"), "archive\n"); + const teardownStartedAt = performance.now(); + try { + await rm(root, { recursive: true, force: true }); + } catch (error) { + teardownErrors.push(error instanceof Error ? error.message : String(error)); + } + await subscription.unsubscribe().catch((error: unknown) => { + teardownErrors.push(error instanceof Error ? error.message : String(error)); + }); + closed = true; + teardownDurations.push(performance.now() - teardownStartedAt); + completed += 1; + } + } finally { + await sharedSubscription.unsubscribe().catch((error: unknown) => { + teardownErrors.push(error instanceof Error ? error.message : String(error)); + }); + await observer.close(); + await rm(base, { recursive: true, force: true }); + } + await new Promise((resolve) => setTimeout(resolve, 25)); + const diagnosticsAfter = observer.getDiagnostics(); + const fileDescriptorsAfter = await readFileDescriptorCount(); + const rssGrowthMiB = (process.memoryUsage().rss - rssBefore) / 1024 / 1024; + const teardownP99Ms = percentile(teardownDurations, 0.99); + const observerBaselineRestored = + diagnosticsAfter.activeObservationCount === diagnosticsBefore.activeObservationCount && + diagnosticsAfter.nativeHandleCount === diagnosticsBefore.nativeHandleCount && + diagnosticsAfter.pendingEventCount === diagnosticsBefore.pendingEventCount && + diagnosticsAfter.reconciliationInFlightCount === diagnosticsBefore.reconciliationInFlightCount; + const assertionsPassed = + completed === ITERATIONS && + teardownErrors.length === 0 && + callbacksAfterClose === 0 && + observerBaselineRestored && + (fileDescriptorsBefore === null || fileDescriptorsAfter === fileDescriptorsBefore) && + rssGrowthMiB < 128 && + teardownP99Ms < 1_000; + return { + backend: "production", + iterations: ITERATIONS, + completed, + teardownErrors: [...new Set(teardownErrors)], + elapsedMs: performance.now() - startedAt, + teardownP99Ms, + callbacksAfterClose, + fileDescriptorsBefore, + fileDescriptorsAfter, + rssGrowthMiB, + observerBaselineRestored, + assertionsPassed, + }; +} + +async function readFileDescriptorCount(): Promise { + if (process.platform !== "linux") return null; + return (await readdir("/proc/self/fd")).length; +} + +function percentile(values: number[], quantile: number): number { + if (values.length === 0) return 0; + const sorted = [...values].sort((left, right) => left - right); + return sorted[Math.min(sorted.length - 1, Math.ceil(sorted.length * quantile) - 1)] ?? 0; +} + +function readPositiveInteger(name: string, fallback: number): number { + const value = Number(process.env[name] ?? fallback); + if (!Number.isInteger(value) || value <= 0) throw new Error(`${name} must be a positive integer`); + return value; +} + +await main(); diff --git a/packages/server/src/server/file-observer/index.test.ts b/packages/server/src/server/file-observer/index.test.ts new file mode 100644 index 000000000..112302321 --- /dev/null +++ b/packages/server/src/server/file-observer/index.test.ts @@ -0,0 +1,362 @@ +import { mkdir, mkdtemp, rename, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, sep } from "node:path"; +import { afterEach, expect, test, vi } from "vitest"; +import { + createFileObserver, + type FileChange, + type FileObserver, + type FileObserverCallback, + type FileObserverOptions, + type FileObserverSubscription, +} from "./index.js"; + +const roots = new Set(); +const observers = new Set(); + +afterEach(async () => { + vi.useRealTimers(); + await Promise.all([...observers].map((observer) => observer.close())); + observers.clear(); + await Promise.all([...roots].map((root) => rm(root, { recursive: true, force: true }))); + roots.clear(); +}); + +test("observes nested files while pruning excluded directories", async () => { + const root = await createRoot(); + const ignored = join(root, "ignored"); + const observed = join(root, "nested"); + await mkdir(ignored); + await mkdir(observed); + const events: FileChange[] = []; + const subscription = await subscribeToFileChanges( + root, + (error, batch) => { + expect(error).toBeNull(); + events.push(...batch); + }, + { ignore: [ignored] }, + ); + + await writeFile(join(ignored, "ignored.txt"), "ignored"); + const observedPath = join(observed, "observed.txt"); + await writeFile(observedPath, "observed"); + + await expect.poll(() => events.map((event) => event.path)).toContain(observedPath); + expect(events.map((event) => event.path)).not.toContain(join(ignored, "ignored.txt")); + await subscription.unsubscribe(); +}); + +test("covers files populated immediately inside a newly created directory", async () => { + const root = await createRoot(); + const events: FileChange[] = []; + const subscription = await subscribeToFileChanges(root, (error, batch) => { + expect(error).toBeNull(); + events.push(...batch); + }); + + const createdDirectory = join(root, "new", "nested"); + await mkdir(createdDirectory, { recursive: true }); + const firstPath = join(createdDirectory, "first.txt"); + await writeFile(firstPath, "first"); + await expect.poll(() => events.map((event) => event.path)).toContain(firstPath); + + const secondPath = join(createdDirectory, "second.txt"); + await writeFile(secondPath, "second"); + await expect.poll(() => events.map((event) => event.path)).toContain(secondPath); + await subscription.unsubscribe(); +}); + +test("re-admits an excluded directory without replacing the subscription", async () => { + const root = await createRoot(); + const ignored = join(root, "ignored"); + await mkdir(ignored); + const events: FileChange[] = []; + const subscription = await subscribeToFileChanges( + root, + (error, batch) => { + expect(error).toBeNull(); + events.push(...batch); + }, + { ignore: [ignored] }, + ); + + await subscription.updateIgnore([]); + const observedPath = join(ignored, "observed.txt"); + await writeFile(observedPath, "observed"); + await expect.poll(() => events.map((event) => event.path)).toContain(observedPath); + await subscription.unsubscribe(); +}); + +test("an ignore update is a barrier for later delivery", async () => { + const root = await createRoot(); + const ignored = join(root, "generated"); + await mkdir(ignored); + const delivered: FileChange[] = []; + const deliveredPaths = new Set(); + const subscription = await subscribeToFileChanges(root, (error, events) => { + expect(error).toBeNull(); + delivered.push(...events); + for (const event of events) deliveredPaths.add(event.path); + }); + + await Promise.all( + Array.from({ length: 100 }, (_, index) => + writeFile(join(ignored, `before-${index}.txt`), `${index}`), + ), + ); + const outsidePaths = Array.from({ length: 1_000 }, (_, index) => + join(root, `outside-${index}.txt`), + ); + await Promise.all([ + Promise.all(outsidePaths.map((path, index) => writeFile(path, `${index}`))), + subscription.updateIgnore([ignored]), + ]); + await expect + .poll(() => outsidePaths.filter((path) => !deliveredPaths.has(path)), { timeout: 10_000 }) + .toEqual([]); + const deliveredAtBarrier = delivered.length; + await Promise.all( + Array.from({ length: 100 }, (_, index) => + writeFile(join(ignored, `after-${index}.txt`), `${index}`), + ), + ); + const sentinel = join(root, "still-observed.txt"); + await writeFile(sentinel, "observed"); + await expect.poll(() => deliveredPaths.has(sentinel)).toBe(true); + + expect( + delivered + .slice(deliveredAtBarrier) + .filter((event) => event.path === ignored || event.path.startsWith(`${ignored}${sep}`)), + ).toEqual([]); + await subscription.unsubscribe(); +}, 20_000); + +test("survives atomic replacement and remains observable", async () => { + const root = await createRoot(); + const target = join(root, "target.txt"); + await writeFile(target, "before"); + const events: FileChange[] = []; + const subscription = await subscribeToFileChanges(root, (error, batch) => { + expect(error).toBeNull(); + events.push(...batch); + }); + + const replacement = join(root, "replacement.txt"); + await writeFile(replacement, "after"); + await rename(replacement, target); + await expect.poll(() => events.map((event) => event.path)).toContain(target); + + events.length = 0; + await writeFile(target, "again"); + await expect.poll(() => events.map((event) => event.path)).toContain(target); + await subscription.unsubscribe(); +}); + +test("classifies a removed file as deleted", async () => { + const root = await createRoot(); + const target = join(root, "removed.txt"); + await writeFile(target, "before"); + const events: FileChange[] = []; + const subscription = await subscribeToFileChanges(root, (error, batch) => { + expect(error).toBeNull(); + events.push(...batch); + }); + + await rm(target); + await expect.poll(() => events.find((event) => event.path === target)?.type).toBe("delete"); + await subscription.unsubscribe(); +}); + +test("observes files moved into the tree with their directory", async () => { + const root = await createRoot(); + const outside = await createRoot(); + const movedFrom = join(outside, "prepared"); + await mkdir(movedFrom); + const movedFile = join(movedFrom, "already-written.txt"); + await writeFile(movedFile, "ready"); + const events: FileChange[] = []; + const subscription = await subscribeToFileChanges(root, (error, batch) => { + expect(error).toBeNull(); + events.push(...batch); + }); + + const movedTo = join(root, "prepared"); + await rename(movedFrom, movedTo); + await expect + .poll(() => events.map((event) => event.path)) + .toContain(join(movedTo, "already-written.txt")); + await subscription.unsubscribe(); +}); + +test("delivers no callbacks after unsubscribe resolves", async () => { + const root = await createRoot(); + const callback = vi.fn(); + const subscription = await subscribeToFileChanges(root, callback); + await subscription.unsubscribe(); + await writeFile(join(root, "after-close.txt"), "closed"); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(callback).not.toHaveBeenCalled(); +}); + +test("deleting a watched root cannot wedge idempotent unsubscription", async () => { + const root = await createRoot(); + const subscription = await subscribeToFileChanges(root, () => undefined); + await rm(root, { recursive: true, force: true }); + await expect(subscription.unsubscribe()).resolves.toBeUndefined(); + await expect(subscription.unsubscribe()).resolves.toBeUndefined(); +}); + +test("observes a thousand concurrent writes and remains healthy after delete and rename churn", async () => { + const root = await createRoot(); + const directories = Array.from({ length: 20 }, (_, index) => join(root, `dir-${index}`)); + await Promise.all(directories.map((directory) => mkdir(directory))); + const observed = new Map>(); + const subscription = await subscribeToFileChanges(root, (error, events) => { + expect(error).toBeNull(); + for (const event of events) { + const types = observed.get(event.path) ?? new Set(); + types.add(event.type); + observed.set(event.path, types); + } + }); + const paths = Array.from({ length: 1_000 }, (_, index) => + join(directories[index % directories.length], `file-${index}.txt`), + ); + + const populatedDirectory = join(root, "populated", "nested"); + await mkdir(populatedDirectory, { recursive: true }); + const populatedPaths = Array.from({ length: 200 }, (_, index) => + join(populatedDirectory, `nested-${index}.txt`), + ); + await Promise.all(populatedPaths.map((path, index) => writeFile(path, `${index}`))); + await expect + .poll(() => populatedPaths.filter((path) => !observed.has(path)), { timeout: 10_000 }) + .toEqual([]); + + await Promise.all(paths.map((path, index) => writeFile(path, `${index}`))); + await expect + .poll(() => paths.filter((path) => !observed.has(path)), { timeout: 10_000 }) + .toEqual([]); + + const removedPaths = paths.slice(0, 100); + await Promise.all(removedPaths.map((path) => rm(path))); + await expect + .poll(() => removedPaths.filter((path) => !observed.get(path)?.has("delete")), { + timeout: 10_000, + }) + .toEqual([]); + + for (let index = 0; index < 10; index += 1) { + const from = directories[index]; + const to = join(root, `renamed-${index}`); + await rename(from, to); + await rm(to, { recursive: true, force: true }); + } + const sentinel = join(directories[15], "still-observed.txt"); + await writeFile(sentinel, "alive"); + await expect.poll(() => observed.has(sentinel)).toBe(true); + await subscription.unsubscribe(); +}, 30_000); + +test("unsubscribe cancels reconciliation queued by a write burst", async () => { + const root = await createRoot(); + const callback = vi.fn(); + const subscription = await subscribeToFileChanges(root, callback); + await Promise.all( + Array.from({ length: 1_000 }, (_, index) => + writeFile(join(root, `closing-${index}.txt`), `${index}`), + ), + ); + + await expect(subscription.unsubscribe()).resolves.toBeUndefined(); + const callsAtClose = callback.mock.calls.length; + await writeFile(join(root, "after-burst-close.txt"), "closed"); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(callback).toHaveBeenCalledTimes(callsAtClose); +}, 20_000); + +test("aggregate diagnostics return to their lifecycle baseline", async () => { + const root = await createRoot(); + const observer = createObserver(); + const subscription = await observer.subscribe(root, () => undefined); + + expect(observer.getDiagnostics()).toMatchObject({ + activeObservationCount: 1, + }); + expect(observer.getDiagnostics().nativeHandleCount).toBeGreaterThan(0); + + await subscription.unsubscribe(); + expect(observer.getDiagnostics()).toMatchObject({ + activeObservationCount: 0, + nativeHandleCount: 0, + nativeTrackedFileCount: 0, + pendingEventCount: 0, + reconciliationInFlightCount: 0, + }); +}); + +test("closing the observer service releases every subscription and rejects reuse", async () => { + const firstRoot = await createRoot(); + const secondRoot = await createRoot(); + const observer = createObserver(); + await observer.subscribe(firstRoot, () => undefined); + await observer.subscribe(secondRoot, () => undefined); + + await observer.close(); + + expect(observer.getDiagnostics()).toEqual({ + activeObservationCount: 0, + nativeHandleCount: 0, + nativeTrackedFileCount: 0, + pendingEventCount: 0, + pendingReconciliationWorkCount: 0, + reconciliationInFlightCount: 0, + reconciliationCount: 0, + scopedReconciliationCount: 0, + fullReconciliationCount: 0, + reconciliationFailureCount: 0, + observerFailureCount: 0, + directoryLimitFailureCount: 0, + nativeEventCount: 0, + nativeChangeEventCount: 0, + nativeRenameEventCount: 0, + nativePathlessEventCount: 0, + nativeClassificationCount: 0, + nativeShallowScanCount: 0, + lastReconciliationDurationMs: 0, + maxReconciliationDurationMs: 0, + }); + await expect(observer.subscribe(firstRoot, () => undefined)).rejects.toThrow( + "File observer is closed", + ); +}); + +async function subscribeToFileChanges( + directory: string, + callback: FileObserverCallback, + options?: FileObserverOptions, +): Promise { + const observer = createObserver(); + const subscription = await observer.subscribe(directory, callback, options); + return { + updateIgnore: (paths) => subscription.updateIgnore(paths), + unsubscribe: async () => { + await subscription.unsubscribe(); + await observer.close(); + }, + }; +} + +function createObserver(): FileObserver { + const observer = createFileObserver(); + observers.add(observer); + return observer; +} + +async function createRoot(): Promise { + const root = await mkdtemp(join(tmpdir(), "paseo-file-observer-")); + roots.add(root); + return root; +} diff --git a/packages/server/src/server/file-observer/index.ts b/packages/server/src/server/file-observer/index.ts new file mode 100644 index 000000000..785161bea --- /dev/null +++ b/packages/server/src/server/file-observer/index.ts @@ -0,0 +1,184 @@ +import { resolve } from "node:path"; +import type { CreateObservationBackend, ObserverMetrics } from "./internal/contracts.js"; +import { createLinuxBackend } from "./internal/linux.js"; +import { createNativeRecursiveBackend } from "./internal/native-recursive.js"; +import { createManagedObservation, type ManagedObservation } from "./internal/observation.js"; +import { createObserverPaths } from "./internal/paths.js"; + +export type FileChangeType = "create" | "update" | "delete"; + +export interface FileChange { + path: string; + type: FileChangeType; +} + +export interface FileObserverOptions { + ignore?: string[]; +} + +export type FileObserverCallback = (error: Error | null, events: FileChange[]) => void; + +export interface FileObserverSubscription { + updateIgnore(paths: string[]): Promise; + unsubscribe(): Promise; +} + +export interface FileObserverDiagnostics { + activeObservationCount: number; + nativeHandleCount: number; + nativeTrackedFileCount: number; + pendingEventCount: number; + pendingReconciliationWorkCount: number; + reconciliationInFlightCount: number; + reconciliationCount: number; + scopedReconciliationCount: number; + fullReconciliationCount: number; + reconciliationFailureCount: number; + observerFailureCount: number; + directoryLimitFailureCount: number; + nativeEventCount: number; + nativeChangeEventCount: number; + nativeRenameEventCount: number; + nativePathlessEventCount: number; + nativeClassificationCount: number; + nativeShallowScanCount: number; + lastReconciliationDurationMs: number; + maxReconciliationDurationMs: number; +} + +export interface FileObserver { + subscribe( + directory: string, + callback: FileObserverCallback, + options?: FileObserverOptions, + ): Promise; + getDiagnostics(): FileObserverDiagnostics; + close(): Promise; +} + +export type SubscribeToFileChanges = FileObserver["subscribe"]; + +export function createFileObserver(): FileObserver { + const platform = process.platform; + const paths = createObserverPaths(platform); + const createBackend: CreateObservationBackend = + platform === "darwin" || platform === "win32" + ? (host) => createNativeRecursiveBackend(host, paths) + : (host) => createLinuxBackend(host, paths); + return new FileObserverService(createBackend, paths); +} + +class FileObserverService implements FileObserver { + private state: + | { + status: "open"; + observations: Set; + metrics: ObserverMetrics; + } + | { status: "closed" } = { + status: "open", + observations: new Set(), + metrics: createObserverMetrics(), + }; + private closePromise: Promise | null = null; + + constructor( + private readonly createBackend: CreateObservationBackend, + private readonly paths: ReturnType, + ) {} + + async subscribe( + directory: string, + callback: FileObserverCallback, + options: FileObserverOptions = {}, + ): Promise { + if (this.state.status === "closed") throw new Error("File observer is closed"); + const state = this.state; + let observation!: ManagedObservation; + observation = createManagedObservation({ + root: resolve(directory), + callback, + options, + paths: this.paths, + metrics: state.metrics, + createBackend: this.createBackend, + onClosed: () => state.observations.delete(observation), + }); + await observation.start(); + if (this.closePromise) { + await observation.close(); + throw new Error("File observer is closed"); + } + state.observations.add(observation); + return observation.subscription(); + } + + getDiagnostics(): FileObserverDiagnostics { + if (this.state.status === "closed") return emptyDiagnostics(); + const { observations, metrics } = this.state; + let nativeHandleCount = 0; + let nativeTrackedFileCount = 0; + let pendingEventCount = 0; + let pendingReconciliationWorkCount = 0; + let reconciliationInFlightCount = 0; + for (const observation of observations) { + const diagnostics = observation.getDiagnostics(); + nativeHandleCount += diagnostics.nativeHandleCount; + nativeTrackedFileCount += diagnostics.nativeTrackedFileCount; + pendingEventCount += diagnostics.pendingEventCount; + pendingReconciliationWorkCount += diagnostics.pendingReconciliationWorkCount; + reconciliationInFlightCount += Number(diagnostics.reconciliationInFlight); + } + return { + activeObservationCount: observations.size, + nativeHandleCount, + nativeTrackedFileCount, + pendingEventCount, + pendingReconciliationWorkCount, + reconciliationInFlightCount, + ...metrics, + }; + } + + close(): Promise { + if (this.closePromise) return this.closePromise; + if (this.state.status === "closed") return Promise.resolve(); + const observations = [...this.state.observations]; + this.state = { status: "closed" }; + this.closePromise = Promise.allSettled( + observations.map((observation) => observation.close()), + ).then(() => undefined); + return this.closePromise; + } +} + +function createObserverMetrics(): ObserverMetrics { + return { + reconciliationCount: 0, + scopedReconciliationCount: 0, + fullReconciliationCount: 0, + reconciliationFailureCount: 0, + observerFailureCount: 0, + directoryLimitFailureCount: 0, + nativeEventCount: 0, + nativeChangeEventCount: 0, + nativeRenameEventCount: 0, + nativePathlessEventCount: 0, + nativeClassificationCount: 0, + nativeShallowScanCount: 0, + lastReconciliationDurationMs: 0, + maxReconciliationDurationMs: 0, + }; +} + +function emptyDiagnostics(): FileObserverDiagnostics { + return { + activeObservationCount: 0, + nativeHandleCount: 0, + nativeTrackedFileCount: 0, + pendingEventCount: 0, + pendingReconciliationWorkCount: 0, + reconciliationInFlightCount: 0, + ...createObserverMetrics(), + }; +} diff --git a/packages/server/src/server/file-observer/internal/contracts.ts b/packages/server/src/server/file-observer/internal/contracts.ts new file mode 100644 index 000000000..fa7a6ddb0 --- /dev/null +++ b/packages/server/src/server/file-observer/internal/contracts.ts @@ -0,0 +1,37 @@ +import type { FileChangeType, FileObserverDiagnostics } from "../index.js"; + +export interface ObserverMetrics extends Omit< + FileObserverDiagnostics, + | "activeObservationCount" + | "nativeHandleCount" + | "nativeTrackedFileCount" + | "pendingEventCount" + | "pendingReconciliationWorkCount" + | "reconciliationInFlightCount" +> {} + +export interface BackendDiagnostics { + nativeHandleCount: number; + nativeTrackedFileCount: number; + pendingReconciliationWorkCount: number; + reconciliationInFlight: boolean; +} + +export interface ObservationBackend { + start(): Promise; + updateIgnore(): Promise; + close(): Promise; + getDiagnostics(): BackendDiagnostics; +} + +export interface ObservationHost { + root: string; + metrics: ObserverMetrics; + isActive(): boolean; + isIgnored(path: string): boolean; + isPathInside(root: string, path: string): boolean; + queueEvent(type: FileChangeType, path: string): void; + fail(error: Error): void; +} + +export type CreateObservationBackend = (host: ObservationHost) => ObservationBackend; diff --git a/packages/server/src/server/file-observer/internal/linux.ts b/packages/server/src/server/file-observer/internal/linux.ts new file mode 100644 index 000000000..70b1e549b --- /dev/null +++ b/packages/server/src/server/file-observer/internal/linux.ts @@ -0,0 +1,270 @@ +import { type Dirent, type FSWatcher, watch } from "node:fs"; +import { readdir, stat } from "node:fs/promises"; +import { resolve } from "node:path"; +import type { BackendDiagnostics, ObservationBackend, ObservationHost } from "./contracts.js"; +import type { ObserverPaths } from "./paths.js"; +import { isMissingPathError, toError } from "./paths.js"; + +const MAX_WATCHED_DIRECTORIES = 5_000; +const RECONCILIATION_DELAY_MS = 50; + +interface DirectorySnapshot { + directories: Set; + filesDiscoveredBeforeCoverage: Set; +} + +export function createLinuxBackend( + host: ObservationHost, + paths: ObserverPaths, +): ObservationBackend { + return new LinuxBackend(host, paths); +} + +class LinuxBackend implements ObservationBackend { + private readonly watchers = new Map(); + private readonly classifications = new Set>(); + private readonly pendingScopes = new Set(); + private reconcileTail: Promise = Promise.resolve(); + private queueDepth = 0; + private inFlight = false; + private timer: NodeJS.Timeout | null = null; + private closePromise: Promise | null = null; + + constructor( + private readonly host: ObservationHost, + private readonly paths: ObserverPaths, + ) {} + + async start(): Promise { + this.watchDirectory(this.host.root); + await this.enqueue([this.host.root]); + } + + async updateIgnore(): Promise { + for (const [directory, watcher] of this.watchers) { + if (directory !== this.host.root && this.host.isIgnored(directory)) { + watcher.close(); + this.watchers.delete(directory); + } + } + this.cancelQueued(); + await this.enqueue([this.host.root]); + } + + close(): Promise { + if (this.closePromise) return this.closePromise; + this.cancelQueued(); + for (const watcher of this.watchers.values()) watcher.close(); + this.watchers.clear(); + this.closePromise = this.finishClose(); + return this.closePromise; + } + + getDiagnostics(): BackendDiagnostics { + return { + nativeHandleCount: this.watchers.size, + nativeTrackedFileCount: 0, + pendingReconciliationWorkCount: + this.pendingScopes.size + + this.classifications.size + + this.queueDepth + + Number(this.timer !== null), + reconciliationInFlight: this.inFlight, + }; + } + + private async finishClose(): Promise { + await this.reconcileTail; + await Promise.allSettled(this.classifications); + for (const watcher of this.watchers.values()) watcher.close(); + this.watchers.clear(); + } + + private watchDirectory(directory: string): void { + if (!this.host.isActive() || this.watchers.has(directory) || this.host.isIgnored(directory)) { + return; + } + if (this.watchers.size >= MAX_WATCHED_DIRECTORIES) { + this.host.metrics.directoryLimitFailureCount += 1; + throw new Error( + `Recursive file observation exceeded ${MAX_WATCHED_DIRECTORIES} directories under ${this.host.root}`, + ); + } + const watcher = watch(directory, (eventType, filename) => { + if (!this.host.isActive()) return; + const path = filename ? resolve(directory, filename.toString()) : directory; + if (!this.host.isIgnored(path)) { + if (eventType === "change") this.host.queueEvent("update", path); + else this.classify(path); + } + if (eventType === "rename" || !filename) this.requestReconcile(path); + }); + this.attachWatcher(directory, watcher); + } + + private attachWatcher(directory: string, watcher: FSWatcher): void { + watcher.on("error", (error) => { + if (!this.host.isActive()) return; + if (this.paths.isExpectedWatchDisappearance(error)) { + watcher.close(); + if (this.watchers.get(directory) === watcher) this.watchers.delete(directory); + if (directory === this.host.root) { + this.host.fail(toError(error)); + return; + } + this.requestReconcile(directory); + return; + } + this.host.fail(toError(error)); + }); + this.watchers.set(directory, watcher); + } + + private requestReconcile(scope: string): void { + if (!this.host.isActive()) return; + this.pendingScopes.add(scope); + if (this.timer) return; + this.timer = setTimeout(() => { + this.timer = null; + const scopes = [...this.pendingScopes]; + this.pendingScopes.clear(); + void this.enqueue(scopes).catch((error: unknown) => this.host.fail(toError(error))); + }, RECONCILIATION_DELAY_MS); + this.timer.unref(); + } + + private enqueue(scopes: string[]): Promise { + this.queueDepth += 1; + const result = this.reconcileTail + .then(() => this.run(scopes)) + .finally(() => { + this.queueDepth -= 1; + }); + this.reconcileTail = result.catch(() => undefined); + return result; + } + + private async run(scopes: string[]): Promise { + if (!this.host.isActive()) return; + this.inFlight = true; + const startedAt = performance.now(); + try { + await this.reconcile(this.paths.collapse(scopes)); + } catch (error) { + this.host.metrics.reconciliationFailureCount += 1; + throw error; + } finally { + const durationMs = performance.now() - startedAt; + this.host.metrics.reconciliationCount += 1; + this.host.metrics.scopedReconciliationCount += 1; + this.host.metrics.lastReconciliationDurationMs = durationMs; + this.host.metrics.maxReconciliationDurationMs = Math.max( + this.host.metrics.maxReconciliationDurationMs, + durationMs, + ); + this.inFlight = false; + } + } + + private async reconcile(scopes: string[]): Promise { + for (const scope of scopes) { + const snapshot = await this.scan(scope); + if (!this.host.isActive()) return; + for (const directory of snapshot.directories) { + if (!this.watchSnapshotDirectory(directory)) snapshot.directories.delete(directory); + } + for (const path of snapshot.filesDiscoveredBeforeCoverage) { + this.host.queueEvent("create", path); + } + for (const [directory, watcher] of this.watchers) { + const disappeared = + directory !== this.host.root && + this.host.isPathInside(scope, directory) && + !snapshot.directories.has(directory); + if (!disappeared) continue; + watcher.close(); + this.watchers.delete(directory); + this.host.queueEvent("delete", directory); + } + } + } + + private watchSnapshotDirectory(directory: string): boolean { + try { + this.watchDirectory(directory); + return true; + } catch (error) { + if (!this.paths.isExpectedWatchDisappearance(error) || directory === this.host.root) { + throw error; + } + return false; + } + } + + private async scan(scope: string): Promise { + let scopeStats; + try { + scopeStats = await stat(scope); + } catch (error) { + if (!isMissingPathError(error) || scope === this.host.root) throw error; + return { directories: new Set(), filesDiscoveredBeforeCoverage: new Set() }; + } + if (!scopeStats.isDirectory()) { + return { directories: new Set(), filesDiscoveredBeforeCoverage: new Set() }; + } + const directories = new Set([scope]); + const filesDiscoveredBeforeCoverage = new Set(); + const pending = [scope]; + while (pending.length > 0 && this.host.isActive()) { + const directory = pending.pop(); + if (!directory) continue; + let entries: Dirent[]; + try { + entries = await readdir(directory, { withFileTypes: true }); + } catch (error) { + if (!isMissingPathError(error)) throw error; + if (directory === this.host.root && this.host.isActive()) throw error; + continue; + } + for (const entry of entries) { + const child = resolve(directory, entry.name); + if (this.host.isIgnored(child)) continue; + if (!entry.isDirectory()) { + if (!this.watchers.has(directory)) filesDiscoveredBeforeCoverage.add(child); + continue; + } + directories.add(child); + if (directories.size > MAX_WATCHED_DIRECTORIES) { + this.host.metrics.directoryLimitFailureCount += 1; + throw new Error( + `Recursive file observation exceeded ${MAX_WATCHED_DIRECTORIES} directories under ${this.host.root}`, + ); + } + pending.push(child); + } + } + return { directories, filesDiscoveredBeforeCoverage }; + } + + private classify(path: string): void { + this.host.metrics.nativeClassificationCount += 1; + let classification!: Promise; + classification = stat(path) + .then(() => this.host.queueEvent("create", path)) + .catch((error: unknown) => { + if (isMissingPathError(error)) { + this.host.queueEvent("delete", path); + return; + } + this.host.fail(toError(error)); + }) + .finally(() => this.classifications.delete(classification)); + this.classifications.add(classification); + } + + private cancelQueued(): void { + if (this.timer) clearTimeout(this.timer); + this.timer = null; + this.pendingScopes.clear(); + } +} diff --git a/packages/server/src/server/file-observer/internal/native-recursive.ts b/packages/server/src/server/file-observer/internal/native-recursive.ts new file mode 100644 index 000000000..e5ef07753 --- /dev/null +++ b/packages/server/src/server/file-observer/internal/native-recursive.ts @@ -0,0 +1,588 @@ +import { type Dirent, type FSWatcher, watch } from "node:fs"; +import { readdir, stat } from "node:fs/promises"; +import { dirname, join, resolve } from "node:path"; +import type { BackendDiagnostics, ObservationBackend, ObservationHost } from "./contracts.js"; +import type { ObserverPaths } from "./paths.js"; +import { isMissingPathError, toError } from "./paths.js"; + +const MAX_TRACKED_ENTRIES = 250_000; +const AUDIT_QUIET_MS = 500; +const AUDIT_MAX_DIRTY_MS = 5_000; +const OPTIONAL_AUDIT_QUIET_MS = 8_000; +const OPTIONAL_AUDIT_MAX_DIRTY_MS = 8_000; +const FULL_AUDIT_QUIET_MS = 30_000; +const FULL_AUDIT_MIN_INTERVAL_MS = 30_000; +const FULL_AUDIT_MAX_DIRTY_MS = 5 * 60_000; + +interface DirectoryEntry { + directories: Set; + files: Set; +} + +interface Inventory { + directories: Set; + entries: Map; + files: Set; +} + +export function createNativeRecursiveBackend( + host: ObservationHost, + paths: ObserverPaths, +): ObservationBackend { + return new NativeRecursiveBackend(host, paths); +} + +class NativeRecursiveBackend implements ObservationBackend { + private watcher: FSWatcher | null = null; + private files = new Set(); + private directories = new Set(); + private entries = new Map(); + private readonly classifications = new Set>(); + private readonly localScopes = new Set(); + private readonly changeScopes = new Set(); + private readonly recursiveScopes = new Set(); + private reconcileTail: Promise = Promise.resolve(); + private queueDepth = 0; + private inFlight = false; + private auditTimer: NodeJS.Timeout | null = null; + private fullAuditTimer: NodeJS.Timeout | null = null; + private fullAuditRequested = false; + private auditDirty = false; + private auditDirtySince: number | null = null; + private auditLastDirtyAt: number | null = null; + private auditQueued = false; + private auditRunning = false; + private safetyAuditPending = false; + private safetyAuditDirtySince: number | null = null; + private safetyAuditLastDirtyAt: number | null = null; + private lastFullAuditAt = Number.NEGATIVE_INFINITY; + private generation = 0; + private closePromise: Promise | null = null; + + constructor( + private readonly host: ObservationHost, + private readonly paths: ObserverPaths, + ) {} + + async start(): Promise { + this.watchRoot(); + this.fullAuditRequested = true; + this.auditQueued = true; + await this.enqueueAudit(false, this.generation); + } + + async updateIgnore(): Promise { + this.generation += 1; + this.cancelAudits(); + this.fullAuditRequested = true; + this.auditQueued = true; + await this.enqueueAudit(true, this.generation); + } + + close(): Promise { + if (this.closePromise) return this.closePromise; + this.cancelAudits(); + this.watcher?.close(); + this.watcher = null; + this.closePromise = this.finishClose(); + return this.closePromise; + } + + getDiagnostics(): BackendDiagnostics { + return { + nativeHandleCount: this.watcher ? 1 : 0, + nativeTrackedFileCount: this.files.size, + pendingReconciliationWorkCount: + this.localScopes.size + + this.changeScopes.size + + this.recursiveScopes.size + + this.classifications.size + + this.queueDepth + + Number(this.auditTimer !== null || this.auditQueued || this.auditDirty), + reconciliationInFlight: this.inFlight, + }; + } + + private async finishClose(): Promise { + await this.reconcileTail; + await Promise.allSettled(this.classifications); + this.watcher?.close(); + this.watcher = null; + this.files.clear(); + this.directories.clear(); + this.entries.clear(); + } + + private watchRoot(): void { + const watcher = watch(this.host.root, { recursive: true }, (eventType, filename) => { + if (!this.host.isActive()) return; + this.host.metrics.nativeEventCount += 1; + if (!filename) { + this.host.metrics.nativePathlessEventCount += 1; + this.host.queueEvent("update", this.host.root); + this.requestAudit(this.host.root); + return; + } + const path = resolve(this.host.root, filename.toString()); + if (this.host.isIgnored(path)) return; + const scope = path === this.host.root ? this.host.root : dirname(path); + if (eventType === "change") { + this.host.metrics.nativeChangeEventCount += 1; + this.host.queueEvent("update", path); + this.requestChangeAudit(scope); + if (this.directories.has(path)) this.requestAudit(path, true); + return; + } + this.host.metrics.nativeRenameEventCount += 1; + const knownDirectory = this.directories.has(path); + this.classify(path, (isDirectory) => { + if (!isDirectory) return; + if (knownDirectory) this.requestAudit(path, true); + else this.requestAudit(scope); + }); + this.requestChangeAudit(scope); + if (knownDirectory) this.requestAudit(scope); + }); + watcher.on("error", (error) => { + if (this.host.isActive()) this.host.fail(toError(error)); + }); + this.watcher = watcher; + } + + private enqueueAudit(emitDiff: boolean, generation: number): Promise { + this.queueDepth += 1; + const result = this.reconcileTail + .then(() => this.runAudit(emitDiff, generation)) + .finally(() => { + this.queueDepth -= 1; + }); + this.reconcileTail = result.catch(() => undefined); + return result; + } + + private async runAudit(emitDiff: boolean, generation: number): Promise { + this.auditQueued = false; + if (!this.canCommit(generation)) return; + const fullAudit = this.fullAuditRequested; + const localScopes = [...this.localScopes]; + const changeScopes = [...this.changeScopes]; + const recursiveScopes = [...this.recursiveScopes]; + this.fullAuditRequested = false; + this.localScopes.clear(); + this.changeScopes.clear(); + this.recursiveScopes.clear(); + if (fullAudit) this.clearSafetyAudit(); + this.auditRunning = true; + this.auditDirty = false; + this.auditDirtySince = null; + this.auditLastDirtyAt = null; + this.inFlight = true; + const startedAt = performance.now(); + try { + if (fullAudit) { + const next = await this.scanTree(this.host.root); + if (!this.canCommit(generation)) return; + if (emitDiff) this.queueDiff(next.files, this.files); + this.replaceInventory(next); + this.lastFullAuditAt = performance.now(); + this.host.metrics.fullReconciliationCount += 1; + } else { + await this.reconcileScopes(localScopes, changeScopes, recursiveScopes, generation); + this.host.metrics.scopedReconciliationCount += 1; + } + } catch (error) { + this.host.metrics.reconciliationFailureCount += 1; + throw error; + } finally { + const durationMs = performance.now() - startedAt; + this.host.metrics.reconciliationCount += 1; + this.host.metrics.lastReconciliationDurationMs = durationMs; + this.host.metrics.maxReconciliationDurationMs = Math.max( + this.host.metrics.maxReconciliationDurationMs, + durationMs, + ); + this.inFlight = false; + this.auditRunning = false; + if (this.safetyAuditPending && this.host.isActive()) this.scheduleFullAudit(); + if (this.auditDirty && this.host.isActive()) this.scheduleAudit(); + } + } + + private async scanTree(root: string): Promise { + const files = new Set(); + const directories = new Set(); + const entries = new Map(); + const pending = [root]; + while (pending.length > 0 && this.host.isActive()) { + const directory = pending.pop(); + if (!directory) continue; + let children: Dirent[]; + try { + children = await readdir(directory, { withFileTypes: true }); + } catch (error) { + if (isMissingPathError(error) && !this.host.isActive()) { + return { directories, entries, files }; + } + if (!isMissingPathError(error) || directory === this.host.root) throw error; + continue; + } + directories.add(directory); + this.assertWithinLimit(files.size + directories.size); + const entry: DirectoryEntry = { directories: new Set(), files: new Set() }; + for (const child of children) { + const path = join(directory, child.name); + if (this.host.isIgnored(path)) continue; + if (child.isDirectory()) { + entry.directories.add(path); + pending.push(path); + } else { + entry.files.add(path); + files.add(path); + this.assertWithinLimit(files.size + directories.size); + } + } + entries.set(directory, entry); + } + return { directories, entries, files }; + } + + private async reconcileScopes( + localScopes: string[], + changeScopes: string[], + recursiveScopes: string[], + generation: number, + ): Promise { + const forcedLocalScopes = new Set(localScopes); + const forcedRecursiveScopes = this.paths.collapse(recursiveScopes); + for (const directory of new Set(localScopes)) { + await this.reconcileDirectory(directory, generation); + if (!this.canCommit(generation)) return; + } + for (const directory of new Set(recursiveScopes)) { + await this.reconcileSubtree(directory, generation); + if (!this.canCommit(generation)) return; + } + for (const directory of this.paths.collapse(changeScopes)) { + const alreadyCovered = + forcedLocalScopes.has(directory) || + forcedRecursiveScopes.some((scope) => this.host.isPathInside(scope, directory)); + if (alreadyCovered) continue; + await this.reconcileDirectory(directory, generation); + if (!this.canCommit(generation)) return; + } + } + + private async reconcileDirectory(directory: string, generation: number): Promise { + if (this.host.isIgnored(directory)) { + this.removeSubtree(directory, false); + return; + } + if (directory !== this.host.root && !this.directories.has(directory)) { + await this.reconcileUnknownSubtree(directory, generation); + return; + } + const next = await this.readDirectory(directory); + if (!this.canCommit(generation)) return; + if (!next) { + this.removeSubtree(directory, true); + return; + } + const previous = this.entries.get(directory) ?? { + directories: new Set(), + files: new Set(), + }; + for (const path of previous.files) { + if (!next.files.has(path)) { + this.files.delete(path); + this.host.queueEvent("delete", path); + } + } + for (const path of previous.directories) { + if (!next.directories.has(path)) this.removeSubtree(path, true); + } + for (const path of next.files) { + if (!previous.files.has(path)) { + this.files.add(path); + this.host.queueEvent("create", path); + } + } + for (const path of next.directories) { + if (previous.directories.has(path)) continue; + const inventory = await this.scanTree(path); + if (!this.canCommit(generation)) return; + if (!inventory.directories.has(path)) { + next.directories.delete(path); + continue; + } + this.mergeInventory(inventory, true); + } + this.directories.add(directory); + this.entries.set(directory, next); + this.assertWithinLimit(this.files.size + this.directories.size); + } + + private async reconcileUnknownSubtree(directory: string, generation: number): Promise { + const inventory = await this.scanTree(directory); + if (!this.canCommit(generation)) return; + if (!inventory.directories.has(directory)) { + this.removeSubtree(directory, true); + return; + } + this.entries.get(dirname(directory))?.directories.add(directory); + this.mergeInventory(inventory, true); + } + + private async readDirectory(directory: string): Promise { + this.host.metrics.nativeShallowScanCount += 1; + let children: Dirent[]; + try { + children = await readdir(directory, { withFileTypes: true }); + } catch (error) { + if (!isMissingPathError(error) || directory === this.host.root) throw error; + return null; + } + const entry: DirectoryEntry = { directories: new Set(), files: new Set() }; + for (const child of children) { + const path = join(directory, child.name); + if (this.host.isIgnored(path)) continue; + if (child.isDirectory()) entry.directories.add(path); + else entry.files.add(path); + } + return entry; + } + + private async reconcileSubtree(directory: string, generation: number): Promise { + if (!this.directories.has(directory) || this.host.isIgnored(directory)) return; + const inventory = await this.scanTree(directory); + if (!this.canCommit(generation)) return; + const previousFiles = new Set( + [...this.files].filter((path) => this.host.isPathInside(directory, path)), + ); + this.queueDiff(inventory.files, previousFiles); + this.removeSubtree(directory, false); + this.mergeInventory(inventory, false); + } + + private queueDiff(nextFiles: Set, previousFiles: Set): void { + for (const path of nextFiles) { + if (!previousFiles.has(path)) this.host.queueEvent("create", path); + } + for (const path of previousFiles) { + if (!nextFiles.has(path)) this.host.queueEvent("delete", path); + } + } + + private replaceInventory(inventory: Inventory): void { + this.files = inventory.files; + this.directories = inventory.directories; + this.entries = inventory.entries; + } + + private mergeInventory(inventory: Inventory, emitCreates: boolean): void { + for (const directory of inventory.directories) this.directories.add(directory); + for (const [directory, entry] of inventory.entries) this.entries.set(directory, entry); + for (const path of inventory.files) { + const added = !this.files.has(path); + this.files.add(path); + if (emitCreates && added) this.host.queueEvent("create", path); + } + this.assertWithinLimit(this.files.size + this.directories.size); + } + + private removeSubtree(root: string, emitDeletes: boolean): void { + for (const path of this.files) { + if (!this.host.isPathInside(root, path)) continue; + this.files.delete(path); + if (emitDeletes) this.host.queueEvent("delete", path); + } + for (const directory of this.directories) { + if (!this.host.isPathInside(root, directory)) continue; + this.directories.delete(directory); + this.entries.delete(directory); + } + const parent = this.entries.get(dirname(root)); + parent?.directories.delete(root); + parent?.files.delete(root); + } + + private requestAudit(scope: string, recursive = false): void { + if (!this.host.isActive()) return; + if (recursive) this.recursiveScopes.add(scope); + else this.localScopes.add(scope); + this.markAuditDirty(); + } + + private requestChangeAudit(scope: string): void { + if (!this.host.isActive()) return; + this.changeScopes.add(scope); + this.markAuditDirty(); + } + + private markAuditDirty(): void { + this.requestSafetyAudit(); + const now = performance.now(); + this.auditDirty = true; + this.auditDirtySince ??= now; + this.auditLastDirtyAt = now; + if (this.auditQueued || this.auditRunning) return; + if (this.auditTimer) clearTimeout(this.auditTimer); + this.auditTimer = null; + this.scheduleAudit(); + } + + private scheduleAudit(): void { + if ( + !this.host.isActive() || + !this.auditDirty || + this.auditQueued || + this.auditRunning || + this.auditTimer + ) { + return; + } + const now = performance.now(); + const optionalOnly = this.isOptionalOnly(); + const quietMs = optionalOnly ? OPTIONAL_AUDIT_QUIET_MS : AUDIT_QUIET_MS; + const maxDirtyMs = optionalOnly ? OPTIONAL_AUDIT_MAX_DIRTY_MS : AUDIT_MAX_DIRTY_MS; + const deadline = Math.min( + (this.auditLastDirtyAt ?? now) + quietMs, + (this.auditDirtySince ?? now) + maxDirtyMs, + ); + this.auditTimer = setTimeout(() => this.runScheduledAudit(), Math.max(0, deadline - now)); + this.auditTimer.unref(); + } + + private runScheduledAudit(): void { + this.auditTimer = null; + if (!this.host.isActive()) return; + const now = performance.now(); + const optionalOnly = this.isOptionalOnly(); + const quietMs = optionalOnly ? OPTIONAL_AUDIT_QUIET_MS : AUDIT_QUIET_MS; + const maxDirtyMs = optionalOnly ? OPTIONAL_AUDIT_MAX_DIRTY_MS : AUDIT_MAX_DIRTY_MS; + const deadline = Math.min( + (this.auditLastDirtyAt ?? now) + quietMs, + (this.auditDirtySince ?? now) + maxDirtyMs, + ); + if (now < deadline) { + this.scheduleAudit(); + return; + } + this.auditQueued = true; + void this.enqueueAudit(true, this.generation).catch((error: unknown) => + this.host.fail(toError(error)), + ); + } + + private isOptionalOnly(): boolean { + return ( + !this.fullAuditRequested && + this.localScopes.size === 0 && + this.recursiveScopes.size === 0 && + this.changeScopes.size > 0 + ); + } + + private requestSafetyAudit(): void { + const now = performance.now(); + this.safetyAuditPending = true; + this.safetyAuditDirtySince ??= now; + this.safetyAuditLastDirtyAt = now; + this.scheduleFullAudit(); + } + + private scheduleFullAudit(): void { + if ( + !this.host.isActive() || + !this.safetyAuditPending || + this.auditRunning || + this.fullAuditTimer + ) { + return; + } + const now = performance.now(); + const deadline = this.fullAuditDeadline(now); + this.fullAuditTimer = setTimeout( + () => this.runScheduledFullAudit(), + Math.max(0, deadline - now), + ); + this.fullAuditTimer.unref(); + } + + private runScheduledFullAudit(): void { + this.fullAuditTimer = null; + if (!this.host.isActive() || !this.safetyAuditPending || this.auditRunning) return; + const now = performance.now(); + if (now < this.fullAuditDeadline(now)) { + this.scheduleFullAudit(); + return; + } + this.safetyAuditPending = false; + this.fullAuditRequested = true; + this.auditDirty = true; + this.auditDirtySince ??= now; + this.scheduleAudit(); + } + + private fullAuditDeadline(now: number): number { + const quietDeadline = (this.safetyAuditLastDirtyAt ?? now) + FULL_AUDIT_QUIET_MS; + const intervalDeadline = this.lastFullAuditAt + FULL_AUDIT_MIN_INTERVAL_MS; + const starvationDeadline = (this.safetyAuditDirtySince ?? now) + FULL_AUDIT_MAX_DIRTY_MS; + return Math.min(starvationDeadline, Math.max(intervalDeadline, quietDeadline)); + } + + private clearSafetyAudit(): void { + this.safetyAuditPending = false; + this.safetyAuditDirtySince = null; + this.safetyAuditLastDirtyAt = null; + if (this.fullAuditTimer) clearTimeout(this.fullAuditTimer); + this.fullAuditTimer = null; + } + + private cancelAudits(): void { + if (this.auditTimer) clearTimeout(this.auditTimer); + if (this.fullAuditTimer) clearTimeout(this.fullAuditTimer); + this.auditTimer = null; + this.fullAuditTimer = null; + this.localScopes.clear(); + this.changeScopes.clear(); + this.recursiveScopes.clear(); + this.fullAuditRequested = false; + this.auditDirty = false; + this.auditDirtySince = null; + this.auditLastDirtyAt = null; + this.safetyAuditPending = false; + this.safetyAuditDirtySince = null; + this.safetyAuditLastDirtyAt = null; + } + + private classify(path: string, onPresent: (isDirectory: boolean) => void): void { + this.host.metrics.nativeClassificationCount += 1; + let classification!: Promise; + classification = stat(path) + .then((stats) => { + this.host.queueEvent("create", path); + if (this.host.isActive() && !this.host.isIgnored(path)) onPresent(stats.isDirectory()); + return undefined; + }) + .catch((error: unknown) => { + if (isMissingPathError(error)) { + this.host.queueEvent("delete", path); + return; + } + this.host.fail(toError(error)); + }) + .finally(() => this.classifications.delete(classification)); + this.classifications.add(classification); + } + + private assertWithinLimit(entryCount: number): void { + if (entryCount <= MAX_TRACKED_ENTRIES) return; + throw new Error( + `Recursive file observation exceeded ${MAX_TRACKED_ENTRIES} entries under ${this.host.root}`, + ); + } + + private canCommit(generation: number): boolean { + return this.host.isActive() && generation === this.generation; + } +} diff --git a/packages/server/src/server/file-observer/internal/observation.ts b/packages/server/src/server/file-observer/internal/observation.ts new file mode 100644 index 000000000..f0c27a504 --- /dev/null +++ b/packages/server/src/server/file-observer/internal/observation.ts @@ -0,0 +1,174 @@ +import { stat } from "node:fs/promises"; +import type { + FileChange, + FileChangeType, + FileObserverCallback, + FileObserverOptions, + FileObserverSubscription, +} from "../index.js"; +import type { + BackendDiagnostics, + CreateObservationBackend, + ObservationBackend, + ObservationHost, + ObserverMetrics, +} from "./contracts.js"; +import type { ObserverPaths } from "./paths.js"; +import { toError } from "./paths.js"; + +const EVENT_BATCH_DELAY_MS = 10; + +export interface ManagedObservation { + start(): Promise; + subscription(): FileObserverSubscription; + close(): Promise; + getDiagnostics(): BackendDiagnostics & { pendingEventCount: number }; +} + +export function createManagedObservation(input: { + root: string; + callback: FileObserverCallback; + options: FileObserverOptions; + paths: ObserverPaths; + metrics: ObserverMetrics; + createBackend: CreateObservationBackend; + onClosed: () => void; +}): ManagedObservation { + return new Observation(input); +} + +class Observation implements ManagedObservation, ObservationHost { + readonly root: string; + readonly metrics: ObserverMetrics; + private ignoredRoots: string[]; + private status: "open" | "failed" | "closed" = "open"; + private readonly backend: ObservationBackend; + private readonly pendingEvents = new Map(); + private flushTimer: NodeJS.Timeout | null = null; + private closePromise: Promise | null = null; + + constructor( + private readonly input: { + root: string; + callback: FileObserverCallback; + options: FileObserverOptions; + paths: ObserverPaths; + metrics: ObserverMetrics; + createBackend: CreateObservationBackend; + onClosed: () => void; + }, + ) { + this.root = input.root; + this.metrics = input.metrics; + this.ignoredRoots = input.paths.normalizeIgnoredRoots(input.root, input.options.ignore ?? []); + this.backend = input.createBackend(this); + } + + async start(): Promise { + const rootStats = await stat(this.root); + if (!rootStats.isDirectory()) { + throw new Error(`File observation requires a directory: ${this.root}`); + } + try { + await this.backend.start(); + } catch (error) { + await this.close(); + throw error; + } + } + + subscription(): FileObserverSubscription { + return { + updateIgnore: (paths) => this.updateIgnore(paths), + unsubscribe: () => this.close(), + }; + } + + async updateIgnore(paths: string[]): Promise { + if (!this.isActive()) return; + const next = this.input.paths.normalizeIgnoredRoots(this.root, paths); + if (samePaths(this.ignoredRoots, next)) return; + this.ignoredRoots = next; + for (const path of this.pendingEvents.keys()) { + if (this.isIgnored(path)) this.pendingEvents.delete(path); + } + try { + await this.backend.updateIgnore(); + } catch (error) { + this.fail(toError(error)); + throw error; + } + } + + close(): Promise { + if (this.closePromise) return this.closePromise; + this.status = "closed"; + this.cancelDelivery(); + this.closePromise = this.backend.close().finally(this.input.onClosed); + return this.closePromise; + } + + getDiagnostics(): BackendDiagnostics & { pendingEventCount: number } { + return { + ...this.backend.getDiagnostics(), + pendingEventCount: this.pendingEvents.size, + }; + } + + isActive(): boolean { + return this.status === "open"; + } + + isIgnored(path: string): boolean { + return this.ignoredRoots.some((root) => this.input.paths.isInside(root, path)); + } + + isPathInside(root: string, path: string): boolean { + return this.input.paths.isInside(root, path); + } + + queueEvent(type: FileChangeType, path: string): void { + if (!this.isActive() || this.isIgnored(path)) return; + const previous = this.pendingEvents.get(path); + this.pendingEvents.set(path, mergeChange(previous, { path, type })); + if (this.flushTimer) return; + this.flushTimer = setTimeout(() => { + this.flushTimer = null; + if (!this.isActive() || this.pendingEvents.size === 0) return; + const events = [...this.pendingEvents.values()]; + this.pendingEvents.clear(); + this.input.callback(null, events); + }, EVENT_BATCH_DELAY_MS); + } + + fail(error: Error): void { + if (!this.isActive()) return; + this.status = "failed"; + this.metrics.observerFailureCount += 1; + this.cancelDelivery(); + this.beginFailureClose(); + this.input.callback(error, []); + } + + private beginFailureClose(): void { + this.closePromise = this.backend.close().finally(this.input.onClosed); + } + + private cancelDelivery(): void { + if (this.flushTimer) clearTimeout(this.flushTimer); + this.flushTimer = null; + this.pendingEvents.clear(); + } +} + +function samePaths(left: string[], right: string[]): boolean { + return left.length === right.length && left.every((path, index) => path === right[index]); +} + +function mergeChange(previous: FileChange | undefined, next: FileChange): FileChange { + if (!previous) return next; + if (previous.type === "create" && next.type === "delete") return { ...next, type: "update" }; + if (previous.type === "delete" && next.type === "create") return { ...next, type: "update" }; + if (previous.type === "create") return previous; + return next; +} diff --git a/packages/server/src/server/file-observer/internal/paths.ts b/packages/server/src/server/file-observer/internal/paths.ts new file mode 100644 index 000000000..71ffffb7d --- /dev/null +++ b/packages/server/src/server/file-observer/internal/paths.ts @@ -0,0 +1,55 @@ +import { resolve, sep } from "node:path"; + +export interface ObserverPaths { + isInside(root: string, path: string): boolean; + normalizeIgnoredRoots(root: string, paths: string[]): string[]; + collapse(paths: string[]): string[]; + isExpectedWatchDisappearance(error: unknown): boolean; +} + +export function createObserverPaths(platform: NodeJS.Platform): ObserverPaths { + function comparable(path: string): string { + return platform === "win32" ? path.toLowerCase() : path; + } + + function isInside(root: string, path: string): boolean { + const comparedRoot = comparable(root); + const comparedPath = comparable(path); + return comparedPath === comparedRoot || comparedPath.startsWith(`${comparedRoot}${sep}`); + } + + function collapse(paths: string[]): string[] { + return [...new Set(paths)] + .sort((left, right) => left.length - right.length) + .filter((path, index, all) => !all.slice(0, index).some((parent) => isInside(parent, path))); + } + + return { + isInside, + collapse, + normalizeIgnoredRoots(root, paths) { + const inside = paths + .map((path) => resolve(path)) + .filter((path) => path !== root && isInside(root, path)); + return collapse(inside); + }, + isExpectedWatchDisappearance(error) { + const code = getErrorCode(error); + return code === "ENOENT" || (platform === "win32" && code === "EPERM"); + }, + }; +} + +export function isMissingPathError(error: unknown): boolean { + const code = getErrorCode(error); + return code === "ENOENT" || code === "ENOTDIR"; +} + +export function toError(value: unknown): Error { + return value instanceof Error ? value : new Error(String(value)); +} + +function getErrorCode(error: unknown): string | null { + if (typeof error !== "object" || error === null || !("code" in error)) return null; + return typeof error.code === "string" ? error.code : null; +} diff --git a/packages/server/src/server/session.create-agent-worktree-autoarchive.e2e.test.ts b/packages/server/src/server/session.create-agent-worktree-autoarchive.e2e.test.ts index 8507a935e..94e45d745 100644 --- a/packages/server/src/server/session.create-agent-worktree-autoarchive.e2e.test.ts +++ b/packages/server/src/server/session.create-agent-worktree-autoarchive.e2e.test.ts @@ -141,12 +141,12 @@ test("create_agent_request creates a worktree and auto-archives both after the f expect(created.cwd).not.toBe(repoDir); const listedWithWorktree = await ctx.client.getPaseoWorktreeList({ cwd: repoDir }); - expect(listedWithWorktree.worktrees).toEqual([ - expect.objectContaining({ - worktreePath: created.cwd, - branchName: "agent-lifecycle-dispatch-test", - }), - ]); + expect(listedWithWorktree.worktrees).toHaveLength(1); + const listedWorktree = listedWithWorktree.worktrees[0]; + expect(listedWorktree?.branchName).toBe("agent-lifecycle-dispatch-test"); + expect(createRealpathAwarePathMatcher(created.cwd)(listedWorktree?.worktreePath ?? "")).toBe( + true, + ); await ctx.client.waitForFinish(created.id, 10000); @@ -154,6 +154,18 @@ test("create_agent_request creates a worktree and auto-archives both after the f // last-reference worktree directory is gone. await expectAgentAbsentFromActiveList(created.id); await expect.poll(() => existsSync(created.cwd), { timeout: 10000, interval: 100 }).toBe(false); + // Archived tabs can continue asking for history. These reads must not recreate + // the removed workspace observation or compromise the next agent lifecycle. + const staleTimelineReads = await Promise.allSettled( + Array.from({ length: 10 }, () => ctx.client.fetchAgentTimeline(created.id, { limit: 20 })), + ); + expect(staleTimelineReads.every((result) => result.status === "rejected")).toBe(true); + const subsequent = await ctx.client.createAgent({ + config: { ...getFullAccessConfig("codex"), cwd: repoDir }, + initialPrompt: "Say done.", + }); + await ctx.client.waitForFinish(subsequent.id, 10_000); + await expectAgentPresentInActiveList(subsequent.id); }, 30000); test("create_agent_request auto-archives a nested workspace from an existing Paseo worktree", async () => { diff --git a/packages/server/src/server/test-utils/workspace-git-service-stub.ts b/packages/server/src/server/test-utils/workspace-git-service-stub.ts index 2cfaf659f..eca511c72 100644 --- a/packages/server/src/server/test-utils/workspace-git-service-stub.ts +++ b/packages/server/src/server/test-utils/workspace-git-service-stub.ts @@ -89,8 +89,30 @@ export function createNoopWorkspaceGitService( fetchInFlightCount: 0, snapshotUpdatedListenerCount: 0, watcherErrorCallbackCount: 0, + fileObserver: { + activeObservationCount: 0, + nativeHandleCount: 0, + nativeTrackedFileCount: 0, + pendingEventCount: 0, + pendingReconciliationWorkCount: 0, + reconciliationInFlightCount: 0, + reconciliationCount: 0, + scopedReconciliationCount: 0, + fullReconciliationCount: 0, + reconciliationFailureCount: 0, + observerFailureCount: 0, + directoryLimitFailureCount: 0, + nativeEventCount: 0, + nativeChangeEventCount: 0, + nativeRenameEventCount: 0, + nativePathlessEventCount: 0, + nativeClassificationCount: 0, + nativeShallowScanCount: 0, + lastReconciliationDurationMs: 0, + maxReconciliationDurationMs: 0, + }, }), - dispose: () => {}, + dispose: async () => {}, ...overrides, }; diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index a6d0d5565..cb01f18dc 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -253,8 +253,30 @@ function createFallbackWorkspaceGitService(): WorkspaceGitService { fetchInFlightCount: 0, snapshotUpdatedListenerCount: 0, watcherErrorCallbackCount: 0, + fileObserver: { + activeObservationCount: 0, + nativeHandleCount: 0, + nativeTrackedFileCount: 0, + pendingEventCount: 0, + pendingReconciliationWorkCount: 0, + reconciliationInFlightCount: 0, + reconciliationCount: 0, + scopedReconciliationCount: 0, + fullReconciliationCount: 0, + reconciliationFailureCount: 0, + observerFailureCount: 0, + directoryLimitFailureCount: 0, + nativeEventCount: 0, + nativeChangeEventCount: 0, + nativeRenameEventCount: 0, + nativePathlessEventCount: 0, + nativeClassificationCount: 0, + nativeShallowScanCount: 0, + lastReconciliationDurationMs: 0, + maxReconciliationDurationMs: 0, + }, }), - dispose: () => {}, + dispose: async () => {}, }; } @@ -1035,7 +1057,7 @@ export class VoiceAssistantWebSocketServer { await Promise.all(cleanupPromises); this.providerSnapshotManager.destroy(); this.checkoutDiffManager.dispose(); - this.workspaceGitService.dispose(); + await this.workspaceGitService.dispose(); this.pendingConnections.clear(); this.sessions.clear(); this.socketIdentities.clear(); 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 f4a7721c4..95d490e91 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 @@ -1,15 +1,13 @@ import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; -import parcelWatcher from "@parcel/watcher"; import type pino from "pino"; import { afterEach, expect, test, vi } from "vitest"; import type { CheckoutSnapshotFacts, CheckoutStatusGit } from "../utils/checkout-git.js"; import { CheckoutDiffManager } from "./checkout-diff-manager.js"; -import { - subscribeToWorkspaceFileChanges, - WorkspaceGitServiceImpl, -} from "./workspace-git-service.js"; +import { createFileObserver } from "./file-observer/index.js"; +import { WorkspaceGitServiceImpl } from "./workspace-git-service.js"; +import type { FileChange, SubscribeToFileChanges } from "./file-observer/index.js"; function createLogger(): pino.Logger { const logger = { @@ -64,7 +62,7 @@ afterEach(async () => { } }); -test("native recursive observation updates tracked state and prunes ignored storms", async () => { +test("recursive observation updates tracked state and prunes ignored storms", async () => { const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-git-observation-"))); const repoDir = path.join(tempDir, "repo"); const trackedPath = path.join(repoDir, "src", "tracked.txt"); @@ -79,14 +77,15 @@ test("native recursive observation updates tracked state and prunes ignored stor writeFileSync(newlyTrackedPath, "base\n"); let activeWatcherCount = 0; + const observer = createFileObserver(); let watcherStartCount = 0; - let onWorkingTreeWatcherStopped: (() => void) | null = null; + let onWorkingTreeIgnoreUpdated: (() => void) | null = null; const deliveredEvents: Array<{ directory: string; - events: parcelWatcher.Event[]; + events: FileChange[]; }> = []; - const subscribe: typeof parcelWatcher.subscribe = async (directory, callback, options) => { - const subscription = await subscribeToWorkspaceFileChanges( + const subscribe: SubscribeToFileChanges = async (directory, callback, options) => { + const subscription = await observer.subscribe( directory, (error, events) => { deliveredEvents.push({ directory, events }); @@ -97,28 +96,50 @@ test("native recursive observation updates tracked state and prunes ignored stor activeWatcherCount += 1; watcherStartCount += 1; return { + updateIgnore: async (paths) => { + await subscription.updateIgnore(paths); + if (directory === repoDir) { + const onUpdated = onWorkingTreeIgnoreUpdated; + onWorkingTreeIgnoreUpdated = null; + onUpdated?.(); + } + }, unsubscribe: async () => { await subscription.unsubscribe(); activeWatcherCount -= 1; - if (directory === repoDir) { - const onStopped = onWorkingTreeWatcherStopped; - onWorkingTreeWatcherStopped = null; - onStopped?.(); - } }, }; }; - const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createFacts(cwd)); - const getCheckoutStatus = vi.fn(async (cwd: string) => createStatus(cwd)); - const getCheckoutShortstat = vi.fn(async () => ({ additions: 0, deletions: 0 })); + const fileObserver = { + subscribe, + getDiagnostics: () => observer.getDiagnostics(), + close: () => observer.close(), + }; let observedPath = trackedPath; let observedRelativePath = "src/tracked.txt"; + const readObservedState = () => { + const contents = readFileSync(observedPath, "utf8"); + const isDirty = contents !== "base\n"; + return { + isDirty, + additions: isDirty ? contents.trim().split("\n").length : 0, + }; + }; + const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createFacts(cwd)); + const getCheckoutStatus = vi.fn(async (cwd: string) => ({ + ...createStatus(cwd), + isDirty: readObservedState().isDirty, + })); + const getCheckoutShortstat = vi.fn(async () => ({ + additions: readObservedState().additions, + deletions: 0, + })); const getCheckoutWorktreeState = vi.fn(async () => { - const additions = readFileSync(observedPath, "utf8").trim().split("\n").length; + const additions = readObservedState().additions; return { isDirty: true, diffStat: { additions, deletions: 0 } }; }); const getCheckoutDiff = vi.fn(async () => { - const additions = readFileSync(observedPath, "utf8").trim().split("\n").length; + const additions = readObservedState().additions; return { diff: "", structured: [ @@ -151,8 +172,8 @@ test("native recursive observation updates tracked state and prunes ignored stor const service = new WorkspaceGitServiceImpl({ logger: createLogger(), paseoHome: path.join(tempDir, "paseo-home"), + fileObserver, deps: { - subscribe, getCheckoutSnapshotFacts, getCheckoutStatus, getCheckoutShortstat, @@ -179,8 +200,8 @@ test("native recursive observation updates tracked state and prunes ignored stor diffSubscription.unsubscribe(); summarySubscription.unsubscribe(); diffManager.dispose(); - service.dispose(); - await vi.waitFor(() => expect(activeWatcherCount).toBe(0), { timeout: 5_000 }); + await service.dispose(); + expect(activeWatcherCount).toBe(0); rmSync(tempDir, { recursive: true, force: true }); }); @@ -294,16 +315,16 @@ test("native recursive observation updates tracked state and prunes ignored stor buildIgnored = false; observedPath = newlyTrackedPath; observedRelativePath = "build/tracked.txt"; - let editedDuringWatcherHandoff = false; - onWorkingTreeWatcherStopped = () => { - editedDuringWatcherHandoff = true; + let editedDuringIgnoreUpdate = false; + onWorkingTreeIgnoreUpdated = () => { + editedDuringIgnoreUpdate = true; writeFileSync(newlyTrackedPath, "first\nsecond\n"); }; writeFileSync(path.join(repoDir, ".git", "index"), "force-added\n"); await vi.waitFor( () => { - expect(editedDuringWatcherHandoff).toBe(true); - expect(watcherStartCount).toBe(3); + expect(editedDuringIgnoreUpdate).toBe(true); + expect(watcherStartCount).toBe(2); expect(activeWatcherCount).toBe(2); expect(summaryListener).toHaveBeenLastCalledWith( expect.objectContaining({ diff --git a/packages/server/src/server/workspace-git-service.observation.test.ts b/packages/server/src/server/workspace-git-service.observation.test.ts index 7f7f0db35..61d90f212 100644 --- a/packages/server/src/server/workspace-git-service.observation.test.ts +++ b/packages/server/src/server/workspace-git-service.observation.test.ts @@ -3,6 +3,7 @@ import type pino from "pino"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import type { CheckoutSnapshotFacts, CheckoutStatusGit } from "../utils/checkout-git.js"; import { CheckoutDiffManager } from "./checkout-diff-manager.js"; +import type { FileObserver } from "./file-observer/index.js"; import { WorkspaceGitServiceImpl } from "./workspace-git-service.js"; const REPO_CWD = path.resolve("/tmp/paseo-observation-repo"); @@ -19,6 +20,7 @@ interface WatchRecord { directory: string; callback: (error: Error | null, events: WatchEvent[]) => void; ignore: Array; + updateIgnore: ReturnType; unsubscribe: ReturnType; } @@ -33,14 +35,19 @@ function createWatcherHarness(harnessOptions?: { failDirectories?: Set } if (harnessOptions?.failDirectories?.has(directory)) { throw new Error(`watch failed: ${directory}`); } + const updateIgnore = vi.fn(async (paths: string[]) => { + const record = records.find((candidate) => candidate.updateIgnore === updateIgnore); + if (record) record.ignore = paths; + }); const unsubscribe = vi.fn(async () => {}); records.push({ directory, callback, ignore: options?.ignore ?? [], + updateIgnore, unsubscribe, }); - return { unsubscribe }; + return { updateIgnore, unsubscribe }; }, ); @@ -142,6 +149,7 @@ function createService( watcher: ReturnType, overrides?: Record, logger: pino.Logger = createLogger(), + fileObserver?: FileObserver, ) { const defaultGetCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const defaultGetCheckoutShortstat = vi.fn(async () => null); @@ -154,6 +162,7 @@ function createService( return new WorkspaceGitServiceImpl({ logger, paseoHome: "/tmp/paseo-home", + fileObserver, deps: { subscribe: watcher.subscribe, getCheckoutSnapshotFacts: vi.fn(async (cwd: string) => createCheckoutFacts(cwd)), @@ -202,6 +211,38 @@ describe("WorkspaceGitService checkout observation", () => { vi.useRealTimers(); }); + test("dispose waits for file observation to finish closing", async () => { + const watcher = createWatcherHarness(); + const closeFinished = createDeferred(); + let closeCalls = 0; + const fileObserver: FileObserver = { + subscribe: watcher.subscribe as FileObserver["subscribe"], + getDiagnostics: () => { + throw new Error("Diagnostics are not used by this lifecycle test"); + }, + close: () => { + closeCalls += 1; + return closeFinished.promise; + }, + }; + const service = createService(watcher, undefined, createLogger(), fileObserver); + + const disposal = service.dispose(); + const repeatedDisposal = service.dispose(); + + expect(disposal).toBeInstanceOf(Promise); + expect(repeatedDisposal).toBe(disposal); + expect(closeCalls).toBe(1); + const status = await Promise.race([ + disposal.then(() => "disposed" as const), + Promise.resolve("pending" as const), + ]); + expect(status).toBe("pending"); + + closeFinished.resolve(); + await disposal; + }); + test("shares one recursive checkout observer between cwd-equivalent consumers", async () => { const watcher = createWatcherHarness(); const runGitCommand = vi.fn(async (args: string[]) => ({ @@ -289,7 +330,10 @@ describe("WorkspaceGitService checkout observation", () => { expect(watcher.subscribe).toHaveBeenCalledTimes(1); }); subscription.unsubscribe(); - openedSubscription.resolve({ unsubscribe: unsubscribeWatcher }); + openedSubscription.resolve({ + updateIgnore: vi.fn(async () => {}), + unsubscribe: unsubscribeWatcher, + }); await vi.waitFor(() => { expect(unsubscribeWatcher).toHaveBeenCalledTimes(1); @@ -1462,14 +1506,23 @@ describe("WorkspaceGitService checkout observation", () => { service.dispose(); }); - test("watcher runtime error is abandoned, counted, and switches to scoped polling", async () => { + test("watcher runtime error is closed, counted, and switches to scoped polling", async () => { const watcher = createWatcherHarness(); + let ignoredDirectories = "node_modules/\n"; + const runGitCommand = vi.fn(async (args: string[]) => ({ + stdout: args[0] === "rev-parse" ? `${REPO_CWD}\n` : ignoredDirectories, + stderr: "", + truncated: false, + exitCode: 0, + signal: null, + })); const getCheckoutSnapshotFacts = vi.fn(async (cwd: string) => createCheckoutFacts(cwd)); const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const service = createService(watcher, { getCheckoutSnapshotFacts, getCheckoutStatus, getWorkspaceGitSelfHealPhaseMs: () => 1_000_000, + runGitCommand, }); const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn()); @@ -1481,7 +1534,7 @@ describe("WorkspaceGitService checkout observation", () => { expect(checkoutWatcher).toBeDefined(); checkoutWatcher?.callback(new Error("watcher stopped"), []); - expect(checkoutWatcher?.unsubscribe).not.toHaveBeenCalled(); + expect(checkoutWatcher?.unsubscribe).toHaveBeenCalledTimes(1); expect(service.getMetrics().watcherErrorCallbackCount).toBe(1); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { @@ -1491,11 +1544,13 @@ describe("WorkspaceGitService checkout observation", () => { expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1); expect(service.getMetrics().workspaceRefreshQueuedCount).toBe(0); + ignoredDirectories = "node_modules/\nbuild/\n"; await vi.advanceTimersByTimeAsync(29_000); await vi.waitFor(() => { expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(2); }); const recoveredWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[1]; + expect(recoveredWatcher?.ignore).toContain(path.join(REPO_CWD, "build")); await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); @@ -1518,7 +1573,10 @@ describe("WorkspaceGitService checkout observation", () => { test("setup-time watcher error defers recovery until subscribe settles", async () => { const watcher = createWatcherHarness(); - const openedSubscription = createDeferred<{ unsubscribe: () => Promise }>(); + const openedSubscription = createDeferred<{ + updateIgnore: (paths: string[]) => Promise; + unsubscribe: () => Promise; + }>(); const erroredUnsubscribe = vi.fn(async () => {}); watcher.subscribe.mockImplementationOnce(async (_directory, callback) => { callback(new Error("watcher stopped during setup"), []); @@ -1539,7 +1597,10 @@ describe("WorkspaceGitService checkout observation", () => { expect(service.getMetrics().workspaceObservationSetupAdmissionActiveCount).toBe(0); expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(1); - openedSubscription.resolve({ unsubscribe: erroredUnsubscribe }); + openedSubscription.resolve({ + updateIgnore: vi.fn(async () => {}), + unsubscribe: erroredUnsubscribe, + }); await vi.waitFor(() => { expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); }); @@ -1554,31 +1615,32 @@ describe("WorkspaceGitService checkout observation", () => { await vi.waitFor(() => { expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(2); }); - expect(erroredUnsubscribe).not.toHaveBeenCalled(); + expect(erroredUnsubscribe).toHaveBeenCalledTimes(1); subscription.unsubscribe(); service.dispose(); }); - test("watcher error during subscription setup is abandoned without native teardown", async () => { + test("watcher error during subscription setup closes the terminal observer", async () => { const watcher = createWatcherHarness(); const erroredUnsubscribe = vi.fn(async () => {}); + const updateIgnore = vi.fn(async () => {}); watcher.subscribe.mockImplementationOnce(async (_directory, callback) => { callback(new Error("watcher stopped during setup"), []); - return { unsubscribe: erroredUnsubscribe }; + return { updateIgnore, unsubscribe: erroredUnsubscribe }; }); const service = createService(watcher); const subscription = await service.requestWorkingTreeWatch(REPO_CWD, vi.fn()); - expect(erroredUnsubscribe).not.toHaveBeenCalled(); + expect(erroredUnsubscribe).toHaveBeenCalledTimes(1); expect(service.getMetrics().watcherErrorCallbackCount).toBe(1); subscription.unsubscribe(); service.dispose(); }); - test("repository watcher runtime error is abandoned, counted, and recovered", async () => { + test("repository watcher runtime error is closed, counted, and recovered", async () => { const watcher = createWatcherHarness(); const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd)); const service = createService(watcher, { @@ -1595,7 +1657,7 @@ describe("WorkspaceGitService checkout observation", () => { repositoryWatcher?.callback(new Error("repository watcher stopped"), []); - expect(repositoryWatcher?.unsubscribe).not.toHaveBeenCalled(); + expect(repositoryWatcher?.unsubscribe).toHaveBeenCalledTimes(1); expect(service.getMetrics().watcherErrorCallbackCount).toBe(1); const statusCallsBeforeReconciliation = getCheckoutStatus.mock.calls.length; @@ -1791,6 +1853,11 @@ describe("WorkspaceGitService checkout observation", () => { await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { expect(runGitCommand).toHaveBeenCalledTimes(3); + expect(checkoutWatcher?.updateIgnore).toHaveBeenCalledTimes(1); + }); + await vi.advanceTimersByTimeAsync(1_000); + await vi.waitFor(() => { + expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); }); expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(1); @@ -1812,7 +1879,7 @@ describe("WorkspaceGitService checkout observation", () => { service.dispose(); }); - test("removing an ignored directory replaces the watcher", async () => { + test("removing an ignored directory updates the watcher without replacement", async () => { const watcher = createWatcherHarness(); let ignoredDirectories = "node_modules/\nbuild/\n"; const runGitCommand = vi.fn(async (args: string[]) => { @@ -1841,14 +1908,15 @@ describe("WorkspaceGitService checkout observation", () => { ignoredDirectories = "node_modules/\n"; await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { - expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(2); + expect(originalWatcher?.updateIgnore).toHaveBeenCalledTimes(1); + expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0); }); - expect(originalWatcher?.unsubscribe).toHaveBeenCalledTimes(1); - const replacementWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[1]; - expect(replacementWatcher?.ignore).not.toContain(path.join(REPO_CWD, "build")); + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(1); + expect(originalWatcher?.unsubscribe).not.toHaveBeenCalled(); + expect(originalWatcher?.ignore).not.toContain(path.join(REPO_CWD, "build")); const statusCallsAfterRefresh = getCheckoutStatus.mock.calls.length; - replacementWatcher?.callback(null, [ + originalWatcher?.callback(null, [ { path: path.join(REPO_CWD, "build", "output.js"), type: "update" }, ]); await vi.advanceTimersByTimeAsync(1_000); @@ -1860,7 +1928,7 @@ describe("WorkspaceGitService checkout observation", () => { service.dispose(); }); - test("ignore watcher teardown failure enters polling with a distinct reason", async () => { + test("ignore watcher update failure enters polling with a distinct reason", async () => { const watcher = createWatcherHarness(); const logger = createLogger(); let ignoredDirectories = "node_modules/\nbuild/\n"; @@ -1888,13 +1956,13 @@ describe("WorkspaceGitService checkout observation", () => { expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0); }); const checkoutWatcher = getWatcherRecordsForDirectory(watcher, REPO_CWD)[0]; - checkoutWatcher?.unsubscribe.mockRejectedValueOnce(new Error("teardown failed")); + checkoutWatcher?.updateIgnore.mockRejectedValueOnce(new Error("update failed")); ignoredDirectories = "node_modules/\n"; await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => { expect(logger.warn).toHaveBeenCalledWith( - expect.objectContaining({ reason: "watcher_teardown_failed" }), + expect.objectContaining({ reason: "watcher_update_failed" }), "Working tree watcher unavailable; using bounded polling fallback", ); }); @@ -1904,6 +1972,10 @@ describe("WorkspaceGitService checkout observation", () => { await vi.waitFor(() => { expect(getCheckoutStatus.mock.calls.length).toBeGreaterThan(statusCallsBeforePoll); }); + await vi.advanceTimersByTimeAsync(25_000); + await vi.waitFor(() => { + expect(getWatcherRecordsForDirectory(watcher, REPO_CWD)).toHaveLength(2); + }); subscription.unsubscribe(); service.dispose(); diff --git a/packages/server/src/server/workspace-git-service.primitive.test.ts b/packages/server/src/server/workspace-git-service.primitive.test.ts index 180fb5416..017ed831d 100644 --- a/packages/server/src/server/workspace-git-service.primitive.test.ts +++ b/packages/server/src/server/workspace-git-service.primitive.test.ts @@ -19,7 +19,6 @@ import { } from "../utils/checkout-git.js"; import { runGitCommand as runGitCommandReal } from "../utils/run-git-command.js"; import { - getWorkspaceFileWatcherBackend, getWorkspaceGitSelfHealPhaseMs, WorkspaceGitServiceImpl, type WorkspaceGitRuntimeSnapshot, @@ -47,7 +46,10 @@ function createDeferred() { } function createAsyncSubscription() { - return { unsubscribe: vi.fn(() => Promise.resolve()) }; + return { + updateIgnore: vi.fn(() => Promise.resolve()), + unsubscribe: vi.fn(() => Promise.resolve()), + }; } async function flushPromises(): Promise { @@ -332,7 +334,10 @@ interface CreateServiceOptions { function buildDefaultServiceDeps() { return { - subscribe: vi.fn(async () => ({ unsubscribe: vi.fn(async () => {}) })), + subscribe: vi.fn(async () => ({ + updateIgnore: vi.fn(async () => {}), + unsubscribe: vi.fn(async () => {}), + })), getCheckoutSnapshotFacts: vi.fn(async (cwd: string) => createCheckoutFacts(cwd)), getCheckoutRefDerivedState: vi.fn(async (_cwd, facts, current) => ({ ...current, @@ -899,20 +904,6 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => { expect(getWorkspaceGitSelfHealPhaseMs("/tmp/staggered-repo")).toBe(9_817); }); - test.each([ - ["darwin", "fs-events"], - ["linux", "inotify"], - ["win32", "windows"], - ] as const)("uses the native %s file watcher backend", (platform, backend) => { - expect(getWorkspaceFileWatcherBackend(platform)).toBe(backend); - }); - - test("rejects unsupported watcher platforms instead of using backend auto-detection", () => { - expect(() => getWorkspaceFileWatcherBackend("freebsd")).toThrow( - "No native workspace file watcher configured for freebsd", - ); - }); - test("self-heal staggers workspaces and settles into a 120 second cadence", async () => { vi.setSystemTime(0); const otherRepoCwd = resolvePath("/tmp/staggered-repo"); @@ -2141,16 +2132,19 @@ describe("WorkspaceGitServiceImpl D2 read methods", () => { service.dispose(); }); - test("working tree observation prunes gitignored directories", async () => { + test("working tree observation prunes ignored trees but retains trees with tracked files", async () => { const tempDir = realpathSync(mkdtempSync(join(tmpdir(), "workspace-git-service-ignored-"))); const repoDir = join(tempDir, "repo"); mkdirSync(join(repoDir, "ignored", "deep"), { recursive: true }); + mkdirSync(join(repoDir, "mixed"), { recursive: true }); mkdirSync(join(repoDir, "kept"), { recursive: true }); execFileSync("git", ["init", "-b", "main"], { cwd: repoDir, stdio: "pipe" }); - writeFileSync(join(repoDir, ".gitignore"), "ignored/\n"); + writeFileSync(join(repoDir, ".gitignore"), "ignored/\nmixed/\n"); writeFileSync(join(repoDir, "ignored", "log.txt"), "noise\n"); writeFileSync(join(repoDir, "ignored", "deep", "log.txt"), "noise\n"); + writeFileSync(join(repoDir, "mixed", "tracked.txt"), "tracked\n"); writeFileSync(join(repoDir, "kept", "file.txt"), "keep\n"); + execFileSync("git", ["add", "-f", "mixed/tracked.txt"], { cwd: repoDir, stdio: "pipe" }); const subscribe = vi.fn(async () => createAsyncSubscription()); diff --git a/packages/server/src/server/workspace-git-service.test.ts b/packages/server/src/server/workspace-git-service.test.ts index a83487e2f..914fde1ff 100644 --- a/packages/server/src/server/workspace-git-service.test.ts +++ b/packages/server/src/server/workspace-git-service.test.ts @@ -200,7 +200,10 @@ function createDeferred() { } function createAsyncSubscription() { - return { unsubscribe: vi.fn(() => Promise.resolve()) }; + return { + updateIgnore: vi.fn(() => Promise.resolve()), + unsubscribe: vi.fn(() => Promise.resolve()), + }; } function createGitHubServiceStub(): ForgeService { @@ -262,7 +265,10 @@ interface CreateServiceTestOptions { function buildDefaultTestServiceDeps() { return { - subscribe: vi.fn(async () => ({ unsubscribe: vi.fn(async () => {}) })), + subscribe: vi.fn(async () => ({ + updateIgnore: vi.fn(async () => {}), + unsubscribe: vi.fn(async () => {}), + })), getCheckoutSnapshotFacts: vi.fn(async (cwd: string) => createCheckoutSnapshotFacts(cwd)), getCheckoutRefDerivedState: vi.fn(async (_cwd, facts, current) => ({ ...current, @@ -1235,7 +1241,11 @@ describe("WorkspaceGitServiceImpl", () => { test("watches nested repository changes through one recursive checkout subscription", async () => { const unsubscribe = vi.fn(async () => {}); - const subscribe = vi.fn(async () => ({ unsubscribe })); + const updateIgnore = vi.fn(async () => {}); + const subscribe = vi.fn(async () => ({ + updateIgnore, + unsubscribe, + })); const service = createService({ subscribe }); const subscription = await service.requestWorkingTreeWatch( @@ -1256,7 +1266,11 @@ describe("WorkspaceGitServiceImpl", () => { test("requestWorkingTreeWatch reference-counts watchers by cwd", async () => { const unsubscribe = vi.fn(async () => {}); - const subscribe = vi.fn(async () => ({ unsubscribe })); + const updateIgnore = vi.fn(async () => {}); + const subscribe = vi.fn(async () => ({ + updateIgnore, + unsubscribe, + })); const service = createService({ subscribe }); const firstListener = vi.fn(); diff --git a/packages/server/src/server/workspace-git-service.ts b/packages/server/src/server/workspace-git-service.ts index 917811c70..33fddb1d8 100644 --- a/packages/server/src/server/workspace-git-service.ts +++ b/packages/server/src/server/workspace-git-service.ts @@ -1,7 +1,6 @@ import { createHash } from "node:crypto"; import { readFile } from "node:fs/promises"; import { join, resolve } from "node:path"; -import parcelWatcher from "@parcel/watcher"; import { LRUCache } from "lru-cache"; import pLimit from "p-limit"; import type pino from "pino"; @@ -58,6 +57,16 @@ import { type WorkspaceGitRemoteRefChange, } from "./workspace-git-fetch.js"; import { deriveProjectSlug } from "./workspace-git-metadata.js"; +import { + type FileChange, + type FileObserverDiagnostics, + type FileObserverCallback, + type FileObserver, + type FileObserverOptions, + type FileObserverSubscription, + type SubscribeToFileChanges, + createFileObserver, +} from "./file-observer/index.js"; import { checkoutLiteFromGitSnapshot } from "./workspace-registry-model.js"; const WORKSPACE_GIT_WATCH_DEBOUNCE_MS = 1_000; @@ -212,7 +221,7 @@ export interface WorkspaceGitService { onWorkspaceStateMayHaveChanged(cwd: string): void; invalidateForge(cwd: string): void; getMetrics(): WorkspaceGitServiceMetrics; - dispose(): void; + dispose(): Promise; } export interface WorkspaceGitServiceMetrics { @@ -233,6 +242,7 @@ export interface WorkspaceGitServiceMetrics { fetchInFlightCount: number; snapshotUpdatedListenerCount: number; watcherErrorCallbackCount: number; + fileObserver: FileObserverDiagnostics; } export type WorkspaceGitListener = (snapshot: WorkspaceGitRuntimeSnapshot) => void; @@ -318,7 +328,7 @@ type WorkspaceGitRefreshState = }; interface WorkspaceGitServiceDependencies { - subscribe: typeof parcelWatcher.subscribe; + subscribe: SubscribeToFileChanges; getCheckoutSnapshotFacts: typeof getCheckoutSnapshotFacts; getCheckoutRefDerivedState: typeof getCheckoutRefDerivedState; getCheckoutStatus: typeof getCheckoutStatus; @@ -351,35 +361,10 @@ interface WorkspaceGitServiceOptions { logger: pino.Logger; paseoHome: string; worktreesRoot?: string; + fileObserver?: FileObserver; deps?: Partial; } -export function getWorkspaceFileWatcherBackend( - platform: NodeJS.Platform, -): parcelWatcher.BackendType { - switch (platform) { - case "darwin": - return "fs-events"; - case "linux": - return "inotify"; - case "win32": - return "windows"; - default: - throw new Error(`No native workspace file watcher configured for ${platform}`); - } -} - -export function subscribeToWorkspaceFileChanges( - directory: string, - callback: parcelWatcher.SubscribeCallback, - options?: parcelWatcher.Options, -): Promise { - return parcelWatcher.subscribe(directory, callback, { - ...options, - backend: getWorkspaceFileWatcherBackend(process.platform), - }); -} - class WorkspaceGitServiceDisposedError extends Error { constructor() { super("WorkspaceGitService is disposed"); @@ -429,13 +414,13 @@ interface RepoGitTarget { repoGitRoot: string; cwd: string; workspaceKeys: Set; - subscription: parcelWatcher.AsyncSubscription | null; + subscription: FileObserverSubscription | null; fallbackPolling: boolean; fallbackPollTimer: NodeJS.Timeout | null; recovery: WatchRecoveryState; intervalId: NodeJS.Timeout | null; fetchInFlight: boolean; - bufferedFetchMetadataEvents: parcelWatcher.Event[]; + bufferedFetchMetadataEvents: FileChange[]; recentFetchRemoteRefChanges: Map< string, { change: WorkspaceGitRemoteRefChange; expiresAtMs: number } @@ -455,7 +440,7 @@ interface WorkingTreeWatchTarget { cwd: string; watchPath: string; repoRoot: string | null; - subscription: parcelWatcher.AsyncSubscription | null; + subscription: FileObserverSubscription | null; ignoredDirectories: Set; ignoredDirectoriesRefreshPromise: Promise | null; aliases: Set; @@ -480,7 +465,7 @@ type WorkingTreeWatchFallbackReason = | "not_a_git_checkout" | "watcher_error" | "watcher_setup_failed" - | "watcher_teardown_failed"; + | "watcher_update_failed"; interface WorkspaceGitAuxiliaryReadCacheEntry { value: T | null; @@ -495,9 +480,11 @@ interface WorkspaceForgePrStatusPollTarget { headRepositoryOwner?: string; } -function buildDefaultWorkspaceGitServiceDeps(): WorkspaceGitServiceDependencies { +function buildDefaultWorkspaceGitServiceDeps( + subscribe: SubscribeToFileChanges, +): WorkspaceGitServiceDependencies { return { - subscribe: subscribeToWorkspaceFileChanges, + subscribe, getCheckoutSnapshotFacts, getCheckoutRefDerivedState, getCheckoutStatus, @@ -519,15 +506,17 @@ function buildDefaultWorkspaceGitServiceDeps(): WorkspaceGitServiceDependencies } function resolveWorkspaceGitServiceDeps( + subscribe: SubscribeToFileChanges, deps: Partial | undefined, ): WorkspaceGitServiceDependencies { - return { ...buildDefaultWorkspaceGitServiceDeps(), ...deps }; + return { ...buildDefaultWorkspaceGitServiceDeps(subscribe), ...deps }; } export class WorkspaceGitServiceImpl implements WorkspaceGitService { private readonly logger: pino.Logger; private readonly paseoHome: string; private readonly worktreesRoot: string | undefined; + private readonly fileObserver: FileObserver; private readonly deps: WorkspaceGitServiceDependencies; private readonly forgeResolver: ForgeResolver; private readonly workspaceRefreshLimit = pLimit({ @@ -540,6 +529,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }); private readonly disposeController = new AbortController(); private disposed = false; + private disposePromise: Promise | null = null; private readonly snapshotUpdatedListeners = new Set(); private readonly workspaceTargets = new Map(); private readonly repoTargets = new Map(); @@ -580,7 +570,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.logger = options.logger.child({ module: "workspace-git-service" }); this.paseoHome = options.paseoHome; this.worktreesRoot = options.worktreesRoot; - this.deps = resolveWorkspaceGitServiceDeps(options.deps); + this.fileObserver = options.fileObserver ?? createFileObserver(); + this.deps = resolveWorkspaceGitServiceDeps( + this.fileObserver.subscribe.bind(this.fileObserver), + options.deps, + ); this.forgeResolver = createForgeResolver({ createService: (forge) => this.deps.forgeOverrides?.[forge] ?? createForgeService(forge), }); @@ -675,6 +669,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { fetchInFlightCount, snapshotUpdatedListenerCount: this.snapshotUpdatedListeners.size, watcherErrorCallbackCount: this.watcherErrorCallbackCount, + fileObserver: this.fileObserver.getDiagnostics(), }; } @@ -963,10 +958,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.forgeResolver.invalidate(resolve(cwd)); } - dispose(): void { - if (this.disposed) { - return; - } + dispose(): Promise { + if (this.disposePromise) return this.disposePromise; this.disposed = true; this.disposeController.abort(new WorkspaceGitServiceDisposedError()); this.workspaceRefreshLimit.clearQueue(); @@ -990,6 +983,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.workingTreeWatchResolutions.clear(); this.workingTreeWatchAliases.clear(); this.snapshotUpdatedListeners.clear(); + this.disposePromise = this.fileObserver.close(); + return this.disposePromise; } private assertNotDisposed(): void { @@ -1323,18 +1318,17 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private async subscribeWithDeadline( watchPath: string, - callback: parcelWatcher.SubscribeCallback, - options: parcelWatcher.Options, + callback: FileObserverCallback, + options: FileObserverOptions, onSubscribeSettled: () => void, - shouldAbandonSubscription: () => boolean, - ): Promise { + ): Promise { this.assertNotDisposed(); const signal = this.disposeController.signal; let outcome: "pending" | "accepted" | "expired" = "pending"; let timeout: NodeJS.Timeout | null = null; let removeAbortListener = () => {}; let unsubscribePromise: Promise | null = null; - let subscriptionPromise: Promise; + let subscriptionPromise: Promise; try { subscriptionPromise = this.deps .subscribe(watchPath, callback, options) @@ -1345,7 +1339,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } void subscriptionPromise.then( (subscription) => { - if ((outcome === "expired" || signal.aborted) && !shouldAbandonSubscription()) { + if (outcome === "expired" || signal.aborted) { unsubscribePromise ??= this.unsubscribeWatcherSubscription(subscription, watchPath); return unsubscribePromise; } @@ -1380,10 +1374,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { ]); if (signal.aborted) { outcome = "expired"; - if (!shouldAbandonSubscription()) { - unsubscribePromise ??= this.unsubscribeWatcherSubscription(subscription, watchPath); - await unsubscribePromise; - } + unsubscribePromise ??= this.unsubscribeWatcherSubscription(subscription, watchPath); + await unsubscribePromise; throw signal.reason; } outcome = "accepted"; @@ -1400,7 +1392,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private async unsubscribeWatcherSubscription( - subscription: parcelWatcher.AsyncSubscription, + subscription: FileObserverSubscription, watchPath: string, ): Promise { try { @@ -1453,9 +1445,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }, { ignore }, markSubscribeSettled, - () => watcherErrored, ); if (watcherErrored) { + await this.unsubscribeWatcherSubscription(subscription, target.watchPath); return false; } if ( @@ -1548,10 +1540,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private degradeWorkingTreeWatch(target: WorkingTreeWatchTarget, reason: "watcher_error"): void { - // COMPAT(parcel-watcher-eintr): added in v0.3.0, remove after 2027-02-09. Unsubscribing an - // errored subscription races native teardown and can strand the Node main thread in - // InotifyBackend::~InotifyBackend(): https://github.com/parcel-bundler/watcher/issues/253 + const subscription = target.subscription; target.subscription = null; + if (subscription) { + void this.unsubscribeWatcherSubscription(subscription, target.watchPath); + } this.notifyWorkingTreeChanged(target, "working-tree-watch-error"); this.startWorkingTreeWatchFallback(target, reason); } @@ -1574,6 +1567,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private async recoverWorkingTreeWatch(target: WorkingTreeWatchTarget): Promise { + if (target.closed || target.subscription) { + return; + } + await this.refreshWorkingTreeIgnoredDirectories(target); if (target.closed || target.subscription) { return; } @@ -1606,7 +1603,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private hasRelevantWorkingTreeEvent( target: WorkingTreeWatchTarget, - events: parcelWatcher.Event[], + events: FileChange[], ): boolean { const gitDir = join(target.watchPath, ".git"); const matchesWatchPath = createRealpathAwarePathMatcher(target.watchPath); @@ -1629,7 +1626,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private refreshWorkingTreeIgnoredDirectories(target: WorkingTreeWatchTarget): Promise { - if (target.closed || target.repoRoot === null || target.fallbackPolling) { + if (target.closed || target.repoRoot === null) { return Promise.resolve(); } if (target.ignoredDirectoriesRefreshPromise) { @@ -1656,34 +1653,27 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { target: WorkingTreeWatchTarget, ): Promise { const ignoredDirectories = await this.loadIgnoredDirs(target.watchPath); - if ( - target.closed || - target.fallbackPolling || - this.haveSamePaths(target.ignoredDirectories, ignoredDirectories) - ) { + if (target.closed || this.haveSamePaths(target.ignoredDirectories, ignoredDirectories)) { return; } - const removedIgnoredDirectory = Array.from(target.ignoredDirectories).some( - (path) => !ignoredDirectories.has(path), - ); target.ignoredDirectories = ignoredDirectories; - if (!removedIgnoredDirectory) { + if (target.fallbackPolling) { return; } - const subscription = target.subscription; if (subscription) { - target.subscription = null; try { - await subscription.unsubscribe(); + await subscription.updateIgnore([join(target.watchPath, ".git"), ...ignoredDirectories]); } catch (error) { + target.subscription = null; if (!target.closed && !target.fallbackPolling) { - this.startWorkingTreeWatchFallback(target, "watcher_teardown_failed"); + this.startWorkingTreeWatchFallback(target, "watcher_update_failed"); + this.scheduleWorkingTreeWatchRecovery(target); } this.logger.warn( { err: error, cwd: target.cwd }, - "Failed to stop working tree watcher while refreshing ignore paths", + "Failed to update working tree watcher ignore paths", ); return; } @@ -1691,8 +1681,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { if (target.closed || target.fallbackPolling) { return; } - - await this.startWorkingTreeSubscription(target); this.notifyWorkingTreeChanged(target, "working-tree-watch-reconfigured"); } @@ -1893,9 +1881,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { }, { ignore }, markSubscribeSettled, - () => watcherErrored, ); if (watcherErrored) { + await this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot); return false; } if ( @@ -1934,9 +1922,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { } private degradeRepoMetadataWatch(target: RepoGitTarget): void { - // COMPAT(parcel-watcher-eintr): added in v0.3.0, remove after 2027-02-09. See - // degradeWorkingTreeWatch for why this errored subscription must be abandoned. + const subscription = target.subscription; target.subscription = null; + if (subscription) { + void this.unsubscribeWatcherSubscription(subscription, target.repoGitRoot); + } this.scheduleRepoMetadataRefresh(target, "git-metadata-watch-error", true); this.startRepoMetadataFallback(target); } @@ -1972,7 +1962,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private routeRepoMetadataEvents( target: RepoGitTarget, - events: parcelWatcher.Event[], + events: FileChange[], ): Map | null { const refreshes = new Map(); const matchesRepoGitRoot = createRealpathAwarePathMatcher(target.repoGitRoot); @@ -1984,7 +1974,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { return refreshes; } - private isFetchRemoteMetadataEvent(target: RepoGitTarget, event: parcelWatcher.Event): boolean { + private isFetchRemoteMetadataEvent(target: RepoGitTarget, event: FileChange): boolean { const relativePath = getRealpathAwareRelativePath(target.repoGitRoot, event.path); const effect = classifyGitMetadataPath("common", relativePath ?? ""); return ( @@ -1996,7 +1986,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private routeRepoMetadataEvent( target: RepoGitTarget, - event: parcelWatcher.Event, + event: FileChange, matchesRepoGitRoot: ReturnType, refreshes: Map, ): boolean { @@ -2053,7 +2043,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { private routePrivateGitDirEvent( target: RepoGitTarget, - event: parcelWatcher.Event, + event: FileChange, matchesRepoGitRoot: ReturnType, refreshes: Map, ): boolean { @@ -3107,7 +3097,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { ); let result: WorkspaceGitFetchResult | null = null; - const eventsBeforeFetchSnapshot: parcelWatcher.Event[] = []; + const eventsBeforeFetchSnapshot: FileChange[] = []; try { result = await this.deps.runGitFetch(target.cwd, { onRefSnapshot: (phase) => { @@ -3180,7 +3170,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService { this.flushFetchMetadataEvents(target, target.bufferedFetchMetadataEvents.splice(0)); } - private flushFetchMetadataEvents(target: RepoGitTarget, events: parcelWatcher.Event[]): void { + private flushFetchMetadataEvents(target: RepoGitTarget, events: FileChange[]): void { if (events.length === 0) { return; } diff --git a/scripts/trace-daemon.mjs b/scripts/trace-daemon.mjs index 3456cf3fe..a4ece424f 100644 --- a/scripts/trace-daemon.mjs +++ b/scripts/trace-daemon.mjs @@ -14,30 +14,11 @@ import { nodeFileTrace } from "@vercel/nft"; import { glob } from "node:fs/promises"; -import { createRequire } from "node:module"; import path from "node:path"; import { pathToFileURL } from "node:url"; const REPO_ROOT = path.resolve(import.meta.dirname, ".."); -function toTracePath(absolutePath) { - return path.relative(REPO_ROOT, absolutePath).split(path.sep).join("/"); -} - -function parcelWatcherNativePackagePath() { - const requireFromRepo = createRequire(path.join(REPO_ROOT, "package.json")); - requireFromRepo("@parcel/watcher"); - - const nativeAddonPath = Object.keys(requireFromRepo.cache) - .map(toTracePath) - .find((file) => /^node_modules\/@parcel\/watcher-[^/]+\/watcher\.node$/.test(file)); - if (!nativeAddonPath) { - throw new Error("@parcel/watcher loaded without a platform native addon"); - } - - return path.posix.dirname(nativeAddonPath); -} - const { sherpaPlatformPackageName } = await import( pathToFileURL( path.join( @@ -84,10 +65,6 @@ const additionalInputs = [ // the Nix derivation builds for one platform at a time and ships only // its own binaries. `node_modules/node-pty/prebuilds/${process.platform}-${process.arch}/**`, - // @parcel/watcher dynamically resolves its optional platform package. npm's - // build closure can contain multiple libc variants, so copy the package the - // wrapper actually loaded rather than every installed optional package. - `${parcelWatcherNativePackagePath()}/**`, // sherpa-onnx-node dynamically resolves a platform-specific native package. // Copy the wrapper plus the host platform package explicitly. "node_modules/sherpa-onnx-node/**",