mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
perf(files): share Windows runtime watches by root (#34858)
* perf(files): share Windows runtime watches by root * test(files): trim duplicate Windows watch pool cases --------- Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
This commit is contained in:
@@ -14,7 +14,7 @@ import type { DirEntry, FsChangeEvent } from '../../../../shared/filesystem-entr
|
||||
import { sortDirEntries } from '../../../../shared/file-name-sort'
|
||||
import { resolveAuthorizedPath } from '../../../ipc/filesystem/filesystem-auth'
|
||||
import { resolveRuntimeAuthorizedPath } from '../chat-scratch-file-root'
|
||||
import { watchWindowsRuntimeFileExplorer } from './windows-file-watch'
|
||||
import { WindowsRuntimeWatchPool } from './windows-runtime-watch-pool'
|
||||
import { beginWatcherInstall } from '../../../ipc/watcher-removal-gate'
|
||||
import {
|
||||
closeFileExplorerWatcherInWatcherProcess,
|
||||
@@ -27,6 +27,8 @@ type RuntimeFileWatchHost = Pick<
|
||||
>
|
||||
|
||||
export class RuntimeFileWatches {
|
||||
private readonly windowsWatches = new WindowsRuntimeWatchPool()
|
||||
|
||||
constructor(private readonly host: RuntimeFileWatchHost) {}
|
||||
|
||||
async readFileExplorerDir(
|
||||
@@ -63,7 +65,7 @@ export class RuntimeFileWatches {
|
||||
throw new Error('not_a_directory')
|
||||
}
|
||||
if (process.platform === 'win32') {
|
||||
const close = watchWindowsRuntimeFileExplorer(rootPath, callback, onTerminalError)
|
||||
const close = await this.windowsWatches.watch(rootPath, callback, onTerminalError, signal)
|
||||
return { unsubscribe: close, rootPaths: [target.path, rootPath] }
|
||||
}
|
||||
// #5308/#8212: isolate native faults and blocking crawls from main/serve.
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { afterEach, beforeEach, expect, it, vi } from 'vitest'
|
||||
import { WindowsRuntimeWatchPool } from './windows-runtime-watch-pool'
|
||||
import { WINDOWS_RUNTIME_FILE_WATCH_CLOSE_DEADLINE_MS } from './windows-file-watch'
|
||||
|
||||
const { watchMock } = vi.hoisted(() => ({ watchMock: vi.fn() }))
|
||||
vi.mock('node:fs', () => ({ watch: watchMock }))
|
||||
|
||||
function installWatcher(close?: () => void) {
|
||||
const watcher = Object.assign(new EventEmitter(), {
|
||||
close: vi.fn(() => (close ? close() : watcher.emit('close')))
|
||||
})
|
||||
let notify = () => undefined
|
||||
watchMock.mockImplementationOnce((_root, _options, callback) => {
|
||||
notify = callback
|
||||
return watcher
|
||||
})
|
||||
return { watcher, notify: () => notify() }
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
watchMock.mockReset()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('shares one native watch and refresh timer across path aliases', async () => {
|
||||
const pool = new WindowsRuntimeWatchPool()
|
||||
const { watcher, notify } = installWatcher()
|
||||
const callbacks = Array.from({ length: 3 }, () => vi.fn())
|
||||
const releases = await Promise.all(
|
||||
callbacks.map((callback, index) =>
|
||||
pool.watch(index % 2 === 0 ? 'C:\\repo' : 'c:/REPO', callback, vi.fn())
|
||||
)
|
||||
)
|
||||
expect(watchMock).toHaveBeenCalledOnce()
|
||||
notify()
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
await releases[0]()
|
||||
expect(watcher.close).not.toHaveBeenCalled()
|
||||
await vi.advanceTimersByTimeAsync(500)
|
||||
expect(callbacks[0]).not.toHaveBeenCalled()
|
||||
for (const callback of callbacks.slice(1)) {
|
||||
expect(callback).toHaveBeenCalledExactlyOnceWith([
|
||||
{ kind: 'overflow', absolutePath: 'C:\\repo' }
|
||||
])
|
||||
}
|
||||
await Promise.all(releases.slice(1).map((release) => release()))
|
||||
expect(watcher.close).toHaveBeenCalledOnce()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('keeps separate roots and runtime owners isolated', async () => {
|
||||
const pool = new WindowsRuntimeWatchPool()
|
||||
const anotherRuntime = new WindowsRuntimeWatchPool()
|
||||
const first = installWatcher()
|
||||
const second = installWatcher()
|
||||
const third = installWatcher()
|
||||
const releaseFirst = await pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
const releaseSecond = await pool.watch('C:\\other', vi.fn(), vi.fn())
|
||||
const releaseThird = await anotherRuntime.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
expect(watchMock).toHaveBeenCalledTimes(3)
|
||||
await releaseFirst()
|
||||
expect(first.watcher.close).toHaveBeenCalledOnce()
|
||||
expect(second.watcher.close).not.toHaveBeenCalled()
|
||||
expect(third.watcher.close).not.toHaveBeenCalled()
|
||||
await Promise.all([releaseSecond(), releaseThird()])
|
||||
})
|
||||
|
||||
it('cancels a successor waiting on physical close without starting another watch', async () => {
|
||||
const pool = new WindowsRuntimeWatchPool()
|
||||
const first = installWatcher(() => undefined)
|
||||
const release = await pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
const closing = release()
|
||||
const controller = new AbortController()
|
||||
const successor = pool.watch('C:\\repo', vi.fn(), vi.fn(), controller.signal)
|
||||
controller.abort(new Error('successor cancelled'))
|
||||
await expect(successor).rejects.toThrow('successor cancelled')
|
||||
expect(watchMock).toHaveBeenCalledOnce()
|
||||
first.watcher.emit('close')
|
||||
await closing
|
||||
const replacement = installWatcher()
|
||||
const releaseReplacement = await pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
expect(watchMock).toHaveBeenCalledTimes(2)
|
||||
await releaseReplacement()
|
||||
expect(replacement.watcher.close).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('retries failed close on the original owner before admitting a replacement', async () => {
|
||||
vi.spyOn(console, 'error').mockImplementation(() => undefined)
|
||||
const pool = new WindowsRuntimeWatchPool()
|
||||
const first = installWatcher()
|
||||
const failure = new Error('native close failed')
|
||||
first.watcher.close.mockImplementationOnce(() => {
|
||||
throw failure
|
||||
})
|
||||
const release = await pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
await expect(release()).rejects.toBe(failure)
|
||||
await expect(pool.watch('C:\\repo', vi.fn(), vi.fn())).rejects.toBe(failure)
|
||||
expect(watchMock).toHaveBeenCalledOnce()
|
||||
await release()
|
||||
expect(first.watcher.close).toHaveBeenCalledTimes(2)
|
||||
installWatcher()
|
||||
const releaseReplacement = await pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
expect(watchMock).toHaveBeenCalledTimes(2)
|
||||
await releaseReplacement()
|
||||
})
|
||||
|
||||
it('retains a timed-out root until late physical exit and keeps cleanup retryable', async () => {
|
||||
const pool = new WindowsRuntimeWatchPool()
|
||||
const first = installWatcher(() => undefined)
|
||||
const release = await pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
const closing = release()
|
||||
const rejected = expect(closing).rejects.toMatchObject({
|
||||
code: 'process_unavailable',
|
||||
physicalExit: expect.any(Promise)
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(WINDOWS_RUNTIME_FILE_WATCH_CLOSE_DEADLINE_MS)
|
||||
await rejected
|
||||
installWatcher()
|
||||
const successor = pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
await Promise.resolve()
|
||||
expect(watchMock).toHaveBeenCalledOnce()
|
||||
const retry = release()
|
||||
first.watcher.emit('close')
|
||||
await retry
|
||||
const releaseSuccessor = await successor
|
||||
expect(watchMock).toHaveBeenCalledTimes(2)
|
||||
expect(first.watcher.close).toHaveBeenCalledOnce()
|
||||
await releaseSuccessor()
|
||||
})
|
||||
|
||||
it('fans out terminal errors once and keeps old cleanup from evicting a new generation', async () => {
|
||||
vi.spyOn(console, 'error').mockImplementation(() => undefined)
|
||||
const pool = new WindowsRuntimeWatchPool()
|
||||
const first = installWatcher()
|
||||
const firstError = vi.fn()
|
||||
const secondError = vi.fn()
|
||||
const releaseFirst = await pool.watch('C:\\repo', vi.fn(), firstError)
|
||||
const releaseSecond = await pool.watch('C:\\repo', vi.fn(), secondError)
|
||||
const failure = new Error('native watcher failed')
|
||||
first.watcher.emit('error', failure)
|
||||
expect(firstError).toHaveBeenCalledExactlyOnceWith(failure)
|
||||
expect(secondError).toHaveBeenCalledExactlyOnceWith(failure)
|
||||
const replacement = installWatcher()
|
||||
const replacementEvents = vi.fn()
|
||||
const releaseReplacement = await pool.watch('C:\\repo', replacementEvents, vi.fn())
|
||||
await Promise.all([releaseFirst(), releaseSecond()])
|
||||
const releaseJoiner = await pool.watch('C:\\repo', vi.fn(), vi.fn())
|
||||
expect(watchMock).toHaveBeenCalledTimes(2)
|
||||
replacement.notify()
|
||||
await vi.advanceTimersByTimeAsync(500)
|
||||
expect(replacementEvents).toHaveBeenCalledOnce()
|
||||
await Promise.all([releaseReplacement(), releaseJoiner()])
|
||||
})
|
||||
|
||||
it('does not create a native watcher for an already aborted subscription', async () => {
|
||||
const pool = new WindowsRuntimeWatchPool()
|
||||
const controller = new AbortController()
|
||||
controller.abort(new Error('request cancelled'))
|
||||
await expect(pool.watch('C:\\repo', vi.fn(), vi.fn(), controller.signal)).rejects.toThrow(
|
||||
'request cancelled'
|
||||
)
|
||||
expect(watchMock).not.toHaveBeenCalled()
|
||||
})
|
||||
@@ -0,0 +1,112 @@
|
||||
import type { FsChangeEvent } from '../../../../shared/filesystem-entry-types'
|
||||
import {
|
||||
throwIfSignalAborted,
|
||||
waitForPromiseWithSignal
|
||||
} from '../../../../shared/abort-signal-reason'
|
||||
import { normalizeRuntimePathForComparison } from '../../../../shared/cross-platform-path'
|
||||
import { isWatcherProcessFailure } from '../../../ipc/filesystem/watcher-process/parcel-watcher-process-failure'
|
||||
import { watchWindowsRuntimeFileExplorer } from './windows-file-watch'
|
||||
|
||||
type Subscriber = {
|
||||
onEvents: (events: FsChangeEvent[]) => void
|
||||
onError: (error: Error) => void
|
||||
}
|
||||
|
||||
type WindowsRootWatch = {
|
||||
subscribers: Set<Subscriber>
|
||||
close: () => Promise<void>
|
||||
closing: Promise<void> | null
|
||||
closeError: Error | null
|
||||
}
|
||||
|
||||
export class WindowsRuntimeWatchPool {
|
||||
private readonly roots = new Map<string, WindowsRootWatch>()
|
||||
|
||||
async watch(
|
||||
rootPath: string,
|
||||
onEvents: Subscriber['onEvents'],
|
||||
onError: Subscriber['onError'],
|
||||
signal?: AbortSignal
|
||||
): Promise<() => Promise<void>> {
|
||||
throwIfSignalAborted(signal)
|
||||
const key = normalizeRuntimePathForComparison(rootPath)
|
||||
let root = this.roots.get(key)
|
||||
if (root?.closing) {
|
||||
await waitForPromiseWithSignal(root.closing, signal)
|
||||
return this.watch(rootPath, onEvents, onError, signal)
|
||||
}
|
||||
if (root?.closeError) {
|
||||
throw root.closeError
|
||||
}
|
||||
if (!root) {
|
||||
root = this.createRoot(key, rootPath)
|
||||
this.roots.set(key, root)
|
||||
}
|
||||
const ownedRoot = root
|
||||
const subscriber = { onEvents, onError }
|
||||
ownedRoot.subscribers.add(subscriber)
|
||||
let releasePromise: Promise<void> | undefined
|
||||
return () => {
|
||||
if (releasePromise) {
|
||||
return releasePromise
|
||||
}
|
||||
ownedRoot.subscribers.delete(subscriber)
|
||||
if (ownedRoot.subscribers.size > 0) {
|
||||
return Promise.resolve()
|
||||
}
|
||||
const attempt = this.closeRoot(key, ownedRoot)
|
||||
releasePromise = attempt
|
||||
void attempt.catch(() => {
|
||||
releasePromise = undefined
|
||||
})
|
||||
return attempt
|
||||
}
|
||||
}
|
||||
|
||||
private createRoot(key: string, rootPath: string): WindowsRootWatch {
|
||||
const subscribers = new Set<Subscriber>()
|
||||
const close = watchWindowsRuntimeFileExplorer(
|
||||
rootPath,
|
||||
(events) => {
|
||||
for (const subscriber of Array.from(subscribers)) {
|
||||
if (subscribers.has(subscriber)) {
|
||||
subscriber.onEvents(events)
|
||||
}
|
||||
}
|
||||
},
|
||||
(error) => {
|
||||
if (this.roots.get(key)?.subscribers === subscribers) {
|
||||
this.roots.delete(key)
|
||||
}
|
||||
const failedSubscribers = Array.from(subscribers)
|
||||
subscribers.clear()
|
||||
for (const subscriber of failedSubscribers) {
|
||||
subscriber.onError(error)
|
||||
}
|
||||
}
|
||||
)
|
||||
return { subscribers, close, closing: null, closeError: null }
|
||||
}
|
||||
|
||||
private closeRoot(key: string, root: WindowsRootWatch): Promise<void> {
|
||||
if (root.closing) {
|
||||
return root.closing
|
||||
}
|
||||
const forget = (): void => {
|
||||
if (this.roots.get(key) === root) {
|
||||
this.roots.delete(key)
|
||||
}
|
||||
}
|
||||
const attempt = root.close().then(forget)
|
||||
root.closing = attempt
|
||||
void attempt.catch((error: unknown) => {
|
||||
root.closeError = error instanceof Error ? error : new Error(String(error))
|
||||
// A timed-out close still owns native locks until its physical exit.
|
||||
root.closing =
|
||||
isWatcherProcessFailure(error) && error.physicalExit
|
||||
? error.physicalExit.then(forget)
|
||||
: null
|
||||
})
|
||||
return attempt
|
||||
}
|
||||
}
|
||||
@@ -182,6 +182,38 @@ describe('RuntimeFileCommands file watching', () => {
|
||||
expect(settled).toBe(true)
|
||||
})
|
||||
|
||||
it('shares one Windows watch across clients without stopping survivors', async () => {
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: 'win32' })
|
||||
const watcher = createWindowsWatcher(() => watcher.emit('close'))
|
||||
let emitChange = () => undefined
|
||||
watchMock.mockImplementation((_rootPath, _options, callback) => {
|
||||
emitChange = callback
|
||||
return watcher
|
||||
})
|
||||
resolveAuthorizedPathMock.mockResolvedValue('C:\\repo')
|
||||
statMock.mockResolvedValue({ isDirectory: () => true })
|
||||
const { commands } = createRuntimeFileCommands('C:\\repo')
|
||||
const callbacks = Array.from({ length: 2 }, () => vi.fn())
|
||||
const releases = await Promise.all(
|
||||
callbacks.map((callback) => commands.watchFileExplorer('id:wt-1', callback))
|
||||
)
|
||||
expect(watchMock).toHaveBeenCalledOnce()
|
||||
|
||||
await releases[0]()
|
||||
expect(watcher.close).not.toHaveBeenCalled()
|
||||
emitChange()
|
||||
await vi.advanceTimersByTimeAsync(WINDOWS_RUNTIME_FILE_WATCH_LEADING_MS)
|
||||
expect(callbacks[0]).not.toHaveBeenCalled()
|
||||
for (const callback of callbacks.slice(1)) {
|
||||
expect(callback).toHaveBeenCalledExactlyOnceWith([
|
||||
{ kind: 'overflow', absolutePath: 'C:\\repo' }
|
||||
])
|
||||
}
|
||||
await Promise.all(releases.slice(1).map((release) => release()))
|
||||
expect(watcher.close).toHaveBeenCalledOnce()
|
||||
expect(_getRuntimeFileWatcherReleaseCountForTests()).toBe(0)
|
||||
})
|
||||
|
||||
it('treats a Windows watcher error before cleanup as physical close', async () => {
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: 'win32' })
|
||||
const watcher = createWindowsWatcher(() => undefined)
|
||||
|
||||
Reference in New Issue
Block a user