fix(fs): run WSL watching in managed Linux host

This commit is contained in:
Jinwoo Hong
2026-07-09 20:30:42 -04:00
parent bc5d410f3a
commit bb7272971b
18 changed files with 1283 additions and 326 deletions
+10 -2
View File
@@ -1,7 +1,7 @@
const { chmodSync, existsSync, readdirSync } = require('node:fs')
const { execFileSync } = require('node:child_process')
const { join, resolve } = require('node:path')
const electronBuilderNativeRebuild = require('./scripts/electron-builder-native-rebuild.cjs')
const electronBuilderBeforeBuild = require('./scripts/electron-builder-before-build.cjs')
const {
assertPackagedDaemonEntryExists,
verifyPackagedDaemonEntryBoots
@@ -66,6 +66,10 @@ module.exports = {
'!Casks{,/**/*}',
'!{AGENTS.md,CLAUDE.md,DEVELOPING.md,bundle-size-progress.md}',
'!out/**/*.test.js',
// Why: these Linux-host artifacts are staged as a Windows-only external
// resource; including either copy in app.asar would duplicate ~62 MB.
'!out/main/wsl-watcher-host.js',
'!out/wsl-watcher{,/**/*}',
'!electron.vite.config.{js,ts,mjs,cjs}',
'!{.eslintcache,eslint.config.mjs,.prettierignore,.prettierrc.yaml,CHANGELOG.md,README.md}',
'!{.env,.env.*,.npmrc,pnpm-lock.yaml}',
@@ -193,6 +197,10 @@ module.exports = {
from: 'native/computer-use-windows/runtime.ps1',
to: 'computer-use-windows/runtime.ps1'
},
{
from: 'out/wsl-watcher',
to: 'wsl-watcher'
},
featureWallResources
]
},
@@ -363,7 +371,7 @@ module.exports = {
afterInstall: 'resources/linux/packaging/after-install.sh',
afterRemove: 'resources/linux/packaging/after-remove.sh'
},
beforeBuild: electronBuilderNativeRebuild,
beforeBuild: electronBuilderBeforeBuild,
// Why: must be true so that electron-builder rebuilds native modules
// (node-pty) for each target architecture when producing dual-arch macOS
// builds (x64 + arm64). With npmRebuild disabled, CI on an arm64 runner
@@ -0,0 +1,32 @@
const { execFileSync } = require('node:child_process')
const { resolve } = require('node:path')
const electronBuilderNativeRebuild = require('./electron-builder-native-rebuild.cjs')
const projectDir = resolve(__dirname, '../..')
function electronBuilderBeforeBuild(context, runner = execFileSync) {
const shouldContinue = electronBuilderNativeRebuild(context)
const preparationArgs = buildWslRuntimePreparationArgs(context)
if (preparationArgs) {
runner(process.execPath, preparationArgs, {
cwd: projectDir,
stdio: 'inherit'
})
}
return shouldContinue
}
function buildWslRuntimePreparationArgs(context) {
const platform =
typeof context?.platform === 'string' ? context.platform : context?.platform?.nodeName
if (platform === 'win32') {
// Why: Windows cannot execute Linux native addons directly, so package a
// verified Linux runtime that can be installed inside each WSL distro.
return ['config/scripts/prepare-wsl-watcher-runtime.mjs']
}
return null
}
module.exports = electronBuilderBeforeBuild
module.exports.default = electronBuilderBeforeBuild
module.exports.buildWslRuntimePreparationArgs = buildWslRuntimePreparationArgs
@@ -6,7 +6,8 @@ import { describe, expect, it } from 'vitest'
const require = createRequire(import.meta.url)
const electronBuilderConfig = require('../electron-builder.config.cjs')
const electronBuilderNativeRebuild = require('./electron-builder-native-rebuild.cjs')
const electronBuilderBeforeBuild = require('./electron-builder-before-build.cjs')
const { buildWslRuntimePreparationArgs } = electronBuilderBeforeBuild
const {
createPackagedRuntimeNodeModuleResources,
findAsarEntry,
@@ -31,7 +32,9 @@ describe('electron-builder config', () => {
'!tests{,/**/*}',
'!Casks{,/**/*}',
'!{AGENTS.md,CLAUDE.md,DEVELOPING.md,bundle-size-progress.md}',
'!out/**/*.test.js'
'!out/**/*.test.js',
'!out/main/wsl-watcher-host.js',
'!out/wsl-watcher{,/**/*}'
])
)
})
@@ -58,6 +61,10 @@ describe('electron-builder config', () => {
expect.objectContaining({
from: 'native/computer-use-windows/runtime.ps1',
to: 'computer-use-windows/runtime.ps1'
}),
expect.objectContaining({
from: 'out/wsl-watcher',
to: 'wsl-watcher'
})
])
)
@@ -131,11 +138,19 @@ describe('electron-builder config', () => {
}
})
it('uses Orca native rebuild hook instead of electron-builder default rebuild', () => {
expect(electronBuilderConfig.beforeBuild).toBe(electronBuilderNativeRebuild)
it('uses Orca build preparation instead of electron-builder default rebuild', () => {
expect(electronBuilderConfig.beforeBuild).toBe(electronBuilderBeforeBuild)
expect(electronBuilderConfig.npmRebuild).toBe(true)
})
it('prepares the managed WSL runtime only for Windows artifacts', () => {
expect(buildWslRuntimePreparationArgs({ platform: { nodeName: 'win32' } })).toEqual([
'config/scripts/prepare-wsl-watcher-runtime.mjs'
])
expect(buildWslRuntimePreparationArgs({ platform: { nodeName: 'darwin' } })).toBeNull()
expect(buildWslRuntimePreparationArgs({ platform: { nodeName: 'linux' } })).toBeNull()
})
it('verifies packaged main runtime deps from Windows-style asar entries', async () => {
const resourcesDir = await mkdtemp(join(tmpdir(), 'orca-runtime-deps-'))
try {
@@ -0,0 +1,125 @@
import { createHash } from 'node:crypto'
import { copyFile, mkdir, readFile, rename, rm, stat, writeFile } from 'node:fs/promises'
import path from 'node:path'
export const WSL_WATCHER_NODE_VERSION = '24.15.0'
const WSL_WATCHER_INSTALL_LAYOUT_VERSION = 1
const rootDir = path.resolve(import.meta.dirname, '../..')
const outputDir = path.join(rootDir, 'out', 'wsl-watcher')
const cacheDir = path.join(rootDir, 'node_modules', '.cache', 'orca-wsl-watcher')
const runtimeArchives = {
x64: {
sha256: '472655581fb851559730c48763e0c9d3bc25975c59d518003fc0849d3e4ba0f6',
watcherPackage: '@parcel/watcher-linux-x64-glibc'
},
arm64: {
sha256: 'f3d5a797b5d210ce8e2cb265544c8e482eaedcb8aa409a8b46da7e8595d0dda0',
watcherPackage: '@parcel/watcher-linux-arm64-glibc'
}
}
function sha256(contents) {
return createHash('sha256').update(contents).digest('hex')
}
async function matchesHash(filename, expected) {
try {
return sha256(await readFile(filename)) === expected
} catch {
return false
}
}
async function downloadRuntime(arch, expectedHash) {
await mkdir(cacheDir, { recursive: true })
const filename = `node-v${WSL_WATCHER_NODE_VERSION}-linux-${arch}.tar.xz`
const cached = path.join(cacheDir, filename)
if (await matchesHash(cached, expectedHash)) {
return cached
}
const response = await fetch(`https://nodejs.org/dist/v${WSL_WATCHER_NODE_VERSION}/${filename}`, {
signal: AbortSignal.timeout(60_000)
})
if (!response.ok) {
throw new Error(`Could not download ${filename}: HTTP ${response.status}`)
}
const contents = Buffer.from(await response.arrayBuffer())
const actualHash = sha256(contents)
if (actualHash !== expectedHash) {
throw new Error(
`Checksum mismatch for ${filename}: expected ${expectedHash}, got ${actualHash}`
)
}
const temporary = `${cached}.${process.pid}.tmp`
await writeFile(temporary, contents)
await rm(cached, { force: true })
await rename(temporary, cached)
return cached
}
async function assertBuildInput(filename, description) {
try {
const info = await stat(filename)
if (info.isFile()) {
return
}
} catch {}
throw new Error(`Missing ${description} at ${filename}. Run pnpm build:electron-vite first.`)
}
export async function prepareWslWatcherRuntime() {
const hostSource = path.join(rootDir, 'out', 'main', 'wsl-watcher-host.js')
await assertBuildInput(hostSource, 'compiled WSL watcher host')
const stageDir = `${outputDir}.${process.pid}.tmp`
await rm(stageDir, { recursive: true, force: true })
await mkdir(stageDir, { recursive: true })
const hostContents = await readFile(hostSource)
await writeFile(path.join(stageDir, 'host.js'), hostContents)
await copyFile(
path.join(rootDir, 'node_modules', '@parcel', 'watcher', 'LICENSE'),
path.join(stageDir, 'parcel-watcher-LICENSE')
)
const versionParts = [
String(WSL_WATCHER_INSTALL_LAYOUT_VERSION),
WSL_WATCHER_NODE_VERSION,
sha256(hostContents)
]
for (const [arch, runtime] of Object.entries(runtimeArchives)) {
const archDir = path.join(stageDir, arch)
await mkdir(archDir, { recursive: true })
const archive = await downloadRuntime(arch, runtime.sha256)
const watcherSource = path.join(
rootDir,
'node_modules',
'@parcel',
runtime.watcherPackage.replace('@parcel/', ''),
'watcher.node'
)
const watcherContents = await readFile(watcherSource)
versionParts.push(arch, runtime.sha256, sha256(watcherContents))
await copyFile(archive, path.join(archDir, 'node.tar.xz'))
await writeFile(path.join(archDir, 'watcher.node'), watcherContents)
}
const manifest = {
protocol: 1,
installLayout: WSL_WATCHER_INSTALL_LAYOUT_VERSION,
nodeVersion: WSL_WATCHER_NODE_VERSION,
bundleVersion: sha256(versionParts.join('\n')).slice(0, 20)
}
await writeFile(path.join(stageDir, 'manifest.json'), `${JSON.stringify(manifest, null, 2)}\n`)
await rm(outputDir, { recursive: true, force: true })
await rename(stageDir, outputDir)
process.stdout.write(
`[prepare-wsl-watcher-runtime] prepared ${manifest.bundleVersion} in ${outputDir}\n`
)
return manifest
}
if (process.argv[1] === import.meta.filename) {
await prepareWslWatcherRuntime()
}
+3
View File
@@ -186,6 +186,9 @@ export default defineConfig({
// Why: forked with ELECTRON_RUN_AS_NODE so @parcel/watcher faults
// can't take down the main process (issue #7547).
'parcel-watcher-process-entry': resolve('src/main/ipc/parcel-watcher-process-entry.ts'),
// Why: WSL needs a Linux-native watcher process beside its filesystem;
// packaging stages this plain-Node entry with the managed Linux runtime.
'wsl-watcher-host': resolve('src/main/ipc/wsl-watcher-host-entry.ts'),
// Why: electron-vite cleans out/main in dev. The dev CLI imports
// this path for `orca agent hooks ...`, so it must survive rebuilds.
'agent-hooks/managed-agent-hook-controls': resolve(
+1
View File
@@ -59,6 +59,7 @@
"audit:localization": "node config/scripts/audit-localization-coverage.mjs",
"build:cli": "tsc -p config/tsconfig.cli.json --outDir out --composite false --incremental false && node config/scripts/verify-cli-bin.mjs --fix-executable --fix-package-json && node config/scripts/install-dev-cli.mjs",
"build:electron-vite": "node config/scripts/run-electron-vite-build.mjs",
"prepare:wsl-watcher": "node config/scripts/prepare-wsl-watcher-runtime.mjs",
"build:web": "node config/scripts/run-vite-web-build.mjs && node config/scripts/verify-web-build.mjs",
"build:desktop": "pnpm run typecheck && pnpm run build:relay && pnpm run build:cli && pnpm run build:electron-vite && pnpm run build:web",
"build": "pnpm run build:desktop && pnpm run build:native",
+45 -77
View File
@@ -1,14 +1,16 @@
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 {
subscribeViaWslWatcherHost,
type WslHostSubscription
} from './filesystem-watcher-wsl-host-client'
import {
buildSnapshotScript,
diffSnapshots,
parseSnapshotFrame,
SNAPSHOT_END,
SNAPSHOT_START,
toWslUncPath,
type WslSnapshot
} from './filesystem-watcher-wsl-snapshot'
@@ -27,7 +29,6 @@ export type WslEngineContext = {
onOverflow: () => void
}
const NATIVE_STARTUP_TIMEOUT_MS = 10_000
const SNAPSHOT_STARTUP_TIMEOUT_MS = 30_000
const MAX_STREAM_BUFFER_CHARS = 10 * 1024 * 1024
@@ -128,85 +129,52 @@ function createChildEngine(
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) })
export function createWslNativeEngine(context: WslEngineContext): WslWatchEngine {
let subscription: WslHostSubscription | null = null
const abortController = new AbortController()
let disposed = false
let stoppedSettled = false
let resolveStopped!: () => void
const stopped = new Promise<void>((resolve) => {
resolveStopped = resolve
})
const settleStopped = (): void => {
if (!stoppedSettled) {
stoppedSettled = true
resolveStopped()
}
}
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()
const ready = subscribeViaWslWatcherHost(
{
distro: context.distro,
linuxPath: context.linuxPath,
ignoreDirs: context.ignoreDirs,
onEvents: context.onEvents,
onOverflow: context.onOverflow,
onStopped: settleStopped
},
abortController.signal
).then((created) => {
if (disposed) {
created.unsubscribe()
} else {
subscription = created
}
})
return {
ready,
stopped,
stop: () => {
if (disposed) {
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')
}
disposed = true
abortController.abort()
subscription?.unsubscribe()
subscription = null
settleStopped()
}
)
}
}
export function createWslSnapshotEngine(context: WslEngineContext): WslWatchEngine {
@@ -0,0 +1,315 @@
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { StringDecoder } from 'node:string_decoder'
import type { Event as WatcherEvent } from '@parcel/watcher'
import { wslHostMessageEvents, type WslHostMessage } from './filesystem-watcher-wsl-host-protocol'
import { ensureWslWatcherRuntime } from './filesystem-watcher-wsl-runtime'
export type WslHostSubscriptionContext = {
distro: string
linuxPath: string
ignoreDirs: readonly string[]
onEvents: (events: WatcherEvent[]) => void
onOverflow: () => void
onStopped: () => void
}
export type WslHostSubscription = {
unsubscribe(): void
}
type PendingResult = {
resolve: () => void
reject: (error: Error) => void
}
type SubscriptionRecord = {
id: number
context: WslHostSubscriptionContext
pending: PendingResult | null
pendingTimer: ReturnType<typeof setTimeout> | null
}
type DistroHost = {
distro: string
process: ChildProcessWithoutNullStreams | null
starting: Promise<void> | null
subscriptions: Map<number, SubscriptionRecord>
nextId: number
streamBuffer: string
stdoutDecoder: StringDecoder
stderrDecoder: StringDecoder
stderrTail: string
}
const hosts = new Map<string, DistroHost>()
const STARTUP_TIMEOUT_MS = 30_000
const SUBSCRIBE_TIMEOUT_MS = 30_000
const MAX_STREAM_BUFFER_CHARS = 10 * 1024 * 1024
function createHost(distro: string): DistroHost {
return {
distro,
process: null,
starting: null,
subscriptions: new Map(),
nextId: 1,
streamBuffer: '',
stdoutDecoder: new StringDecoder('utf8'),
stderrDecoder: new StringDecoder('utf8'),
stderrTail: ''
}
}
function settleRecordFailure(host: DistroHost, record: SubscriptionRecord, error: Error): void {
host.subscriptions.delete(record.id)
if (record.pendingTimer) {
clearTimeout(record.pendingTimer)
record.pendingTimer = null
}
if (record.pending) {
record.pending.reject(error)
record.pending = null
} else {
record.context.onStopped()
}
if (host.subscriptions.size === 0) {
hosts.delete(host.distro)
const child = host.process
host.process = null
child?.kill()
} else {
send(host, { op: 'unsubscribe', id: record.id })
}
}
function handleMessage(host: DistroHost, message: WslHostMessage, settleReady: () => void): void {
if (message.op === 'ready') {
if (message.protocol === 1) {
settleReady()
} else {
host.process?.kill()
}
return
}
if (message.op === 'protocol-error') {
host.process?.kill()
return
}
if (!Number.isSafeInteger(message.id)) {
return
}
const record = host.subscriptions.get(message.id as number)
if (!record) {
return
}
if (message.op === 'subscribed') {
if (record.pendingTimer) {
clearTimeout(record.pendingTimer)
record.pendingTimer = null
}
record.pending?.resolve()
record.pending = null
} else if (message.op === 'subscribe-failed') {
settleRecordFailure(host, record, new Error(String(message.message ?? 'subscribe failed')))
} else if (message.op === 'watch-error') {
settleRecordFailure(host, record, new Error(String(message.message ?? 'watch failed')))
} else if (message.op === 'events') {
const events = wslHostMessageEvents(message, record.context)
if (events.length > 0) {
record.context.onEvents(events)
}
}
}
function handleHostGone(host: DistroHost, child: ChildProcessWithoutNullStreams): void {
if (host.process !== child) {
return
}
host.process = null
hosts.delete(host.distro)
const error = new Error(
`Managed WSL watcher exited${host.stderrTail.trim() ? `: ${host.stderrTail.trim()}` : ''}`
)
for (const record of host.subscriptions.values()) {
if (record.pendingTimer) {
clearTimeout(record.pendingTimer)
record.pendingTimer = null
}
if (record.pending) {
record.pending.reject(error)
record.pending = null
} else {
record.context.onStopped()
}
}
host.subscriptions.clear()
}
function send(host: DistroHost, message: object): boolean {
try {
if (!host.process?.stdin.writable) {
return false
}
host.process.stdin.write(`${JSON.stringify(message)}\n`)
return true
} catch {
return false
}
}
async function startHost(host: DistroHost): Promise<void> {
const runtime = await ensureWslWatcherRuntime(host.distro)
const child = spawn(
'wsl.exe',
['-d', host.distro, '--exec', runtime.nodePath, runtime.hostPath],
{ stdio: ['pipe', 'pipe', 'pipe'], windowsHide: true }
)
host.process = child
host.streamBuffer = ''
host.stderrTail = ''
await new Promise<void>((resolve, reject) => {
let settled = false
const settle = (error?: Error): void => {
if (settled) {
return
}
settled = true
clearTimeout(timer)
if (error) {
reject(error)
} else {
resolve()
}
}
const timer = setTimeout(() => {
settle(new Error(`Timed out starting managed WSL watcher for ${host.distro}`))
child.kill()
}, STARTUP_TIMEOUT_MS)
child.stdout.on('data', (chunk: Buffer) => {
host.streamBuffer += host.stdoutDecoder.write(chunk)
if (host.streamBuffer.length > MAX_STREAM_BUFFER_CHARS) {
settle(new Error('Managed WSL watcher protocol buffer overflow'))
child.kill()
return
}
let newline = host.streamBuffer.indexOf('\n')
while (newline !== -1) {
const line = host.streamBuffer.slice(0, newline)
host.streamBuffer = host.streamBuffer.slice(newline + 1)
try {
handleMessage(host, JSON.parse(line) as WslHostMessage, () => settle())
} catch (error) {
settle(error instanceof Error ? error : new Error(String(error)))
child.kill()
return
}
newline = host.streamBuffer.indexOf('\n')
}
})
child.stderr.on('data', (chunk: Buffer) => {
host.stderrTail = (host.stderrTail + host.stderrDecoder.write(chunk)).slice(-4096)
})
child.once('error', (error) => {
settle(error)
handleHostGone(host, child)
})
child.once('close', (code, signal) => {
settle(new Error(`Managed WSL watcher exited before ready (${code ?? signal})`))
handleHostGone(host, child)
})
})
}
function ensureHost(distro: string): Promise<DistroHost> {
let host = hosts.get(distro)
if (!host) {
host = createHost(distro)
hosts.set(distro, host)
}
if (!host.starting) {
host.starting = startHost(host)
.catch((error) => {
if (host?.process) {
host.process.kill()
}
hosts.delete(distro)
throw error
})
.finally(() => {
if (host) {
host.starting = null
}
})
}
return host.starting.then(() => host!)
}
export async function subscribeViaWslWatcherHost(
context: WslHostSubscriptionContext,
signal?: AbortSignal
): Promise<WslHostSubscription> {
if (signal?.aborted) {
throw new Error('Managed WSL watcher subscription cancelled')
}
const host = await ensureHost(context.distro)
if (signal?.aborted) {
if (host.subscriptions.size === 0) {
hosts.delete(host.distro)
const child = host.process
host.process = null
child?.kill()
}
throw new Error('Managed WSL watcher subscription cancelled')
}
const id = host.nextId++
const record: SubscriptionRecord = { id, context, pending: null, pendingTimer: null }
const ready = new Promise<void>((resolve, reject) => {
record.pending = { resolve, reject }
})
host.subscriptions.set(id, record)
record.pendingTimer = setTimeout(() => {
settleRecordFailure(host, record, new Error('Timed out subscribing managed WSL watcher'))
}, SUBSCRIBE_TIMEOUT_MS)
signal?.addEventListener(
'abort',
() => {
if (host.subscriptions.has(id)) {
settleRecordFailure(host, record, new Error('Managed WSL watcher subscription cancelled'))
}
},
{ once: true }
)
if (
!send(host, {
op: 'subscribe',
id,
dir: context.linuxPath,
ignoreDirs: context.ignoreDirs
})
) {
settleRecordFailure(host, record, new Error('Managed WSL watcher is unavailable'))
}
await ready
return {
unsubscribe: () => {
if (!host.subscriptions.delete(id)) {
return
}
if (host.subscriptions.size === 0) {
hosts.delete(host.distro)
const child = host.process
host.process = null
child?.kill()
} else {
send(host, { op: 'unsubscribe', id })
}
}
}
}
export function resetWslWatcherHostsForTest(): void {
for (const host of hosts.values()) {
host.process?.kill()
}
hosts.clear()
}
@@ -0,0 +1,37 @@
import type { Event as WatcherEvent } from '@parcel/watcher'
import { toWslUncPath } from './filesystem-watcher-wsl-snapshot'
export type WslHostMessage =
| { op: 'ready'; protocol?: unknown }
| { op: 'subscribed'; id?: unknown }
| { op: 'subscribe-failed'; id?: unknown; message?: unknown }
| { op: 'events'; id?: unknown; events?: unknown }
| { op: 'watch-error'; id?: unknown; message?: unknown }
| { op: 'unsubscribed'; id?: unknown }
| { op: 'protocol-error'; message?: unknown }
export function wslHostMessageEvents(
message: WslHostMessage,
context: { distro: string; linuxPath: string }
): WatcherEvent[] {
if (message.op !== 'events' || !Array.isArray(message.events)) {
return []
}
const root = context.linuxPath.replace(/\/+$/, '') || '/'
const prefix = root === '/' ? '/' : `${root}/`
const events: WatcherEvent[] = []
for (const event of message.events) {
const candidate = event as { type?: unknown; path?: unknown }
if (
(candidate.type === 'create' || candidate.type === 'update' || candidate.type === 'delete') &&
typeof candidate.path === 'string' &&
(candidate.path === root || candidate.path.startsWith(prefix))
) {
events.push({
type: candidate.type,
path: toWslUncPath(candidate.path, context.distro)
})
}
}
return events
}
@@ -2,6 +2,7 @@ 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'
import { ensureWslWatcherRuntime } from './filesystem-watcher-wsl-runtime'
const execFileAsync = promisify(execFile)
const runLive = process.platform === 'win32' && process.env.ORCA_WSL_WATCHER_LIVE === '1'
@@ -19,7 +20,9 @@ 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('mkdir', '-p', `${fixturePath}/packages/app/node_modules/pkg`)
await runWsl('touch', `${fixturePath}/docs/deep/README.md`)
await runWsl('touch', `${fixturePath}/packages/app/node_modules/pkg/ignored.js`)
})
afterAll(async () => {
@@ -31,6 +34,8 @@ describe.skipIf(!runLive)('WSL watcher live integration', () => {
it('reports a deep external edit without periodic polling events', async () => {
const rootPath = `\\\\wsl.localhost\\${distro}${fixturePath.replace(/\//g, '\\')}`
const runtime = await ensureWslWatcherRuntime(distro)
await execFileAsync('wsl.exe', ['-d', distro, '--', 'test', '-x', runtime.nodePath])
const batches: WatchedRoot[] = []
const root = await createWslWatcher(rootPath, rootPath, {
ignoreDirs: ['node_modules', '.git'],
@@ -48,8 +53,16 @@ describe.skipIf(!runLive)('WSL watcher live integration', () => {
path: `${rootPath}\\docs\\deep\\README.md`
})
},
{ timeout: 5_000 }
// Native event delivery should not need to wait for the 5-second scan fallback.
{ timeout: 2_000 }
)
root.batch.events = []
batches.length = 0
await runWsl('touch', `${fixturePath}/packages/app/node_modules/pkg/ignored.js`)
await new Promise((resolve) => setTimeout(resolve, 750))
expect(batches).toHaveLength(0)
expect(root.batch.events).toHaveLength(0)
} finally {
await root.subscription.unsubscribe()
}
@@ -1,173 +0,0 @@
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,21 @@
import { describe, expect, it } from 'vitest'
import { INSTALL_SCRIPT, parseRunningWslDistros } from './filesystem-watcher-wsl-runtime'
describe('managed WSL watcher runtime', () => {
it('parses running distro output across WSL encodings', () => {
expect(parseRunningWslDistros('U\0b\0u\0n\0t\0u\0\r\0\n\0* Debian\r\n')).toEqual([
'Ubuntu',
'Debian'
])
})
it('installs a verified architecture-specific host without a user runtime', () => {
expect(INSTALL_SCRIPT).toContain('case "$(uname -m)"')
expect(INSTALL_SCRIPT).toContain('getconf GNU_LIBC_VERSION')
expect(INSTALL_SCRIPT).toContain('node-v$node_version-linux-$arch/bin/node')
expect(INSTALL_SCRIPT).toContain('node-v$node_version-linux-$arch/LICENSE')
expect(INSTALL_SCRIPT).toContain('parcel-watcher-LICENSE')
expect(INSTALL_SCRIPT).toContain('"$stage/node" "$stage/host.js" --check')
expect(INSTALL_SCRIPT).not.toContain('python')
})
})
@@ -0,0 +1,247 @@
import { execFile, spawn } from 'node:child_process'
import { existsSync } from 'node:fs'
import { readFile } from 'node:fs/promises'
import { join, resolve } from 'node:path'
import { promisify } from 'node:util'
type WslWatcherManifest = {
protocol: 1
installLayout: 1
nodeVersion: string
bundleVersion: string
}
export type InstalledWslWatcherRuntime = {
nodePath: string
hostPath: string
}
const execFileAsync = promisify(execFile)
const installationPromises = new Map<string, Promise<InstalledWslWatcherRuntime>>()
const runningQueries = new Map<string, Promise<boolean>>()
const INSTALL_SCRIPT = String.raw`
set -efu
bundle_windows=$1
version=$2
node_version=$3
case "$(uname -m)" in
x86_64) arch=x64 ;;
aarch64|arm64) arch=arm64 ;;
*) echo "unsupported WSL architecture: $(uname -m)" >&2; exit 70 ;;
esac
if ! getconf GNU_LIBC_VERSION >/dev/null 2>&1; then
echo "managed WSL watcher requires a glibc distro" >&2
exit 71
fi
bundle_linux=$(wslpath -u "$bundle_windows")
archive="$bundle_linux/$arch/node.tar.xz"
watcher="$bundle_linux/$arch/watcher.node"
host="$bundle_linux/host.js"
watcher_license="$bundle_linux/parcel-watcher-LICENSE"
test -f "$archive" && test -f "$watcher" && test -f "$host" && test -f "$watcher_license"
base="$HOME/.local/share/orca/wsl-watcher"
install="$base/$version/$arch"
complete="$install/.complete"
if test -x "$install/node" && test -f "$complete"; then
printf '%s\n%s\n' "$install/node" "$install/host.js"
exit 0
fi
mkdir -p "$base/$version"
lock="$base/$version/.install-$arch.lock"
attempt=0
while ! mkdir "$lock" 2>/dev/null; do
if test -r "$lock/pid"; then
owner=$(cat "$lock/pid" 2>/dev/null || true)
if test -n "$owner" && ! kill -0 "$owner" 2>/dev/null; then
rm -rf -- "$lock"
continue
fi
fi
if test -x "$install/node" && test -f "$complete"; then
printf '%s\n%s\n' "$install/node" "$install/host.js"
exit 0
fi
attempt=$((attempt + 1))
test "$attempt" -lt 300 || { echo "timed out waiting for WSL watcher install" >&2; exit 72; }
sleep 0.1
done
printf '%s\n' "$$" >"$lock/pid"
if test -x "$install/node" && test -f "$complete"; then
rm -rf -- "$lock"
printf '%s\n%s\n' "$install/node" "$install/host.js"
exit 0
fi
stage="$base/version-$version-$arch-$$.tmp"
cleanup() { rm -rf -- "$stage" "$lock"; }
trap cleanup EXIT HUP INT TERM
rm -rf -- "$stage"
mkdir -p "$stage"
tar -xJf "$archive" -C "$stage" --strip-components=2 \
"node-v$node_version-linux-$arch/bin/node"
tar -xJf "$archive" -C "$stage" --strip-components=1 \
"node-v$node_version-linux-$arch/LICENSE"
cp "$host" "$stage/host.js"
cp "$watcher" "$stage/watcher.node"
cp "$watcher_license" "$stage/parcel-watcher-LICENSE"
chmod 700 "$stage/node" "$stage/host.js"
"$stage/node" "$stage/host.js" --check >/dev/null
printf '%s\n' "$version" >"$stage/.complete"
rm -rf -- "$install"
mv "$stage" "$install"
printf '%s\n%s\n' "$install/node" "$install/host.js"
`
function resolveBundlePath(): string {
if (process.resourcesPath) {
const packaged = join(process.resourcesPath, 'wsl-watcher')
if (existsSync(packaged)) {
return packaged
}
}
const development = resolve(process.cwd(), 'out', 'wsl-watcher')
if (existsSync(development)) {
return development
}
// In development __dirname is out/main and the prepared bundle is out/wsl-watcher.
return resolve(__dirname, '..', 'wsl-watcher')
}
function validateManifest(value: unknown): WslWatcherManifest {
const manifest = value as Partial<WslWatcherManifest>
if (
manifest?.protocol !== 1 ||
manifest.installLayout !== 1 ||
typeof manifest.nodeVersion !== 'string' ||
!/^\d+\.\d+\.\d+$/.test(manifest.nodeVersion) ||
typeof manifest.bundleVersion !== 'string' ||
!/^[a-f0-9]{20}$/.test(manifest.bundleVersion)
) {
throw new Error('Invalid managed WSL watcher manifest')
}
return manifest as WslWatcherManifest
}
async function installRuntime(distro: string): Promise<InstalledWslWatcherRuntime> {
const bundlePath = resolveBundlePath()
const manifest = validateManifest(
JSON.parse(await readFile(join(bundlePath, 'manifest.json'), 'utf8')) as unknown
)
const stdout = await runInstaller(distro, [
bundlePath,
manifest.bundleVersion,
manifest.nodeVersion
])
const [nodePath, hostPath] = stdout
.replaceAll(String.fromCharCode(0), '')
.split(/\r?\n/)
.map((line) => line.trim())
.filter(Boolean)
if (!nodePath?.startsWith('/') || !hostPath?.startsWith('/')) {
throw new Error('Managed WSL watcher installer returned invalid paths')
}
return { nodePath, hostPath }
}
function runInstaller(distro: string, args: string[]): Promise<string> {
return new Promise((resolvePromise, rejectPromise) => {
const child = spawn('wsl.exe', ['-d', distro, '--exec', 'sh', '-s', '--', ...args], {
stdio: ['pipe', 'pipe', 'pipe'],
windowsHide: true
})
let stdout = ''
let stderr = ''
let settled = false
const settle = (error?: Error): void => {
if (settled) {
return
}
settled = true
clearTimeout(timer)
if (error) {
rejectPromise(error)
} else {
resolvePromise(stdout)
}
}
const timer = setTimeout(() => {
child.kill()
settle(new Error(`Timed out installing managed WSL watcher for ${distro}`))
}, 120_000)
child.stdout.on('data', (chunk: Buffer) => {
stdout += chunk.toString('utf8')
if (stdout.length > 64 * 1024) {
child.kill()
settle(new Error('Managed WSL watcher installer produced too much output'))
}
})
child.stderr.on('data', (chunk: Buffer) => {
stderr = (stderr + chunk.toString('utf8')).slice(-4096)
})
child.once('error', (error) => settle(error))
child.once('close', (code, signal) => {
if (code === 0) {
settle()
} else {
settle(
new Error(
`Managed WSL watcher install failed (${code ?? signal})${stderr.trim() ? `: ${stderr.trim()}` : ''}`
)
)
}
})
child.stdin.once('error', (error) => settle(error))
child.stdin.end(INSTALL_SCRIPT)
})
}
export function ensureWslWatcherRuntime(distro: string): Promise<InstalledWslWatcherRuntime> {
let pending = installationPromises.get(distro)
if (!pending) {
pending = installRuntime(distro).catch((error) => {
installationPromises.delete(distro)
throw error
})
installationPromises.set(distro, pending)
}
return pending
}
async function queryWslDistroRunning(distro: string): Promise<boolean> {
try {
const { stdout } = await execFileAsync('wsl.exe', ['--list', '--running', '--quiet'], {
encoding: 'utf8',
timeout: 5_000,
maxBuffer: 64 * 1024
})
return parseRunningWslDistros(stdout).some(
(runningDistro) => runningDistro.toLowerCase() === distro.toLowerCase()
)
} catch {
return false
}
}
export function parseRunningWslDistros(output: string): string[] {
return output
.replaceAll(String.fromCharCode(0), '')
.split(/\r?\n/)
.map((line) => line.trim().replace(/^\*\s*/, ''))
.filter(Boolean)
}
export function isWslDistroRunning(distro: string): Promise<boolean> {
let pending = runningQueries.get(distro)
if (!pending) {
pending = queryWslDistroRunning(distro).finally(() => runningQueries.delete(distro))
runningQueries.set(distro, pending)
}
return pending
}
export function resetWslWatcherRuntimeForTest(): void {
installationPromises.clear()
runningQueries.clear()
}
export { INSTALL_SCRIPT }
@@ -1,7 +1,7 @@
import type { Event as WatcherEvent } from '@parcel/watcher'
import { validateWslWatcherIgnoreDirs } from './filesystem-watcher-wsl-native-script'
const POLL_INTERVAL_SECONDS = 5
const SAFE_IGNORE_NAME = /^[A-Za-z0-9_.-]+$/
export const SNAPSHOT_START = '\x1e'
export const SNAPSHOT_END = '\x1f'
@@ -13,6 +13,14 @@ type WslSnapshotEntry = {
export type WslSnapshot = Map<string, WslSnapshotEntry>
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}`)
}
}
}
export function toWslUncPath(linuxPath: string, distro: string): string {
return `\\\\wsl.localhost\\${distro}${linuxPath.replace(/\//g, '\\')}`
}
+107 -65
View File
@@ -2,27 +2,45 @@ 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 { ensureRuntimeMock, isDistroRunningMock, spawnMock } = vi.hoisted(() => ({
ensureRuntimeMock: vi.fn(),
isDistroRunningMock: vi.fn(),
spawnMock: vi.fn()
}))
vi.mock('child_process', () => ({ spawn: spawnMock }))
vi.mock('node:child_process', () => ({ spawn: spawnMock }))
vi.mock('./filesystem-watcher-wsl-runtime', () => ({
ensureWslWatcherRuntime: ensureRuntimeMock,
isWslDistroRunning: isDistroRunningMock
}))
import { createWslWatcher } from './filesystem-watcher-wsl'
import type { WatchedRoot, WslWatcherDeps } from './filesystem-watcher-wsl'
import { resetWslWatcherHostsForTest } from './filesystem-watcher-wsl-host-client'
import { SNAPSHOT_END, SNAPSHOT_START } from './filesystem-watcher-wsl-snapshot'
const ROOT_KEY = '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo'
const SECOND_ROOT_KEY = '\\\\wsl.localhost\\Ubuntu\\home\\me\\other'
class FakeChildProcess extends EventEmitter {
stdin = new PassThrough()
stdout = new PassThrough()
stderr = new PassThrough()
stdinData = ''
kill = vi.fn(() => {
this.emit('close', null, 'SIGTERM')
return true
})
constructor() {
super()
this.stdin.on('data', (chunk: Buffer) => {
this.stdinData += chunk.toString('utf8')
})
}
}
function nativeMessage(message: unknown): string {
function hostMessage(message: unknown): string {
return `${JSON.stringify(message)}\n`
}
@@ -46,38 +64,59 @@ function queueChildren(...children: FakeChildProcess[]): void {
}
}
async function readyHost(child: FakeChildProcess, ids = [1]): Promise<void> {
child.stdout.write(hostMessage({ op: 'ready', protocol: 1 }))
await vi.waitFor(() => expect(child.stdinData).toContain('"op":"subscribe"'))
for (const id of ids) {
child.stdout.write(hostMessage({ op: 'subscribed', id }))
}
}
async function startNativeWatcher(
deps = makeDeps()
): Promise<{ child: FakeChildProcess; root: WatchedRoot; deps: ReturnType<typeof makeDeps> }> {
const child = new FakeChildProcess()
queueChildren(child)
const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, deps)
child.stdout.write(nativeMessage({ type: 'ready' }))
await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledOnce())
await readyHost(child)
return { child, root: await promise, deps }
}
describe('createWslWatcher', () => {
beforeEach(() => {
spawnMock.mockReset()
ensureRuntimeMock.mockReset().mockResolvedValue({
nodePath: '/home/me/.local/share/orca/wsl-watcher/version/x64/node',
hostPath: '/home/me/.local/share/orca/wsl-watcher/version/x64/host.js'
})
isDistroRunningMock.mockReset().mockResolvedValue(true)
})
afterEach(() => {
resetWslWatcherHostsForTest()
vi.useRealTimers()
vi.restoreAllMocks()
})
it('starts native Linux events without requiring an inotify utility package', async () => {
const { child } = await startNativeWatcher()
it('starts Orca managed Linux events without Python or a distro Node dependency', async () => {
const { child, root } = await startNativeWatcher()
expect(spawnMock).toHaveBeenCalledWith(
'wsl.exe',
['-d', 'Ubuntu', '--', 'python3', '-u', '-', '/home/me/repo', 'node_modules', '.git'],
[
'-d',
'Ubuntu',
'--exec',
'/home/me/.local/share/orca/wsl-watcher/version/x64/node',
'/home/me/.local/share/orca/wsl-watcher/version/x64/host.js'
],
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')
expect(child.stdinData).toContain('"dir":"/home/me/repo"')
expect(child.stdinData).toContain('"ignoreDirs":["node_modules",".git"]')
expect(child.stdinData).not.toContain('python')
await root.subscription.unsubscribe()
})
it('streams nested native changes as UNC watcher events', async () => {
@@ -85,13 +124,14 @@ describe('createWslWatcher', () => {
const { child, root } = await startNativeWatcher(makeDeps(scheduleBatchFlush))
child.stdout.write(
nativeMessage({
type: 'events',
hostMessage({
op: 'events',
id: 1,
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']
{ type: 'update', path: '/home/me/repo/docs/deep/README.md' },
{ type: 'create', path: '/home/me/repo/new.txt' },
{ type: 'delete', path: '/home/me/repo/old.txt' },
{ type: 'update', path: '/home/me/repository/outside.txt' }
]
})
)
@@ -102,20 +142,40 @@ describe('createWslWatcher', () => {
{ type: 'create', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\new.txt' },
{ type: 'delete', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\old.txt' }
])
await root.subscription.unsubscribe()
})
it('uses recursive snapshots when the native runtime is unavailable', async () => {
const native = new FakeChildProcess()
it('shares one Linux host across roots in the same distro', async () => {
const child = new FakeChildProcess()
queueChildren(child)
const firstPromise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps())
const secondPromise = createWslWatcher(SECOND_ROOT_KEY, SECOND_ROOT_KEY, makeDeps())
await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledOnce())
child.stdout.write(hostMessage({ op: 'ready', protocol: 1 }))
await vi.waitFor(() => {
expect(child.stdinData).toContain('"dir":"/home/me/repo"')
expect(child.stdinData).toContain('"dir":"/home/me/other"')
})
child.stdout.write(hostMessage({ op: 'subscribed', id: 1 }))
child.stdout.write(hostMessage({ op: 'subscribed', id: 2 }))
const [first, second] = await Promise.all([firstPromise, secondPromise])
await first.subscription.unsubscribe()
expect(child.kill).not.toHaveBeenCalled()
await second.subscription.unsubscribe()
expect(child.kill).toHaveBeenCalledOnce()
})
it('uses recursive snapshots when the managed runtime is unavailable', async () => {
ensureRuntimeMock.mockRejectedValueOnce(new Error('managed runtime unavailable'))
const snapshot = new FakeChildProcess()
queueChildren(native, snapshot)
queueChildren(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))
await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledOnce())
snapshot.stdout.write(snapshotFrame([]))
await promise
const root = await promise
expect(spawnMock.mock.calls[1]?.[1]).toEqual([
expect(spawnMock.mock.calls[0]?.[1]).toEqual([
'-d',
'Ubuntu',
'--',
@@ -124,19 +184,18 @@ describe('createWslWatcher', () => {
'--',
'/home/me/repo'
])
const script = snapshot.stdin.read()?.toString('utf8') ?? ''
expect(script).toContain('find "$root" -mindepth 1')
expect(script).not.toContain('-maxdepth')
expect(snapshot.stdinData).toContain('find "$root" -mindepth 1')
expect(snapshot.stdinData).not.toContain('-maxdepth')
await root.subscription.unsubscribe()
})
it('diffs fallback snapshots into create, update, and delete events', async () => {
ensureRuntimeMock.mockRejectedValueOnce(new Error('managed runtime unavailable'))
const scheduleBatchFlush = vi.fn<ScheduleBatchFlush>()
const native = new FakeChildProcess()
const snapshot = new FakeChildProcess()
queueChildren(native, snapshot)
queueChildren(snapshot)
const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps(scheduleBatchFlush))
native.emit('error', new Error('spawn failed'))
await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledTimes(2))
await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledOnce())
snapshot.stdout.write(
snapshotFrame([
['f', '1.0', '/home/me/repo/README.md'],
@@ -156,32 +215,34 @@ describe('createWslWatcher', () => {
{ type: 'create', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\new.txt' },
{ type: 'delete', path: '\\\\wsl.localhost\\Ubuntu\\home\\me\\repo\\old.txt' }
])
await root.subscription.unsubscribe()
})
it('refreshes and restarts an active watcher after WSL exits', async () => {
it('waits for an intentionally stopped distro before restarting', async () => {
vi.useFakeTimers()
isDistroRunningMock.mockResolvedValueOnce(false).mockResolvedValue(true)
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' }))
await vi.advanceTimersByTimeAsync(0)
await readyHost(first)
const root = await promise
first.emit('close', null, 'SIGTERM')
await vi.advanceTimersByTimeAsync(0)
await vi.advanceTimersByTimeAsync(500)
expect(spawnMock).toHaveBeenCalledOnce()
await vi.advanceTimersByTimeAsync(5_000)
expect(spawnMock).toHaveBeenCalledTimes(2)
await readyHost(second)
expect(root.batch.overflowed).toBe(true)
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 refresh', async () => {
it('kills the shared WSL host on the last unsubscribe without emitting a refresh', async () => {
const scheduleBatchFlush = vi.fn<ScheduleBatchFlush>()
const { child, root } = await startNativeWatcher(makeDeps(scheduleBatchFlush))
@@ -191,31 +252,12 @@ describe('createWslWatcher', () => {
expect(scheduleBatchFlush).not.toHaveBeenCalled()
})
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)
await root.subscription.unsubscribe()
expect(restarting.kill).toHaveBeenCalledOnce()
await vi.advanceTimersByTimeAsync(10_000)
expect(spawnMock).toHaveBeenCalledTimes(2)
})
it('rejects only after both native and snapshot startup fail', async () => {
const native = new FakeChildProcess()
it('rejects only after both managed native watching and snapshot startup fail', async () => {
ensureRuntimeMock.mockRejectedValueOnce(new Error('managed runtime unavailable'))
const snapshot = new FakeChildProcess()
queueChildren(native, snapshot)
queueChildren(snapshot)
const promise = createWslWatcher(ROOT_KEY, ROOT_KEY, makeDeps())
native.emit('error', new Error('native failed'))
await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledTimes(2))
await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledOnce())
snapshot.stderr.write('find failed')
snapshot.emit('close', 1, null)
+16 -3
View File
@@ -8,6 +8,7 @@ import {
type WslEngineContext,
type WslWatchEngine
} from './filesystem-watcher-wsl-engine'
import { isWslDistroRunning } from './filesystem-watcher-wsl-runtime'
import { parseWslUncPath } from '../../shared/wsl-paths'
export type WatcherSubscription = {
@@ -34,6 +35,7 @@ export type WslWatcherDeps = {
const RESTART_DELAYS_MS = [500, 1_000, 2_000, 5_000, 10_000] as const
const STABLE_ENGINE_MS = 30_000
const STOPPED_DISTRO_RECHECK_MS = 5_000
function markOverflow(root: WatchedRoot): void {
if (root.batch.timer) {
@@ -117,10 +119,21 @@ export async function createWslWatcher(
const scheduleRestart = (delay: number): void => {
restartTimer = setTimeout(() => {
restartTimer = null
void installEngine().catch(() => {
if (!disposed) {
scheduleRestart(maximumRestartDelay)
void isWslDistroRunning(wsl.distro).then((running) => {
if (disposed) {
return
}
if (!running) {
// Why: restarting a watcher through `wsl.exe -d` would undo an
// intentional WSL shutdown; the running-distro query does not wake it.
scheduleRestart(STOPPED_DISTRO_RECHECK_MS)
return
}
void installEngine().catch(() => {
if (!disposed) {
scheduleRestart(maximumRestartDelay)
}
})
})
}, delay)
}
@@ -0,0 +1,69 @@
import { PassThrough } from 'node:stream'
import { describe, expect, it, vi } from 'vitest'
import { buildIgnoreGlobs, startWslWatcherHost } from './wsl-watcher-host-entry'
describe('WSL watcher Linux host', () => {
it('builds nested directory exclusions without treating names as regex', () => {
const [nodeModules, dotted] = buildIgnoreGlobs(['node_modules', '.cache'])
expect(nodeModules).toBe('^(?:.*/)?node_modules(?:/.*)?$')
expect(dotted).toBe('^(?:.*/)?\\.cache(?:/.*)?$')
expect(new RegExp(nodeModules!).test('/repo/packages/app/node_modules/pkg/file.js')).toBe(true)
expect(new RegExp(dotted!).test('/repo/.cache/item')).toBe(true)
})
it('multiplexes subscriptions and native events over JSON lines', async () => {
const input = new PassThrough()
const output = new PassThrough()
let outputText = ''
output.on('data', (chunk: Buffer) => {
outputText += chunk.toString('utf8')
})
let callback: ((error: Error | null, events: unknown[]) => void) | undefined
const subscribe = vi.fn(async (_dir, createdCallback) => {
callback = createdCallback
})
const unsubscribe = vi.fn(async () => undefined)
const exit = vi.fn()
startWslWatcherHost({ subscribe, unsubscribe } as never, input, output, exit)
expect(outputText).toContain('"op":"ready"')
input.write(
`${JSON.stringify({
op: 'subscribe',
id: 7,
dir: '/home/me/repo',
ignoreDirs: ['node_modules']
})}\n`
)
await vi.waitFor(() => expect(outputText).toContain('"op":"subscribed","id":7'))
expect(subscribe).toHaveBeenCalledWith('/home/me/repo', expect.any(Function), {
ignoreGlobs: ['^(?:.*/)?node_modules(?:/.*)?$']
})
callback?.(null, [{ type: 'update', path: '/home/me/repo/README.md' }])
expect(outputText).toContain('"op":"events","id":7')
input.write(`${JSON.stringify({ op: 'unsubscribe', id: 7 })}\n`)
await vi.waitFor(() => expect(outputText).toContain('"op":"unsubscribed","id":7'))
expect(unsubscribe).toHaveBeenCalledOnce()
input.end()
await vi.waitFor(() => expect(exit).toHaveBeenCalledWith(0))
})
it('rejects malformed commands without invoking the native binding', async () => {
const input = new PassThrough()
const output = new PassThrough()
let outputText = ''
output.on('data', (chunk: Buffer) => {
outputText += chunk.toString('utf8')
})
const subscribe = vi.fn()
startWslWatcherHost({ subscribe, unsubscribe: vi.fn() } as never, input, output, vi.fn())
input.write('{not-json}\n')
await vi.waitFor(() => expect(outputText).toContain('"op":"protocol-error"'))
expect(subscribe).not.toHaveBeenCalled()
input.end()
})
})
+213
View File
@@ -0,0 +1,213 @@
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { createInterface, type Interface } from 'node:readline'
import type { Readable, Writable } from 'node:stream'
type WatcherEvent = {
type: 'create' | 'update' | 'delete'
path: string
}
type NativeWatcherOptions = {
ignoreGlobs?: string[]
}
type NativeWatcherBinding = {
subscribe(
dir: string,
callback: (error: Error | null, events: WatcherEvent[]) => void,
options: NativeWatcherOptions
): Promise<void>
unsubscribe(
dir: string,
callback: (error: Error | null, events: WatcherEvent[]) => void,
options: NativeWatcherOptions
): Promise<void>
}
type HostCommand =
| { op: 'subscribe'; id: number; dir: string; ignoreDirs: string[] }
| { op: 'unsubscribe'; id: number }
type Subscription = {
dir: string
callback: (error: Error | null, events: WatcherEvent[]) => void
options: NativeWatcherOptions
ready: Promise<void>
}
type HostOutput = Pick<Writable, 'write'>
const CANARY_INTERVAL_MS = 10_000
const CANARY_EVENT_TIMEOUT_MS = 5_000
const CANARY_MAX_MISSES = 2
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error)
}
function escapeRegex(value: string): string {
return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')
}
export function buildIgnoreGlobs(ignoreDirs: readonly string[]): string[] {
return ignoreDirs.map((dir) => `^(?:.*/)?${escapeRegex(dir)}(?:/.*)?$`)
}
function parseCommand(line: string): HostCommand | null {
try {
const command = JSON.parse(line) as Partial<HostCommand>
if (!Number.isSafeInteger(command.id) || (command.id ?? 0) <= 0) {
return null
}
if (command.op === 'unsubscribe') {
return { op: 'unsubscribe', id: command.id! }
}
if (
command.op === 'subscribe' &&
typeof command.dir === 'string' &&
command.dir.startsWith('/') &&
Array.isArray(command.ignoreDirs) &&
command.ignoreDirs.every((dir) => typeof dir === 'string')
) {
return {
op: 'subscribe',
id: command.id!,
dir: command.dir,
ignoreDirs: command.ignoreDirs
}
}
} catch {
return null
}
return null
}
export function startWslWatcherHost(
binding: NativeWatcherBinding,
input: Readable,
output: HostOutput,
exit: (code: number) => void = process.exit
): { close(): Promise<void> } {
const subscriptions = new Map<number, Subscription>()
const send = (message: object): void => {
output.write(`${JSON.stringify(message)}\n`)
}
const subscribe = (command: Extract<HostCommand, { op: 'subscribe' }>): void => {
if (subscriptions.has(command.id)) {
send({ op: 'subscribe-failed', id: command.id, message: 'duplicate subscription id' })
return
}
const options = { ignoreGlobs: buildIgnoreGlobs(command.ignoreDirs) }
const callback = (error: Error | null, events: WatcherEvent[]): void => {
if (error) {
send({ op: 'watch-error', id: command.id, message: errorMessage(error) })
} else if (events.length > 0) {
send({ op: 'events', id: command.id, events })
}
}
const ready = binding
.subscribe(command.dir, callback, options)
.then(() => send({ op: 'subscribed', id: command.id }))
.catch((error: unknown) => {
subscriptions.delete(command.id)
send({ op: 'subscribe-failed', id: command.id, message: errorMessage(error) })
})
subscriptions.set(command.id, { dir: command.dir, callback, options, ready })
}
const unsubscribe = async (id: number): Promise<void> => {
const subscription = subscriptions.get(id)
subscriptions.delete(id)
try {
await subscription?.ready
if (subscription) {
await binding.unsubscribe(subscription.dir, subscription.callback, subscription.options)
}
} catch (error) {
process.stderr.write(`[wsl-watcher-host] unsubscribe ${id}: ${errorMessage(error)}\n`)
}
send({ op: 'unsubscribed', id })
}
let closing: Promise<void> | null = null
const close = (): Promise<void> => {
if (!closing) {
closing = Promise.all(Array.from(subscriptions.keys(), unsubscribe)).then(() => undefined)
}
return closing
}
const lines: Interface = createInterface({ input, crlfDelay: Infinity })
lines.on('line', (line) => {
const command = parseCommand(line)
if (!command) {
send({ op: 'protocol-error', message: 'invalid command' })
} else if (command.op === 'subscribe') {
subscribe(command)
} else {
void unsubscribe(command.id)
}
})
lines.once('close', () => {
void close().finally(() => exit(0))
})
send({ op: 'ready', protocol: 1 })
return { close }
}
async function startCanary(binding: NativeWatcherBinding): Promise<void> {
const dir = mkdtempSync(join(tmpdir(), 'orca-wsl-watcher-'))
const options: NativeWatcherOptions = {}
let lastEventAt = 0
const callback = (error: Error | null): void => {
if (!error) {
lastEventAt = Date.now()
}
}
await binding.subscribe(dir, callback, options)
process.once('exit', () => {
try {
rmSync(dir, { recursive: true, force: true })
} catch {
// Temporary canary cleanup is best-effort during process exit.
}
})
let misses = 0
setInterval(() => {
const probedAt = Date.now()
try {
writeFileSync(join(dir, 'canary.txt'), String(probedAt))
} catch {
return
}
setTimeout(() => {
if (lastEventAt >= probedAt) {
misses = 0
} else if (++misses >= CANARY_MAX_MISSES) {
process.stderr.write('[wsl-watcher-host] native event delivery stalled\n')
process.exit(2)
}
}, CANARY_EVENT_TIMEOUT_MS)
}, CANARY_INTERVAL_MS)
}
function main(): void {
const binding = require('./watcher.node') as NativeWatcherBinding
if (process.argv.includes('--check')) {
process.stdout.write('ok\n')
return
}
startWslWatcherHost(binding, process.stdin, process.stdout)
// Why: the native watcher can remain alive while its delivery thread stalls;
// a private canary converts that silent failure into a recoverable host exit.
void startCanary(binding).catch((error: unknown) => {
process.stderr.write(`[wsl-watcher-host] canary unavailable: ${errorMessage(error)}\n`)
})
}
if (require.main === module) {
main()
}