mirror of
https://github.com/stablyai/orca.git
synced 2026-10-17 00:03:08 +00:00
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.
This commit is contained in:
Jinwoo Hong
@@ -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<void>
|
||||
stopped: Promise<void>
|
||||
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<void>((resolve, reject) => {
|
||||
resolveReady = resolve
|
||||
rejectReady = reject
|
||||
})
|
||||
const stopped = new Promise<void>((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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
@@ -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<string> {
|
||||
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()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -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)
|
||||
`
|
||||
}
|
||||
@@ -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<string, WslSnapshotEntry>
|
||||
|
||||
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
|
||||
}
|
||||
@@ -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<typeof vi.fn<ScheduleBatchFlush>>
|
||||
|
||||
function makeDeps(
|
||||
scheduleBatchFlush: ScheduleBatchFlushMock = vi.fn<ScheduleBatchFlush>()
|
||||
): WslWatcherDeps & {
|
||||
scheduleBatchFlush: ScheduleBatchFlushMock
|
||||
watchedRoots: Map<string, WatchedRoot>
|
||||
} {
|
||||
return {
|
||||
ignoreDirs: ['node_modules', '.git'],
|
||||
scheduleBatchFlush,
|
||||
watchedRoots: new Map()
|
||||
scheduleBatchFlush = vi.fn<ScheduleBatchFlush>()
|
||||
): WslWatcherDeps & { scheduleBatchFlush: ReturnType<typeof vi.fn<ScheduleBatchFlush>> } {
|
||||
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<WatchedRoot>
|
||||
deps: ReturnType<typeof makeDeps>
|
||||
} {
|
||||
async function startNativeWatcher(
|
||||
deps = makeDeps()
|
||||
): Promise<{ child: FakeChildProcess; root: WatchedRoot; deps: ReturnType<typeof makeDeps> }> {
|
||||
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<WatchedRoot>
|
||||
): Promise<WatchedRoot> {
|
||||
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<ScheduleBatchFlush>()
|
||||
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<ScheduleBatchFlush>()
|
||||
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<ScheduleBatchFlush>()
|
||||
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<ScheduleBatchFlush>()
|
||||
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')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<string, WatchedRoot>
|
||||
}
|
||||
|
||||
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<string, WslSnapshotEntry>
|
||||
|
||||
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<typeof setTimeout> | null = null
|
||||
let stabilityTimer: ReturnType<typeof setTimeout> | 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<void>((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<WslWatchEngine> => {
|
||||
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<void>
|
||||
const scheduleRestart = (delay: number): void => {
|
||||
restartTimer = setTimeout(() => {
|
||||
restartTimer = null
|
||||
void installEngine().catch(() => {
|
||||
if (!disposed) {
|
||||
scheduleRestart(maximumRestartDelay)
|
||||
}
|
||||
})
|
||||
}, delay)
|
||||
}
|
||||
|
||||
installEngine = async (): Promise<void> => {
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user