From bc5d410f3a23d2bca7c32b389f78ebfc314f31f9 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:25:40 -0400 Subject: [PATCH] fix(fs): use native events for WSL watcher Replace periodic WSL snapshots with a recursive inotify delta stream so nested external edits are not depth-dependent. Keep full recursive snapshots as a compatibility fallback and restart active subscriptions after the WSL process exits. --- src/main/ipc/filesystem-watcher-wsl-engine.ts | 255 ++++++++++++ .../ipc/filesystem-watcher-wsl-live.test.ts | 57 +++ .../filesystem-watcher-wsl-native-script.ts | 173 ++++++++ .../ipc/filesystem-watcher-wsl-snapshot.ts | 91 +++++ src/main/ipc/filesystem-watcher-wsl.test.ts | 228 ++++++----- src/main/ipc/filesystem-watcher-wsl.ts | 369 +++++------------- src/main/ipc/filesystem-watcher.ts | 7 +- 7 files changed, 816 insertions(+), 364 deletions(-) create mode 100644 src/main/ipc/filesystem-watcher-wsl-engine.ts create mode 100644 src/main/ipc/filesystem-watcher-wsl-live.test.ts create mode 100644 src/main/ipc/filesystem-watcher-wsl-native-script.ts create mode 100644 src/main/ipc/filesystem-watcher-wsl-snapshot.ts diff --git a/src/main/ipc/filesystem-watcher-wsl-engine.ts b/src/main/ipc/filesystem-watcher-wsl-engine.ts new file mode 100644 index 00000000000..dcdcb74efe1 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-wsl-engine.ts @@ -0,0 +1,255 @@ +import { spawn } from 'node:child_process' +import { StringDecoder } from 'node:string_decoder' +import type { Event as WatcherEvent } from '@parcel/watcher' +import { buildWslNativeWatcherScript } from './filesystem-watcher-wsl-native-script' +import { + buildSnapshotScript, + diffSnapshots, + parseSnapshotFrame, + SNAPSHOT_END, + SNAPSHOT_START, + toWslUncPath, + type WslSnapshot +} from './filesystem-watcher-wsl-snapshot' + +export type WslWatchEngine = { + ready: Promise + stopped: Promise + stop(): void +} + +export type WslEngineContext = { + distro: string + linuxPath: string + worktreePath: string + ignoreDirs: readonly string[] + onEvents: (events: WatcherEvent[]) => void + onOverflow: () => void +} + +const NATIVE_STARTUP_TIMEOUT_MS = 10_000 +const SNAPSHOT_STARTUP_TIMEOUT_MS = 30_000 +const MAX_STREAM_BUFFER_CHARS = 10 * 1024 * 1024 + +function createChildEngine( + args: string[], + stdin: string, + startupTimeoutMs: number, + worktreePath: string, + handleStdout: (chunk: Buffer, settleReady: (error?: Error) => void, stop: () => void) => void +): WslWatchEngine { + const child = spawn('wsl.exe', args, { + stdio: ['pipe', 'pipe', 'pipe'], + windowsHide: true + }) + let disposed = false + let readySettled = false + let stoppedSettled = false + let stderrTail = '' + const stderrDecoder = new StringDecoder('utf8') + let resolveReady!: () => void + let rejectReady!: (error: Error) => void + let resolveStopped!: () => void + const ready = new Promise((resolve, reject) => { + resolveReady = resolve + rejectReady = reject + }) + const stopped = new Promise((resolve) => { + resolveStopped = resolve + }) + const settleReady = (error?: Error): void => { + if (readySettled) { + return + } + readySettled = true + clearTimeout(startupTimer) + if (error) { + rejectReady(error) + } else { + resolveReady() + } + } + const settleStopped = (): void => { + if (stoppedSettled) { + return + } + stoppedSettled = true + resolveStopped() + } + const stop = (): void => { + if (disposed) { + return + } + disposed = true + child.kill() + } + const startupTimer = setTimeout(() => { + settleReady(new Error(`Timed out starting WSL watcher for ${worktreePath}`)) + stop() + }, startupTimeoutMs) + + child.stdin.on('error', (error) => { + if (!readySettled) { + settleReady(error) + } + }) + child.stdout.on('data', (chunk: Buffer) => { + if (!disposed) { + handleStdout(chunk, settleReady, stop) + } + }) + child.stderr.on('data', (chunk: Buffer) => { + stderrTail = (stderrTail + stderrDecoder.write(chunk)).slice(-4096) + }) + child.stdout.on('error', (error) => { + if (!readySettled) { + settleReady(error) + } else if (!disposed) { + stop() + } + }) + child.stderr.on('error', () => undefined) + child.once('error', (error) => { + if (!readySettled) { + settleReady(error) + } else if (!disposed) { + stop() + } + settleStopped() + }) + child.once('close', (code, signal) => { + if (!readySettled) { + const suffix = stderrTail.trim() ? `: ${stderrTail.trim()}` : '' + settleReady(new Error(`WSL watcher exited before ready (${code ?? signal})${suffix}`)) + } + settleStopped() + }) + child.stdin.end(stdin) + return { ready, stopped, stop } +} + +type NativeMessage = + | { type: 'ready' } + | { type: 'overflow' } + | { type: 'error'; message?: unknown } + | { type: 'events'; events?: unknown } + +function nativeEvents(message: NativeMessage, context: WslEngineContext): WatcherEvent[] { + if (message.type !== 'events' || !Array.isArray(message.events)) { + return [] + } + const rootPrefix = context.linuxPath === '/' ? '/' : `${context.linuxPath.replace(/\/+$/, '')}/` + const events: WatcherEvent[] = [] + for (const item of message.events) { + if ( + Array.isArray(item) && + (item[0] === 'create' || item[0] === 'update' || item[0] === 'delete') && + typeof item[1] === 'string' && + (item[1] === context.linuxPath || item[1].startsWith(rootPrefix)) + ) { + events.push({ type: item[0], path: toWslUncPath(item[1], context.distro) }) + } + } + return events +} + +export function createWslNativeEngine(context: WslEngineContext): WslWatchEngine { + let streamBuffer = '' + const decoder = new StringDecoder('utf8') + const args = [ + '-d', + context.distro, + '--', + 'python3', + '-u', + '-', + context.linuxPath, + ...context.ignoreDirs + ] + return createChildEngine( + args, + buildWslNativeWatcherScript(), + NATIVE_STARTUP_TIMEOUT_MS, + context.worktreePath, + (chunk, settleReady, stop) => { + streamBuffer += decoder.write(chunk) + if (streamBuffer.length > MAX_STREAM_BUFFER_CHARS) { + context.onOverflow() + stop() + return + } + let newline = streamBuffer.indexOf('\n') + while (newline !== -1) { + const line = streamBuffer.slice(0, newline) + streamBuffer = streamBuffer.slice(newline + 1) + try { + const message = JSON.parse(line) as NativeMessage + if (message.type === 'ready') { + settleReady() + } else if (message.type === 'overflow') { + context.onOverflow() + } else if (message.type === 'error') { + settleReady(new Error(String(message.message ?? 'native WSL watcher failed'))) + stop() + } else { + const events = nativeEvents(message, context) + if (events.length > 0) { + context.onEvents(events) + } + } + } catch (error) { + context.onOverflow() + settleReady(error instanceof Error ? error : new Error(String(error))) + stop() + return + } + newline = streamBuffer.indexOf('\n') + } + } + ) +} + +export function createWslSnapshotEngine(context: WslEngineContext): WslWatchEngine { + let streamBuffer = '' + let previous: WslSnapshot | null = null + const decoder = new StringDecoder('utf8') + return createChildEngine( + ['-d', context.distro, '--', 'sh', '-s', '--', context.linuxPath], + buildSnapshotScript(context.ignoreDirs), + SNAPSHOT_STARTUP_TIMEOUT_MS, + context.worktreePath, + (chunk, settleReady, stop) => { + streamBuffer += decoder.write(chunk) + while (true) { + const start = streamBuffer.indexOf(SNAPSHOT_START) + if (start === -1) { + streamBuffer = streamBuffer.slice(-1) + return + } + if (start > 0) { + streamBuffer = streamBuffer.slice(start) + } + const end = streamBuffer.indexOf(SNAPSHOT_END, 1) + if (end === -1) { + if (streamBuffer.length > MAX_STREAM_BUFFER_CHARS) { + context.onOverflow() + stop() + } + return + } + const next = parseSnapshotFrame(streamBuffer.slice(1, end), context.distro) + streamBuffer = streamBuffer.slice(end + 1) + if (!previous) { + previous = next + settleReady() + } else { + const events = diffSnapshots(previous, next) + previous = next + if (events.length > 0) { + context.onEvents(events) + } + } + } + } + ) +} diff --git a/src/main/ipc/filesystem-watcher-wsl-live.test.ts b/src/main/ipc/filesystem-watcher-wsl-live.test.ts new file mode 100644 index 00000000000..92ac849d420 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-wsl-live.test.ts @@ -0,0 +1,57 @@ +import { execFile } from 'node:child_process' +import { promisify } from 'node:util' +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' +import { createWslWatcher, type WatchedRoot } from './filesystem-watcher-wsl' + +const execFileAsync = promisify(execFile) +const runLive = process.platform === 'win32' && process.env.ORCA_WSL_WATCHER_LIVE === '1' +const fixturePath = `/tmp/orca-wsl-watcher-${process.pid}` +let distro = '' + +async function runWsl(...args: string[]): Promise { + const result = await execFileAsync('wsl.exe', ['--', ...args], { + encoding: 'utf8' + }) + return result.stdout +} + +describe.skipIf(!runLive)('WSL watcher live integration', () => { + beforeAll(async () => { + distro = (await runWsl('printenv', 'WSL_DISTRO_NAME')).trim() + await runWsl('mkdir', '-p', `${fixturePath}/docs/deep`) + await runWsl('touch', `${fixturePath}/docs/deep/README.md`) + }) + + afterAll(async () => { + // Why: keep recursive cleanup constrained to the unique test directory. + if (fixturePath.startsWith('/tmp/orca-wsl-watcher-')) { + await runWsl('rm', '-rf', '--', fixturePath) + } + }) + + it('reports a deep external edit without periodic polling events', async () => { + const rootPath = `\\\\wsl.localhost\\${distro}${fixturePath.replace(/\//g, '\\')}` + const batches: WatchedRoot[] = [] + const root = await createWslWatcher(rootPath, rootPath, { + ignoreDirs: ['node_modules', '.git'], + scheduleBatchFlush: (_rootKey, watchedRoot) => batches.push(watchedRoot) + }) + try { + await new Promise((resolve) => setTimeout(resolve, 2_500)) + expect(batches).toHaveLength(0) + + await runWsl('touch', `${fixturePath}/docs/deep/README.md`) + await vi.waitFor( + () => { + expect(root.batch.events).toContainEqual({ + type: 'update', + path: `${rootPath}\\docs\\deep\\README.md` + }) + }, + { timeout: 5_000 } + ) + } finally { + await root.subscription.unsubscribe() + } + }) +}) diff --git a/src/main/ipc/filesystem-watcher-wsl-native-script.ts b/src/main/ipc/filesystem-watcher-wsl-native-script.ts new file mode 100644 index 00000000000..2fd9d321fa3 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-wsl-native-script.ts @@ -0,0 +1,173 @@ +const SAFE_IGNORE_NAME = /^[A-Za-z0-9_.-]+$/ + +export function validateWslWatcherIgnoreDirs(ignoreDirs: readonly string[]): void { + for (const name of ignoreDirs) { + if (!SAFE_IGNORE_NAME.test(name)) { + throw new Error(`Unsupported WSL watcher ignore name: ${name}`) + } + } +} + +/** + * Build a third-party-package-free recursive watcher for the Linux side of WSL. + * + * The script uses Python's standard library to call inotify directly. It emits + * newline-delimited JSON so arbitrary Linux path characters stay framed while + * only change deltas cross the WSL boundary. + */ +export function buildWslNativeWatcherScript(): string { + return String.raw`import ctypes +import json +import os +import select +import struct +import sys + +IN_MODIFY = 0x00000002 +IN_ATTRIB = 0x00000004 +IN_CLOSE_WRITE = 0x00000008 +IN_MOVED_FROM = 0x00000040 +IN_MOVED_TO = 0x00000080 +IN_CREATE = 0x00000100 +IN_DELETE = 0x00000200 +IN_DELETE_SELF = 0x00000400 +IN_MOVE_SELF = 0x00000800 +IN_Q_OVERFLOW = 0x00004000 +IN_IGNORED = 0x00008000 +IN_ISDIR = 0x40000000 +WATCH_MASK = (IN_MODIFY | IN_ATTRIB | IN_CLOSE_WRITE | IN_MOVED_FROM | + IN_MOVED_TO | IN_CREATE | IN_DELETE | IN_DELETE_SELF | IN_MOVE_SELF) +MAX_EVENTS = 5000 +EVENT = struct.Struct("iIII") + +root = os.path.abspath(sys.argv[1]) +ignored = set(sys.argv[2:]) +libc = ctypes.CDLL(None, use_errno=True) +libc.inotify_init1.argtypes = [ctypes.c_int] +libc.inotify_init1.restype = ctypes.c_int +libc.inotify_add_watch.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_uint32] +libc.inotify_add_watch.restype = ctypes.c_int +libc.inotify_rm_watch.argtypes = [ctypes.c_int, ctypes.c_int] +libc.inotify_rm_watch.restype = ctypes.c_int + +def emit(message): + sys.stdout.write(json.dumps(message, ensure_ascii=True, separators=(",", ":")) + "\n") + sys.stdout.flush() + +def fail(message): + emit({"type": "error", "message": str(message)}) + raise SystemExit(1) + +fd = libc.inotify_init1(os.O_CLOEXEC | os.O_NONBLOCK) +if fd < 0: + fail(OSError(ctypes.get_errno(), os.strerror(ctypes.get_errno()))) + +watch_paths = {} +path_watches = {} + +def add_watch(path): + if path in path_watches: + return + wd = libc.inotify_add_watch(fd, os.fsencode(path), WATCH_MASK) + if wd < 0: + error = ctypes.get_errno() + raise OSError(error, os.strerror(error), path) + old_path = watch_paths.get(wd) + if old_path is not None: + path_watches.pop(old_path, None) + watch_paths[wd] = path + path_watches[path] = wd + +def add_tree(path): + pending = [path] + while pending: + directory = pending.pop() + add_watch(directory) + try: + with os.scandir(directory) as entries: + for entry in entries: + if entry.name in ignored: + continue + try: + if entry.is_dir(follow_symlinks=False): + pending.append(entry.path) + except OSError: + pass + except FileNotFoundError: + pass + +def forget_watch(wd): + path = watch_paths.pop(wd, None) + if path is not None: + path_watches.pop(path, None) + +def remove_tree(path): + prefix = path + os.sep + for watched_path, wd in list(path_watches.items()): + if watched_path != path and not watched_path.startswith(prefix): + continue + libc.inotify_rm_watch(fd, wd) + forget_watch(wd) + +def event_type(mask): + if mask & (IN_DELETE | IN_MOVED_FROM | IN_DELETE_SELF | IN_MOVE_SELF): + return "delete" + if mask & (IN_CREATE | IN_MOVED_TO): + return "create" + if mask & (IN_MODIFY | IN_ATTRIB | IN_CLOSE_WRITE): + return "update" + return None + +try: + if not os.path.isdir(root): + fail("watch root is not a directory: " + root) + add_tree(root) + emit({"type": "ready"}) + while True: + select.select([fd], [], []) + data = os.read(fd, 1024 * 1024) + offset = 0 + changes = [] + overflowed = False + while offset + EVENT.size <= len(data): + wd, mask, cookie, name_length = EVENT.unpack_from(data, offset) + offset += EVENT.size + raw_name = data[offset:offset + name_length].split(b"\0", 1)[0] + offset += name_length + if mask & IN_Q_OVERFLOW: + overflowed = True + continue + directory = watch_paths.get(wd) + if directory is None: + continue + if mask & IN_IGNORED: + forget_watch(wd) + continue + name = os.fsdecode(raw_name) if raw_name else "" + if name in ignored: + continue + path = os.path.join(directory, name) if name else directory + if mask & IN_ISDIR and mask & (IN_DELETE | IN_MOVED_FROM): + remove_tree(path) + if mask & IN_ISDIR and mask & (IN_CREATE | IN_MOVED_TO): + add_tree(path) + kind = event_type(mask) + if kind is not None: + changes.append([kind, path]) + if len(changes) > MAX_EVENTS: + changes = [] + overflowed = True + if path == root and mask & (IN_DELETE_SELF | IN_MOVE_SELF): + overflowed = True + emit({"type": "overflow"}) + raise SystemExit(1) + if overflowed: + emit({"type": "overflow"}) + elif changes: + emit({"type": "events", "events": changes}) +except (OSError, ValueError) as error: + fail(error) +finally: + os.close(fd) +` +} diff --git a/src/main/ipc/filesystem-watcher-wsl-snapshot.ts b/src/main/ipc/filesystem-watcher-wsl-snapshot.ts new file mode 100644 index 00000000000..5672274ff74 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-wsl-snapshot.ts @@ -0,0 +1,91 @@ +import type { Event as WatcherEvent } from '@parcel/watcher' +import { validateWslWatcherIgnoreDirs } from './filesystem-watcher-wsl-native-script' + +const POLL_INTERVAL_SECONDS = 5 +export const SNAPSHOT_START = '\x1e' +export const SNAPSHOT_END = '\x1f' + +type WslSnapshotEntry = { + path: string + type: string + mtime: string +} + +export type WslSnapshot = Map + +export function toWslUncPath(linuxPath: string, distro: string): string { + return `\\\\wsl.localhost\\${distro}${linuxPath.replace(/\//g, '\\')}` +} + +function buildPruneExpression(ignoreDirs: readonly string[]): string { + validateWslWatcherIgnoreDirs(ignoreDirs) + if (ignoreDirs.length === 0) { + return '' + } + const names = ignoreDirs.map((name) => `-name '${name}'`).join(' -o ') + return `\\( -type d \\( ${names} \\) -prune \\) -o` +} + +export function buildSnapshotScript(ignoreDirs: readonly string[]): string { + const prune = buildPruneExpression(ignoreDirs) + return [ + 'set -efu', + 'root=$1', + 'while :; do', + " printf '\\036'", + ' if [ -d "$root" ]; then', + // Why: this is the compatibility path; correctness still requires every + // nested open file while high-churn directories are pruned during traversal. + ` find "$root" -mindepth 1 ${prune} -printf '%y\\t%T@\\t%p\\0' 2>/dev/null || true`, + ' fi', + " printf '\\037'", + ` sleep ${POLL_INTERVAL_SECONDS} || exit 0`, + 'done' + ].join('\n') +} + +export function parseSnapshotFrame(frame: string, distro: string): WslSnapshot { + const snapshot: WslSnapshot = new Map() + for (const rawEntry of frame.split('\0')) { + if (!rawEntry) { + continue + } + const firstTab = rawEntry.indexOf('\t') + const secondTab = firstTab === -1 ? -1 : rawEntry.indexOf('\t', firstTab + 1) + if (firstTab <= 0 || secondTab <= firstTab + 1) { + continue + } + const linuxPath = rawEntry.slice(secondTab + 1) + if (!linuxPath.startsWith('/')) { + continue + } + const entry: WslSnapshotEntry = { + type: rawEntry.slice(0, firstTab), + mtime: rawEntry.slice(firstTab + 1, secondTab), + path: toWslUncPath(linuxPath, distro) + } + snapshot.set(entry.path, entry) + } + return snapshot +} + +export function diffSnapshots(prev: WslSnapshot, next: WslSnapshot): WatcherEvent[] { + const events: WatcherEvent[] = [] + for (const [entryPath, nextEntry] of next) { + const prevEntry = prev.get(entryPath) + if (!prevEntry) { + events.push({ type: 'create', path: entryPath } as WatcherEvent) + } else if (prevEntry.type !== nextEntry.type) { + events.push({ type: 'delete', path: entryPath } as WatcherEvent) + events.push({ type: 'create', path: entryPath } as WatcherEvent) + } else if (prevEntry.mtime !== nextEntry.mtime) { + events.push({ type: 'update', path: entryPath } as WatcherEvent) + } + } + for (const entryPath of prev.keys()) { + if (!next.has(entryPath)) { + events.push({ type: 'delete', path: entryPath } as WatcherEvent) + } + } + return events +} diff --git a/src/main/ipc/filesystem-watcher-wsl.test.ts b/src/main/ipc/filesystem-watcher-wsl.test.ts index d6d0fb2ab17..b7db3dc7c5e 100644 --- a/src/main/ipc/filesystem-watcher-wsl.test.ts +++ b/src/main/ipc/filesystem-watcher-wsl.test.ts @@ -2,19 +2,14 @@ import { EventEmitter } from 'node:events' import { PassThrough } from 'node:stream' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { spawnMock } = vi.hoisted(() => ({ - spawnMock: vi.fn() -})) +const { spawnMock } = vi.hoisted(() => ({ spawnMock: vi.fn() })) -vi.mock('child_process', () => ({ - spawn: spawnMock -})) +vi.mock('child_process', () => ({ spawn: spawnMock })) import { createWslWatcher } from './filesystem-watcher-wsl' import type { WatchedRoot, WslWatcherDeps } from './filesystem-watcher-wsl' +import { SNAPSHOT_END, SNAPSHOT_START } from './filesystem-watcher-wsl-snapshot' -const SNAPSHOT_START = '\x1e' -const SNAPSHOT_END = '\x1f' const ROOT_KEY = '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo' class FakeChildProcess extends EventEmitter { @@ -27,6 +22,10 @@ class FakeChildProcess extends EventEmitter { }) } +function nativeMessage(message: unknown): string { + return `${JSON.stringify(message)}\n` +} + function snapshotFrame(entries: [type: string, mtime: string, path: string][]): string { return `${SNAPSHOT_START}${entries .map(([type, mtime, entryPath]) => `${type}\t${mtime}\t${entryPath}\0`) @@ -34,38 +33,27 @@ function snapshotFrame(entries: [type: string, mtime: string, path: string][]): } type ScheduleBatchFlush = (rootKey: string, root: WatchedRoot) => void -type ScheduleBatchFlushMock = ReturnType> function makeDeps( - scheduleBatchFlush: ScheduleBatchFlushMock = vi.fn() -): WslWatcherDeps & { - scheduleBatchFlush: ScheduleBatchFlushMock - watchedRoots: Map -} { - return { - ignoreDirs: ['node_modules', '.git'], - scheduleBatchFlush, - watchedRoots: new Map() + scheduleBatchFlush = vi.fn() +): WslWatcherDeps & { scheduleBatchFlush: ReturnType> } { + return { ignoreDirs: ['node_modules', '.git'], scheduleBatchFlush } +} + +function queueChildren(...children: FakeChildProcess[]): void { + for (const child of children) { + spawnMock.mockReturnValueOnce(child) } } -function startWatcher(deps = makeDeps()): { - child: FakeChildProcess - promise: Promise - deps: ReturnType -} { +async function startNativeWatcher( + deps = makeDeps() +): Promise<{ child: FakeChildProcess; root: WatchedRoot; deps: ReturnType }> { const child = new FakeChildProcess() - spawnMock.mockReturnValueOnce(child) + queueChildren(child) const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, deps) - return { child, promise, deps } -} - -async function resolveInitialSnapshot( - child: FakeChildProcess, - promise: Promise -): Promise { - child.stdout.write(snapshotFrame([['f', '1.0', '/home/me/repo/README.md']])) - return promise + child.stdout.write(nativeMessage({ type: 'ready' })) + return { child, root: await promise, deps } } describe('createWslWatcher', () => { @@ -74,55 +62,95 @@ describe('createWslWatcher', () => { }) afterEach(() => { + vi.useRealTimers() vi.restoreAllMocks() }) - it('spawns a WSL-native snapshot process for the distro path', async () => { - const { child, promise } = startWatcher() - child.stdout.write(snapshotFrame([])) - - await promise + it('starts native Linux events without requiring an inotify utility package', async () => { + const { child } = await startNativeWatcher() expect(spawnMock).toHaveBeenCalledWith( 'wsl.exe', - ['-d', 'Ubuntu', '--', 'sh', '-s', '--', '/home/me/repo'], - expect.objectContaining({ - stdio: ['pipe', 'pipe', 'pipe'], - windowsHide: true - }) + ['-d', 'Ubuntu', '--', 'python3', '-u', '-', '/home/me/repo', 'node_modules', '.git'], + expect.objectContaining({ stdio: ['pipe', 'pipe', 'pipe'], windowsHide: true }) ) + const script = child.stdin.read()?.toString('utf8') ?? '' + expect(script).toContain('libc.inotify_add_watch') + expect(script).not.toContain('inotifywait') + expect(script).not.toContain('while :; do') }) - it('scans below the first two levels so nested open files stay fresh', async () => { - const { child, promise } = startWatcher() - child.stdout.write(snapshotFrame([])) - - await promise - - const snapshotScript = child.stdin.read()?.toString('utf8') ?? '' - expect(snapshotScript).toContain('find "$root" -mindepth 1') - expect(snapshotScript).not.toContain('-maxdepth') - }) - - it('diffs WSL snapshots into create, update, and delete events', async () => { - const scheduleBatchFlush = vi.fn() - const { child, promise } = startWatcher(makeDeps(scheduleBatchFlush)) + it('streams nested native changes as UNC watcher events', async () => { + const scheduleBatchFlush = vi.fn() + const { child, root } = await startNativeWatcher(makeDeps(scheduleBatchFlush)) child.stdout.write( + nativeMessage({ + type: 'events', + events: [ + ['update', '/home/me/repo/docs/deep/README.md'], + ['create', '/home/me/repo/new.txt'], + ['delete', '/home/me/repo/old.txt'], + ['update', '/home/me/repository/outside.txt'] + ] + }) + ) + + expect(scheduleBatchFlush).toHaveBeenCalledOnce() + expect(root.batch.events).toEqual([ + { type: 'update', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\docs\\deep\\README.md' }, + { type: 'create', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\new.txt' }, + { type: 'delete', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\old.txt' } + ]) + }) + + it('uses recursive snapshots when the native runtime is unavailable', async () => { + const native = new FakeChildProcess() + const snapshot = new FakeChildProcess() + queueChildren(native, snapshot) + const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps()) + + native.stdout.write(nativeMessage({ type: 'error', message: 'python3: not found' })) + await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledTimes(2)) + snapshot.stdout.write(snapshotFrame([])) + await promise + + expect(spawnMock.mock.calls[1]?.[1]).toEqual([ + '-d', + 'Ubuntu', + '--', + 'sh', + '-s', + '--', + '/home/me/repo' + ]) + const script = snapshot.stdin.read()?.toString('utf8') ?? '' + expect(script).toContain('find "$root" -mindepth 1') + expect(script).not.toContain('-maxdepth') + }) + + it('diffs fallback snapshots into create, update, and delete events', async () => { + const scheduleBatchFlush = vi.fn() + const native = new FakeChildProcess() + const snapshot = new FakeChildProcess() + queueChildren(native, snapshot) + const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps(scheduleBatchFlush)) + native.emit('error', new Error('spawn failed')) + await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledTimes(2)) + snapshot.stdout.write( snapshotFrame([ ['f', '1.0', '/home/me/repo/README.md'], ['f', '1.0', '/home/me/repo/old.txt'] ]) ) const root = await promise - child.stdout.write( + snapshot.stdout.write( snapshotFrame([ ['f', '2.0', '/home/me/repo/README.md'], ['f', '1.0', '/home/me/repo/new.txt'] ]) ) - expect(scheduleBatchFlush).toHaveBeenCalledOnce() expect(root.batch.events).toEqual([ { type: 'update', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\README.md' }, { type: 'create', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\new.txt' }, @@ -130,25 +158,32 @@ describe('createWslWatcher', () => { ]) }) - it('turns watcher exit into an overflow refresh without retaining UNC paths', async () => { - const scheduleBatchFlush = vi.fn() - const deps = makeDeps(scheduleBatchFlush) - const { child, promise } = startWatcher(deps) - const root = await resolveInitialSnapshot(child, promise) - deps.watchedRoots.set(ROOT_KEY, root) + it('refreshes and restarts an active watcher after WSL exits', async () => { + vi.useFakeTimers() + const scheduleBatchFlush = vi.fn() + const first = new FakeChildProcess() + const second = new FakeChildProcess() + queueChildren(first, second) + const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps(scheduleBatchFlush)) + first.stdout.write(nativeMessage({ type: 'ready' })) + const root = await promise - child.emit('close', null, 'SIGTERM') - - expect(scheduleBatchFlush).toHaveBeenCalledOnce() - expect(root.batch.events).toEqual([]) + first.emit('close', null, 'SIGTERM') + await vi.advanceTimersByTimeAsync(0) expect(root.batch.overflowed).toBe(true) - expect(deps.watchedRoots.has(ROOT_KEY)).toBe(false) + expect(scheduleBatchFlush).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(500) + expect(spawnMock).toHaveBeenCalledTimes(2) + second.stdout.write(nativeMessage({ type: 'ready' })) + await vi.advanceTimersByTimeAsync(0) + + await root.subscription.unsubscribe() + expect(second.kill).toHaveBeenCalledOnce() }) - it('kills the WSL child on unsubscribe without emitting a shutdown refresh', async () => { - const scheduleBatchFlush = vi.fn() - const { child, promise } = startWatcher(makeDeps(scheduleBatchFlush)) - const root = await resolveInitialSnapshot(child, promise) + it('kills the WSL child on unsubscribe without emitting a refresh', async () => { + const scheduleBatchFlush = vi.fn() + const { child, root } = await startNativeWatcher(makeDeps(scheduleBatchFlush)) await root.subscription.unsubscribe() @@ -156,31 +191,34 @@ describe('createWslWatcher', () => { expect(scheduleBatchFlush).not.toHaveBeenCalled() }) - it('rejects when the WSL process exits before the first snapshot', async () => { - const { child, promise } = startWatcher() + it('cancels a watcher that is still restarting', async () => { + vi.useFakeTimers() + const first = new FakeChildProcess() + const restarting = new FakeChildProcess() + queueChildren(first, restarting) + const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps()) + first.stdout.write(nativeMessage({ type: 'ready' })) + const root = await promise + first.emit('close', null, 'SIGTERM') + await vi.advanceTimersByTimeAsync(500) - child.stderr.write('find failed') - child.emit('close', 1, null) + await root.subscription.unsubscribe() - await expect(promise).rejects.toThrow('WSL watcher exited before first snapshot') + expect(restarting.kill).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(10_000) + expect(spawnMock).toHaveBeenCalledTimes(2) }) - it('does not emit a shutdown refresh after a startup error', async () => { - const scheduleBatchFlush = vi.fn() - const { child, promise } = startWatcher(makeDeps(scheduleBatchFlush)) + it('rejects only after both native and snapshot startup fail', async () => { + const native = new FakeChildProcess() + const snapshot = new FakeChildProcess() + queueChildren(native, snapshot) + const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps()) + native.emit('error', new Error('native failed')) + await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledTimes(2)) + snapshot.stderr.write('find failed') + snapshot.emit('close', 1, null) - child.emit('error', new Error('spawn failed')) - child.emit('close', 1, null) - - await expect(promise).rejects.toThrow('spawn failed') - expect(scheduleBatchFlush).not.toHaveBeenCalled() - }) - - it('rejects startup when WSL exits before reading the snapshot script', async () => { - const { child, promise } = startWatcher() - - child.stdin.emit('error', new Error('write EPIPE')) - - await expect(promise).rejects.toThrow('write EPIPE') + await expect(promise).rejects.toThrow('WSL watcher exited before ready') }) }) diff --git a/src/main/ipc/filesystem-watcher-wsl.ts b/src/main/ipc/filesystem-watcher-wsl.ts index 92e677247d9..677e4be450b 100644 --- a/src/main/ipc/filesystem-watcher-wsl.ts +++ b/src/main/ipc/filesystem-watcher-wsl.ts @@ -1,15 +1,13 @@ -/** - * WSL-native file watcher for WSL paths. - * - * Why: polling \\wsl.localhost from Windows keeps waking the distro after - * `wsl --shutdown`, which can make WSL look wedged. Keep the polling process - * inside the distro so shutdown kills it instead of Orca restarting WSL. - */ -import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process' -import { StringDecoder } from 'node:string_decoder' +/** WSL file watching with native Linux events and a recursive scan fallback. */ import type { WebContents } from 'electron' import type { Event as WatcherEvent } from '@parcel/watcher' import { queueWatcherEvents } from './filesystem-watcher-event-batch' +import { + createWslNativeEngine, + createWslSnapshotEngine, + type WslEngineContext, + type WslWatchEngine +} from './filesystem-watcher-wsl-engine' import { parseWslUncPath } from '../../shared/wsl-paths' export type WatcherSubscription = { @@ -32,115 +30,12 @@ export type WatchedRoot = { export type WslWatcherDeps = { ignoreDirs: string[] scheduleBatchFlush: (rootKey: string, root: WatchedRoot) => void - watchedRoots: Map } -const POLL_INTERVAL_MS = 2000 -const POLL_INTERVAL_SECONDS = Math.max(1, Math.ceil(POLL_INTERVAL_MS / 1000)) -const STARTUP_TIMEOUT_MS = 10_000 -const SNAPSHOT_START = '\x1e' -const SNAPSHOT_END = '\x1f' -const MAX_STREAM_BUFFER_CHARS = 10 * 1024 * 1024 +const RESTART_DELAYS_MS = [500, 1_000, 2_000, 5_000, 10_000] as const +const STABLE_ENGINE_MS = 30_000 -type WslSnapshotEntry = { - path: string - type: string - mtime: string -} - -type WslSnapshot = Map - -function toWslUncPath(linuxPath: string, distro: string): string { - return `\\\\wsl.localhost\\${distro}${linuxPath.replace(/\//g, '\\')}` -} - -function quoteSafeFindName(name: string): string { - if (!/^[A-Za-z0-9_.-]+$/.test(name)) { - throw new Error(`Unsupported WSL watcher ignore name: ${name}`) - } - return `'${name}'` -} - -function buildPruneExpression(ignoreDirs: readonly string[]): string { - if (ignoreDirs.length === 0) { - return '' - } - const names = ignoreDirs.map((name) => `-name ${quoteSafeFindName(name)}`).join(' -o ') - return `\\( -type d \\( ${names} \\) -prune \\) -o` -} - -function buildSnapshotScript(ignoreDirs: readonly string[]): string { - const prune = buildPruneExpression(ignoreDirs) - return [ - 'set -efu', - 'root=$1', - 'while :; do', - " printf '\\036'", - ' if [ -d "$root" ]; then', - // Why: open editor tabs can live at any depth, and the scan runs natively - // inside WSL with high-churn directories pruned before traversal. - ` find "$root" -mindepth 1 ${prune} -printf '%y\\t%T@\\t%p\\0' 2>/dev/null || true`, - ' fi', - " printf '\\037'", - ` sleep ${POLL_INTERVAL_SECONDS} || exit 0`, - 'done' - ].join('\n') -} - -function parseSnapshotFrame(frame: string, distro: string): WslSnapshot { - const snapshot: WslSnapshot = new Map() - for (const rawEntry of frame.split('\0')) { - if (!rawEntry) { - continue - } - const firstTab = rawEntry.indexOf('\t') - const secondTab = firstTab === -1 ? -1 : rawEntry.indexOf('\t', firstTab + 1) - if (firstTab <= 0 || secondTab <= firstTab + 1) { - continue - } - const linuxPath = rawEntry.slice(secondTab + 1) - if (!linuxPath.startsWith('/')) { - continue - } - const entry: WslSnapshotEntry = { - type: rawEntry.slice(0, firstTab), - mtime: rawEntry.slice(firstTab + 1, secondTab), - path: toWslUncPath(linuxPath, distro) - } - snapshot.set(entry.path, entry) - } - return snapshot -} - -function diffSnapshots(prev: WslSnapshot, next: WslSnapshot): WatcherEvent[] { - const events: WatcherEvent[] = [] - - for (const [entryPath, nextEntry] of next) { - const prevEntry = prev.get(entryPath) - if (!prevEntry) { - events.push({ type: 'create', path: entryPath } as WatcherEvent) - continue - } - if (prevEntry.type !== nextEntry.type) { - events.push({ type: 'delete', path: entryPath } as WatcherEvent) - events.push({ type: 'create', path: entryPath } as WatcherEvent) - continue - } - if (prevEntry.mtime !== nextEntry.mtime) { - events.push({ type: 'update', path: entryPath } as WatcherEvent) - } - } - - for (const entryPath of prev.keys()) { - if (!next.has(entryPath)) { - events.push({ type: 'delete', path: entryPath } as WatcherEvent) - } - } - - return events -} - -function markOverflowWithoutUncStat(root: WatchedRoot): void { +function markOverflow(root: WatchedRoot): void { if (root.batch.timer) { clearTimeout(root.batch.timer) root.batch.timer = null @@ -158,179 +53,123 @@ export async function createWslWatcher( if (!wsl) { throw new Error(`Not a WSL path: ${worktreePath}`) } - const distro = wsl.distro - const linuxPath = wsl.linuxPath const root: WatchedRoot = { subscription: null!, listeners: new Map(), batch: { events: [], overflowed: false, timer: null, firstEventAt: 0 } } - + let activeEngine: WslWatchEngine | null = null + let startingEngine: WslWatchEngine | null = null let disposed = false - let prevSnapshot: WslSnapshot | null = null - let stopped = false - let streamBuffer = '' - const stdoutDecoder = new StringDecoder('utf8') - const stderrDecoder = new StringDecoder('utf8') - let stderrTail = '' + let nativeUnavailable = false + let restartAttempt = 0 + let restartTimer: ReturnType | null = null + let stabilityTimer: ReturnType | null = null + const maximumRestartDelay = RESTART_DELAYS_MS.at(-1) ?? 10_000 - let resolveInitial!: () => void - let rejectInitial!: (error: Error) => void - let initialSettled = false - const initialSnapshotReady = new Promise((resolve, reject) => { - resolveInitial = resolve - rejectInitial = reject - }) - - function settleInitial(error?: Error): void { - if (initialSettled) { - return - } - initialSettled = true - if (error) { - rejectInitial(error) - } else { - resolveInitial() - } - } - - function signalWatcherStopped(): void { - if (stopped) { - return - } - if (!prevSnapshot) { - return - } - stopped = true - markOverflowWithoutUncStat(root) - deps.scheduleBatchFlush(rootKey, root) - deps.watchedRoots.delete(rootKey) - } - - function ingestFrame(frame: string): void { - const nextSnapshot = parseSnapshotFrame(frame, distro) - if (!prevSnapshot) { - prevSnapshot = nextSnapshot - settleInitial() - return - } - const events = diffSnapshots(prevSnapshot, nextSnapshot) - prevSnapshot = nextSnapshot - - if (events.length > 0) { + const context: WslEngineContext = { + distro: wsl.distro, + linuxPath: wsl.linuxPath, + worktreePath, + ignoreDirs: deps.ignoreDirs, + onEvents: (events) => { queueWatcherEvents(root.batch, events) deps.scheduleBatchFlush(rootKey, root) + }, + onOverflow: () => { + markOverflow(root) + deps.scheduleBatchFlush(rootKey, root) } } - function drainFrames(): void { - while (true) { - const start = streamBuffer.indexOf(SNAPSHOT_START) - if (start === -1) { - streamBuffer = streamBuffer.slice(-1) - return - } - if (start > 0) { - streamBuffer = streamBuffer.slice(start) - } - const end = streamBuffer.indexOf(SNAPSHOT_END, 1) - if (end === -1) { - if (streamBuffer.length > MAX_STREAM_BUFFER_CHARS) { - streamBuffer = '' - markOverflowWithoutUncStat(root) - deps.scheduleBatchFlush(rootKey, root) + const startEngine = async (): Promise => { + if (!nativeUnavailable) { + const native = createWslNativeEngine(context) + startingEngine = native + try { + await native.ready + startingEngine = null + return native + } catch (error) { + startingEngine = null + nativeUnavailable = true + native.stop() + if (disposed) { + throw error } + } + } + const snapshot = createWslSnapshotEngine(context) + startingEngine = snapshot + try { + await snapshot.ready + } catch (error) { + snapshot.stop() + throw error + } finally { + startingEngine = null + } + return snapshot + } + + let installEngine: () => Promise + const scheduleRestart = (delay: number): void => { + restartTimer = setTimeout(() => { + restartTimer = null + void installEngine().catch(() => { + if (!disposed) { + scheduleRestart(maximumRestartDelay) + } + }) + }, delay) + } + + installEngine = async (): Promise => { + const engine = await startEngine() + if (disposed) { + engine.stop() + return + } + activeEngine = engine + stabilityTimer = setTimeout(() => { + restartAttempt = 0 + }, STABLE_ENGINE_MS) + void engine.stopped.then(() => { + if (disposed || activeEngine !== engine) { return } - const frame = streamBuffer.slice(1, end) - streamBuffer = streamBuffer.slice(end + 1) - ingestFrame(frame) - } - } - - let child: ChildProcessWithoutNullStreams - try { - child = spawn('wsl.exe', ['-d', distro, '--', 'sh', '-s', '--', linuxPath], { - stdio: ['pipe', 'pipe', 'pipe'], - windowsHide: true + activeEngine = null + if (stabilityTimer) { + clearTimeout(stabilityTimer) + } + markOverflow(root) + deps.scheduleBatchFlush(rootKey, root) + const delay = + RESTART_DELAYS_MS.at(Math.min(restartAttempt, RESTART_DELAYS_MS.length - 1)) ?? + maximumRestartDelay + restartAttempt += 1 + // Why: WSL shutdowns and transient distro failures must not permanently + // orphan an active renderer subscription. + scheduleRestart(delay) }) - } catch (error) { - throw error instanceof Error ? error : new Error(String(error)) } - const startupTimer = setTimeout(() => { - settleInitial(new Error(`Timed out starting WSL watcher for ${worktreePath}`)) - child.kill() - }, STARTUP_TIMEOUT_MS) - - child.stdin.on('error', (error) => { - // Why: WSL can exit before reading the script; handle EPIPE here so the - // startup failure rejects the watcher instead of crashing on a stream error. - if (!initialSettled) { - settleInitial(error) - } - }) - - child.stdout.on('data', (chunk: Buffer) => { - if (disposed) { - return - } - streamBuffer += stdoutDecoder.write(chunk) - drainFrames() - }) - - child.stderr.on('data', (chunk: Buffer) => { - stderrTail = (stderrTail + stderrDecoder.write(chunk)).slice(-4096) - }) - - child.stdout.on('error', (error) => { - if (!initialSettled) { - settleInitial(error) - return - } - if (!disposed) { - signalWatcherStopped() - } - }) - - child.stderr.on('error', () => { - // Ignore diagnostic stream failures; stdout/close determine watcher state. - }) - - child.once('error', (error) => { - if (!initialSettled) { - settleInitial(error) - return - } - if (!disposed) { - signalWatcherStopped() - } - }) - - child.once('close', (code, signal) => { - if (!initialSettled) { - const suffix = stderrTail.trim() ? `: ${stderrTail.trim()}` : '' - settleInitial( - new Error(`WSL watcher exited before first snapshot (${code ?? signal})${suffix}`) - ) - return - } - if (!disposed) { - signalWatcherStopped() - } - }) - - child.stdin.end(buildSnapshotScript(deps.ignoreDirs)) - - await initialSnapshotReady.finally(() => clearTimeout(startupTimer)) - + await installEngine() root.subscription = { unsubscribe: async () => { disposed = true - child.kill() + if (restartTimer) { + clearTimeout(restartTimer) + } + if (stabilityTimer) { + clearTimeout(stabilityTimer) + } + startingEngine?.stop() + startingEngine = null + activeEngine?.stop() + activeEngine = null } } - return root } diff --git a/src/main/ipc/filesystem-watcher.ts b/src/main/ipc/filesystem-watcher.ts index 56ab2aa2981..1134fbf55d0 100644 --- a/src/main/ipc/filesystem-watcher.ts +++ b/src/main/ipc/filesystem-watcher.ts @@ -500,13 +500,12 @@ async function doInstallLocalWatcher( } try { - // Why: WSL paths use one snapshot subprocess inside the Linux distro so - // `wsl --shutdown` can kill it; native Windows paths use @parcel/watcher. + // Why: WSL paths are watched inside Linux so native kernel events remain + // reliable; native host paths use the platform watcher process. root = isWslPath(worktreePath) ? await createWslWatcher(rootKey, worktreePath, { ignoreDirs: WATCHER_IGNORE_DIRS, - scheduleBatchFlush, - watchedRoots + scheduleBatchFlush }) : await createWatcher(rootKey, rootKey) } catch {