From e3c47eff1720c5ec925e46e381d94e2b4b10a653 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:34:59 -0700 Subject: [PATCH] Fix ssh watcher isolation (#8463) * fix(ssh): isolate relay filesystem watchers * Fix relay watcher fault-harness pid file and in-process fallback isolati - Use exclusive ('wx') creation for the fault-harness pid file so a leaked ORCA_WATCHER_CHILD_PID_FILE env var can't clobber an existing file, and have the harness remove the file after reading a replacement pid. - Force useInProcessVitestFallback to false in the relay watcher pool so a leaked VITEST env var can never load the native watcher addon in-process on the relay; fail closed instead when the isolated child is missing. - Thread an injectable RelayWatcherProcessPool into FsHandler/ RelayFilesystemWatchRegistry for tests, and add coverage for both fixes. --- .github/workflows/release-cut.yml | 6 + .github/workflows/release-mac-build.yml | 6 + config/reliability-gates.jsonc | 89 ++++- config/scripts/build-relay.mjs | 34 +- ...package-electron-runtime-contract.test.mjs | 40 +++ .../scripts/relay-watcher-fault-harness.mjs | 286 ++++++++++++++++ src/main/ipc/parcel-watcher-child-launch.ts | 11 + .../ipc/parcel-watcher-host-subscriptions.ts | 15 + .../ipc/parcel-watcher-process-failure.ts | 7 + ...rcel-watcher-process-supervisor-options.ts | 4 + .../ipc/parcel-watcher-process-supervisor.ts | 42 +-- src/main/ipc/parcel-watcher-process.test.ts | 41 ++- .../parcel-watcher-supervisor-subscribe.ts | 8 +- .../ssh/ssh-relay-versioned-install.test.ts | 3 +- src/main/ssh/ssh-relay-versioned-install.ts | 7 +- src/main/ssh/ssh-remote-commands.ts | 5 +- src/relay/fs-handler.test.ts | 7 +- src/relay/fs-handler.ts | 154 +-------- .../relay-filesystem-watch-registry.test.ts | 170 ++++++++++ src/relay/relay-filesystem-watch-registry.ts | 312 ++++++++++++++++++ src/relay/relay-watcher-process-pool.ts | 26 ++ src/relay/subprocess.test.ts | 13 +- 22 files changed, 1092 insertions(+), 194 deletions(-) create mode 100644 config/scripts/relay-watcher-fault-harness.mjs create mode 100644 src/main/ipc/parcel-watcher-process-supervisor-options.ts create mode 100644 src/relay/relay-filesystem-watch-registry.test.ts create mode 100644 src/relay/relay-filesystem-watch-registry.ts create mode 100644 src/relay/relay-watcher-process-pool.ts diff --git a/.github/workflows/release-cut.yml b/.github/workflows/release-cut.yml index b9d3dea86..605645186 100644 --- a/.github/workflows/release-cut.yml +++ b/.github/workflows/release-cut.yml @@ -871,6 +871,12 @@ jobs: node config/scripts/runtime-file-watcher-fault-harness.mjs ELECTRON_RUN_AS_NODE=1 pnpm exec electron config/scripts/runtime-file-watcher-fault-harness.mjs + - name: Gate SSH relay watcher process isolation + run: | + # Why: the remote native watcher shares a daemon with live PTYs. + # Kill only its child and require both PTY and watch recovery before packaging. + node config/scripts/relay-watcher-fault-harness.mjs + - name: Publish release artifacts (Linux) if: matrix.platform == 'linux-x64' || matrix.platform == 'linux-arm64' uses: nick-fields/retry@v4 diff --git a/.github/workflows/release-mac-build.yml b/.github/workflows/release-mac-build.yml index d08584c10..bd5c5dc9c 100644 --- a/.github/workflows/release-mac-build.yml +++ b/.github/workflows/release-mac-build.yml @@ -127,6 +127,12 @@ jobs: node config/scripts/runtime-file-watcher-fault-harness.mjs ELECTRON_RUN_AS_NODE=1 pnpm exec electron config/scripts/runtime-file-watcher-fault-harness.mjs + - name: Gate SSH relay watcher process isolation + run: | + # Why: the remote native watcher shares a daemon with live PTYs. + # Kill only its child and require both PTY and watch recovery before packaging. + node config/scripts/relay-watcher-fault-harness.mjs + - name: Publish release artifacts (macOS) uses: nick-fields/retry@v4 with: diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 542fe8218..10fbfd0b5 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -3750,7 +3750,7 @@ }, { "id": "runtime-files.watcher-process-isolation", - "title": "Runtime file watcher faults stay process-isolated and recover without host starvation", + "title": "Runtime and SSH relay watcher faults stay process-isolated without disrupting host services", "maturity": "experimental", "protection": "partial", "owner": "runtime-platform", @@ -3759,7 +3759,8 @@ "headless orca serve", "paired file explorer", "runtime editor external-change detection", - "desktop filesystem watcher" + "desktop filesystem watcher", + "SSH relay filesystem watcher and live PTYs" ], "platforms": [ "macos", @@ -3779,19 +3780,22 @@ "remote-runtime", "ssh" ], - "coverageNotes": "Deterministic tests cover a four-child healthy runtime pool plus at most four bounded quarantine children, one quarantine attempt per watch lifetime, queue-aware cancellation, bounded initial and crash-resubscribe crawls, physical desktop-install abort, stale-generation rejection, per-child stat bounds, final-overflow RPC delivery, and renderer eviction before terminal callbacks can retry. A built-entry macOS harness drives the real supervisor, kills its child with SIGSEGV, then requires automatic resubscription and a later event. Windows keeps the existing fs.watch runtime path; SSH shared-watch setup uses registration-owned cancellation so one caller cannot cancel another caller's watch.", + "coverageNotes": "Deterministic tests cover a four-child healthy runtime pool plus at most four bounded quarantine children, one quarantine attempt per watch lifetime, queue-aware cancellation, bounded initial and crash-resubscribe crawls, physical desktop-install abort, stale-generation rejection, per-child stat bounds, final-overflow RPC delivery, and renderer eviction before terminal callbacks can retry. A built-entry macOS harness drives the real supervisor, kills its child with SIGSEGV, then requires automatic resubscription and a later event. SSH now bundles the same child boundary beside relay.js, requires both artifacts for install completeness, and preserves registration-owned same-root cancellation. A built-relay harness kills that child and requires the original relay PID and existing PTY to survive before overflow and precise events resume.", "motivatingLinks": [ "https://github.com/stablyai/orca/issues/5308", "https://github.com/stablyai/orca/issues/7547", "https://github.com/stablyai/orca/issues/8212" ], - "invariant": "Local runtime roots use at most four shared watcher children while healthy and at most four additional quarantine children after shard failure. Cancelling queued setup removes only that request; cancelling an active crawl restarts one shard and restores its healthy records. Setup deadlines start only when the child owns the crawl; the runtime RPC transport owns the normal 15-second initial budget, while a 60-second recovery deadline quarantines a root that would otherwise pin its shard. A crash-fused shared shard moves affected roots into the bounded quarantine pool, and a second fused quarantine ends those watch lifetimes instead of spawning replacements indefinitely. Every interrupted setup emits overflow after recovery, directory metadata work stays bounded to eight concurrent operations per child, destructive desktop cleanup physically aborts pending native setup, and terminal renderer watches are evicted before error callbacks can retry.", - "oracle": "Fill the four-child healthy pool with multiple roots, inject IPC disconnect-before-exit, queued and active setup cancellation, initial and crash-resubscribe timeouts, and shard-level terminal failure, then assert healthy roots are restored, affected roots spread across no more than four quarantine supervisors, a second quarantine fuse creates no replacement, stale generations stop delivering, and existing same-root clients receive overflow plus later events without error/end. Abort a pending desktop install and require worktree close and app shutdown to settle before the native crawl resolves. Synchronously retry from a renderer terminal-error callback and require a fresh stream that survives the old stream's end. Run eight simultaneous metadata batches in one child and assert no more than eight stats are active there. Drive a real built child through the source supervisor, kill it with SIGSEGV, and require automatic resubscription plus a post-crash event while the harness parent remains alive.", + "invariant": "Local runtime and SSH relay roots use at most four shared watcher children while healthy and at most four additional quarantine children after shard failure. Cancelling queued setup removes only that request; cancelling an active crawl restarts one shard and restores its healthy records. Setup deadlines start only when the child owns the crawl; the runtime RPC transport owns the normal 15-second initial budget, while a 60-second recovery deadline quarantines a root that would otherwise pin its shard. A crash-fused shared shard moves affected roots into the bounded quarantine pool, and a second fused quarantine ends those watch lifetimes instead of spawning replacements indefinitely. Every interrupted setup emits overflow after recovery, event delivery stays bounded, and the SSH relay, PTYs, Git, and other filesystem operations never load the watcher native addon or share its fault boundary.", + "oracle": "Fill the four-child healthy pool with multiple roots, inject IPC disconnect-before-exit, queued and active setup cancellation, initial and crash-resubscribe timeouts, and shard-level terminal failure, then assert healthy roots are restored, affected roots spread across no more than four quarantine supervisors, a second quarantine fuse creates no replacement, stale generations stop delivering, and existing same-root clients receive overflow plus later events without error/end. Abort a pending desktop install and require worktree close and app shutdown to settle before the native crawl resolves. Verify relay build hashes and remote install probes require both relay.js and relay-watcher.js. Launch the built relay with a real PTY and Parcel watch, kill only the watcher child, and require the same relay PID and PTY to answer before a replacement child delivers overflow and a post-crash event.", "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/parcel-watcher-process.test.ts src/main/ipc/runtime-watcher-process-pool.test.ts src/main/ipc/parcel-watcher-process-entry.test.ts src/main/ipc/parcel-watcher-event-delivery.test.ts src/main/ipc/filesystem-watcher-local-unsubscribe.test.ts src/main/ipc/filesystem-watcher-wsl.test.ts src/main/runtime/file-watcher-host.test.ts src/main/runtime/rpc/methods/files.test.ts src/main/providers/ssh-filesystem-provider.test.ts src/renderer/src/runtime/runtime-file-client.test.ts", "pnpm run build:electron-vite", "node config/scripts/runtime-file-watcher-fault-harness.mjs", - "ELECTRON_RUN_AS_NODE=1 pnpm exec electron config/scripts/runtime-file-watcher-fault-harness.mjs" + "ELECTRON_RUN_AS_NODE=1 pnpm exec electron config/scripts/runtime-file-watcher-fault-harness.mjs", + "pnpm exec vitest run --config config/vitest.config.ts src/relay/relay-filesystem-watch-registry.test.ts src/relay/fs-handler.test.ts src/relay/subprocess.test.ts src/main/ssh/ssh-relay-versioned-install.test.ts config/scripts/package-electron-runtime-contract.test.mjs", + "pnpm run build:relay", + "node config/scripts/relay-watcher-fault-harness.mjs" ], "testFiles": [ "src/main/ipc/parcel-watcher-process.test.ts", @@ -3804,7 +3808,13 @@ "src/main/runtime/rpc/methods/files.test.ts", "src/main/providers/ssh-filesystem-provider.test.ts", "src/renderer/src/runtime/runtime-file-client.test.ts", - "config/scripts/runtime-file-watcher-fault-harness.mjs" + "config/scripts/runtime-file-watcher-fault-harness.mjs", + "src/relay/relay-filesystem-watch-registry.test.ts", + "src/relay/fs-handler.test.ts", + "src/relay/subprocess.test.ts", + "src/main/ssh/ssh-relay-versioned-install.test.ts", + "config/scripts/package-electron-runtime-contract.test.mjs", + "config/scripts/relay-watcher-fault-harness.mjs" ], "assertionRefs": [ { @@ -3814,7 +3824,8 @@ "IPC disconnect recovery restores every existing root before accepting new setup", "queued abort removes one request without restarting healthy roots, while active abort or crawl-scoped timeout restores the shard", "a crash-resubscribe crawl deadline ends only the stuck root and tears down its shard process", - "missing built entries fail closed and unavailable canary storage remains optional" + "missing built entries fail closed and unavailable canary storage remains optional", + "fault-harness PID reporting cannot clobber an existing filesystem path" ] }, { @@ -3885,6 +3896,39 @@ "a later same-root listener opens a fresh server subscription instead of reusing a dead stream" ] }, + { + "file": "src/relay/relay-filesystem-watch-registry.test.ts", + "assertions": [ + "replacement emits overflow before later ordered events", + "terminal shard failure preserves same-root clients through recovery", + "pending setup aborts only after the last same-root client leaves", + "a missing child fails closed even if the relay inherits a VITEST environment" + ] + }, + { + "file": "src/relay/fs-handler.test.ts", + "assertions": [ + "shared watches retain the twenty-root budget and release after every client leaves" + ] + }, + { + "file": "src/relay/subprocess.test.ts", + "assertions": [ + "the real relay bundle serves PTY, filesystem, and Git requests with its adjacent watcher child" + ] + }, + { + "file": "src/main/ssh/ssh-relay-versioned-install.test.ts", + "assertions": [ + "remote completeness requires relay.js, relay-watcher.js, and the install sentinel" + ] + }, + { + "file": "config/scripts/package-electron-runtime-contract.test.mjs", + "assertions": [ + "every release platform packages and gates the hashed watcher child" + ] + }, { "file": "config/scripts/runtime-file-watcher-fault-harness.mjs", "assertions": [ @@ -3892,17 +3936,42 @@ "the harness parent survives watcher-child SIGSEGV", "the real supervisor automatically replaces the child, resubscribes, and delivers a post-crash event" ] + }, + { + "file": "config/scripts/relay-watcher-fault-harness.mjs", + "assertions": [ + "the built relay and existing PTY survive watcher-child fault injection", + "replacement emits overflow and a precise post-crash event" + ] } ], "evidenceRuns": [ + { + "date": "2026-07-12", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/relay/relay-filesystem-watch-registry.test.ts src/relay/fs-handler.test.ts src/relay/subprocess.test.ts src/main/ssh/ssh-relay-versioned-install.test.ts config/scripts/package-electron-runtime-contract.test.mjs", + "result": "passed", + "durationSeconds": 4.5, + "summary": "Five relay registry, handler, bundled subprocess, install, and package contract files passed with 114 tests." + }, + { + "date": "2026-07-12", + "runner": "local", + "platform": "macos", + "command": "node config/scripts/relay-watcher-fault-harness.mjs", + "result": "passed", + "durationSeconds": 0.61, + "summary": "The relay PID and existing PTY survived watcher-child SIGSEGV; overflow and precise events resumed from a replacement child." + }, { "date": "2026-07-12", "runner": "local", "platform": "macos", "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/parcel-watcher-process.test.ts src/main/ipc/runtime-watcher-process-pool.test.ts src/main/ipc/parcel-watcher-process-entry.test.ts src/main/ipc/parcel-watcher-event-delivery.test.ts src/main/ipc/filesystem-watcher-local-unsubscribe.test.ts src/main/ipc/filesystem-watcher-wsl.test.ts src/main/runtime/file-watcher-host.test.ts src/main/runtime/rpc/methods/files.test.ts src/main/providers/ssh-filesystem-provider.test.ts src/renderer/src/runtime/runtime-file-client.test.ts", "result": "passed", - "durationSeconds": 1.73, - "summary": "10 focused supervisor, pool, child-entry, delivery, desktop/WSL cleanup, runtime-host, RPC, SSH ownership, and renderer-client test files passed with 222 tests." + "durationSeconds": 1.7, + "summary": "10 focused supervisor, pool, child-entry, delivery, desktop/WSL cleanup, runtime-host, RPC, SSH ownership, and renderer-client test files passed with 223 tests." }, { "date": "2026-07-12", diff --git a/config/scripts/build-relay.mjs b/config/scripts/build-relay.mjs index c9cabcea8..cb7e01c97 100644 --- a/config/scripts/build-relay.mjs +++ b/config/scripts/build-relay.mjs @@ -1,9 +1,9 @@ #!/usr/bin/env node /** - * Bundle the relay daemon into a single relay.js file per platform. + * Bundle the relay daemon and its crash-isolated watcher child per platform. * - * The relay runs on remote hosts via `node relay.js`, so it must be a - * self-contained CommonJS bundle with no external dependencies beyond + * The relay runs on remote hosts via `node relay.js`, so both outputs use + * self-contained CommonJS bundles with no external dependencies beyond * Node.js built-ins. Native addons (node-pty, @parcel/watcher) are * marked external and expected to be installed on the remote or * gracefully degraded. @@ -17,6 +17,7 @@ const __dirname = import.meta.dirname // Why: the script lives under config/scripts, so go two levels up to reach the repo root. const ROOT = join(__dirname, '..', '..') const RELAY_ENTRY = join(ROOT, 'src', 'relay', 'relay.ts') +const WATCHER_ENTRY = join(ROOT, 'src', 'main', 'ipc', 'parcel-watcher-process-entry.ts') const PLATFORMS = [ 'linux-x64', @@ -42,7 +43,22 @@ for (const platform of PLATFORMS) { outfile: join(outDir, 'relay.js'), // Native addons cannot be bundled — they must exist on the remote host. // The relay gracefully degrades when they are absent. - external: ['node-pty', '@parcel/watcher'], + external: ['node-pty', '@parcel/watcher', 'electron'], + sourcemap: false, + minify: true, + define: { + 'process.env.NODE_ENV': '"production"' + } + }) + + await build({ + entryPoints: [WATCHER_ENTRY], + bundle: true, + platform: 'node', + target: 'node18', + format: 'cjs', + outfile: join(outDir, 'relay-watcher.js'), + external: ['@parcel/watcher'], sourcemap: false, minify: true, define: { @@ -51,9 +67,15 @@ for (const platform of PLATFORMS) { }) // Why: include a content hash so the deploy check detects code changes - // even when RELAY_VERSION hasn't been bumped (common during development). + // even when RELAY_VERSION hasn't been bumped. Hash both process artifacts + // so a watcher-only change always deploys beside the matching relay host. const relayContent = readFileSync(join(outDir, 'relay.js')) - const hash = createHash('sha256').update(relayContent).digest('hex').slice(0, 12) + const watcherContent = readFileSync(join(outDir, 'relay-watcher.js')) + const hash = createHash('sha256') + .update(relayContent) + .update(watcherContent) + .digest('hex') + .slice(0, 12) writeFileSync(join(outDir, '.version'), `${RELAY_VERSION}+${hash}`) console.log(`Built relay for ${platform} → ${outDir}/relay.js`) diff --git a/config/scripts/package-electron-runtime-contract.test.mjs b/config/scripts/package-electron-runtime-contract.test.mjs index 241cad088..e4a118a31 100644 --- a/config/scripts/package-electron-runtime-contract.test.mjs +++ b/config/scripts/package-electron-runtime-contract.test.mjs @@ -122,6 +122,46 @@ describe('Electron runtime package contract', () => { ) }) + it('packages and release-gates the SSH relay watcher child', () => { + const relayBuild = readFileSync(join(projectDir, 'config/scripts/build-relay.mjs'), 'utf8') + const builderConfig = readFileSync( + join(projectDir, 'config/electron-builder.config.cjs'), + 'utf8' + ) + const remoteCommands = readFileSync( + join(projectDir, 'src/main/ssh/ssh-remote-commands.ts'), + 'utf8' + ) + const releaseWorkflow = parse( + readFileSync(join(projectDir, '.github/workflows/release-cut.yml'), 'utf8') + ) + const macWorkflow = parse( + readFileSync(join(projectDir, '.github/workflows/release-mac-build.yml'), 'utf8') + ) + + expect(relayBuild).toContain("'parcel-watcher-process-entry.ts'") + expect(relayBuild).toContain("outfile: join(outDir, 'relay-watcher.js')") + expect(relayBuild).toContain("readFileSync(join(outDir, 'relay-watcher.js'))") + expect(builderConfig).toContain("from: 'out/relay'") + expect(remoteCommands).toContain("joinRemotePath(host, remoteRelayDir, 'relay-watcher.js')") + + const assertRelayGate = (steps, publishStepName) => { + const names = steps.map((step) => step.name) + const gate = steps.find((step) => step.name === 'Gate SSH relay watcher process isolation') + expect(gate['continue-on-error']).toBeUndefined() + expect(gate.run).toContain('node config/scripts/relay-watcher-fault-harness.mjs') + expect(names.indexOf('Build app')).toBeLessThan(names.indexOf(gate.name)) + expect(names.indexOf(gate.name)).toBeLessThan(names.indexOf(publishStepName)) + } + + assertRelayGate(releaseWorkflow.jobs.build.steps, 'Publish release artifacts (Linux)') + assertRelayGate(macWorkflow.jobs['build-mac'].steps, 'Publish release artifacts (macOS)') + const releaseNames = releaseWorkflow.jobs.build.steps.map((step) => step.name) + expect(releaseNames.indexOf('Gate SSH relay watcher process isolation')).toBeLessThan( + releaseNames.indexOf('Build Windows release artifacts') + ) + }) + it('pins the Windows release builder to the VS 2022 runner image', () => { const releaseWorkflow = parse( readFileSync(join(projectDir, '.github/workflows/release-cut.yml'), 'utf8') diff --git a/config/scripts/relay-watcher-fault-harness.mjs b/config/scripts/relay-watcher-fault-harness.mjs new file mode 100644 index 000000000..c86ba7649 --- /dev/null +++ b/config/scripts/relay-watcher-fault-harness.mjs @@ -0,0 +1,286 @@ +import { build } from 'esbuild' +import { spawn } from 'node:child_process' +import { createRequire } from 'node:module' +import { existsSync } from 'node:fs' +import { mkdtemp, readFile, realpath, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { dirname, join, resolve } from 'node:path' + +const WAIT_TIMEOUT_MS = 20_000 +const require = createRequire(import.meta.url) + +function withTimeout(promise, label, stderr) { + return new Promise((resolvePromise, rejectPromise) => { + const timer = setTimeout(() => { + rejectPromise(new Error(`Timed out waiting for ${label}\n${stderr()}`)) + }, WAIT_TIMEOUT_MS) + promise.then( + (value) => { + clearTimeout(timer) + resolvePromise(value) + }, + (error) => { + clearTimeout(timer) + rejectPromise(error) + } + ) + }) +} + +function pollUntil(readValue, label, stderr) { + const deadline = Date.now() + WAIT_TIMEOUT_MS + return new Promise((resolveValue, rejectValue) => { + const poll = async () => { + try { + const value = await readValue() + if (value !== undefined) { + resolveValue(value) + return + } + } catch {} + if (Date.now() >= deadline) { + rejectValue(new Error(`Timed out waiting for ${label}\n${stderr()}`)) + return + } + setTimeout(poll, 10) + } + void poll() + }) +} + +function waitForExit(proc) { + if (proc.exitCode !== null || proc.signalCode !== null) { + return Promise.resolve() + } + return new Promise((resolveExit) => proc.once('exit', resolveExit)) +} + +async function loadProtocol(bundleDir) { + const outfile = join(bundleDir, 'relay-protocol.cjs') + await build({ + entryPoints: [resolve('src/relay/protocol.ts')], + bundle: true, + platform: 'node', + format: 'cjs', + outfile, + logLevel: 'silent' + }) + return require(outfile) +} + +function createRelayClient(entryPath, args, env, protocol) { + const proc = spawn(process.execPath, [entryPath, ...args], { + cwd: dirname(entryPath), + env, + stdio: ['pipe', 'pipe', 'pipe'] + }) + const messages = [] + let stderr = '' + let nextSequence = 1 + let stdoutBuffer = Buffer.alloc(0) + let ready = false + let resolveReady + const sentinelReceived = new Promise((resolvePromise) => { + resolveReady = resolvePromise + }) + const decoder = new protocol.FrameDecoder((frame) => { + if (frame.type === protocol.MessageType.Regular) { + messages.push(protocol.parseJsonRpcMessage(frame.payload)) + } + }) + proc.stderr.on('data', (chunk) => { + stderr = `${stderr}${String(chunk)}`.slice(-8_000) + }) + proc.stdout.on('data', (chunk) => { + if (ready) { + decoder.feed(chunk) + return + } + stdoutBuffer = Buffer.concat([stdoutBuffer, chunk]) + const sentinel = Buffer.from(protocol.RELAY_SENTINEL) + const index = stdoutBuffer.indexOf(sentinel) + if (index < 0) { + return + } + ready = true + resolveReady() + const remainder = stdoutBuffer.subarray(index + sentinel.length) + if (remainder.length > 0) { + decoder.feed(remainder) + } + }) + + const waitForMessage = (startIndex, predicate, label) => + pollUntil( + () => messages.slice(startIndex).find(predicate), + label, + () => stderr + ) + + const request = async (method, params = {}) => { + const id = nextSequence++ + const startIndex = messages.length + proc.stdin.write(protocol.encodeJsonRpcFrame({ jsonrpc: '2.0', id, method, params }, id, 0)) + const response = await waitForMessage( + startIndex, + (message) => message.id === id, + `response to ${method}` + ) + if (response.error) { + throw new Error(`${method} failed: ${response.error.message}`) + } + return response.result + } + + const notify = (method, params = {}) => { + const sequence = nextSequence++ + proc.stdin.write(protocol.encodeJsonRpcFrame({ jsonrpc: '2.0', method, params }, sequence, 0)) + } + + return { + proc, + request, + notify, + sentinelReceived: withTimeout(sentinelReceived, 'relay sentinel', () => stderr), + messageCount: () => messages.length, + waitForNotification: (startIndex, method, predicate = () => true) => + waitForMessage( + startIndex, + (message) => message.method === method && predicate(message.params ?? {}), + `${method} notification` + ), + stderr: () => stderr + } +} + +async function waitForWatcherPid(pidFile, previousPid, stderr) { + return pollUntil( + async () => { + const pid = Number((await readFile(pidFile, 'utf8')).trim()) + if (!Number.isInteger(pid) || pid <= 0 || pid === previousPid) { + return undefined + } + // Why: replacement children reuse this exclusive path after fault injection. + await rm(pidFile, { force: true }) + return pid + }, + previousPid ? 'replacement watcher child pid' : 'initial watcher child pid', + stderr + ) +} + +function includesWatchPath(params, targetPath) { + return Array.isArray(params.events) + ? params.events.some((event) => event.absolutePath === targetPath) + : false +} + +async function main() { + const platform = `${process.platform}-${process.arch}` + const relayEntry = resolve('out', 'relay', platform, 'relay.js') + const watcherEntry = resolve('out', 'relay', platform, 'relay-watcher.js') + if (!existsSync(relayEntry) || !existsSync(watcherEntry)) { + throw new Error(`Missing built relay artifacts for ${platform}; run pnpm run build:relay first`) + } + + let tempRoot + let relay + try { + tempRoot = await mkdtemp(join(tmpdir(), 'orca-relay-watcher-fault-')) + const watchRoot = await realpath(tempRoot) + const pidFile = join(tempRoot, 'watcher.pid') + const protocol = await loadProtocol(tempRoot) + const socketPath = + process.platform === 'win32' + ? `\\\\.\\pipe\\orca-relay-watcher-fault-${process.pid}-${Date.now()}` + : join(tempRoot, 'relay.sock') + relay = createRelayClient( + relayEntry, + ['--sock-path', socketPath, '--endpoint-dir', join(tempRoot, 'agent-hooks')], + { ...process.env, ORCA_WATCHER_CHILD_PID_FILE: pidFile }, + protocol + ) + await relay.sentinelReceived + + const spawned = await relay.request('pty.spawn', { cols: 80, rows: 24, cwd: watchRoot }) + const beforePtyMarker = `ORCA_PTY_BEFORE_${Date.now()}` + let startIndex = relay.messageCount() + relay.notify('pty.data', { id: spawned.id, data: `echo ${beforePtyMarker}\r` }) + await relay.waitForNotification( + startIndex, + 'pty.data', + (params) => params.id === spawned.id && String(params.data).includes(beforePtyMarker) + ) + + await relay.request('fs.watch', { rootPath: watchRoot }) + const firstWatcherPid = await waitForWatcherPid(pidFile, undefined, relay.stderr) + const beforePath = join(watchRoot, 'before.txt') + startIndex = relay.messageCount() + await writeFile(beforePath, 'before') + await relay.waitForNotification(startIndex, 'fs.changed', (params) => + includesWatchPath(params, beforePath) + ) + + const faultSignal = process.platform === 'win32' ? 'SIGTERM' : 'SIGSEGV' + startIndex = relay.messageCount() + process.kill(firstWatcherPid, faultSignal) + const replacementWatcherPid = await waitForWatcherPid(pidFile, firstWatcherPid, relay.stderr) + await relay.waitForNotification(startIndex, 'fs.changed', (params) => + Array.isArray(params.events) + ? params.events.some( + (event) => event.kind === 'overflow' && event.absolutePath === watchRoot + ) + : false + ) + + const status = await relay.request('relay.status') + if (status.pid !== relay.proc.pid) { + throw new Error('relay.status did not come from the original surviving relay process') + } + const afterPtyMarker = `ORCA_PTY_AFTER_${Date.now()}` + startIndex = relay.messageCount() + relay.notify('pty.data', { id: spawned.id, data: `echo ${afterPtyMarker}\r` }) + await relay.waitForNotification( + startIndex, + 'pty.data', + (params) => params.id === spawned.id && String(params.data).includes(afterPtyMarker) + ) + + const afterPath = join(watchRoot, 'after.txt') + startIndex = relay.messageCount() + await writeFile(afterPath, 'after') + await relay.waitForNotification(startIndex, 'fs.changed', (params) => + includesWatchPath(params, afterPath) + ) + + relay.notify('fs.unwatch', { rootPath: watchRoot }) + await relay.request('pty.shutdown', { id: spawned.id }) + console.log( + JSON.stringify({ + relayPid: relay.proc.pid, + killedWatcherPid: firstWatcherPid, + replacementWatcherPid, + faultSignal, + relaySurvived: true, + existingPtySurvived: true, + overflowRefreshDelivered: true, + postCrashEventDelivered: true + }) + ) + } finally { + if (relay && relay.proc.exitCode === null && relay.proc.signalCode === null) { + relay.proc.kill('SIGTERM') + try { + await withTimeout(waitForExit(relay.proc), 'relay shutdown', relay.stderr) + } catch { + relay.proc.kill('SIGKILL') + await withTimeout(waitForExit(relay.proc), 'forced relay shutdown', relay.stderr) + } + } + if (tempRoot) { + await rm(tempRoot, { recursive: true, force: true }) + } + } +} + +await main() diff --git a/src/main/ipc/parcel-watcher-child-launch.ts b/src/main/ipc/parcel-watcher-child-launch.ts index cf8e7348b..6a72bb981 100644 --- a/src/main/ipc/parcel-watcher-child-launch.ts +++ b/src/main/ipc/parcel-watcher-child-launch.ts @@ -1,4 +1,5 @@ import { fork, type ChildProcess } from 'node:child_process' +import { writeFileSync } from 'node:fs' import { createWatcherCanaryDirectory, removeWatcherCanaryDirectory @@ -33,6 +34,16 @@ export function launchWatcherChild( console.error('[parcel-watcher-process] failed to fork watcher process:', error) return null } + const faultHarnessPidFile = process.env.ORCA_WATCHER_CHILD_PID_FILE + if (faultHarnessPidFile && child.pid) { + try { + // Why: exclusive creation lets the harness identify the child without a + // leaked test-only environment variable clobbering an existing file. + writeFileSync(faultHarnessPidFile, String(child.pid), { flag: 'wx' }) + } catch { + // Fault-injection observability must never affect watcher availability. + } + } child.stderr?.on('data', (chunk: Buffer) => { console.error('[parcel-watcher-process]', String(chunk).trimEnd()) }) diff --git a/src/main/ipc/parcel-watcher-host-subscriptions.ts b/src/main/ipc/parcel-watcher-host-subscriptions.ts index fd6b9a025..71a3d8286 100644 --- a/src/main/ipc/parcel-watcher-host-subscriptions.ts +++ b/src/main/ipc/parcel-watcher-host-subscriptions.ts @@ -100,6 +100,21 @@ export function resolvePendingWatcherUnsubscribes( pendingUnsubscribes.clear() } +export function disposeWatcherSupervisorSubscriptions( + records: Map, + pendingUnsubscribes: Map void>, + cancelledSubscribesAwaitingChild: Set, + error: Error +): void { + for (const record of records.values()) { + resetPendingSubscribeAttempt(record) + takePendingSubscribe(record)?.reject(error) + } + resolvePendingWatcherUnsubscribes(pendingUnsubscribes) + cancelledSubscribesAwaitingChild.clear() + records.clear() +} + export type CreateHostWatcherSubscriptionOptions = { record: WatcherProcessSubscriptionRecord records: Map diff --git a/src/main/ipc/parcel-watcher-process-failure.ts b/src/main/ipc/parcel-watcher-process-failure.ts index b93f2f375..80eb79ddd 100644 --- a/src/main/ipc/parcel-watcher-process-failure.ts +++ b/src/main/ipc/parcel-watcher-process-failure.ts @@ -20,6 +20,13 @@ export class WatcherProcessFailure extends Error { } } +export function watcherHostFailure( + message: string, + code: WatcherProcessFailureCode +): WatcherProcessFailure { + return new WatcherProcessFailure(message, 'supervisor', code) +} + export function isWatcherProcessFailure(error: unknown): error is WatcherProcessFailure { return error instanceof WatcherProcessFailure } diff --git a/src/main/ipc/parcel-watcher-process-supervisor-options.ts b/src/main/ipc/parcel-watcher-process-supervisor-options.ts new file mode 100644 index 000000000..47f4282a7 --- /dev/null +++ b/src/main/ipc/parcel-watcher-process-supervisor-options.ts @@ -0,0 +1,4 @@ +export type WatcherProcessSupervisorOptions = { + entryPath?: string + useInProcessVitestFallback?: boolean +} diff --git a/src/main/ipc/parcel-watcher-process-supervisor.ts b/src/main/ipc/parcel-watcher-process-supervisor.ts index 40a9c193b..9be9b232b 100644 --- a/src/main/ipc/parcel-watcher-process-supervisor.ts +++ b/src/main/ipc/parcel-watcher-process-supervisor.ts @@ -5,6 +5,7 @@ import { removeWatcherCanaryDirectory } from './parcel-watcher-canary-directory' import { launchWatcherChild } from './parcel-watcher-child-launch' import { WatcherProcessCrashFuse } from './parcel-watcher-crash-fuse' import { + disposeWatcherSupervisorSubscriptions, failAllWatcherSubscriptions, handleWatcherHostMessage, reportWatcherTerminalError, @@ -16,7 +17,8 @@ import { startPendingSubscribeTimeout, takePendingSubscribe } from './parcel-watcher-pending-subscribe' -import { WatcherProcessFailure } from './parcel-watcher-process-failure' +import { watcherHostFailure } from './parcel-watcher-process-failure' +import type { WatcherProcessFailure } from './parcel-watcher-process-failure' import type { HostToWatcherMessage, WatcherProcessSubscribeOptions, @@ -28,6 +30,7 @@ import type { WatcherProcessSubscription, WatcherProcessSubscriptionRecord } from './parcel-watcher-process-subscription' +import type { WatcherProcessSupervisorOptions } from './parcel-watcher-process-supervisor-options' import { subscribeThroughWatcherSupervisor } from './parcel-watcher-supervisor-subscribe' export type { @@ -35,6 +38,7 @@ export type { WatcherProcessHooks, WatcherProcessSubscription } from './parcel-watcher-process-subscription' +export type { WatcherProcessSupervisorOptions } from './parcel-watcher-process-supervisor-options' export class WatcherProcessSupervisor { private child: ChildProcess | null = null @@ -45,6 +49,9 @@ export class WatcherProcessSupervisor { private readonly records = new Map() private readonly pendingUnsubscribes = new Map void>() private readonly cancelledSubscribesAwaitingChild = new Set() + + constructor(private readonly options: WatcherProcessSupervisorOptions = {}) {} + subscribe( dir: string, callback: WatcherProcessCallback, @@ -57,6 +64,8 @@ export class WatcherProcessSupervisor { opts, hooks, shutdownRequested: this.shutdownRequested, + entryPath: this.options.entryPath ?? getWatcherProcessEntryPath(), + useInProcessVitestFallback: this.options.useInProcessVitestFallback ?? true, allocateId: () => this.nextSubscriptionId++, records: this.records, pendingUnsubscribes: this.pendingUnsubscribes, @@ -73,19 +82,13 @@ export class WatcherProcessSupervisor { this.shutdownRequested = true const proc = this.child this.child = null - const error = new WatcherProcessFailure( - 'file watcher supervisor disposed', - 'supervisor', - 'supervisor_disposed' + const error = watcherHostFailure('file watcher supervisor disposed', 'supervisor_disposed') + disposeWatcherSupervisorSubscriptions( + this.records, + this.pendingUnsubscribes, + this.cancelledSubscribesAwaitingChild, + error ) - for (const record of this.records.values()) { - resetPendingSubscribeAttempt(record) - const pending = takePendingSubscribe(record) - pending?.reject(error) - } - resolvePendingWatcherUnsubscribes(this.pendingUnsubscribes) - this.cancelledSubscribesAwaitingChild.clear() - this.records.clear() proc?.kill() this.canaryDir = removeWatcherCanaryDirectory(this.canaryDir) } @@ -96,7 +99,9 @@ export class WatcherProcessSupervisor { this.crashFuse.reset() } - private ensureWatcherProcess(entryPath = getWatcherProcessEntryPath()): ChildProcess | null { + private ensureWatcherProcess( + entryPath = this.options.entryPath ?? getWatcherProcessEntryPath() + ): ChildProcess | null { if (this.shutdownRequested) { return null } @@ -200,11 +205,7 @@ export class WatcherProcessSupervisor { ) failAllWatcherSubscriptions( this.records, - new WatcherProcessFailure( - 'file watcher process crashed repeatedly', - 'supervisor', - 'supervisor_crash_fuse' - ) + watcherHostFailure('file watcher process crashed repeatedly', 'supervisor_crash_fuse') ) this.canaryDir = removeWatcherCanaryDirectory(this.canaryDir) return @@ -279,9 +280,8 @@ export class WatcherProcessSupervisor { if (!replacement) { failAllWatcherSubscriptions( this.records, - new WatcherProcessFailure( + watcherHostFailure( 'file watcher process unavailable after subscription cancellation', - 'supervisor', 'process_unavailable' ) ) diff --git a/src/main/ipc/parcel-watcher-process.test.ts b/src/main/ipc/parcel-watcher-process.test.ts index 278918400..1809ad484 100644 --- a/src/main/ipc/parcel-watcher-process.test.ts +++ b/src/main/ipc/parcel-watcher-process.test.ts @@ -4,21 +4,28 @@ import { EventEmitter } from 'node:events' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { forkMock, existsSyncMock, mkdtempSyncMock, parcelSubscribeMock, rmSyncMock } = vi.hoisted( - () => ({ - forkMock: vi.fn(), - existsSyncMock: vi.fn(), - mkdtempSyncMock: vi.fn(() => '/tmp/orca-watcher-canary-supervisor-test'), - parcelSubscribeMock: vi.fn(), - rmSyncMock: vi.fn() - }) -) +const { + forkMock, + existsSyncMock, + mkdtempSyncMock, + parcelSubscribeMock, + rmSyncMock, + writeFileSyncMock +} = vi.hoisted(() => ({ + forkMock: vi.fn(), + existsSyncMock: vi.fn(), + mkdtempSyncMock: vi.fn(() => '/tmp/orca-watcher-canary-supervisor-test'), + parcelSubscribeMock: vi.fn(), + rmSyncMock: vi.fn(), + writeFileSyncMock: vi.fn() +})) vi.mock('node:child_process', () => ({ fork: forkMock })) vi.mock('node:fs', () => ({ existsSync: existsSyncMock, mkdtempSync: mkdtempSyncMock, - rmSync: rmSyncMock + rmSync: rmSyncMock, + writeFileSync: writeFileSyncMock })) vi.mock('@parcel/watcher', () => ({ subscribe: parcelSubscribeMock })) @@ -35,6 +42,7 @@ type SentMessage = { op: string; id: number; dir?: string } class FakeChild extends EventEmitter { connected = true + pid = 1234 sent: SentMessage[] = [] stderr = new EventEmitter() kill = vi.fn(() => { @@ -94,6 +102,19 @@ describe('subscribeViaWatcherProcess', () => { expect(callback).toHaveBeenCalledWith(null, events) }) + it('creates the fault-harness pid file without clobbering an existing path', async () => { + vi.stubEnv('ORCA_WATCHER_CHILD_PID_FILE', '/tmp/orca-watcher.pid') + + const promise = subscribeViaWatcherProcess('/repo', vi.fn(), {}) + const child = currentChild() + + expect(writeFileSyncMock).toHaveBeenCalledWith('/tmp/orca-watcher.pid', '1234', { + flag: 'wx' + }) + ackSubscribe(child) + await promise + }) + it('forwards watcher errors to the callback', async () => { const callback = vi.fn() const promise = subscribeViaWatcherProcess('/repo', callback, {}) diff --git a/src/main/ipc/parcel-watcher-supervisor-subscribe.ts b/src/main/ipc/parcel-watcher-supervisor-subscribe.ts index 3f0fd8ad4..abfa9ffe5 100644 --- a/src/main/ipc/parcel-watcher-supervisor-subscribe.ts +++ b/src/main/ipc/parcel-watcher-supervisor-subscribe.ts @@ -1,6 +1,5 @@ import type { ChildProcess } from 'node:child_process' import { existsSync } from 'node:fs' -import { getWatcherProcessEntryPath } from './parcel-watcher-entry-path' import { createHostWatcherSubscription } from './parcel-watcher-host-subscriptions' import { subscribeWithInProcessWatcher } from './parcel-watcher-in-process-fallback' import { installPendingSubscribeControls } from './parcel-watcher-pending-subscribe' @@ -22,6 +21,8 @@ type WatcherSupervisorSubscribeOptions = { opts: WatcherProcessSubscribeOptions hooks: WatcherProcessHooks shutdownRequested: boolean + entryPath: string + useInProcessVitestFallback: boolean allocateId: () => number records: Map pendingUnsubscribes: Map void> @@ -42,6 +43,8 @@ export function subscribeThroughWatcherSupervisor({ opts, hooks, shutdownRequested, + entryPath, + useInProcessVitestFallback, allocateId, records, pendingUnsubscribes, @@ -72,10 +75,9 @@ export function subscribeThroughWatcherSupervisor({ } // Why: under Vitest we cannot fork a real watcher child, so exercise the // subscription path in-process (against mocked @parcel/watcher) instead. - if (process.env.VITEST) { + if (process.env.VITEST && useInProcessVitestFallback) { return subscribeWithInProcessWatcher(dir, callback, opts, hooks) } - const entryPath = getWatcherProcessEntryPath() if (!existsSync(entryPath)) { return Promise.reject( new WatcherProcessFailure( diff --git a/src/main/ssh/ssh-relay-versioned-install.test.ts b/src/main/ssh/ssh-relay-versioned-install.test.ts index 8cc46e8b1..244ae80a4 100644 --- a/src/main/ssh/ssh-relay-versioned-install.test.ts +++ b/src/main/ssh/ssh-relay-versioned-install.test.ts @@ -108,11 +108,12 @@ describe('isRelayAlreadyInstalled', () => { ).rejects.toBe(sessionLimitError) }) - it('checks for relay.js AND .install-complete in addition to the dir', async () => { + it('checks for both relay process artifacts and .install-complete', async () => { mockExec.mockResolvedValueOnce('OK') await isRelayAlreadyInstalled(conn, '/r') const cmd = mockExec.mock.calls.at(-1)?.[1] ?? '' expect(cmd).toContain('relay.js') + expect(cmd).toContain('relay-watcher.js') expect(cmd).toContain('.install-complete') }) }) diff --git a/src/main/ssh/ssh-relay-versioned-install.ts b/src/main/ssh/ssh-relay-versioned-install.ts index 5ebac3a07..a669f0b78 100644 --- a/src/main/ssh/ssh-relay-versioned-install.ts +++ b/src/main/ssh/ssh-relay-versioned-install.ts @@ -126,10 +126,9 @@ export function computeRemoteRelayDir( /** * Probe whether a fully-installed relay already exists at remoteRelayDir. * - * "Fully installed" means: the directory exists, contains relay.js, AND - * contains the .install-complete sentinel written at the end of a successful - * install. A directory missing .install-complete is either mid-install (lock - * held) or a crashed-install partial — either way we re-run the deploy. + * "Fully installed" means: the directory contains relay.js, its isolated + * relay-watcher.js child, and the .install-complete sentinel written at the + * end of a successful install. Missing artifacts force a complete re-deploy. */ export async function isRelayAlreadyInstalled( conn: SshConnection, diff --git a/src/main/ssh/ssh-remote-commands.ts b/src/main/ssh/ssh-remote-commands.ts index 01ab8b266..339f4aef5 100644 --- a/src/main/ssh/ssh-remote-commands.ts +++ b/src/main/ssh/ssh-remote-commands.ts @@ -59,11 +59,13 @@ export function probeRelayInstalledCommand( remoteRelayDir: string ): string { const relayJs = joinRemotePath(host, remoteRelayDir, 'relay.js') + const relayWatcherJs = joinRemotePath(host, remoteRelayDir, 'relay-watcher.js') const installComplete = joinRemotePath(host, remoteRelayDir, '.install-complete') if (!isWindowsRemoteHost(host)) { return ( `test -d ${shellEscape(remoteRelayDir)} ` + `&& test -f ${shellEscape(relayJs)} ` + + `&& test -f ${shellEscape(relayWatcherJs)} ` + `&& test -f ${shellEscape(installComplete)} ` + `&& echo OK || echo MISSING` ) @@ -72,8 +74,9 @@ export function probeRelayInstalledCommand( [ `$dir = ${powerShellLiteral(remoteRelayDir)}`, `$relay = ${powerShellLiteral(relayJs)}`, + `$watcher = ${powerShellLiteral(relayWatcherJs)}`, `$complete = ${powerShellLiteral(installComplete)}`, - "if ((Test-Path -LiteralPath $dir -PathType Container) -and (Test-Path -LiteralPath $relay -PathType Leaf) -and (Test-Path -LiteralPath $complete -PathType Leaf)) { 'OK' } else { 'MISSING' }" + "if ((Test-Path -LiteralPath $dir -PathType Container) -and (Test-Path -LiteralPath $relay -PathType Leaf) -and (Test-Path -LiteralPath $watcher -PathType Leaf) -and (Test-Path -LiteralPath $complete -PathType Leaf)) { 'OK' } else { 'MISSING' }" ].join('; ') ) } diff --git a/src/relay/fs-handler.test.ts b/src/relay/fs-handler.test.ts index 7d5354cf4..63e399072 100644 --- a/src/relay/fs-handler.test.ts +++ b/src/relay/fs-handler.test.ts @@ -7,6 +7,7 @@ import * as fs from 'node:fs/promises' import * as path from 'node:path' import { mkdtempSync, writeFileSync, mkdirSync, symlinkSync } from 'node:fs' import { tmpdir } from 'node:os' +import { subscribeWithInProcessWatcher } from '../main/ipc/parcel-watcher-in-process-fallback' const { mockSubscribe } = vi.hoisted(() => ({ mockSubscribe: vi.fn() @@ -121,7 +122,11 @@ describe('FsHandler', () => { tmpDir = mkdtempSync(path.join(tmpdir(), 'relay-fs-')) dispatcher = createMockDispatcher() const ctx = new RelayContext() - handler = new FsHandler(dispatcher as unknown as RelayDispatcher, ctx) + handler = new FsHandler(dispatcher as unknown as RelayDispatcher, ctx, { + dispose: vi.fn(), + forgetRoot: vi.fn(), + subscribe: subscribeWithInProcessWatcher + }) }) afterEach(async () => { diff --git a/src/relay/fs-handler.ts b/src/relay/fs-handler.ts index 6074fad2e..38f0dec9b 100644 --- a/src/relay/fs-handler.ts +++ b/src/relay/fs-handler.ts @@ -1,5 +1,5 @@ /* eslint-disable max-lines -- Why: relay filesystem request handling shares - path expansion, file IO, search, streaming reads, Space scans, and watch lifecycle state. */ + path expansion, file IO, search, streaming reads, and Space scans. */ import { readdir, writeFile, stat, lstat, mkdir, rename, cp, rm, realpath } from 'node:fs/promises' import { execFile } from 'node:child_process' import { tmpdir } from 'node:os' @@ -34,17 +34,8 @@ import { RelayStreamRegistry } from './fs-stream-registry' import { scanWorkspaceSpaceDirectory } from './workspace-space-scan' import { buildRelayCommandEnv } from './relay-command-env' import { assertNoClobberRenameDestinationAvailable } from '../shared/filesystem-rename-collision' -import { - WATCHER_IGNORE_DIRS, - buildParcelWatcherIgnoreOptions -} from '../main/ipc/filesystem-watcher-ignore' - -type WatchState = { - rootPath: string - unwatchFn: (() => void) | null - setupPromise: Promise | null - clients: Map boolean> -} +import { RelayFilesystemWatchRegistry } from './relay-filesystem-watch-registry' +import type { RelayWatcherProcessPool } from './relay-watcher-process-pool' async function isDirectoryEntry( dirPath: string, @@ -85,15 +76,19 @@ function fileStatFromLstat(stats: Awaited>) { export class FsHandler { private dispatcher: RelayDispatcher - private watches = new Map() + private watchRegistry: RelayFilesystemWatchRegistry private streamRegistry = new RelayStreamRegistry() private listFilesScans = new ListFilesScanCoordinator() - constructor(dispatcher: RelayDispatcher, _context: RelayContext) { + constructor( + dispatcher: RelayDispatcher, + _context: RelayContext, + watcherPool?: RelayWatcherProcessPool + ) { this.dispatcher = dispatcher + this.watchRegistry = new RelayFilesystemWatchRegistry(dispatcher, watcherPool) this.registerHandlers() - this.dispatcher.onClientDetached?.((clientId) => { - this.releaseClientWatches(clientId) + this.dispatcher.onClientDetached?.(() => { // Why: a detached client's fs.streamAck frames will never arrive; wake // any pump parked on the ack window so it re-checks staleness and exits // instead of stranding its open file handle. @@ -122,8 +117,12 @@ export class FsHandler { this.dispatcher.onRequest('fs.search', (p) => this.search(p)) this.dispatcher.onRequest('fs.listFiles', (p, c) => this.listFiles(p, c)) this.dispatcher.onRequest('fs.workspaceSpaceScan', (p, c) => this.workspaceSpaceScan(p, c)) - this.dispatcher.onRequest('fs.watch', (p, context) => this.watch(p, context)) - this.dispatcher.onNotification('fs.unwatch', (p, context) => this.unwatch(p, context)) + this.dispatcher.onRequest('fs.watch', (p, context) => + this.watchRegistry.watch(expandTilde(p.rootPath as string), context) + ) + this.dispatcher.onNotification('fs.unwatch', (p, context) => + this.watchRegistry.unwatch(expandTilde(p.rootPath as string), context) + ) this.dispatcher.onNotification('fs.cancelStream', (p) => this.cancelStream(p)) this.dispatcher.onNotification('fs.streamAck', (p) => this.streamAck(p)) } @@ -414,125 +413,8 @@ export class FsHandler { return scanWorkspaceSpaceDirectory(rootPath, context) } - private async watch(params: Record, context?: RequestContext) { - const rootPath = expandTilde(params.rootPath as string) - - this.releaseStaleWatches() - - const existing = this.watches.get(rootPath) - if (existing) { - if ([...existing.clients.values()].some((isStale) => !isStale())) { - existing.clients.set(context?.clientId ?? 0, context?.isStale ?? (() => false)) - if (existing.setupPromise) { - await existing.setupPromise - } - return - } - existing.unwatchFn?.() - this.watches.delete(rootPath) - } - - if (this.watches.size >= 20) { - throw new Error('Maximum number of file watchers reached') - } - - const watchState: WatchState = { - rootPath, - unwatchFn: null, - setupPromise: null, - clients: new Map([[context?.clientId ?? 0, context?.isStale ?? (() => false)]]) - } - this.watches.set(rootPath, watchState) - - const setupPromise = (async () => { - const watcher = await import('@parcel/watcher') - const subscription = await watcher.subscribe( - rootPath, - (err, events) => { - if (err) { - this.dispatcher.notify('fs.changed', { - events: [{ kind: 'overflow', absolutePath: rootPath }] - }) - return - } - const mapped = events.map((evt) => ({ - kind: evt.type, - absolutePath: evt.path - })) - this.dispatcher.notify('fs.changed', { events: mapped }) - }, - // Why: align remote watchers with the shared nested exclusion so - // generated trees neither exhaust inotify nor trigger slow glob regexes. - buildParcelWatcherIgnoreOptions(WATCHER_IGNORE_DIRS) - ) - watchState.unwatchFn = () => { - void subscription.unsubscribe() - } - if ( - [...watchState.clients.values()].every((isStale) => isStale()) || - this.watches.get(rootPath) !== watchState - ) { - // Why: if the only requesting client reconnects while watcher setup is - // in flight, no client can later balance it with fs.unwatch. Tear down - // only this request's subscription so a newer replacement watch for the - // same root is not removed. - void subscription.unsubscribe() - if (this.watches.get(rootPath) === watchState) { - this.watches.delete(rootPath) - } - } - })() - watchState.setupPromise = setupPromise - - try { - await setupPromise - } catch { - if (this.watches.get(rootPath) === watchState) { - this.watches.delete(rootPath) - } - // @parcel/watcher not available -- polling fallback would go here - process.stderr.write('[relay] File watcher not available, fs.changed events disabled\n') - } - } - - private unwatch(params: Record, context?: RequestContext): void { - const rootPath = expandTilde(params.rootPath as string) - const state = this.watches.get(rootPath) - if (state) { - this.releaseWatchClient(rootPath, state, context?.clientId ?? 0) - } - } - - private releaseClientWatches(clientId: number): void { - for (const [rootPath, state] of this.watches) { - this.releaseWatchClient(rootPath, state, clientId) - } - } - - private releaseStaleWatches(): void { - for (const [rootPath, state] of this.watches) { - if ([...state.clients.values()].some((isStale) => !isStale())) { - continue - } - state.unwatchFn?.() - this.watches.delete(rootPath) - } - } - - private releaseWatchClient(rootPath: string, state: WatchState, clientId: number): void { - state.clients.delete(clientId) - if (state.clients.size > 0) { - return - } - state.unwatchFn?.() - this.watches.delete(rootPath) - } - dispose(): void { - for (const [, state] of this.watches) { - state.unwatchFn?.() - } - this.watches.clear() + this.watchRegistry.dispose() void this.streamRegistry.disposeAll() } } diff --git a/src/relay/relay-filesystem-watch-registry.test.ts b/src/relay/relay-filesystem-watch-registry.test.ts new file mode 100644 index 000000000..5f1e46eec --- /dev/null +++ b/src/relay/relay-filesystem-watch-registry.test.ts @@ -0,0 +1,170 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { WatcherProcessFailure } from '../main/ipc/parcel-watcher-process-failure' +import type { + WatcherProcessCallback, + WatcherProcessHooks, + WatcherProcessSubscription +} from '../main/ipc/parcel-watcher-process-subscription' +import type { RelayDispatcher, RequestContext } from './dispatcher' +import { RelayFilesystemWatchRegistry } from './relay-filesystem-watch-registry' +import { createRelayWatcherProcessPool } from './relay-watcher-process-pool' + +type InstalledWatch = { + callback: WatcherProcessCallback + hooks: WatcherProcessHooks + unsubscribe: ReturnType Promise>> +} + +class FakeWatcherPool { + readonly installed: InstalledWatch[] = [] + readonly dispose = vi.fn() + readonly forgetRoot = vi.fn() + + async subscribe( + _rootPath: string, + callback: WatcherProcessCallback, + _options: object, + hooks: WatcherProcessHooks + ): Promise { + const unsubscribe = vi.fn(async () => undefined) + this.installed.push({ callback, hooks, unsubscribe }) + return { unsubscribe } + } +} + +function createDispatcher() { + const detached = new Set<(clientId: number) => void>() + return { + notify: vi.fn(), + onClientDetached: vi.fn((listener: (clientId: number) => void) => { + detached.add(listener) + return () => detached.delete(listener) + }) + } +} + +function context(clientId: number): RequestContext { + return { clientId, isStale: () => false } +} + +describe('RelayFilesystemWatchRegistry', () => { + let dispatcher: ReturnType + let pool: FakeWatcherPool + let registry: RelayFilesystemWatchRegistry + + beforeEach(() => { + dispatcher = createDispatcher() + pool = new FakeWatcherPool() + registry = new RelayFilesystemWatchRegistry(dispatcher as unknown as RelayDispatcher, pool) + }) + + it('emits overflow around child replacement and resumes ordered event delivery', async () => { + await registry.watch('/repo', context(1)) + const first = pool.installed[0] + + first.callback(null, [{ type: 'create', path: '/repo/before.txt' }]) + first.hooks.onInterruption?.() + first.callback(null, [{ type: 'update', path: '/repo/after-resubscribe.txt' }]) + + expect(dispatcher.notify.mock.calls).toEqual([ + ['fs.changed', { events: [{ kind: 'create', absolutePath: '/repo/before.txt' }] }], + ['fs.changed', { events: [{ kind: 'overflow', absolutePath: '/repo' }] }], + ['fs.changed', { events: [{ kind: 'update', absolutePath: '/repo/after-resubscribe.txt' }] }] + ]) + }) + + it('moves a terminal shard failure into recovery without dropping shared clients', async () => { + await registry.watch('/repo', context(1)) + await registry.watch('/repo', context(2)) + const first = pool.installed[0] + first.hooks.onTerminalError?.( + new WatcherProcessFailure( + 'file watcher process crashed repeatedly', + 'supervisor', + 'supervisor_crash_fuse' + ) + ) + await Promise.resolve() + + expect(pool.installed).toHaveLength(2) + pool.installed[1].callback(null, [{ type: 'create', path: '/repo/recovered.txt' }]) + expect(dispatcher.notify).toHaveBeenNthCalledWith(1, 'fs.changed', { + events: [{ kind: 'overflow', absolutePath: '/repo' }] + }) + expect(dispatcher.notify).toHaveBeenNthCalledWith(2, 'fs.changed', { + events: [{ kind: 'create', absolutePath: '/repo/recovered.txt' }] + }) + + registry.unwatch('/repo', context(1)) + expect(pool.installed[1].unsubscribe).not.toHaveBeenCalled() + registry.unwatch('/repo', context(2)) + expect(pool.installed[1].unsubscribe).toHaveBeenCalledTimes(1) + }) + + it('aborts a pending crawl only after the last same-root client leaves', async () => { + let sharedSignal: AbortSignal | undefined + let rejectSubscribe: ((error: Error) => void) | undefined + vi.spyOn(pool, 'subscribe').mockImplementation( + (_rootPath, _callback, _options, hooks): Promise => + new Promise((_resolve, reject) => { + sharedSignal = hooks.signal + rejectSubscribe = reject + hooks.signal?.addEventListener( + 'abort', + () => + reject( + new WatcherProcessFailure( + 'file watcher subscription aborted', + 'subscription', + 'subscribe_aborted' + ) + ), + { once: true } + ) + }) + ) + const firstAbort = new AbortController() + const secondAbort = new AbortController() + const first = registry.watch('/repo', { + ...context(1), + signal: firstAbort.signal + }) + const second = registry.watch('/repo', { + ...context(2), + signal: secondAbort.signal + }) + + firstAbort.abort() + await first + expect(sharedSignal?.aborted).toBe(false) + + secondAbort.abort() + await second + expect(sharedSignal?.aborted).toBe(true) + expect(rejectSubscribe).toBeDefined() + }) +}) + +describe('createRelayWatcherProcessPool', () => { + it('fails closed instead of loading the native watcher in the relay process', async () => { + const previousVitest = process.env.VITEST + process.env.VITEST = 'true' + const pool = createRelayWatcherProcessPool( + join(tmpdir(), `missing-relay-watcher-${process.pid}.js`) + ) + try { + await expect(pool.subscribe('/repo', vi.fn(), {}, {})).rejects.toMatchObject({ + code: 'entry_missing' + }) + } finally { + pool.dispose() + if (previousVitest === undefined) { + delete process.env.VITEST + } else { + process.env.VITEST = previousVitest + } + } + }) +}) diff --git a/src/relay/relay-filesystem-watch-registry.ts b/src/relay/relay-filesystem-watch-registry.ts new file mode 100644 index 000000000..a1ccda058 --- /dev/null +++ b/src/relay/relay-filesystem-watch-registry.ts @@ -0,0 +1,312 @@ +import type { RelayDispatcher, RequestContext } from './dispatcher' +import { MAX_BATCHED_WATCHER_EVENTS } from '../main/ipc/filesystem-watcher-event-batch' +import { isWatcherProcessFailure } from '../main/ipc/parcel-watcher-process-failure' +import type { + WatcherProcessEvent, + WatcherProcessSubscription +} from '../main/ipc/parcel-watcher-process' +import { + WATCHER_IGNORE_DIRS, + buildParcelWatcherIgnoreOptions +} from '../main/ipc/filesystem-watcher-ignore' +import { + createRelayWatcherProcessPool, + type RelayWatcherProcessPool +} from './relay-watcher-process-pool' + +const MAX_RELAY_WATCH_ROOTS = 20 +const RELAY_WATCH_CRAWL_TIMEOUT_MS = 60_000 +const RELAY_WATCH_OPTIONS = buildParcelWatcherIgnoreOptions(WATCHER_IGNORE_DIRS) + +type RelayWatchState = { + rootPath: string + clients: Map boolean> + setupPromise: Promise + subscription: WatcherProcessSubscription | null + abortController: AbortController + generation: number + closed: boolean +} + +function overflowEvent(rootPath: string): Record { + return { events: [{ kind: 'overflow', absolutePath: rootPath }] } +} + +function createWatchAbortError(): Error { + const error = new Error('Request "fs.watch" was cancelled') + error.name = 'AbortError' + return error +} + +function shouldRetryInitialWatch(error: unknown): boolean { + return ( + isWatcherProcessFailure(error) && + error.code !== 'entry_missing' && + error.code !== 'subscribe_aborted' && + error.code !== 'supervisor_disposed' && + (error.scope === 'supervisor' || error.code === 'subscribe_timeout') + ) +} + +export class RelayFilesystemWatchRegistry { + private readonly watches = new Map() + + constructor( + private readonly dispatcher: RelayDispatcher, + private readonly watcherPool: RelayWatcherProcessPool = createRelayWatcherProcessPool() + ) { + this.dispatcher.onClientDetached?.((clientId) => this.releaseClientWatches(clientId)) + } + + async watch(rootPath: string, context?: RequestContext): Promise { + this.releaseStaleWatches() + const clientId = context?.clientId ?? 0 + const isStale = context?.isStale ?? (() => false) + const existing = this.watches.get(rootPath) + if (existing) { + existing.clients.set(clientId, isStale) + await this.awaitSetupForClient(existing, clientId, context) + return + } + + if (this.watches.size >= MAX_RELAY_WATCH_ROOTS) { + throw new Error('Maximum number of file watchers reached') + } + + const state: RelayWatchState = { + rootPath, + clients: new Map([[clientId, isStale]]), + setupPromise: Promise.resolve(), + subscription: null, + abortController: new AbortController(), + generation: 0, + closed: false + } + this.watches.set(rootPath, state) + state.setupPromise = this.startInitialWatch(state) + await this.awaitSetupForClient(state, clientId, context) + } + + unwatch(rootPath: string, context?: RequestContext): void { + const state = this.watches.get(rootPath) + if (state) { + this.releaseWatchClient(state, context?.clientId ?? 0) + } + } + + dispose(): void { + for (const state of Array.from(this.watches.values())) { + this.closeWatch(state) + } + this.watcherPool.dispose() + } + + private async startInitialWatch(state: RelayWatchState): Promise { + try { + await this.subscribeState(state) + } catch (firstError) { + if (!state.closed && shouldRetryInitialWatch(firstError)) { + try { + await this.subscribeState(state) + this.emitOverflow(state) + return + } catch (quarantineError) { + this.closeWatch(state) + throw quarantineError + } + } + this.closeWatch(state) + throw firstError + } + } + + private subscribeState(state: RelayWatchState): Promise { + const generation = ++state.generation + const emitOverflow = (): void => { + if (state.generation === generation) { + this.emitOverflow(state) + } + } + return this.watcherPool + .subscribe( + state.rootPath, + (error, events) => { + if (state.closed || state.generation !== generation) { + return + } + if (error) { + process.stderr.write( + `[relay] File watcher error for ${state.rootPath}: ${error.message}\n` + ) + emitOverflow() + return + } + this.emitEvents(state, events) + }, + RELAY_WATCH_OPTIONS, + { + delivery: { maxEventsPerBatch: MAX_BATCHED_WATCHER_EVENTS }, + onInterruption: emitOverflow, + onOverflow: emitOverflow, + onTerminalError: (error) => this.recoverWatch(state, generation, error), + signal: state.abortController.signal, + subscribeTimeoutMs: RELAY_WATCH_CRAWL_TIMEOUT_MS + } + ) + .then(async (subscription) => { + if ( + state.closed || + state.generation !== generation || + this.watches.get(state.rootPath) !== state + ) { + await subscription.unsubscribe() + return + } + state.subscription = subscription + }) + } + + private recoverWatch(state: RelayWatchState, failedGeneration: number, error: Error): void { + if (state.closed || state.generation !== failedGeneration) { + return + } + state.subscription = null + this.emitOverflow(state) + const recovery = this.subscribeState(state) + state.setupPromise = recovery + void recovery.catch((recoveryError: unknown) => { + if (!state.closed) { + const message = recoveryError instanceof Error ? recoveryError.message : error.message + process.stderr.write( + `[relay] File watcher disabled after bounded recovery for ${state.rootPath}: ${message}\n` + ) + this.closeWatch(state) + } + }) + } + + private emitEvents(state: RelayWatchState, events: readonly WatcherProcessEvent[]): void { + if (state.closed || events.length === 0) { + return + } + this.dispatcher.notify('fs.changed', { + events: events.map((event) => ({ + kind: event.type, + absolutePath: event.path, + ...(event.isDirectory === undefined ? {} : { isDirectory: event.isDirectory }) + })) + }) + } + + private emitOverflow(state: RelayWatchState): void { + if (!state.closed) { + this.dispatcher.notify('fs.changed', overflowEvent(state.rootPath)) + } + } + + private async awaitSetupForClient( + state: RelayWatchState, + clientId: number, + context?: RequestContext + ): Promise { + try { + await this.awaitSetupWithAbort(state.setupPromise, context?.signal) + } catch (error) { + this.releaseWatchClient(state, clientId) + const expectedAbort = + (error instanceof Error && error.name === 'AbortError') || + (isWatcherProcessFailure(error) && error.code === 'subscribe_aborted') + if (!expectedAbort && error instanceof Error) { + process.stderr.write( + `[relay] File watcher not available for ${state.rootPath}: ${error.message}\n` + ) + throw error + } + return + } + if (context?.isStale()) { + this.releaseWatchClient(state, clientId) + } + } + + private awaitSetupWithAbort(setupPromise: Promise, signal?: AbortSignal): Promise { + if (!signal) { + return setupPromise + } + if (signal.aborted) { + return Promise.reject(createWatchAbortError()) + } + return new Promise((resolve, reject) => { + const onAbort = (): void => { + cleanup() + reject(createWatchAbortError()) + } + const cleanup = (): void => signal.removeEventListener('abort', onAbort) + signal.addEventListener('abort', onAbort, { once: true }) + setupPromise.then( + () => { + cleanup() + resolve() + }, + (error) => { + cleanup() + reject(error) + } + ) + }) + } + + private releaseClientWatches(clientId: number): void { + for (const state of Array.from(this.watches.values())) { + this.releaseWatchClient(state, clientId) + } + } + + private releaseStaleWatches(): void { + for (const state of Array.from(this.watches.values())) { + for (const [clientId, isStale] of state.clients) { + if (isStale()) { + state.clients.delete(clientId) + } + } + if (state.clients.size === 0) { + this.closeWatch(state) + } + } + } + + private releaseWatchClient(state: RelayWatchState, clientId: number): void { + state.clients.delete(clientId) + if (!state.closed && state.clients.size === 0) { + this.closeWatch(state) + } + } + + private closeWatch(state: RelayWatchState): void { + if (state.closed) { + return + } + state.closed = true + state.generation++ + state.abortController.abort() + state.clients.clear() + if (this.watches.get(state.rootPath) === state) { + this.watches.delete(state.rootPath) + } + const subscription = state.subscription + state.subscription = null + if (subscription) { + // Why: a child can die during unwatch; release quarantine history even + // when physical teardown reports that already-contained failure. + void subscription.unsubscribe().then( + () => this.watcherPool.forgetRoot(state.rootPath), + () => this.watcherPool.forgetRoot(state.rootPath) + ) + return + } + void state.setupPromise.then( + () => this.watcherPool.forgetRoot(state.rootPath), + () => this.watcherPool.forgetRoot(state.rootPath) + ) + } +} diff --git a/src/relay/relay-watcher-process-pool.ts b/src/relay/relay-watcher-process-pool.ts new file mode 100644 index 000000000..af6cb8285 --- /dev/null +++ b/src/relay/relay-watcher-process-pool.ts @@ -0,0 +1,26 @@ +import { join } from 'node:path' +import { RuntimeWatcherProcessPool } from '../main/ipc/runtime-watcher-process-pool' +import { WatcherProcessSupervisor } from '../main/ipc/parcel-watcher-process-supervisor' + +export type RelayWatcherProcessPool = Pick< + RuntimeWatcherProcessPool, + 'dispose' | 'forgetRoot' | 'subscribe' +> + +export function getRelayWatcherProcessEntryPath(): string { + return join(__dirname, 'relay-watcher.js') +} + +export function createRelayWatcherProcessPool( + entryPath = getRelayWatcherProcessEntryPath() +): RelayWatcherProcessPool { + return new RuntimeWatcherProcessPool({ + createSupervisor: () => + new WatcherProcessSupervisor({ + entryPath, + // Why: a leaked VITEST environment must never move the native addon + // back into the relay when its crash-isolation child is missing. + useInProcessVitestFallback: false + }) + }) +} diff --git a/src/relay/subprocess.test.ts b/src/relay/subprocess.test.ts index 4d19542b9..219addc02 100644 --- a/src/relay/subprocess.test.ts +++ b/src/relay/subprocess.test.ts @@ -10,6 +10,7 @@ import { getEndpointFileName } from '../shared/agent-hook-listener' import { relayTestSocketPath } from './relay-test-socket-path' const RELAY_TS_ENTRY = path.resolve(__dirname, 'relay.ts') +const WATCHER_TS_ENTRY = path.resolve(__dirname, '../main/ipc/parcel-watcher-process-entry.ts') let bundleDir: string let relayEntry: string const spawnedSocketDirs: string[] = [] @@ -24,7 +25,17 @@ beforeAll(async () => { target: 'node18', format: 'cjs', outfile: relayEntry, - external: ['node-pty', '@parcel/watcher'], + external: ['node-pty', '@parcel/watcher', 'electron'], + sourcemap: false + }) + await build({ + entryPoints: [WATCHER_TS_ENTRY], + bundle: true, + platform: 'node', + target: 'node18', + format: 'cjs', + outfile: path.join(bundleDir, 'relay-watcher.js'), + external: ['@parcel/watcher'], sourcemap: false }) }, 30_000)