mirror of
https://github.com/stablyai/orca.git
synced 2026-10-16 08:03:09 +00:00
Run the file watcher off the main process; make git non-interactive (#5308)
Root-cause replacements for the earlier timeout-based stopgaps. Watcher: @parcel/watcher's Linux/Windows brute-force backend recursively walks the whole tree on a libuv threadpool thread before subscribe() resolves. In the main/serve process that crawl (huge non-git home dir on NFS) starved the shared 4-thread libuv pool, so static-asset serving and RPC crypto — everything else needing that pool — hung while the event loop stayed alive. Move the watcher into a worker thread (file-watcher-worker.ts + file-watcher-host.ts), mirroring the existing theme-parser/STT workers. The crawl now runs on the worker's own libuv pool and cannot starve the main process by construction — no timeout needed. The per-event stat fanout stays bounded (8 concurrent) inside the worker; worker crash/error surfaces an overflow so the client refreshes. SSH watching (remote host) and the Windows fs.watch path are unchanged. Git: drop the default timeout; instead force non-interactive execution (GIT_TERMINAL_PROMPT=0, empty GIT/SSH_ASKPASS, ssh BatchMode=yes) so a read-path call that hits a credential/host-key prompt fails fast instead of blocking on stdin forever. BatchMode does not change host trust. Verified end-to-end against the built worker under the Electron runtime: native @parcel/watcher loads in the worker, a real fs change is delivered as an FsChangeEvent, and unsubscribe/terminate is clean. Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Jinwoo-H
Orca
@@ -181,6 +181,7 @@ export default defineConfig({
|
||||
'computer-sidecar': resolve('src/main/computer/sidecar-entry.ts'),
|
||||
'stt-worker': resolve('src/main/speech/stt-worker.ts'),
|
||||
'warp-theme-parser-worker': resolve('src/main/warp-themes/warp-theme-parser-worker.ts'),
|
||||
'file-watcher-worker': resolve('src/main/runtime/file-watcher-worker.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(
|
||||
|
||||
@@ -13,7 +13,7 @@ vi.mock('node:child_process', () => ({
|
||||
spawn: spawnMock
|
||||
}))
|
||||
|
||||
import { commandExecFileAsync, gitExecFileAsync, GIT_DEFAULT_TIMEOUT_MS } from './runner'
|
||||
import { commandExecFileAsync, gitExecFileAsync } from './runner'
|
||||
|
||||
type MockChildProcess = EventEmitter & {
|
||||
stdout: EventEmitter
|
||||
@@ -211,48 +211,23 @@ describe('runner execFile timeout handling', () => {
|
||||
expect(child.kill).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// Issue #5308: a hung read-path git command (non-git folder on a wedged NFS
|
||||
// mount, or git waiting on a credential prompt) with NO explicit timeout must
|
||||
// still self-resolve at the default ceiling instead of hanging forever and
|
||||
// wedging the serve runtime for all clients.
|
||||
it('rejects a hung read-path git command at the default timeout when none is set', async () => {
|
||||
// Issue #5308: git read-path calls must be forced non-interactive so a
|
||||
// credential / SSH host-key prompt fails fast instead of blocking forever on
|
||||
// stdin and wedging the serve runtime for all clients.
|
||||
it('runs git non-interactively so a prompt fails fast instead of hanging', async () => {
|
||||
const child = createMockChildProcess(1234)
|
||||
execFileMock.mockReturnValue(child)
|
||||
|
||||
const promise = gitExecFileAsync(['worktree', 'list', '--porcelain', '-z'], {
|
||||
cwd: '/home5/Brian'
|
||||
})
|
||||
const rejection = expect(promise).rejects.toThrow('git timed out.')
|
||||
await vi.advanceTimersByTimeAsync(GIT_DEFAULT_TIMEOUT_MS)
|
||||
|
||||
await rejection
|
||||
expect(child.kill).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// Long-running network subcommands opt out of the default ceiling so large
|
||||
// clones/fetches aren't killed mid-transfer.
|
||||
it('does not apply the default timeout to network subcommands', async () => {
|
||||
const child = createMockChildProcess(1234)
|
||||
let execCallback: ((err: Error | null, stdout: string, stderr: string) => void) | undefined
|
||||
execFileMock.mockImplementation((_cmd, _args, _opts, cb) => {
|
||||
execCallback = cb
|
||||
let capturedEnv: NodeJS.ProcessEnv | undefined
|
||||
execFileMock.mockImplementation((_cmd, _args, opts, cb) => {
|
||||
capturedEnv = opts.env
|
||||
cb(null, '', '')
|
||||
return child
|
||||
})
|
||||
|
||||
let settled = false
|
||||
const promise = gitExecFileAsync(['fetch', '--prune'], { cwd: '/repo' })
|
||||
.catch(() => undefined)
|
||||
.finally(() => {
|
||||
settled = true
|
||||
})
|
||||
// Advance well past the read-path ceiling; fetch must still be running.
|
||||
await vi.advanceTimersByTimeAsync(GIT_DEFAULT_TIMEOUT_MS * 5)
|
||||
expect(settled).toBe(false)
|
||||
expect(child.kill).not.toHaveBeenCalled()
|
||||
await gitExecFileAsync(['worktree', 'list', '--porcelain', '-z'], { cwd: '/home5/Brian' })
|
||||
|
||||
// Let the fetch "finish" so the test doesn't leak a pending promise.
|
||||
execCallback?.(null, '', '')
|
||||
await promise
|
||||
expect(settled).toBe(true)
|
||||
expect(capturedEnv?.GIT_TERMINAL_PROMPT).toBe('0')
|
||||
expect(capturedEnv?.GIT_ASKPASS).toBe('')
|
||||
expect(capturedEnv?.SSH_ASKPASS).toBe('')
|
||||
expect(capturedEnv?.GIT_SSH_COMMAND).toContain('BatchMode=yes')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,55 +0,0 @@
|
||||
// Reproduction harness for issue #5308: a hung git subprocess on the
|
||||
// worktree-resolution / status read path (e.g. a non-git folder on a wedged
|
||||
// NFS mount, or git waiting on a credential/stdin prompt) must not hang
|
||||
// forever. Without a default timeout the promise never settles, the libuv
|
||||
// subprocess slot is held, and the serve runtime stops answering every client.
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { resolveGitDefaultTimeoutMs, GIT_DEFAULT_TIMEOUT_MS } from './runner'
|
||||
|
||||
describe('resolveGitDefaultTimeoutMs', () => {
|
||||
it('honors an explicit caller timeout verbatim', () => {
|
||||
expect(resolveGitDefaultTimeoutMs(['status'], 1234)).toBe(1234)
|
||||
// Explicit 0 means "caller opted out" — never override it.
|
||||
expect(resolveGitDefaultTimeoutMs(['status'], 0)).toBe(0)
|
||||
})
|
||||
|
||||
it('applies the default timeout to read-path commands with no explicit value', () => {
|
||||
// These are the commands on the hot worktree-resolution / status path that
|
||||
// wedged the runtime in #5308. They should never run unbounded.
|
||||
for (const args of [
|
||||
['worktree', 'list', '--porcelain', '-z'],
|
||||
['status', '--porcelain'],
|
||||
['rev-parse', '--show-toplevel'],
|
||||
['symbolic-ref', '--short', 'HEAD'],
|
||||
['config', '--get', 'user.name']
|
||||
]) {
|
||||
expect(resolveGitDefaultTimeoutMs(args, undefined)).toBe(GIT_DEFAULT_TIMEOUT_MS)
|
||||
}
|
||||
})
|
||||
|
||||
it('exempts long-running network subcommands from the default timeout', () => {
|
||||
// Clone/fetch/pull/push are legitimately slow and must not be capped by the
|
||||
// generic read-path default — capping them would break large-repo workflows.
|
||||
for (const args of [
|
||||
['clone', 'https://example.com/repo.git'],
|
||||
['fetch', '--prune'],
|
||||
['pull', '--rebase', 'origin', 'main'],
|
||||
['push', 'origin', 'HEAD'],
|
||||
['remote', 'update'],
|
||||
['submodule', 'update', '--init'],
|
||||
['lfs', 'pull']
|
||||
]) {
|
||||
expect(resolveGitDefaultTimeoutMs(args, undefined)).toBeUndefined()
|
||||
}
|
||||
})
|
||||
|
||||
it('looks past global flags to find the subcommand', () => {
|
||||
// `git -c key=val -C /path fetch` is still a fetch.
|
||||
expect(
|
||||
resolveGitDefaultTimeoutMs(['-c', 'gc.auto=0', '-C', '/repo', 'fetch'], undefined)
|
||||
).toBeUndefined()
|
||||
expect(
|
||||
resolveGitDefaultTimeoutMs(['-c', 'core.pager=cat', 'status', '--porcelain'], undefined)
|
||||
).toBe(GIT_DEFAULT_TIMEOUT_MS)
|
||||
})
|
||||
})
|
||||
+26
-56
@@ -480,64 +480,32 @@ export function gitOptionalLocksDisabledEnv(
|
||||
}
|
||||
}
|
||||
|
||||
// Why: a git subprocess that never exits (waiting on a credential/stdin
|
||||
// prompt, or a wedged NFS/network filesystem under the cwd) otherwise leaves
|
||||
// gitExecFileAsync pending forever. On the headless `serve` runtime these
|
||||
// read-path calls pile up and the whole runtime stops answering clients
|
||||
// (issue #5308). A generous ceiling kills a truly-hung process without
|
||||
// tripping on normal local git work.
|
||||
export const GIT_DEFAULT_TIMEOUT_MS = 60_000
|
||||
|
||||
// Subcommands that legitimately run longer than the read-path ceiling — they
|
||||
// move data over the network or recurse into submodules/LFS. Capping these
|
||||
// would break large-repo clones/fetches, so they opt out of the default.
|
||||
const GIT_UNBOUNDED_SUBCOMMANDS = new Set([
|
||||
'clone',
|
||||
'fetch',
|
||||
'pull',
|
||||
'push',
|
||||
'remote',
|
||||
'submodule',
|
||||
'lfs'
|
||||
])
|
||||
|
||||
// Global flags that precede the subcommand: skip them (and their values, for
|
||||
// the ones that take an argument) to find the real subcommand.
|
||||
const GIT_GLOBAL_FLAGS_WITH_VALUE = new Set(['-c', '-C', '--git-dir', '--work-tree', '--namespace'])
|
||||
|
||||
function findGitSubcommand(args: string[]): string | undefined {
|
||||
for (let i = 0; i < args.length; i++) {
|
||||
const arg = args[i]
|
||||
if (GIT_GLOBAL_FLAGS_WITH_VALUE.has(arg)) {
|
||||
i++ // skip the flag's value
|
||||
continue
|
||||
}
|
||||
if (arg.startsWith('-')) {
|
||||
continue // valueless global flag (e.g. --no-pager)
|
||||
}
|
||||
return arg
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the effective timeout for a git invocation. An explicit caller
|
||||
* timeout (including an explicit 0 opt-out) always wins; otherwise read-path
|
||||
* commands get GIT_DEFAULT_TIMEOUT_MS and long-running network subcommands
|
||||
* stay unbounded.
|
||||
* Force git to be non-interactive so it fails fast instead of blocking forever
|
||||
* on a prompt. Without this, a git read-path call (status, worktree list, …)
|
||||
* that hits an auth/credential prompt or an SSH host-key confirmation hangs on
|
||||
* stdin with no terminal to answer it; on the headless `serve` runtime those
|
||||
* stuck calls pile up and the runtime stops answering all clients (issue #5308).
|
||||
*
|
||||
* - GIT_TERMINAL_PROMPT=0: git refuses to prompt for credentials and errors out.
|
||||
* - GIT_ASKPASS / SSH_ASKPASS='': disable any GUI/askpass credential helper that
|
||||
* would otherwise pop a prompt and block.
|
||||
* - GIT_SSH_COMMAND BatchMode=yes: SSH fails instead of waiting on an
|
||||
* interactive password/host-key prompt. BatchMode does NOT change host trust
|
||||
* (an unknown host still errors, it just won't hang). Only added when the
|
||||
* caller hasn't set its own GIT_SSH_COMMAND.
|
||||
*/
|
||||
export function resolveGitDefaultTimeoutMs(
|
||||
args: string[],
|
||||
explicitTimeout: number | undefined
|
||||
): number | undefined {
|
||||
if (explicitTimeout !== undefined) {
|
||||
return explicitTimeout
|
||||
export function nonInteractiveGitEnv(env: NodeJS.ProcessEnv = process.env): NodeJS.ProcessEnv {
|
||||
const next: NodeJS.ProcessEnv = {
|
||||
...env,
|
||||
GIT_TERMINAL_PROMPT: '0',
|
||||
GIT_ASKPASS: env.GIT_ASKPASS ?? '',
|
||||
SSH_ASKPASS: env.SSH_ASKPASS ?? ''
|
||||
}
|
||||
const subcommand = findGitSubcommand(args)
|
||||
if (subcommand && GIT_UNBOUNDED_SUBCOMMANDS.has(subcommand)) {
|
||||
return undefined
|
||||
if (!next.GIT_SSH_COMMAND) {
|
||||
next.GIT_SSH_COMMAND = 'ssh -o BatchMode=yes'
|
||||
}
|
||||
return GIT_DEFAULT_TIMEOUT_MS
|
||||
return next
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -560,8 +528,10 @@ export async function gitExecFileAsync(
|
||||
cwd: resolved.cwd,
|
||||
encoding: (options.encoding ?? 'utf-8') as BufferEncoding,
|
||||
maxBuffer: options.maxBuffer,
|
||||
timeout: resolveGitDefaultTimeoutMs(args, options.timeout),
|
||||
env: options.env
|
||||
timeout: options.timeout,
|
||||
// Why: never let a git read-path call block on an interactive prompt
|
||||
// (issue #5308) — fail fast instead of hanging the runtime.
|
||||
env: nonInteractiveGitEnv(options.env)
|
||||
})
|
||||
return { stdout: stdout as string, stderr: stderr as string }
|
||||
}
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { FsChangeEvent } from '../../shared/types'
|
||||
|
||||
type MockWorker = {
|
||||
terminated: boolean
|
||||
postedMessages: unknown[]
|
||||
workerData: unknown
|
||||
on(event: string, listener: (arg?: unknown) => void): MockWorker
|
||||
postMessage(message: unknown): void
|
||||
terminate(): Promise<number>
|
||||
emit(event: string, arg?: unknown): void
|
||||
}
|
||||
|
||||
const workerState = vi.hoisted(() => {
|
||||
const instances: MockWorker[] = []
|
||||
class MockWorkerImpl {
|
||||
terminated = false
|
||||
postedMessages: unknown[] = []
|
||||
workerData: unknown
|
||||
private listeners = new Map<string, ((arg?: unknown) => void)[]>()
|
||||
|
||||
constructor(_workerPath: string, options: { workerData?: unknown }) {
|
||||
this.workerData = options.workerData
|
||||
instances.push(this as unknown as MockWorker)
|
||||
}
|
||||
|
||||
on(event: string, listener: (arg?: unknown) => void): this {
|
||||
const list = this.listeners.get(event) ?? []
|
||||
list.push(listener)
|
||||
this.listeners.set(event, list)
|
||||
return this
|
||||
}
|
||||
|
||||
postMessage(message: unknown): void {
|
||||
this.postedMessages.push(message)
|
||||
}
|
||||
|
||||
async terminate(): Promise<number> {
|
||||
this.terminated = true
|
||||
return 0
|
||||
}
|
||||
|
||||
emit(event: string, arg?: unknown): void {
|
||||
for (const listener of this.listeners.get(event) ?? []) {
|
||||
listener(arg)
|
||||
}
|
||||
}
|
||||
}
|
||||
return { instances, MockWorkerImpl }
|
||||
})
|
||||
|
||||
vi.mock('electron', () => ({
|
||||
app: { isPackaged: false }
|
||||
}))
|
||||
|
||||
vi.mock('worker_threads', () => ({
|
||||
Worker: workerState.MockWorkerImpl
|
||||
}))
|
||||
|
||||
import { watchFileExplorerInWorker } from './file-watcher-host'
|
||||
|
||||
function lastWorker(): MockWorker {
|
||||
const worker = workerState.instances.at(-1)
|
||||
if (!worker) {
|
||||
throw new Error('no worker spawned')
|
||||
}
|
||||
return worker
|
||||
}
|
||||
|
||||
describe('watchFileExplorerInWorker', () => {
|
||||
beforeEach(() => {
|
||||
workerState.instances.length = 0
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('resolves to an unsubscribe fn once the worker reports ready', async () => {
|
||||
const promise = watchFileExplorerInWorker('/repo', vi.fn())
|
||||
const worker = lastWorker()
|
||||
expect(worker.workerData).toMatchObject({ rootPath: '/repo' })
|
||||
|
||||
worker.emit('message', { type: 'ready' })
|
||||
const dispose = await promise
|
||||
expect(typeof dispose).toBe('function')
|
||||
})
|
||||
|
||||
it('forwards worker events to the callback only after ready', async () => {
|
||||
const onEvents = vi.fn<(events: FsChangeEvent[]) => void>()
|
||||
const promise = watchFileExplorerInWorker('/repo', onEvents)
|
||||
const worker = lastWorker()
|
||||
worker.emit('message', { type: 'ready' })
|
||||
await promise
|
||||
|
||||
const events: FsChangeEvent[] = [
|
||||
{ kind: 'update', absolutePath: '/repo/a.txt', isDirectory: false }
|
||||
]
|
||||
worker.emit('message', { type: 'events', events })
|
||||
expect(onEvents).toHaveBeenCalledWith(events)
|
||||
})
|
||||
|
||||
it('rejects if the worker errors before the crawl goes live', async () => {
|
||||
const promise = watchFileExplorerInWorker('/repo', vi.fn())
|
||||
const worker = lastWorker()
|
||||
worker.emit('message', { type: 'error', message: 'addon missing' })
|
||||
|
||||
await expect(promise).rejects.toThrow('addon missing')
|
||||
expect(worker.terminated).toBe(true)
|
||||
})
|
||||
|
||||
it('rejects if the worker exits before ready', async () => {
|
||||
const promise = watchFileExplorerInWorker('/repo', vi.fn())
|
||||
const worker = lastWorker()
|
||||
worker.emit('exit', 1)
|
||||
|
||||
await expect(promise).rejects.toThrow(/exited before ready/)
|
||||
})
|
||||
|
||||
it('emits an overflow if a live worker crashes', async () => {
|
||||
const onEvents = vi.fn<(events: FsChangeEvent[]) => void>()
|
||||
const promise = watchFileExplorerInWorker('/repo', onEvents)
|
||||
const worker = lastWorker()
|
||||
worker.emit('message', { type: 'ready' })
|
||||
await promise
|
||||
|
||||
worker.emit('error', new Error('boom'))
|
||||
expect(onEvents).toHaveBeenCalledWith([{ kind: 'overflow', absolutePath: '/repo' }])
|
||||
})
|
||||
|
||||
it('unsubscribes and terminates the worker on dispose', async () => {
|
||||
const promise = watchFileExplorerInWorker('/repo', vi.fn())
|
||||
const worker = lastWorker()
|
||||
worker.emit('message', { type: 'ready' })
|
||||
const dispose = await promise
|
||||
|
||||
dispose()
|
||||
expect(worker.postedMessages).toContainEqual({ type: 'unsubscribe' })
|
||||
expect(worker.terminated).toBe(true)
|
||||
|
||||
// Idempotent: a second dispose does nothing further.
|
||||
dispose()
|
||||
expect(
|
||||
worker.postedMessages.filter((m) => (m as { type?: string }).type === 'unsubscribe')
|
||||
).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('stops forwarding events after dispose', async () => {
|
||||
const onEvents = vi.fn<(events: FsChangeEvent[]) => void>()
|
||||
const promise = watchFileExplorerInWorker('/repo', onEvents)
|
||||
const worker = lastWorker()
|
||||
worker.emit('message', { type: 'ready' })
|
||||
const dispose = await promise
|
||||
dispose()
|
||||
onEvents.mockClear()
|
||||
|
||||
worker.emit('message', {
|
||||
type: 'events',
|
||||
events: [{ kind: 'update', absolutePath: '/repo/a.txt' }]
|
||||
})
|
||||
expect(onEvents).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,110 @@
|
||||
// Why: spawns the file-watcher worker thread and adapts it to the synchronous
|
||||
// `watchFileExplorer` contract (a promise that resolves to an unsubscribe fn
|
||||
// once the recursive crawl is live). Running @parcel/watcher in the worker
|
||||
// keeps its blocking initial crawl off the main process's libuv pool so a huge
|
||||
// non-git tree can't wedge the `serve` runtime (issue #5308).
|
||||
import { Worker } from 'worker_threads'
|
||||
import { join } from 'path'
|
||||
import { app } from 'electron'
|
||||
import type { FsChangeEvent } from '../../shared/types'
|
||||
import type { FileWatcherHostMessage, FileWatcherWorkerMessage } from './file-watcher-worker'
|
||||
|
||||
// Mirrors VS Code's predefined recursive-watch excludes: skip churny generated
|
||||
// trees at crawl time so the watcher never traverses them.
|
||||
const RUNTIME_FILE_WATCH_IGNORE = [
|
||||
'.git',
|
||||
'node_modules',
|
||||
'dist',
|
||||
'build',
|
||||
'.next',
|
||||
'.cache',
|
||||
'__pycache__',
|
||||
'target',
|
||||
'.venv'
|
||||
]
|
||||
|
||||
function getFileWatcherWorkerPath(): string {
|
||||
if (app.isPackaged) {
|
||||
return join(process.resourcesPath, 'app.asar', 'out', 'main', 'file-watcher-worker.js')
|
||||
}
|
||||
return join(__dirname, 'file-watcher-worker.js')
|
||||
}
|
||||
|
||||
/** Start a recursive file watch in a worker thread. Resolves to an unsubscribe
|
||||
* function once the worker reports the crawl is live; rejects if the worker
|
||||
* fails to start the watch. */
|
||||
export function watchFileExplorerInWorker(
|
||||
rootPath: string,
|
||||
callback: (events: FsChangeEvent[]) => void
|
||||
): Promise<() => void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const worker = new Worker(getFileWatcherWorkerPath(), {
|
||||
workerData: { rootPath, ignore: RUNTIME_FILE_WATCH_IGNORE }
|
||||
})
|
||||
|
||||
let ready = false
|
||||
let disposed = false
|
||||
|
||||
const dispose = (): void => {
|
||||
if (disposed) {
|
||||
return
|
||||
}
|
||||
disposed = true
|
||||
// Ask the worker to unsubscribe its native watcher, then terminate as a
|
||||
// backstop in case the worker is wedged and never closes its own port.
|
||||
try {
|
||||
worker.postMessage({ type: 'unsubscribe' } satisfies FileWatcherHostMessage)
|
||||
} catch {
|
||||
// Worker already gone — terminate below covers it.
|
||||
}
|
||||
void worker.terminate()
|
||||
}
|
||||
|
||||
worker.on('message', (message: FileWatcherWorkerMessage) => {
|
||||
if (message.type === 'ready') {
|
||||
ready = true
|
||||
resolve(dispose)
|
||||
return
|
||||
}
|
||||
if (message.type === 'events') {
|
||||
if (!disposed) {
|
||||
callback(message.events)
|
||||
}
|
||||
return
|
||||
}
|
||||
if (message.type === 'error') {
|
||||
if (!ready) {
|
||||
// The crawl never went live — fail the watch so the caller knows.
|
||||
disposed = true
|
||||
void worker.terminate()
|
||||
reject(new Error(message.message))
|
||||
return
|
||||
}
|
||||
// Already live: a mid-stream watcher error. Tell the renderer to
|
||||
// refresh; the worker also emits an overflow event alongside this.
|
||||
console.error('[runtime-files.watch] worker error', { rootPath, error: message.message })
|
||||
}
|
||||
})
|
||||
|
||||
worker.on('error', (err) => {
|
||||
if (!ready) {
|
||||
disposed = true
|
||||
reject(err)
|
||||
return
|
||||
}
|
||||
// A live worker crashed: surface an overflow so the renderer re-reads,
|
||||
// rather than silently going stale.
|
||||
console.error('[runtime-files.watch] worker crashed', { rootPath, err })
|
||||
if (!disposed) {
|
||||
callback([{ kind: 'overflow', absolutePath: rootPath }])
|
||||
}
|
||||
})
|
||||
|
||||
worker.on('exit', (code) => {
|
||||
if (!ready && !disposed) {
|
||||
disposed = true
|
||||
reject(new Error(`file watcher worker exited before ready (code ${code})`))
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
// Why: on Linux/Windows @parcel/watcher uses a brute-force backend that
|
||||
// recursively walks the whole tree on a libuv threadpool thread before
|
||||
// subscribe() resolves. On a huge tree backed by slow storage (a home dir on
|
||||
// NFS opened as a worktree) that crawl can run for minutes. Running it here, in
|
||||
// a dedicated worker thread, keeps it off the main/`serve` process's libuv pool
|
||||
// so it can never starve static-asset serving, RPC crypto, or other clients
|
||||
// (issue #5308). The worker owns the subscribe, the per-event stat fanout, and
|
||||
// the event batching; the main thread only relays results.
|
||||
import { stat } from 'fs/promises'
|
||||
import { parentPort, workerData } from 'worker_threads'
|
||||
import type * as ParcelWatcher from '@parcel/watcher'
|
||||
import type { FsChangeEvent } from '../../shared/types'
|
||||
|
||||
const RUNTIME_FILE_WATCH_EVENT_STAT_LIMIT = 200
|
||||
const RUNTIME_FILE_WATCH_STAT_CONCURRENCY = 8
|
||||
|
||||
type FileWatcherWorkerData = {
|
||||
rootPath: string
|
||||
ignore: string[]
|
||||
}
|
||||
|
||||
// Messages the worker sends back to the host.
|
||||
export type FileWatcherWorkerMessage =
|
||||
| { type: 'ready' }
|
||||
| { type: 'events'; events: FsChangeEvent[] }
|
||||
| { type: 'error'; message: string }
|
||||
|
||||
// Messages the host sends to the worker.
|
||||
export type FileWatcherHostMessage = { type: 'unsubscribe' }
|
||||
|
||||
const data = workerData as FileWatcherWorkerData
|
||||
|
||||
if (!parentPort) {
|
||||
throw new Error('File watcher worker must run with a parent port.')
|
||||
}
|
||||
|
||||
const port = parentPort
|
||||
|
||||
/** Run an async mapper over items with a bounded number in flight at once, so a
|
||||
* large batch can't occupy every libuv threadpool thread in this worker. */
|
||||
async function mapWithConcurrency<T, R>(
|
||||
items: readonly T[],
|
||||
limit: number,
|
||||
mapper: (item: T) => Promise<R>
|
||||
): Promise<R[]> {
|
||||
const results = Array.from<R>({ length: items.length })
|
||||
let cursor = 0
|
||||
const worker = async (): Promise<void> => {
|
||||
while (cursor < items.length) {
|
||||
const index = cursor++
|
||||
results[index] = await mapper(items[index])
|
||||
}
|
||||
}
|
||||
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker))
|
||||
return results
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
let watcher: typeof ParcelWatcher
|
||||
try {
|
||||
watcher = await import('@parcel/watcher')
|
||||
} catch (err) {
|
||||
port.postMessage({
|
||||
type: 'error',
|
||||
message: err instanceof Error ? err.message : String(err)
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
return
|
||||
}
|
||||
|
||||
const subscription = await watcher.subscribe(
|
||||
data.rootPath,
|
||||
(err, events) => {
|
||||
if (err) {
|
||||
// Surface the error and ask the renderer to refresh from scratch.
|
||||
port.postMessage({
|
||||
type: 'error',
|
||||
message: err.message
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
port.postMessage({
|
||||
type: 'events',
|
||||
events: [{ kind: 'overflow', absolutePath: data.rootPath }]
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
return
|
||||
}
|
||||
// Why: large watcher batches usually mean a generated directory or branch
|
||||
// switch. Avoid stat fanout and ask the renderer to refresh.
|
||||
if (events.length > RUNTIME_FILE_WATCH_EVENT_STAT_LIMIT) {
|
||||
port.postMessage({
|
||||
type: 'events',
|
||||
events: [{ kind: 'overflow', absolutePath: data.rootPath }]
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
return
|
||||
}
|
||||
void mapWithConcurrency(
|
||||
events,
|
||||
RUNTIME_FILE_WATCH_STAT_CONCURRENCY,
|
||||
async (event): Promise<FsChangeEvent> => {
|
||||
let isDirectory = false
|
||||
try {
|
||||
isDirectory = (await stat(event.path)).isDirectory()
|
||||
} catch {
|
||||
isDirectory = false
|
||||
}
|
||||
return { kind: event.type, absolutePath: event.path, isDirectory }
|
||||
}
|
||||
).then((mapped) => {
|
||||
port.postMessage({ type: 'events', events: mapped } satisfies FileWatcherWorkerMessage)
|
||||
})
|
||||
},
|
||||
{ ignore: data.ignore }
|
||||
)
|
||||
|
||||
// The crawl finished and the subscription is live.
|
||||
port.postMessage({ type: 'ready' } satisfies FileWatcherWorkerMessage)
|
||||
|
||||
port.on('message', (message: FileWatcherHostMessage) => {
|
||||
if (message.type === 'unsubscribe') {
|
||||
void subscription.unsubscribe().finally(() => {
|
||||
port.close()
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
void main().catch((err: unknown) => {
|
||||
port.postMessage({
|
||||
type: 'error',
|
||||
message: err instanceof Error ? err.message : String(err)
|
||||
} satisfies FileWatcherWorkerMessage)
|
||||
})
|
||||
@@ -2,15 +2,14 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as Fs from 'fs'
|
||||
import type * as FsPromises from 'fs/promises'
|
||||
import type * as FilesystemAuth from '../ipc/filesystem-auth'
|
||||
import type { FsChangeEvent } from '../../shared/types'
|
||||
|
||||
const { resolveAuthorizedPathMock, statMock, subscribeParcelWatcherMock, watchMock } = vi.hoisted(
|
||||
() => ({
|
||||
resolveAuthorizedPathMock: vi.fn(),
|
||||
statMock: vi.fn(),
|
||||
subscribeParcelWatcherMock: vi.fn(),
|
||||
watchMock: vi.fn()
|
||||
})
|
||||
)
|
||||
const { resolveAuthorizedPathMock, statMock, watchMock, watchInWorkerMock } = vi.hoisted(() => ({
|
||||
resolveAuthorizedPathMock: vi.fn(),
|
||||
statMock: vi.fn(),
|
||||
watchMock: vi.fn(),
|
||||
watchInWorkerMock: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('fs', async () => {
|
||||
const actual = await vi.importActual<typeof Fs>('fs')
|
||||
@@ -28,8 +27,9 @@ vi.mock('fs/promises', async () => {
|
||||
}
|
||||
})
|
||||
|
||||
vi.mock('@parcel/watcher', () => ({
|
||||
subscribe: subscribeParcelWatcherMock
|
||||
// The local (non-Windows, non-SSH) watch path now delegates to a worker thread.
|
||||
vi.mock('./file-watcher-host', () => ({
|
||||
watchFileExplorerInWorker: watchInWorkerMock
|
||||
}))
|
||||
|
||||
vi.mock('../ipc/filesystem-auth', async () => {
|
||||
@@ -65,8 +65,8 @@ describe('RuntimeFileCommands file watching', () => {
|
||||
vi.useFakeTimers()
|
||||
resolveAuthorizedPathMock.mockReset()
|
||||
statMock.mockReset()
|
||||
subscribeParcelWatcherMock.mockReset()
|
||||
watchMock.mockReset()
|
||||
watchInWorkerMock.mockReset()
|
||||
Object.defineProperty(process, 'platform', {
|
||||
configurable: true,
|
||||
value: originalPlatform
|
||||
@@ -103,6 +103,8 @@ describe('RuntimeFileCommands file watching', () => {
|
||||
const unsubscribe = await commands.watchFileExplorer('id:wt-1', onEvents)
|
||||
|
||||
expect(watchMock).toHaveBeenCalledWith('C:\\repo', { recursive: true }, expect.any(Function))
|
||||
// Windows path does not go through the worker.
|
||||
expect(watchInWorkerMock).not.toHaveBeenCalled()
|
||||
const emit = listener as (() => void) | null
|
||||
expect(emit).not.toBeNull()
|
||||
|
||||
@@ -119,17 +121,59 @@ describe('RuntimeFileCommands file watching', () => {
|
||||
expect(close).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('tracks native Parcel watcher unsubscribe work so shutdown can await it', async () => {
|
||||
// Issue #5308: the local recursive watch runs in a worker thread so
|
||||
// @parcel/watcher's blocking initial crawl can't starve the serve runtime.
|
||||
it('delegates local recursive watching to the worker thread', async () => {
|
||||
resolveAuthorizedPathMock.mockResolvedValue('/home5/Brian')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
|
||||
const captured: { cb?: (events: FsChangeEvent[]) => void } = {}
|
||||
const workerDispose = vi.fn()
|
||||
watchInWorkerMock.mockImplementation((_rootPath, cb) => {
|
||||
captured.cb = cb
|
||||
return Promise.resolve(workerDispose)
|
||||
})
|
||||
|
||||
const onEvents = vi.fn()
|
||||
const { commands } = createRuntimeFileCommands('/home5/Brian')
|
||||
const unsubscribe = await commands.watchFileExplorer('id:wt-1', onEvents)
|
||||
|
||||
expect(watchInWorkerMock).toHaveBeenCalledWith('/home5/Brian', expect.any(Function))
|
||||
|
||||
// Events surfaced by the worker reach the caller.
|
||||
captured.cb?.([{ kind: 'update', absolutePath: '/home5/Brian/a.txt', isDirectory: false }])
|
||||
expect(onEvents).toHaveBeenCalledWith([
|
||||
{ kind: 'update', absolutePath: '/home5/Brian/a.txt', isDirectory: false }
|
||||
])
|
||||
|
||||
// Unsubscribe tears the worker down (dispose runs on the shutdown-drain
|
||||
// microtask, so await the drain before asserting).
|
||||
unsubscribe()
|
||||
await awaitRuntimeFileWatcherUnsubscribes()
|
||||
expect(workerDispose).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('propagates a worker watch failure to the caller', async () => {
|
||||
resolveAuthorizedPathMock.mockResolvedValue('/repo')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
let resolveUnsubscribe: () => void = () => {}
|
||||
const unsubscribeMock = vi.fn(
|
||||
watchInWorkerMock.mockRejectedValue(new Error('worker_failed'))
|
||||
const { commands } = createRuntimeFileCommands('/repo')
|
||||
|
||||
await expect(commands.watchFileExplorer('id:wt-1', vi.fn())).rejects.toThrow('worker_failed')
|
||||
})
|
||||
|
||||
it('tracks worker unsubscribe work so shutdown can await it', async () => {
|
||||
resolveAuthorizedPathMock.mockResolvedValue('/repo')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
|
||||
let resolveDispose: () => void = () => {}
|
||||
const disposeMock = vi.fn(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
resolveUnsubscribe = resolve
|
||||
resolveDispose = resolve
|
||||
})
|
||||
)
|
||||
subscribeParcelWatcherMock.mockResolvedValue({ unsubscribe: unsubscribeMock })
|
||||
watchInWorkerMock.mockResolvedValue(disposeMock)
|
||||
const { commands } = createRuntimeFileCommands('/repo')
|
||||
|
||||
const unsubscribe = await commands.watchFileExplorer('id:wt-1', vi.fn())
|
||||
@@ -141,94 +185,11 @@ describe('RuntimeFileCommands file watching', () => {
|
||||
})
|
||||
await Promise.resolve()
|
||||
|
||||
expect(unsubscribeMock).toHaveBeenCalledTimes(1)
|
||||
expect(disposeMock).toHaveBeenCalledTimes(1)
|
||||
expect(drained).toBe(false)
|
||||
|
||||
resolveUnsubscribe()
|
||||
resolveDispose()
|
||||
await drainPromise
|
||||
expect(drained).toBe(true)
|
||||
})
|
||||
|
||||
// Issue #5308: @parcel/watcher's Linux/Windows brute-force backend recursively
|
||||
// crawls the whole tree on a libuv threadpool thread before subscribe()
|
||||
// resolves. On a huge/slow root (a home dir on NFS opened as a worktree) that
|
||||
// crawl can run for minutes, starving all other async fs/crypto work and
|
||||
// wedging the serve runtime. The crawl must be time-bounded.
|
||||
it('fails the watch if the initial subscribe crawl exceeds the timeout', async () => {
|
||||
resolveAuthorizedPathMock.mockResolvedValue('/home5/Brian')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
|
||||
// subscribe() never resolves — simulates an unbounded recursive crawl.
|
||||
const unsubscribeMock = vi.fn(async () => {})
|
||||
let resolveSubscribe: (sub: { unsubscribe: () => Promise<void> }) => void = () => {}
|
||||
subscribeParcelWatcherMock.mockImplementation(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
resolveSubscribe = resolve
|
||||
})
|
||||
)
|
||||
const { commands } = createRuntimeFileCommands('/home5/Brian')
|
||||
|
||||
const watchPromise = commands.watchFileExplorer('id:wt-1', vi.fn())
|
||||
const rejection = expect(watchPromise).rejects.toThrow('watch_subscribe_timeout')
|
||||
await vi.advanceTimersByTimeAsync(15_000)
|
||||
await rejection
|
||||
|
||||
// If the slow crawl ever finishes, its subscription is dropped, not leaked.
|
||||
resolveSubscribe({ unsubscribe: unsubscribeMock })
|
||||
await vi.runOnlyPendingTimersAsync()
|
||||
expect(unsubscribeMock).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('bounds concurrent stat() calls when resolving a watcher event batch', async () => {
|
||||
resolveAuthorizedPathMock.mockResolvedValue('/repo')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
|
||||
let onBatch: ((err: Error | null, events: { path: string; type: string }[]) => void) | null =
|
||||
null
|
||||
subscribeParcelWatcherMock.mockImplementation((_root, cb) => {
|
||||
onBatch = cb
|
||||
return Promise.resolve({ unsubscribe: vi.fn(async () => {}) })
|
||||
})
|
||||
|
||||
// Track how many stat() calls are running concurrently. Each call resolves
|
||||
// on the next microtask, so workers immediately pull their next item —
|
||||
// peak concurrency reflects the limiter, not artificial gating.
|
||||
let inFlight = 0
|
||||
let peakInFlight = 0
|
||||
statMock.mockImplementation(async (targetPath: string) => {
|
||||
// The root directory check at watch setup must report a directory.
|
||||
if (targetPath === '/repo') {
|
||||
return { isDirectory: () => true }
|
||||
}
|
||||
inFlight++
|
||||
peakInFlight = Math.max(peakInFlight, inFlight)
|
||||
await Promise.resolve()
|
||||
inFlight--
|
||||
return { isDirectory: () => false }
|
||||
})
|
||||
|
||||
const onEvents = vi.fn()
|
||||
const { commands } = createRuntimeFileCommands('/repo')
|
||||
await commands.watchFileExplorer('id:wt-1', onEvents)
|
||||
expect(onBatch).not.toBeNull()
|
||||
|
||||
// 50 events in one batch (under the 200 overflow cap) must not run 50 stats
|
||||
// at once — the threadpool only has a few threads.
|
||||
const events = Array.from({ length: 50 }, (_unused, i) => ({
|
||||
path: `/repo/file-${i}`,
|
||||
type: 'update'
|
||||
}))
|
||||
onBatch?.(null, events)
|
||||
await vi.runAllTimersAsync()
|
||||
|
||||
expect(onEvents).toHaveBeenCalledWith(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ absolutePath: '/repo/file-0', kind: 'update' })
|
||||
])
|
||||
)
|
||||
// 50 event stats + 1 root-directory check at setup.
|
||||
expect(statMock).toHaveBeenCalledTimes(51)
|
||||
expect(peakInFlight).toBeLessThanOrEqual(8)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -14,7 +14,7 @@ const {
|
||||
renameMock,
|
||||
resolveAuthorizedPathMock,
|
||||
statMock,
|
||||
subscribeParcelWatcherMock,
|
||||
watchInWorkerMock,
|
||||
checkRgAvailableMock,
|
||||
wslAwareSpawnMock,
|
||||
watchMock
|
||||
@@ -25,7 +25,7 @@ const {
|
||||
renameMock: vi.fn(),
|
||||
resolveAuthorizedPathMock: vi.fn(),
|
||||
statMock: vi.fn(),
|
||||
subscribeParcelWatcherMock: vi.fn(),
|
||||
watchInWorkerMock: vi.fn(),
|
||||
wslAwareSpawnMock: vi.fn(),
|
||||
watchMock: vi.fn()
|
||||
}))
|
||||
@@ -49,8 +49,8 @@ vi.mock('fs/promises', async () => {
|
||||
}
|
||||
})
|
||||
|
||||
vi.mock('@parcel/watcher', () => ({
|
||||
subscribe: subscribeParcelWatcherMock
|
||||
vi.mock('./file-watcher-host', () => ({
|
||||
watchFileExplorerInWorker: watchInWorkerMock
|
||||
}))
|
||||
|
||||
vi.mock('../ipc/filesystem-auth', async () => {
|
||||
@@ -160,7 +160,7 @@ describe('RuntimeFileCommands', () => {
|
||||
renameMock.mockReset()
|
||||
resolveAuthorizedPathMock.mockReset()
|
||||
statMock.mockReset()
|
||||
subscribeParcelWatcherMock.mockReset()
|
||||
watchInWorkerMock.mockReset()
|
||||
watchMock.mockReset()
|
||||
checkRgAvailableMock.mockReset()
|
||||
wslAwareSpawnMock.mockReset()
|
||||
@@ -364,64 +364,19 @@ describe('RuntimeFileCommands', () => {
|
||||
expect(close).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('tracks native Parcel watcher unsubscribe work so shutdown can await it', async () => {
|
||||
it('delegates local recursive watching to the worker thread', async () => {
|
||||
resolveAuthorizedPathMock.mockResolvedValue('/repo')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
let resolveUnsubscribe: () => void = () => {}
|
||||
const unsubscribeMock = vi.fn(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
resolveUnsubscribe = resolve
|
||||
})
|
||||
)
|
||||
subscribeParcelWatcherMock.mockResolvedValue({ unsubscribe: unsubscribeMock })
|
||||
const dispose = vi.fn()
|
||||
watchInWorkerMock.mockResolvedValue(dispose)
|
||||
const { commands } = createRuntimeFileCommands()
|
||||
|
||||
const unsubscribe = await commands.watchFileExplorer('id:wt-1', vi.fn())
|
||||
expect(watchInWorkerMock).toHaveBeenCalledWith('/repo', expect.any(Function))
|
||||
|
||||
unsubscribe()
|
||||
|
||||
let drained = false
|
||||
const drainPromise = awaitRuntimeFileWatcherUnsubscribes().then(() => {
|
||||
drained = true
|
||||
})
|
||||
await Promise.resolve()
|
||||
|
||||
expect(unsubscribeMock).toHaveBeenCalledTimes(1)
|
||||
expect(drained).toBe(false)
|
||||
|
||||
resolveUnsubscribe()
|
||||
await drainPromise
|
||||
expect(drained).toBe(true)
|
||||
})
|
||||
|
||||
it('collapses large Parcel watcher batches to an overflow refresh', async () => {
|
||||
resolveAuthorizedPathMock.mockResolvedValue('/repo')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
type ParcelCallback = (err: Error | null, events: { type: 'create'; path: string }[]) => void
|
||||
const parcelCallbackRef: { current: ParcelCallback | null } = { current: null }
|
||||
subscribeParcelWatcherMock.mockImplementation(async (_rootPath, callback) => {
|
||||
parcelCallbackRef.current = callback as ParcelCallback
|
||||
return { unsubscribe: vi.fn() }
|
||||
})
|
||||
const { commands } = createRuntimeFileCommands()
|
||||
const onEvents = vi.fn()
|
||||
|
||||
await commands.watchFileExplorer('id:wt-1', onEvents)
|
||||
statMock.mockClear()
|
||||
if (!parcelCallbackRef.current) {
|
||||
throw new Error('Parcel watcher callback was not registered')
|
||||
}
|
||||
parcelCallbackRef.current(
|
||||
null,
|
||||
Array.from({ length: 201 }, (_, index) => ({
|
||||
type: 'create',
|
||||
path: `/repo/generated-${index}.txt`
|
||||
}))
|
||||
)
|
||||
await Promise.resolve()
|
||||
|
||||
expect(statMock).not.toHaveBeenCalled()
|
||||
expect(onEvents).toHaveBeenCalledWith([{ kind: 'overflow', absolutePath: '/repo' }])
|
||||
await awaitRuntimeFileWatcherUnsubscribes()
|
||||
expect(dispose).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('settles and detaches runtime rg searches when timeout kill is ignored', async () => {
|
||||
|
||||
@@ -30,11 +30,7 @@ import type {
|
||||
RuntimeFilePreviewResult,
|
||||
RuntimeFileReadResult
|
||||
} from '../../shared/runtime-types'
|
||||
import type {
|
||||
AsyncSubscription,
|
||||
subscribe as ParcelSubscribe,
|
||||
SubscribeCallback
|
||||
} from '@parcel/watcher'
|
||||
import { watchFileExplorerInWorker } from './file-watcher-host'
|
||||
import { wslAwareSpawn } from '../git/runner'
|
||||
import { parseWslPath, toWindowsWslPath } from '../wsl'
|
||||
import { isENOENT, resolveAuthorizedPath } from '../ipc/filesystem-auth'
|
||||
@@ -65,19 +61,6 @@ const MOBILE_FILE_LIST_LIMIT = 5000
|
||||
const MOBILE_FILE_READ_MAX_BYTES = 512 * 1024
|
||||
const RUNTIME_PREVIEWABLE_BINARY_MAX_BYTES = 10 * 1024 * 1024
|
||||
const WINDOWS_RUNTIME_FILE_WATCH_DEBOUNCE_MS = 150
|
||||
const RUNTIME_FILE_WATCH_EVENT_STAT_LIMIT = 200
|
||||
// Why: on Linux/Windows @parcel/watcher uses a brute-force backend that
|
||||
// recursively walks the whole tree on its libuv threadpool thread before
|
||||
// subscribe() resolves. On a huge tree backed by slow storage (e.g. a home
|
||||
// directory on NFS opened as a worktree) that crawl can run for minutes and
|
||||
// hold a threadpool slot, starving every other async fs/crypto op — on the
|
||||
// headless `serve` runtime this wedges all clients (issue #5308). Cap the
|
||||
// initial crawl so an over-large root fails the watch instead of the runtime;
|
||||
// the explorer still works via manual reads, just without live updates.
|
||||
const RUNTIME_FILE_WATCH_SUBSCRIBE_TIMEOUT_MS = 15_000
|
||||
// Why: bound concurrent stat() calls per watcher batch so a burst can't
|
||||
// occupy every libuv threadpool thread at once (same starvation risk).
|
||||
const RUNTIME_FILE_WATCH_STAT_CONCURRENCY = 8
|
||||
// Why: runtime files.watch subscriptions are cleaned up through synchronous RPC
|
||||
// callbacks. Track native Parcel unsubscribe work so app shutdown can drain it.
|
||||
const pendingRuntimeFileWatcherUnsubscribes = new Set<Promise<void>>()
|
||||
@@ -311,10 +294,11 @@ export class RuntimeFileCommands {
|
||||
if (process.platform === 'win32') {
|
||||
return watchWindowsRuntimeFileExplorer(rootPath, callback)
|
||||
}
|
||||
const watcher = await import('@parcel/watcher')
|
||||
const subscription = await subscribeWithCrawlTimeout(watcher, rootPath, callback)
|
||||
// Why: the watcher runs in a worker thread so @parcel/watcher's blocking
|
||||
// recursive crawl can't starve the main/`serve` process (issue #5308).
|
||||
const dispose = await watchFileExplorerInWorker(rootPath, callback)
|
||||
return () => {
|
||||
trackRuntimeFileWatcherUnsubscribe(rootPath, () => subscription.unsubscribe())
|
||||
trackRuntimeFileWatcherUnsubscribe(rootPath, async () => dispose())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -827,94 +811,6 @@ export class RuntimeFileCommands {
|
||||
}
|
||||
}
|
||||
|
||||
const RUNTIME_FILE_WATCH_IGNORE = [
|
||||
'.git',
|
||||
'node_modules',
|
||||
'dist',
|
||||
'build',
|
||||
'.next',
|
||||
'.cache',
|
||||
'__pycache__',
|
||||
'target',
|
||||
'.venv'
|
||||
]
|
||||
|
||||
/** Run an async mapper over items with a bounded number in flight at once. */
|
||||
async function mapWithConcurrency<T, R>(
|
||||
items: readonly T[],
|
||||
limit: number,
|
||||
mapper: (item: T) => Promise<R>
|
||||
): Promise<R[]> {
|
||||
const results = Array.from<R>({ length: items.length })
|
||||
let cursor = 0
|
||||
const worker = async (): Promise<void> => {
|
||||
while (cursor < items.length) {
|
||||
const index = cursor++
|
||||
results[index] = await mapper(items[index])
|
||||
}
|
||||
}
|
||||
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker))
|
||||
return results
|
||||
}
|
||||
|
||||
/** Subscribe to a Parcel watcher, but bound the initial recursive crawl so a
|
||||
* pathologically large/slow root fails the watch instead of holding a libuv
|
||||
* threadpool slot indefinitely (issue #5308). */
|
||||
async function subscribeWithCrawlTimeout(
|
||||
watcher: { subscribe: typeof ParcelSubscribe },
|
||||
rootPath: string,
|
||||
callback: (events: FsChangeEvent[]) => void
|
||||
): Promise<AsyncSubscription> {
|
||||
const onBatch: SubscribeCallback = (err, events) => {
|
||||
if (err) {
|
||||
console.error('[runtime-files.watch] watcher error', { rootPath, err })
|
||||
callback([{ kind: 'overflow', absolutePath: rootPath }])
|
||||
return
|
||||
}
|
||||
// Why: large watcher batches usually mean a generated directory or branch
|
||||
// switch. Avoid stat fanout and ask the renderer to refresh.
|
||||
if (events.length > RUNTIME_FILE_WATCH_EVENT_STAT_LIMIT) {
|
||||
callback([{ kind: 'overflow', absolutePath: rootPath }])
|
||||
return
|
||||
}
|
||||
void mapWithConcurrency(
|
||||
events,
|
||||
RUNTIME_FILE_WATCH_STAT_CONCURRENCY,
|
||||
async (event): Promise<FsChangeEvent> => {
|
||||
let isDirectory = false
|
||||
try {
|
||||
isDirectory = (await stat(event.path)).isDirectory()
|
||||
} catch {
|
||||
isDirectory = false
|
||||
}
|
||||
return { kind: event.type, absolutePath: event.path, isDirectory }
|
||||
}
|
||||
).then(callback)
|
||||
}
|
||||
|
||||
const subscribePromise = watcher.subscribe(rootPath, onBatch, {
|
||||
ignore: RUNTIME_FILE_WATCH_IGNORE
|
||||
})
|
||||
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
const timeout = new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(() => {
|
||||
// Best-effort: if the slow subscribe ever resolves, drop it so the native
|
||||
// crawl thread isn't leaked.
|
||||
void subscribePromise.then((sub) => sub.unsubscribe()).catch(() => {})
|
||||
reject(new Error('watch_subscribe_timeout'))
|
||||
}, RUNTIME_FILE_WATCH_SUBSCRIBE_TIMEOUT_MS)
|
||||
})
|
||||
|
||||
try {
|
||||
return await Promise.race([subscribePromise, timeout])
|
||||
} finally {
|
||||
if (timer) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function watchWindowsRuntimeFileExplorer(
|
||||
rootPath: string,
|
||||
callback: (events: FsChangeEvent[]) => void
|
||||
|
||||
Reference in New Issue
Block a user