mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
Watch only the files Orca needs in its settings folder and .git, instead of the whole folders (#32410)
* perf(watch): stop watching all of userData and .git for a handful of files On macOS a directory fs.watch is a recursive FSEvents stream. Two watches used that to notice a few files: - The Active Server preference watched the whole userData directory for orca-environments.json, so Orca's own writes there (agent-hooks status, terminal history, browser caches; ~10 events/s measured) were routed back to it. - The shallow git metadata watch ran fs.watch on each repo's whole common .git dir (and .git/logs), receiving objects/, refs/, lfs/ and every linked worktree's churn to see six files. Both now watch the individual files (kqueue on macOS, no fseventsd) through a new rebinding file watch: events and a 30s sweep re-stat the path, rebind when a rename-replace or deletion moved it, and pick up files created later. Events are confirmed by a stat (size/mtime/ctime/inode) because macOS raises an attribute event for every mmap read of the index and packed-refs; the watch is one-shot with a backing-off window so a file every git process maps costs about one rebind a second instead of a wakeup per read. The registry path is unchanged, so CLI/serve readers and older peers are unaffected. Linux and Windows keep the non-recursive directory watch for shallow subscriptions. Co-Authored-By: Claude <noreply@anthropic.com> * fix(watch): don't hang unsubscribing an errored watcher; find late metadata files sooner - Node closes an errored fs.watch handle without emitting 'close', and a later close() is a no-op, so awaiting 'close' could hang a watcher-child unsubscribe and every subscribe queued behind it. Errored watchers are now dropped without a close, and closeFileSystemWatcher also settles on 'error' or after 1s. Applies to both the macOS file watch and the Linux/Windows directory watch. - Missing metadata files (packed-refs from pack-refs/gc) are re-statted every 2s instead of waiting for the 30s rebind sweep. - On macOS the delivery probe watches a file, matching the kqueue path the shallow subscriptions now use. Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -179,6 +179,9 @@ const NODE_RUNTIME_INCLUDE = [
|
||||
'src/main/runtime/mobile-cloud-relay/relay-control-client.test.ts',
|
||||
'src/main/ipc/filesystem-watcher-ignore-real.test.ts',
|
||||
'src/main/ipc/runtime-environments/runtime-environment-preference.test.ts',
|
||||
// Real fs.watch file bindings; Bun's watcher is not the one Electron runs.
|
||||
'src/main/ipc/rebinding-file-watch.test.ts',
|
||||
'src/main/ipc/parcel-watcher-shallow-file-subscription.test.ts',
|
||||
'src/main/ipc/runtime-environments/runtime-environment-transport-routing-tailscale-hint.test.ts',
|
||||
'src/renderer/src/components/emulator-pane/emulator-device-frame-visibility.test.tsx',
|
||||
'src/main/claude/claude-agent-sdk-contract-pins.test.ts',
|
||||
|
||||
@@ -13,6 +13,8 @@ const { detectShallowWatchDeliveryMock, statMock, subscribeMock, watchMock, writ
|
||||
vi.mock('node:fs', () => ({
|
||||
mkdtempSync: vi.fn(() => '/tmp/orca-watcher-canary-test'),
|
||||
rmSync: vi.fn(),
|
||||
// Shallow roots and macOS metadata files are stat-ed before watching.
|
||||
statSync: vi.fn(() => ({ isDirectory: () => true, dev: 1n, ino: 1n, birthtimeNs: 0n })),
|
||||
watch: watchMock,
|
||||
writeFileSync: writeFileSyncMock
|
||||
}))
|
||||
@@ -143,8 +145,14 @@ describe('parcel watcher process canary', () => {
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
|
||||
expect(subscribeMock).toHaveBeenCalledTimes(1)
|
||||
watcherCallbacks.get('/repo/.git')?.('change', 'HEAD')
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
// macOS binds the metadata file itself; other platforms bind the directory.
|
||||
if (process.platform === 'darwin') {
|
||||
watcherCallbacks.get('/repo/.git/HEAD')?.('change', 'HEAD')
|
||||
} else {
|
||||
watcherCallbacks.get('/repo/.git')?.('change', 'HEAD')
|
||||
}
|
||||
// macOS confirms the event with a stat after a short coalescing window.
|
||||
await vi.advanceTimersByTimeAsync(200)
|
||||
expect(sendMock).toHaveBeenCalledWith(
|
||||
{
|
||||
op: 'events',
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
import { mkdirSync, mkdtempSync, renameSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { PRIMARY_CHECKOUT_METADATA_FILES } from './worktree-git-common-metadata-files'
|
||||
import {
|
||||
MISSING_FILE_SWEEP_INTERVAL_MS,
|
||||
startShallowFileWatcher
|
||||
} from './parcel-watcher-shallow-file-subscription'
|
||||
import type { ShallowWatcherSubscription } from './parcel-watcher-shallow-subscription'
|
||||
|
||||
let commonDir: string
|
||||
let subscription: ShallowWatcherSubscription | null
|
||||
let paths: string[]
|
||||
|
||||
beforeEach(() => {
|
||||
commonDir = mkdtempSync(join(tmpdir(), 'orca-shallow-file-watch-'))
|
||||
for (const directory of ['logs', 'objects/ab', 'refs/heads', 'worktrees/feature', 'lfs']) {
|
||||
mkdirSync(join(commonDir, directory), { recursive: true })
|
||||
}
|
||||
// packed-refs and config.worktree are deliberately absent at start.
|
||||
for (const name of ['HEAD', 'index', 'config', 'logs/HEAD']) {
|
||||
writeFileSync(join(commonDir, name), 'initial')
|
||||
}
|
||||
subscription = null
|
||||
paths = []
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await subscription?.unsubscribe()
|
||||
rmSync(commonDir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
function start(): void {
|
||||
subscription = startShallowFileWatcher(
|
||||
commonDir,
|
||||
PRIMARY_CHECKOUT_METADATA_FILES,
|
||||
(events) => paths.push(...events.map((event) => event.path)),
|
||||
(error) => {
|
||||
throw error
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
function replaceLikeGit(name: string, content: string): void {
|
||||
const path = join(commonDir, name)
|
||||
writeFileSync(`${path}.lock`, content)
|
||||
renameSync(`${path}.lock`, path)
|
||||
}
|
||||
|
||||
async function settle(): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, 150))
|
||||
}
|
||||
|
||||
async function expectEventFor(name: string): Promise<void> {
|
||||
await vi.waitFor(() => expect(paths).toContain(join(commonDir, name)), { timeout: 5_000 })
|
||||
}
|
||||
|
||||
describe('shallow file watcher', () => {
|
||||
it('ignores object, ref, lfs and linked-worktree churn under the common dir', async () => {
|
||||
start()
|
||||
await settle()
|
||||
paths = []
|
||||
|
||||
writeFileSync(join(commonDir, 'objects/ab/cdef'), 'blob')
|
||||
writeFileSync(join(commonDir, 'refs/heads/topic'), 'sha')
|
||||
writeFileSync(join(commonDir, 'worktrees/feature/index'), 'index')
|
||||
writeFileSync(join(commonDir, 'worktrees/feature/HEAD'), 'ref')
|
||||
writeFileSync(join(commonDir, 'lfs/object'), 'large')
|
||||
writeFileSync(join(commonDir, 'FETCH_HEAD'), 'fetched')
|
||||
await settle()
|
||||
|
||||
expect(paths).toEqual([])
|
||||
})
|
||||
|
||||
it('reports git lock-file replacements of the index every time', async () => {
|
||||
start()
|
||||
await settle()
|
||||
|
||||
for (const content of ['one', 'two', 'three']) {
|
||||
paths = []
|
||||
replaceLikeGit('index', content)
|
||||
await expectEventFor('index')
|
||||
await settle()
|
||||
}
|
||||
})
|
||||
|
||||
it('reports appends to the nested reflog', async () => {
|
||||
start()
|
||||
await settle()
|
||||
paths = []
|
||||
|
||||
writeFileSync(join(commonDir, 'logs/HEAD'), 'initial\nmoved', { flag: 'w' })
|
||||
|
||||
await expectEventFor('logs/HEAD')
|
||||
})
|
||||
|
||||
it('picks up a metadata file created after start when a sibling changes', async () => {
|
||||
start()
|
||||
await settle()
|
||||
paths = []
|
||||
|
||||
writeFileSync(join(commonDir, 'packed-refs'), 'packed')
|
||||
replaceLikeGit('HEAD', 'ref: refs/heads/topic')
|
||||
|
||||
await expectEventFor('HEAD')
|
||||
await expectEventFor('packed-refs')
|
||||
paths = []
|
||||
replaceLikeGit('packed-refs', 'repacked')
|
||||
await expectEventFor('packed-refs')
|
||||
})
|
||||
|
||||
it('finds a metadata file created later with no sibling write, as pack-refs does', async () => {
|
||||
start()
|
||||
await settle()
|
||||
paths = []
|
||||
|
||||
replaceLikeGit('packed-refs', 'packed')
|
||||
|
||||
await vi.waitFor(() => expect(paths).toContain(join(commonDir, 'packed-refs')), {
|
||||
timeout: MISSING_FILE_SWEEP_INTERVAL_MS + 3_000
|
||||
})
|
||||
})
|
||||
|
||||
it('reports a missing root as a watch error', async () => {
|
||||
const onError = vi.fn()
|
||||
subscription = startShallowFileWatcher(join(commonDir, 'missing'), ['HEAD'], () => {}, onError)
|
||||
expect(onError).toHaveBeenCalledOnce()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,101 @@
|
||||
import { statSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import type { Event as ParcelWatcherEvent } from '@parcel/watcher'
|
||||
import type { ShallowWatcherSubscription } from './parcel-watcher-shallow-subscription'
|
||||
import { watchFileAcrossReplacement, type RebindingFileWatch } from './rebinding-file-watch'
|
||||
|
||||
// Why: bindings that missed a replacement are found by this sweep. Primary git
|
||||
// metadata also has its own backstop poll, so this bounds staleness, not latency.
|
||||
const REBIND_SWEEP_INTERVAL_MS = 30_000
|
||||
// Why: a file created later with no sibling write (packed-refs from pack-refs or
|
||||
// gc) is only found by a sweep. Re-statting just the missing paths is one ENOENT
|
||||
// each, so they get the git poll cadence instead of waiting for the full sweep.
|
||||
export const MISSING_FILE_SWEEP_INTERVAL_MS = 2_000
|
||||
|
||||
/** One watch per included file, so a `.git` root never streams objects/, refs/
|
||||
* or worktrees/ churn. Used on macOS, where a directory watch is recursive. */
|
||||
export function startShallowFileWatcher(
|
||||
rootPath: string,
|
||||
relativePaths: readonly string[],
|
||||
onEvents: (events: ParcelWatcherEvent[]) => void,
|
||||
onError: (error: Error) => void
|
||||
): ShallowWatcherSubscription {
|
||||
let disposed = false
|
||||
let reportedError = false
|
||||
const reportError = (error: unknown): void => {
|
||||
if (disposed || reportedError) {
|
||||
return
|
||||
}
|
||||
reportedError = true
|
||||
onError(error instanceof Error ? error : new Error(String(error)))
|
||||
}
|
||||
|
||||
try {
|
||||
if (!statSync(rootPath).isDirectory()) {
|
||||
throw new Error(`Shallow watcher root is not a directory: ${rootPath}`)
|
||||
}
|
||||
} catch (error) {
|
||||
reportError(error)
|
||||
}
|
||||
|
||||
const emitUpdate = (path: string): void => {
|
||||
if (!disposed) {
|
||||
onEvents([{ type: 'update', path }])
|
||||
}
|
||||
}
|
||||
|
||||
const watches = new Map<string, RebindingFileWatch>()
|
||||
const refreshAll = (onlyUnbound: boolean): void => {
|
||||
for (const fileWatch of watches.values()) {
|
||||
if (!onlyUnbound || !fileWatch.isBound()) {
|
||||
void fileWatch.refresh()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const paths = new Set(
|
||||
relativePaths
|
||||
.map((relativePath) =>
|
||||
relativePath.split(process.platform === 'win32' ? /[\\/]+/ : /\/+/).filter(Boolean)
|
||||
)
|
||||
.filter((parts) => parts.length > 0)
|
||||
.map((parts) => join(rootPath, ...parts))
|
||||
)
|
||||
for (const path of paths) {
|
||||
watches.set(
|
||||
path,
|
||||
watchFileAcrossReplacement(
|
||||
path,
|
||||
() => {
|
||||
emitUpdate(path)
|
||||
// Git writes these files together, so one event is a cheap moment to
|
||||
// look for a sibling that has just been created.
|
||||
refreshAll(true)
|
||||
},
|
||||
reportError
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
const sweepTimer = setInterval(() => {
|
||||
if (!disposed) {
|
||||
refreshAll(false)
|
||||
}
|
||||
}, REBIND_SWEEP_INTERVAL_MS)
|
||||
sweepTimer.unref?.()
|
||||
const missingFileTimer = setInterval(() => {
|
||||
if (!disposed) {
|
||||
refreshAll(true)
|
||||
}
|
||||
}, MISSING_FILE_SWEEP_INTERVAL_MS)
|
||||
missingFileTimer.unref?.()
|
||||
|
||||
return {
|
||||
unsubscribe: async () => {
|
||||
disposed = true
|
||||
clearInterval(sweepTimer)
|
||||
clearInterval(missingFileTimer)
|
||||
await Promise.all([...watches.values()].map((fileWatch) => fileWatch.close()))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -41,7 +41,7 @@ vi.mock('node:fs/promises', async () => {
|
||||
return { ...actual, stat: vi.fn(actual.stat) }
|
||||
})
|
||||
|
||||
import { startShallowWatcher } from './parcel-watcher-shallow-subscription'
|
||||
import { startShallowDirectoryWatcher } from './parcel-watcher-shallow-subscription'
|
||||
|
||||
function emit(path: string, fileName: string): void {
|
||||
watcherState.get(path)?.callback('change', fileName)
|
||||
@@ -60,7 +60,7 @@ describe('shallow watcher subscription', () => {
|
||||
await mkdir(join(root, 'logs'))
|
||||
const events: string[] = []
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
const subscription = startShallowWatcher(
|
||||
const subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
['HEAD', 'config', 'logs/HEAD'],
|
||||
(nextEvents) => {
|
||||
@@ -94,7 +94,7 @@ describe('shallow watcher subscription', () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-watcher-'))
|
||||
try {
|
||||
const events: string[] = []
|
||||
const subscription = startShallowWatcher(
|
||||
const subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
['HEAD'],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -113,12 +113,34 @@ describe('shallow watcher subscription', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('unsubscribes promptly after the runtime closed an errored watcher', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-watcher-'))
|
||||
try {
|
||||
const onError = vi.fn()
|
||||
const subscription = startShallowDirectoryWatcher(root, ['HEAD'], () => {}, onError)
|
||||
const state = watcherState.get(root)
|
||||
if (!state) {
|
||||
throw new Error('root was not watched')
|
||||
}
|
||||
// Node closes the handle on error without 'close'; a later close() is a no-op.
|
||||
state.watcher.close = () => {}
|
||||
state.watcher.emit('error', new Error('EPERM'))
|
||||
expect(onError).toHaveBeenCalledOnce()
|
||||
|
||||
const startedAt = Date.now()
|
||||
await subscription.unsubscribe()
|
||||
expect(Date.now() - startedAt).toBeLessThan(500)
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it.skipIf(process.platform === 'win32')(
|
||||
'preserves a literal backslash in an included filename',
|
||||
async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-watcher-'))
|
||||
const events: string[] = []
|
||||
const subscription = startShallowWatcher(
|
||||
const subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
['plugin\\name'],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -141,7 +163,7 @@ describe('shallow watcher subscription', () => {
|
||||
try {
|
||||
await mkdir(join(root, 'logs'))
|
||||
const events: string[] = []
|
||||
const subscription = startShallowWatcher(
|
||||
const subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
['HEAD', 'logs/HEAD'],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -169,7 +191,7 @@ describe('shallow watcher subscription', () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-watcher-'))
|
||||
try {
|
||||
await mkdir(join(root, 'logs'))
|
||||
const subscription = startShallowWatcher(
|
||||
const subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
['logs/HEAD'],
|
||||
() => {},
|
||||
@@ -190,11 +212,11 @@ describe('shallow watcher subscription', () => {
|
||||
vi.useFakeTimers()
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-watcher-'))
|
||||
const logs = join(root, 'logs')
|
||||
let subscription: ReturnType<typeof startShallowWatcher> | undefined
|
||||
let subscription: ReturnType<typeof startShallowDirectoryWatcher> | undefined
|
||||
try {
|
||||
await mkdir(logs)
|
||||
const events: string[] = []
|
||||
subscription = startShallowWatcher(
|
||||
subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
['HEAD', 'logs/HEAD'],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -237,7 +259,7 @@ describe('shallow watcher subscription', () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-large-identity-'))
|
||||
const directory = join(root, parent)
|
||||
const previousDirectory = join(dirname(directory), `${basename(directory)}-previous`)
|
||||
let subscription: ReturnType<typeof startShallowWatcher> | undefined
|
||||
let subscription: ReturnType<typeof startShallowDirectoryWatcher> | undefined
|
||||
try {
|
||||
if (parent) {
|
||||
await mkdir(directory)
|
||||
@@ -276,7 +298,7 @@ describe('shallow watcher subscription', () => {
|
||||
options?.bigint ? nextBigInt : nextNumeric
|
||||
)
|
||||
const events: string[] = []
|
||||
subscription = startShallowWatcher(
|
||||
subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
[join(parent, 'HEAD')],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -321,7 +343,7 @@ describe('shallow watcher subscription', () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-inode-less-'))
|
||||
const directory = join(root, parent)
|
||||
const previousDirectory = join(dirname(directory), `${basename(directory)}-previous`)
|
||||
let subscription: ReturnType<typeof startShallowWatcher> | undefined
|
||||
let subscription: ReturnType<typeof startShallowDirectoryWatcher> | undefined
|
||||
try {
|
||||
if (parent) {
|
||||
await mkdir(directory)
|
||||
@@ -352,7 +374,7 @@ describe('shallow watcher subscription', () => {
|
||||
.mockResolvedValueOnce(replacement)
|
||||
.mockResolvedValueOnce(healthyReplacement)
|
||||
const events: string[] = []
|
||||
subscription = startShallowWatcher(
|
||||
subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
[join(parent, 'HEAD')],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -402,7 +424,7 @@ describe('shallow watcher subscription', () => {
|
||||
vi.useFakeTimers()
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-unavailable-identity-'))
|
||||
const directory = join(root, parent)
|
||||
let subscription: ReturnType<typeof startShallowWatcher> | undefined
|
||||
let subscription: ReturnType<typeof startShallowDirectoryWatcher> | undefined
|
||||
try {
|
||||
if (parent) {
|
||||
await mkdir(directory)
|
||||
@@ -425,7 +447,7 @@ describe('shallow watcher subscription', () => {
|
||||
}
|
||||
const events: string[] = []
|
||||
const onError = vi.fn()
|
||||
subscription = startShallowWatcher(
|
||||
subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
[join(parent, 'HEAD')],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -471,7 +493,7 @@ describe('shallow watcher subscription', () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-known-identity-'))
|
||||
const directory = join(root, parent)
|
||||
const previousDirectory = join(dirname(directory), `${basename(directory)}-previous`)
|
||||
let subscription: ReturnType<typeof startShallowWatcher> | undefined
|
||||
let subscription: ReturnType<typeof startShallowDirectoryWatcher> | undefined
|
||||
try {
|
||||
if (parent) {
|
||||
await mkdir(directory)
|
||||
@@ -494,7 +516,7 @@ describe('shallow watcher subscription', () => {
|
||||
vi.mocked(stat).mockResolvedValue(unknown)
|
||||
const events: string[] = []
|
||||
const onError = vi.fn()
|
||||
subscription = startShallowWatcher(
|
||||
subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
[join(parent, 'HEAD')],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
@@ -545,7 +567,7 @@ describe('shallow watcher subscription', () => {
|
||||
it('resyncs included files when an initially missing directory becomes watchable', async () => {
|
||||
vi.useFakeTimers()
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-shallow-watcher-'))
|
||||
let subscription: ReturnType<typeof startShallowWatcher> | undefined
|
||||
let subscription: ReturnType<typeof startShallowDirectoryWatcher> | undefined
|
||||
try {
|
||||
const defaultWatch = vi.mocked(watch).getMockImplementation()
|
||||
if (!defaultWatch) {
|
||||
@@ -557,7 +579,7 @@ describe('shallow watcher subscription', () => {
|
||||
throw new Error('ENOENT: logs directory does not exist')
|
||||
})
|
||||
const events: string[] = []
|
||||
subscription = startShallowWatcher(
|
||||
subscription = startShallowDirectoryWatcher(
|
||||
root,
|
||||
['HEAD', 'logs/HEAD'],
|
||||
(nextEvents) => events.push(...nextEvents.map((event) => event.path)),
|
||||
|
||||
@@ -2,6 +2,8 @@ import { statSync, watch, type FSWatcher } from 'node:fs'
|
||||
import { stat } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import type { Event as ParcelWatcherEvent } from '@parcel/watcher'
|
||||
import { startShallowFileWatcher } from './parcel-watcher-shallow-file-subscription'
|
||||
import { closeFileSystemWatcher } from './rebinding-file-watch'
|
||||
import { watcherDirectoryIdentity } from './watcher-directory-identity'
|
||||
|
||||
export type ShallowWatcherSubscription = {
|
||||
@@ -15,18 +17,20 @@ export type ShallowWatcherSubscription = {
|
||||
// bounded cadence. Two stats per interval is the whole steady-state cost.
|
||||
const REBIND_CHECK_INTERVAL_MS = 30_000
|
||||
|
||||
function closeFileSystemWatcher(watcher: FSWatcher): Promise<void> {
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
watcher.once('close', resolve)
|
||||
try {
|
||||
watcher.close()
|
||||
} catch {
|
||||
resolve()
|
||||
}
|
||||
return promise
|
||||
export function startShallowWatcher(
|
||||
rootPath: string,
|
||||
relativePaths: readonly string[],
|
||||
onEvents: (events: ParcelWatcherEvent[]) => void,
|
||||
onError: (error: Error) => void
|
||||
): ShallowWatcherSubscription {
|
||||
// Why: on macOS fs.watch(dir) is a recursive FSEvents stream, so a `.git`
|
||||
// root would stream every object, ref and linked-worktree write to fseventsd.
|
||||
return process.platform === 'darwin'
|
||||
? startShallowFileWatcher(rootPath, relativePaths, onEvents, onError)
|
||||
: startShallowDirectoryWatcher(rootPath, relativePaths, onEvents, onError)
|
||||
}
|
||||
|
||||
export function startShallowWatcher(
|
||||
export function startShallowDirectoryWatcher(
|
||||
rootPath: string,
|
||||
relativePaths: readonly string[],
|
||||
onEvents: (events: ParcelWatcherEvent[]) => void,
|
||||
@@ -103,7 +107,14 @@ export function startShallowWatcher(
|
||||
emitUpdates(parent, [name])
|
||||
}
|
||||
})
|
||||
watcher.on('error', reportError)
|
||||
watcher.on('error', (error) => {
|
||||
// The runtime already closed it; unsubscribe must not wait on its 'close'.
|
||||
if (watchers.get(parent) === watcher) {
|
||||
watchers.delete(parent)
|
||||
boundIdentities.delete(parent)
|
||||
}
|
||||
reportError(error)
|
||||
})
|
||||
watchers.set(parent, watcher)
|
||||
return true
|
||||
} catch (error) {
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
import { EventEmitter } from 'node:events'
|
||||
import type * as Fs from 'node:fs'
|
||||
import { mkdtempSync, rmSync, watch, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { nativeEvents, nativeFailures } = vi.hoisted(() => {
|
||||
const nativeEvents: (() => void)[] = []
|
||||
const nativeFailures: ((error: Error) => void)[] = []
|
||||
return { nativeEvents, nativeFailures }
|
||||
})
|
||||
|
||||
vi.mock('node:fs', async () => {
|
||||
const actual = await vi.importActual<typeof Fs>('node:fs')
|
||||
return {
|
||||
...actual,
|
||||
watch: vi.fn((_path: string, _options: unknown, callback: () => void) => {
|
||||
nativeEvents.push(callback)
|
||||
// Mirrors Node: an error closes the handle without 'close', and a later close() is a no-op.
|
||||
let handleOpen = true
|
||||
const watcher: EventEmitter = Object.assign(new EventEmitter(), {
|
||||
close: () => {
|
||||
if (handleOpen) {
|
||||
handleOpen = false
|
||||
watcher.emit('close')
|
||||
}
|
||||
}
|
||||
})
|
||||
nativeFailures.push((error) => {
|
||||
handleOpen = false
|
||||
watcher.emit('error', error)
|
||||
})
|
||||
return watcher
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
import {
|
||||
closeFileSystemWatcher,
|
||||
watchFileAcrossReplacement,
|
||||
type RebindingFileWatch
|
||||
} from './rebinding-file-watch'
|
||||
|
||||
let directory: string
|
||||
let fileWatch: RebindingFileWatch | null
|
||||
|
||||
beforeEach(() => {
|
||||
directory = mkdtempSync(join(tmpdir(), 'orca-rebinding-noise-'))
|
||||
nativeEvents.length = 0
|
||||
nativeFailures.length = 0
|
||||
fileWatch = null
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await fileWatch?.close()
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function settle(): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, 250))
|
||||
}
|
||||
|
||||
describe('watchFileAcrossReplacement event filtering', () => {
|
||||
it('drops attribute-only events such as macOS mmap reads of git metadata', async () => {
|
||||
const path = join(directory, 'packed-refs')
|
||||
writeFileSync(path, 'refs')
|
||||
const onChange = vi.fn()
|
||||
fileWatch = watchFileAcrossReplacement(path, onChange, (error) => {
|
||||
throw error
|
||||
})
|
||||
await settle()
|
||||
|
||||
for (let read = 0; read < 50; read++) {
|
||||
nativeEvents.at(-1)?.()
|
||||
}
|
||||
await settle()
|
||||
expect(onChange).not.toHaveBeenCalled()
|
||||
|
||||
writeFileSync(path, 'refs and more')
|
||||
for (let read = 0; read < 50; read++) {
|
||||
nativeEvents.at(-1)?.()
|
||||
}
|
||||
await settle()
|
||||
expect(onChange).toHaveBeenCalledOnce()
|
||||
})
|
||||
})
|
||||
|
||||
describe('watchFileAcrossReplacement after a watcher error', () => {
|
||||
it('closes promptly and rebinds once the runtime has closed the errored watcher', async () => {
|
||||
const path = join(directory, 'index')
|
||||
writeFileSync(path, 'index')
|
||||
const onError = vi.fn()
|
||||
fileWatch = watchFileAcrossReplacement(path, vi.fn(), onError)
|
||||
await settle()
|
||||
|
||||
nativeFailures.at(-1)?.(new Error('EPERM'))
|
||||
expect(onError).toHaveBeenCalledOnce()
|
||||
expect(fileWatch.isBound()).toBe(false)
|
||||
await fileWatch.refresh()
|
||||
expect(fileWatch.isBound()).toBe(true)
|
||||
|
||||
nativeFailures.at(-1)?.(new Error('EPERM'))
|
||||
const startedAt = Date.now()
|
||||
await fileWatch.close()
|
||||
fileWatch = null
|
||||
// Well under the bounded close fallback: the errored watcher is not awaited at all.
|
||||
expect(Date.now() - startedAt).toBeLessThan(500)
|
||||
})
|
||||
|
||||
it('bounds closing a watcher whose close event never arrives', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const watcher = watch(join(directory, 'config'), { persistent: false }, () => {})
|
||||
watcher.on('error', () => {})
|
||||
nativeFailures.at(-1)?.(new Error('EPERM'))
|
||||
const settled = vi.fn()
|
||||
void closeFileSystemWatcher(watcher).then(settled)
|
||||
await vi.advanceTimersByTimeAsync(999)
|
||||
expect(settled).not.toHaveBeenCalled()
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
expect(settled).toHaveBeenCalledOnce()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,116 @@
|
||||
import { mkdtempSync, renameSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { watchFileAcrossReplacement, type RebindingFileWatch } from './rebinding-file-watch'
|
||||
|
||||
let directory: string
|
||||
let fileWatch: RebindingFileWatch | null
|
||||
|
||||
beforeEach(() => {
|
||||
directory = mkdtempSync(join(tmpdir(), 'orca-rebinding-file-watch-'))
|
||||
fileWatch = null
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await fileWatch?.close()
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
function replaceAtomically(path: string, content: string): void {
|
||||
const temporary = `${path}.${Math.random().toString(36).slice(2)}.tmp`
|
||||
writeFileSync(temporary, content)
|
||||
renameSync(temporary, path)
|
||||
}
|
||||
|
||||
async function settle(): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, 300))
|
||||
}
|
||||
|
||||
describe('watchFileAcrossReplacement', () => {
|
||||
it('keeps reporting writes across repeated atomic replacement', async () => {
|
||||
const path = join(directory, 'index')
|
||||
writeFileSync(path, '0')
|
||||
const onChange = vi.fn()
|
||||
fileWatch = watchFileAcrossReplacement(path, onChange, (error) => {
|
||||
throw error
|
||||
})
|
||||
await settle()
|
||||
onChange.mockClear()
|
||||
|
||||
for (const content of ['1', '2', '3']) {
|
||||
const before = onChange.mock.calls.length
|
||||
replaceAtomically(path, content)
|
||||
await vi.waitFor(() => expect(onChange.mock.calls.length).toBeGreaterThan(before), {
|
||||
timeout: 5_000
|
||||
})
|
||||
// Let the post-event re-stat rebind to the new inode before the next write.
|
||||
await settle()
|
||||
}
|
||||
const before = onChange.mock.calls.length
|
||||
writeFileSync(path, 'in place')
|
||||
await vi.waitFor(() => expect(onChange.mock.calls.length).toBeGreaterThan(before), {
|
||||
timeout: 5_000
|
||||
})
|
||||
})
|
||||
|
||||
it('ignores writes to siblings and subdirectories of the watched file', async () => {
|
||||
const path = join(directory, 'HEAD')
|
||||
writeFileSync(path, 'ref: refs/heads/main')
|
||||
const onChange = vi.fn()
|
||||
fileWatch = watchFileAcrossReplacement(path, onChange, (error) => {
|
||||
throw error
|
||||
})
|
||||
await settle()
|
||||
onChange.mockClear()
|
||||
|
||||
writeFileSync(join(directory, 'FETCH_HEAD'), 'noise')
|
||||
replaceAtomically(join(directory, 'last-status.json'), '{}')
|
||||
await settle()
|
||||
|
||||
expect(onChange).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('binds a file created after the watch started once refreshed', async () => {
|
||||
const path = join(directory, 'packed-refs')
|
||||
const onChange = vi.fn()
|
||||
fileWatch = watchFileAcrossReplacement(path, onChange, (error) => {
|
||||
throw error
|
||||
})
|
||||
expect(fileWatch.isBound()).toBe(false)
|
||||
await fileWatch.refresh()
|
||||
expect(onChange).not.toHaveBeenCalled()
|
||||
|
||||
writeFileSync(path, 'created')
|
||||
await fileWatch.refresh()
|
||||
expect(onChange).toHaveBeenCalledOnce()
|
||||
expect(fileWatch.isBound()).toBe(true)
|
||||
await settle()
|
||||
onChange.mockClear()
|
||||
|
||||
writeFileSync(path, 'updated')
|
||||
await vi.waitFor(() => expect(onChange).toHaveBeenCalled(), { timeout: 5_000 })
|
||||
})
|
||||
|
||||
it('drops the binding when the file is deleted and reports nothing after close', async () => {
|
||||
const path = join(directory, 'config')
|
||||
writeFileSync(path, 'a')
|
||||
const onChange = vi.fn()
|
||||
fileWatch = watchFileAcrossReplacement(path, onChange, (error) => {
|
||||
throw error
|
||||
})
|
||||
await settle()
|
||||
rmSync(path)
|
||||
await vi.waitFor(() => expect(onChange).toHaveBeenCalled(), { timeout: 5_000 })
|
||||
expect(fileWatch.isBound()).toBe(false)
|
||||
|
||||
writeFileSync(path, 'b')
|
||||
await fileWatch.refresh()
|
||||
expect(fileWatch.isBound()).toBe(true)
|
||||
await fileWatch.close()
|
||||
onChange.mockClear()
|
||||
writeFileSync(path, 'c')
|
||||
await settle()
|
||||
expect(onChange).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,195 @@
|
||||
import { statSync, watch, type BigIntStats, type FSWatcher } from 'node:fs'
|
||||
import { stat } from 'node:fs/promises'
|
||||
|
||||
// Why: watching one file instead of its directory keeps macOS off fseventsd — a
|
||||
// file watch is a kqueue registration there, while a directory watch is a
|
||||
// recursive FSEvents stream that delivers every write beneath it. A file watch
|
||||
// binds an inode, though: rename-replace (git lock files, atomic JSON writes)
|
||||
// leaves it deaf and a missing file cannot be watched at all, so events and
|
||||
// owner sweeps re-stat the path and rebind when it names a new file.
|
||||
|
||||
export type RebindingFileWatch = {
|
||||
/** Re-stats the path, rebinding and reporting a change if it differs from the last look. */
|
||||
refresh: () => Promise<void>
|
||||
isBound: () => boolean
|
||||
close: () => Promise<void>
|
||||
}
|
||||
|
||||
// Why: macOS raises a vnode attribute event for every mmap read (git maps the
|
||||
// index and packed-refs; ~120/s measured on a busy repo), so events are
|
||||
// coalesced and only a stat that differs counts as a change. An access-time
|
||||
// update leaves mtime, ctime and size alone. Each event that proves to be
|
||||
// noise doubles the window, so a file that every git process maps costs about
|
||||
// one rebind a second; a real change restores the short window.
|
||||
const EVENT_COALESCE_MS = 200
|
||||
const MAX_EVENT_COALESCE_MS = 1_600
|
||||
// Why: the native registration only lands on a later loop turn, and a stat
|
||||
// already queued can predate a replacement in that gap, which would leave the
|
||||
// binding on a dead inode. A timer cannot fire before that turn.
|
||||
const BIND_VERIFY_DELAY_MS = 50
|
||||
|
||||
type FileSnapshot = { identity: string; stamp: string }
|
||||
|
||||
// Why: Node closes an errored watcher's handle without emitting 'close', and a
|
||||
// later close() is then a silent no-op, so waiting on 'close' alone can hang.
|
||||
const WATCHER_CLOSE_TIMEOUT_MS = 1_000
|
||||
|
||||
export function closeFileSystemWatcher(watcher: FSWatcher): Promise<void> {
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
const timer = setTimeout(resolve, WATCHER_CLOSE_TIMEOUT_MS)
|
||||
timer.unref?.()
|
||||
watcher.once('close', resolve)
|
||||
watcher.once('error', () => resolve())
|
||||
try {
|
||||
watcher.close()
|
||||
} catch {
|
||||
resolve()
|
||||
}
|
||||
return promise.finally(() => clearTimeout(timer))
|
||||
}
|
||||
|
||||
function fileSnapshot(entry: BigIntStats): FileSnapshot {
|
||||
return {
|
||||
// Volumes without inode ids fall back to birth time, which a rename-replace also changes.
|
||||
identity:
|
||||
entry.ino !== 0n ? `${entry.dev}:${entry.ino}` : `${entry.dev}:birth:${entry.birthtimeNs}`,
|
||||
stamp: `${entry.size}:${entry.mtimeNs}:${entry.ctimeNs}`
|
||||
}
|
||||
}
|
||||
|
||||
function isMissingPathError(error: unknown): boolean {
|
||||
const code = error instanceof Error && 'code' in error ? error.code : undefined
|
||||
return code === 'ENOENT' || code === 'ENOTDIR'
|
||||
}
|
||||
|
||||
export function watchFileAcrossReplacement(
|
||||
path: string,
|
||||
onChange: () => void,
|
||||
onError: (error: Error) => void
|
||||
): RebindingFileWatch {
|
||||
let watcher: FSWatcher | null = null
|
||||
let boundIdentity: string | null = null
|
||||
let lastSeen: FileSnapshot | null = null
|
||||
let closed = false
|
||||
let refreshing: Promise<void> | null = null
|
||||
let refreshAgain = false
|
||||
let coalesceTimer: ReturnType<typeof setTimeout> | undefined
|
||||
let coalesceMs = EVENT_COALESCE_MS
|
||||
let verifyTimer: ReturnType<typeof setTimeout> | undefined
|
||||
const closing = new Set<Promise<void>>()
|
||||
|
||||
const unbind = (): void => {
|
||||
const current = watcher
|
||||
watcher = null
|
||||
boundIdentity = null
|
||||
if (current) {
|
||||
const pending = closeFileSystemWatcher(current).finally(() => closing.delete(pending))
|
||||
closing.add(pending)
|
||||
}
|
||||
}
|
||||
|
||||
const scheduleCoalescedRefresh = (): void => {
|
||||
if (coalesceTimer === undefined) {
|
||||
coalesceTimer = setTimeout(() => {
|
||||
coalesceTimer = undefined
|
||||
coalesceMs = Math.min(coalesceMs * 2, MAX_EVENT_COALESCE_MS)
|
||||
void refresh()
|
||||
}, coalesceMs)
|
||||
coalesceTimer.unref?.()
|
||||
}
|
||||
}
|
||||
|
||||
const bind = (identity: string): void => {
|
||||
try {
|
||||
const next = watch(path, { persistent: false }, () => {
|
||||
if (!closed && watcher === next) {
|
||||
// One-shot: a hot file costs one rebind per window instead of a wakeup
|
||||
// per read. The refresh stat covers writes made while disarmed.
|
||||
unbind()
|
||||
scheduleCoalescedRefresh()
|
||||
}
|
||||
})
|
||||
next.on('error', (error) => {
|
||||
if (!closed && watcher === next) {
|
||||
// The runtime already closed it; close() would wait for a 'close' that never comes.
|
||||
watcher = null
|
||||
boundIdentity = null
|
||||
onError(error)
|
||||
}
|
||||
})
|
||||
watcher = next
|
||||
boundIdentity = identity
|
||||
clearTimeout(verifyTimer)
|
||||
verifyTimer = setTimeout(() => void refresh(), BIND_VERIFY_DELAY_MS)
|
||||
verifyTimer.unref?.()
|
||||
} catch (error) {
|
||||
if (!isMissingPathError(error)) {
|
||||
onError(error instanceof Error ? error : new Error(String(error)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const readSnapshot = async (): Promise<FileSnapshot | null> => {
|
||||
try {
|
||||
return fileSnapshot(await stat(path, { bigint: true }))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
const refreshOnce = async (): Promise<void> => {
|
||||
const current = await readSnapshot()
|
||||
if (closed) {
|
||||
return
|
||||
}
|
||||
const changed = current?.identity !== lastSeen?.identity || current?.stamp !== lastSeen?.stamp
|
||||
lastSeen = current
|
||||
if (current?.identity !== boundIdentity) {
|
||||
unbind()
|
||||
if (current) {
|
||||
// A replacement between the stat and this bind leaves the identity stale,
|
||||
// which the bind's verify pass treats as a change: the harmless direction.
|
||||
bind(current.identity)
|
||||
}
|
||||
}
|
||||
if (changed) {
|
||||
coalesceMs = EVENT_COALESCE_MS
|
||||
onChange()
|
||||
}
|
||||
}
|
||||
|
||||
const refresh = (): Promise<void> => {
|
||||
if (refreshing) {
|
||||
refreshAgain = true
|
||||
return refreshing
|
||||
}
|
||||
refreshing = (async () => {
|
||||
do {
|
||||
refreshAgain = false
|
||||
await refreshOnce()
|
||||
} while (refreshAgain && !closed)
|
||||
})().finally(() => {
|
||||
refreshing = null
|
||||
})
|
||||
return refreshing
|
||||
}
|
||||
|
||||
try {
|
||||
lastSeen = fileSnapshot(statSync(path, { bigint: true }))
|
||||
bind(lastSeen.identity)
|
||||
} catch {
|
||||
// Missing until a refresh finds it.
|
||||
}
|
||||
|
||||
return {
|
||||
refresh,
|
||||
isBound: () => watcher !== null,
|
||||
close: async () => {
|
||||
closed = true
|
||||
clearTimeout(coalesceTimer)
|
||||
clearTimeout(verifyTimer)
|
||||
unbind()
|
||||
await Promise.all(closing)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -91,6 +91,30 @@ describe('Active Server after a saved host is removed', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps following the registry after earlier replacements', async () => {
|
||||
const kept = saveEnvironment('kept')
|
||||
const removed = saveEnvironment('removed')
|
||||
const store = preferenceStore(removed.id)
|
||||
const stopWatching = watchRuntimeEnvironmentPreference(store, userDataPath)
|
||||
try {
|
||||
await new Promise((resolve) => setTimeout(resolve, 150))
|
||||
// Unrelated replacements move the registry to new inodes first.
|
||||
saveEnvironment('first')
|
||||
await new Promise((resolve) => setTimeout(resolve, 150))
|
||||
saveEnvironment('second')
|
||||
await new Promise((resolve) => setTimeout(resolve, 150))
|
||||
expect(store.getSettings().activeRuntimeEnvironmentId).toBe(removed.id)
|
||||
|
||||
removeEnvironment(userDataPath, removed.id)
|
||||
await vi.waitFor(() => expect(store.getSettings().activeRuntimeEnvironmentId).toBeNull(), {
|
||||
timeout: 5_000
|
||||
})
|
||||
expect(listEnvironments(userDataPath).map(({ id }) => id)).toContain(kept.id)
|
||||
} finally {
|
||||
stopWatching()
|
||||
}
|
||||
})
|
||||
|
||||
it('repairs a CLI removal before settings can route to the deleted pairing', () => {
|
||||
const environment = saveEnvironment()
|
||||
const store = preferenceStore(environment.id)
|
||||
|
||||
@@ -1,11 +1,10 @@
|
||||
import { watch, type FSWatcher } from 'node:fs'
|
||||
import { basename } from 'node:path'
|
||||
import type { GlobalSettings } from '../../../shared/global-settings-types'
|
||||
import {
|
||||
getEnvironmentStorePath,
|
||||
listEnvironments
|
||||
} from '../../../shared/runtime-environment-store'
|
||||
import type { Store } from '../../persistence'
|
||||
import { watchFileAcrossReplacement } from '../rebinding-file-watch'
|
||||
|
||||
type PreferenceStore = Pick<Store, 'getSettings' | 'updateSettings'>
|
||||
|
||||
@@ -40,11 +39,14 @@ export function readSettingsWithRuntimeEnvironmentPreference(
|
||||
return reconcileRuntimeEnvironmentPreference(store, environments)
|
||||
}
|
||||
|
||||
// Why: finds a registry created, or replaced unseen, after the watch started.
|
||||
// Settings reads repair on their own, so this only bounds renderer staleness.
|
||||
const REGISTRY_REBIND_SWEEP_MS = 30_000
|
||||
|
||||
export function watchRuntimeEnvironmentPreference(
|
||||
store: PreferenceStore,
|
||||
userDataPath: string
|
||||
): () => void {
|
||||
const registryName = basename(getEnvironmentStorePath(userDataPath))
|
||||
const repairPreference = (): void => {
|
||||
try {
|
||||
readSettingsWithRuntimeEnvironmentPreference(store, userDataPath)
|
||||
@@ -54,23 +56,18 @@ export function watchRuntimeEnvironmentPreference(
|
||||
}
|
||||
// Reconcile changes made while the native watch was being installed.
|
||||
const initialCheck = setImmediate(repairPreference)
|
||||
let watcher: FSWatcher | undefined
|
||||
try {
|
||||
// Watch the directory because registry writes atomically replace the file.
|
||||
watcher = watch(userDataPath, { persistent: false }, (_event, filename) => {
|
||||
if (filename === null || filename.toString() === registryName) {
|
||||
repairPreference()
|
||||
}
|
||||
})
|
||||
watcher.on('error', (error) => {
|
||||
watcher?.close()
|
||||
console.warn('[runtime-environments] Active Server preference watch failed:', error)
|
||||
})
|
||||
} catch (error) {
|
||||
console.warn('[runtime-environments] Active Server preference watch failed:', error)
|
||||
}
|
||||
// Why: watch the registry file, not userData. On macOS a directory watch is a
|
||||
// recursive FSEvents stream that Orca's own userData writes keep busy.
|
||||
const registryWatch = watchFileAcrossReplacement(
|
||||
getEnvironmentStorePath(userDataPath),
|
||||
repairPreference,
|
||||
(error) => console.warn('[runtime-environments] Active Server preference watch failed:', error)
|
||||
)
|
||||
const sweep = setInterval(() => void registryWatch.refresh(), REGISTRY_REBIND_SWEEP_MS)
|
||||
sweep.unref?.()
|
||||
return () => {
|
||||
clearImmediate(initialCheck)
|
||||
watcher?.close()
|
||||
clearInterval(sweep)
|
||||
void registryWatch.close()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const h = vi.hoisted(() => ({
|
||||
@@ -25,7 +26,11 @@ import {
|
||||
resetShallowWatchDeliveryProbeForTests
|
||||
} from './shallow-watch-delivery-probe'
|
||||
|
||||
const originalPlatform = process.platform
|
||||
|
||||
beforeEach(() => {
|
||||
// Write counts below are for the directory probe; darwin is covered separately.
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: 'linux' })
|
||||
vi.useFakeTimers()
|
||||
vi.resetAllMocks()
|
||||
resetShallowWatchDeliveryProbeForTests()
|
||||
@@ -44,6 +49,7 @@ beforeEach(() => {
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: originalPlatform })
|
||||
vi.clearAllTimers()
|
||||
vi.useRealTimers()
|
||||
resetShallowWatchDeliveryProbeForTests()
|
||||
@@ -156,4 +162,21 @@ describe('shallow probe resource ownership', () => {
|
||||
await expect(detectShallowWatchDelivery()).resolves.toBe(true)
|
||||
expect(h.watch).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('probes a file watch on macOS, where shallow subscriptions watch files', async () => {
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: 'darwin' })
|
||||
h.write.mockImplementation(async () => {
|
||||
if (h.watch.mock.calls.length > 0) {
|
||||
h.deliver()
|
||||
}
|
||||
})
|
||||
|
||||
await expect(measureShallowWatchDelivery()).resolves.toBe(true)
|
||||
|
||||
const probePath = join('/fake-shallow-probe', 'probe')
|
||||
expect(h.write.mock.calls[0]).toEqual([probePath, '0'])
|
||||
expect(h.write.mock.invocationCallOrder[0]).toBeLessThan(h.watch.mock.invocationCallOrder[0])
|
||||
expect(h.watch.mock.calls[0][0]).toBe(probePath)
|
||||
expectReleased()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -32,8 +32,17 @@ export async function measureShallowWatchDelivery(timeoutMs = PROBE_TIMEOUT_MS):
|
||||
let directory: string | null = null
|
||||
try {
|
||||
directory = await mkdtemp(join(tmpdir(), 'orca-shallow-probe-'))
|
||||
const probePath = join(directory, 'probe')
|
||||
// macOS shallow subscriptions watch files (kqueue), not directories (FSEvents),
|
||||
// so probe the same notification path they depend on.
|
||||
const watchesFiles = process.platform === 'darwin'
|
||||
if (watchesFiles) {
|
||||
await writeFile(probePath, '0')
|
||||
}
|
||||
const { promise, resolve } = Promise.withResolvers<boolean>()
|
||||
const watcher = watch(directory, { persistent: false }, () => resolve(true))
|
||||
const watcher = watch(watchesFiles ? probePath : directory, { persistent: false }, () =>
|
||||
resolve(true)
|
||||
)
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
try {
|
||||
watcher.on('error', () => resolve(false))
|
||||
@@ -41,8 +50,8 @@ export async function measureShallowWatchDelivery(timeoutMs = PROBE_TIMEOUT_MS):
|
||||
// has no other pending work at probe time.
|
||||
timer = setTimeout(() => resolve(false), timeoutMs)
|
||||
// Two writes: some backends coalesce the creation of the first entry.
|
||||
await writeFile(join(directory, 'probe'), '1')
|
||||
await writeFile(join(directory, 'probe'), '2')
|
||||
await writeFile(probePath, '1')
|
||||
await writeFile(probePath, '2')
|
||||
return await promise
|
||||
} finally {
|
||||
if (timer !== undefined) {
|
||||
|
||||
Reference in New Issue
Block a user