mirror of
https://github.com/stablyai/orca.git
synced 2026-09-29 08:03:20 +00:00
fix(watchers): own shallow binding teardown completion
This commit is contained in:
Merge Sim
@@ -135,6 +135,82 @@ describe('parcel watcher process canary', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it.each(['error', 'delayed-close', 'throw'] as const)(
|
||||
'advances the real child lifecycle queue after shallow %s teardown',
|
||||
async (schedule) => {
|
||||
detectShallowWatchDeliveryMock.mockResolvedValue(true)
|
||||
const handles: (EventEmitter & { close: () => void })[] = []
|
||||
watchMock.mockImplementation(() => {
|
||||
const watcher = new EventEmitter() as EventEmitter & {
|
||||
close: () => void
|
||||
}
|
||||
watcher.close = () => {
|
||||
watcher.emit('close')
|
||||
}
|
||||
handles.push(watcher)
|
||||
return watcher
|
||||
})
|
||||
const sendMock = vi.fn()
|
||||
process.send = sendMock
|
||||
await import('./parcel-watcher-process-entry')
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
for (const id of [1, 2]) {
|
||||
process.emit('message', {
|
||||
op: 'subscribe',
|
||||
id,
|
||||
dir: `/folder-${id}`,
|
||||
opts: { mode: 'shallow', include: ['HEAD'] }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
}
|
||||
handles[0].close = () => {
|
||||
if (schedule === 'throw') {
|
||||
throw new Error('handle still active')
|
||||
}
|
||||
}
|
||||
if (schedule === 'error') {
|
||||
handles[0].emit('error', new Error('terminal'))
|
||||
}
|
||||
process.emit('message', { op: 'unsubscribe', id: 1 })
|
||||
process.emit('message', {
|
||||
op: 'subscribe',
|
||||
id: 3,
|
||||
dir: '/folder-3',
|
||||
opts: { mode: 'shallow', include: ['HEAD'] }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
if (schedule === 'delayed-close') {
|
||||
expect(sendMock).not.toHaveBeenCalledWith({
|
||||
op: 'subscribe-started',
|
||||
id: 3
|
||||
})
|
||||
handles[0].emit('close')
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
}
|
||||
expect(sendMock).toHaveBeenCalledWith({ op: 'subscribed', id: 3 })
|
||||
if (schedule === 'throw') {
|
||||
expect(sendMock).toHaveBeenCalledWith({
|
||||
op: 'unsubscribe-failed',
|
||||
id: 1,
|
||||
message: 'handle still active'
|
||||
})
|
||||
expect(sendMock).not.toHaveBeenCalledWith({
|
||||
op: 'unsubscribed',
|
||||
id: 1
|
||||
})
|
||||
handles[0].emit('close')
|
||||
} else {
|
||||
expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 1 })
|
||||
}
|
||||
for (const id of [1, 2, 3]) {
|
||||
process.emit('message', { op: 'unsubscribe', id })
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 2 })
|
||||
expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 3 })
|
||||
}
|
||||
)
|
||||
|
||||
it('does not restart while a native subscription is still crawling', async () => {
|
||||
subscribeMock
|
||||
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import { statSync, watch, type FSWatcher } from 'node:fs'
|
||||
import { statSync, watch } from 'node:fs'
|
||||
import { stat } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import type { Event as ParcelWatcherEvent } from '@parcel/watcher'
|
||||
import { createShallowWatcherBinding, type ShallowWatcherBinding } from './shallow-watcher-binding'
|
||||
|
||||
export type ShallowWatcherSubscription = {
|
||||
unsubscribe: () => Promise<void>
|
||||
@@ -14,17 +15,6 @@ 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[],
|
||||
@@ -44,8 +34,9 @@ export function startShallowWatcher(
|
||||
pathsByDirectory.set(parent, fileNames)
|
||||
}
|
||||
|
||||
const watchers = new Map<string, FSWatcher>()
|
||||
const boundIdentities = new Map<string, string>()
|
||||
const watchers = new Map<string, ShallowWatcherBinding>()
|
||||
const ownedBindings = new Set<ShallowWatcherBinding>()
|
||||
let unsubscribePromise: Promise<void> | undefined
|
||||
let disposed = false
|
||||
let reportedError = false
|
||||
|
||||
@@ -73,13 +64,15 @@ export function startShallowWatcher(
|
||||
return
|
||||
}
|
||||
watchers.delete(parent)
|
||||
boundIdentities.delete(parent)
|
||||
void closeFileSystemWatcher(existing)
|
||||
void existing.close().catch(reportError)
|
||||
}
|
||||
const directoryPath = join(rootPath, parent)
|
||||
// Read before watch so a replacement in the gap is detected by the next sweep.
|
||||
const identity = directoryIdentitySync(parent)
|
||||
try {
|
||||
let binding: ShallowWatcherBinding
|
||||
const watcher = watch(directoryPath, { persistent: false }, (eventType, fileName) => {
|
||||
if (disposed) {
|
||||
if (disposed || watchers.get(parent) !== binding) {
|
||||
return
|
||||
}
|
||||
const name = fileName?.toString()
|
||||
@@ -100,8 +93,19 @@ export function startShallowWatcher(
|
||||
emitUpdates(parent, [name])
|
||||
}
|
||||
})
|
||||
watcher.on('error', reportError)
|
||||
watchers.set(parent, watcher)
|
||||
binding = createShallowWatcherBinding(
|
||||
watcher,
|
||||
identity,
|
||||
() => {
|
||||
ownedBindings.delete(binding)
|
||||
if (watchers.get(parent) === binding) {
|
||||
watchers.delete(parent)
|
||||
}
|
||||
},
|
||||
reportError
|
||||
)
|
||||
ownedBindings.add(binding)
|
||||
watchers.set(parent, binding)
|
||||
} catch (error) {
|
||||
// Nested metadata directories may not exist until Git creates them.
|
||||
if (parent === '') {
|
||||
@@ -129,18 +133,18 @@ export function startShallowWatcher(
|
||||
}
|
||||
|
||||
const refreshBinding = async (parent: string, fileNames: Set<string>): Promise<void> => {
|
||||
const existing = watchers.get(parent)
|
||||
const identity = await directoryIdentity(parent)
|
||||
if (disposed || identity === null) {
|
||||
if (disposed || identity === null || watchers.get(parent) !== existing) {
|
||||
return
|
||||
}
|
||||
const bound = boundIdentities.get(parent)
|
||||
const bound = existing?.identity
|
||||
if (bound === identity) {
|
||||
return
|
||||
}
|
||||
// Either the directory appeared after we started, or it was replaced while
|
||||
// watched. Both leave the old binding deaf, so rebind and resync.
|
||||
watchDirectory(parent, fileNames, true)
|
||||
boundIdentities.set(parent, identity)
|
||||
if (bound !== undefined) {
|
||||
emitUpdates(parent, fileNames)
|
||||
}
|
||||
@@ -157,22 +161,19 @@ export function startShallowWatcher(
|
||||
rebindTimer.unref?.()
|
||||
|
||||
for (const [parent, fileNames] of pathsByDirectory) {
|
||||
// Why: read identity BEFORE binding. If the directory is replaced in the gap,
|
||||
// the recorded identity is stale and the first sweep rebinds — the harmless
|
||||
// direction. Reading after would pin the dead inode's watcher to the new
|
||||
// identity, and the sweep would then never rebind it.
|
||||
const identityBeforeBind = directoryIdentitySync(parent)
|
||||
watchDirectory(parent, fileNames)
|
||||
if (watchers.has(parent) && identityBeforeBind !== null) {
|
||||
boundIdentities.set(parent, identityBeforeBind)
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
unsubscribe: async () => {
|
||||
disposed = true
|
||||
clearInterval(rebindTimer)
|
||||
await Promise.all([...watchers.values()].map(closeFileSystemWatcher))
|
||||
unsubscribe: () => {
|
||||
if (!unsubscribePromise) {
|
||||
disposed = true
|
||||
clearInterval(rebindTimer)
|
||||
unsubscribePromise = Promise.all([...ownedBindings].map((binding) => binding.close())).then(
|
||||
() => undefined
|
||||
)
|
||||
}
|
||||
return unsubscribePromise
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { watchMock, statMock, statSyncMock } = vi.hoisted(() => ({
|
||||
watchMock: vi.fn(),
|
||||
statMock: vi.fn(),
|
||||
statSyncMock: vi.fn()
|
||||
}))
|
||||
vi.mock('node:fs', () => ({ watch: watchMock, statSync: statSyncMock }))
|
||||
vi.mock('node:fs/promises', () => ({ stat: statMock }))
|
||||
import { startShallowWatcher } from './parcel-watcher-shallow-subscription'
|
||||
|
||||
class Watcher extends EventEmitter {
|
||||
close = vi.fn(() => {
|
||||
this.emit('close')
|
||||
})
|
||||
constructor(readonly change: (event: string, name: string) => void) {
|
||||
super()
|
||||
}
|
||||
}
|
||||
const root = join('folder', 'metadata')
|
||||
const identity = (ino = 1) => ({ dev: 1, ino, isDirectory: () => true })
|
||||
let created: Watcher[]
|
||||
const subscribe = (onError = vi.fn(), onEvents = vi.fn()) =>
|
||||
startShallowWatcher(root, ['HEAD', 'logs/HEAD'], onEvents, onError)
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
created = []
|
||||
statMock.mockReset().mockResolvedValue(identity())
|
||||
statSyncMock.mockReset().mockReturnValue(identity())
|
||||
watchMock.mockReset().mockImplementation((_path, _options, change) => {
|
||||
const watcher = new Watcher(change)
|
||||
created.push(watcher)
|
||||
return watcher
|
||||
})
|
||||
})
|
||||
afterEach(() => {
|
||||
vi.clearAllTimers()
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
describe('shallow watcher binding ownership', () => {
|
||||
it('reuses unsubscribe completion during and after delayed normal close', async () => {
|
||||
const subscription = subscribe()
|
||||
created[0].close.mockImplementation(() => {})
|
||||
const first = subscription.unsubscribe()
|
||||
const settled = vi.fn()
|
||||
void first.then(settled)
|
||||
expect(subscription.unsubscribe()).toBe(first)
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
expect(settled).not.toHaveBeenCalled()
|
||||
created[0].emit('close')
|
||||
await first
|
||||
expect(subscription.unsubscribe()).toBe(first)
|
||||
await subscription.unsubscribe()
|
||||
expect(created[0].close).toHaveBeenCalledTimes(1)
|
||||
expect(settled).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it.each(['close', 'error'] as const)(
|
||||
'remembers prior terminal %s without awaiting another close',
|
||||
async (event) => {
|
||||
const onError = vi.fn()
|
||||
const subscription = subscribe(onError)
|
||||
created[0].close.mockImplementation(() => {})
|
||||
created[0].emit(event, new Error('native terminal'))
|
||||
await subscription.unsubscribe()
|
||||
await subscription.unsubscribe()
|
||||
expect(created[0].close).not.toHaveBeenCalled()
|
||||
expect(onError).toHaveBeenCalledTimes(event === 'error' ? 1 : 0)
|
||||
}
|
||||
)
|
||||
|
||||
it('settles an in-flight close on terminal error without a close event', async () => {
|
||||
const subscription = subscribe()
|
||||
created[0].close.mockImplementation(() => {})
|
||||
const completion = subscription.unsubscribe()
|
||||
created[0].emit('error', new Error('native terminal'))
|
||||
await completion
|
||||
created[0].emit('close')
|
||||
expect(subscription.unsubscribe()).toBe(completion)
|
||||
})
|
||||
|
||||
it('rejects throwing close without terminal evidence and reuses that failure', async () => {
|
||||
const subscription = subscribe()
|
||||
const failure = new Error('still live')
|
||||
created[0].close.mockImplementation(() => {
|
||||
throw failure
|
||||
})
|
||||
const completion = subscription.unsubscribe()
|
||||
await expect(completion).rejects.toBe(failure)
|
||||
expect(subscription.unsubscribe()).toBe(completion)
|
||||
await expect(subscription.unsubscribe()).rejects.toBe(failure)
|
||||
expect(created[0].close).toHaveBeenCalledTimes(1)
|
||||
expect(created[1].close).toHaveBeenCalledTimes(1)
|
||||
created[0].emit('close')
|
||||
})
|
||||
|
||||
it.each(['close', 'error'] as const)(
|
||||
'accepts terminal %s established before close throws',
|
||||
async (event) => {
|
||||
const subscription = subscribe()
|
||||
created[0].close.mockImplementation(() => {
|
||||
created[0].emit(event, new Error('terminal'))
|
||||
throw new Error('after terminal')
|
||||
})
|
||||
await subscription.unsubscribe()
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['close', 'error'] as const)(
|
||||
'isolates late old %s and updates from replacement identity',
|
||||
async (event) => {
|
||||
const onEvents = vi.fn()
|
||||
const subscription = subscribe(vi.fn(), onEvents)
|
||||
const old = created[1]
|
||||
old.close.mockImplementation(() => {})
|
||||
created[0].change('rename', 'logs')
|
||||
const replacement = created[2]
|
||||
old.emit(event, new Error('old terminal'))
|
||||
onEvents.mockClear()
|
||||
old.change('change', 'HEAD')
|
||||
expect(onEvents).not.toHaveBeenCalled()
|
||||
replacement.change('change', 'HEAD')
|
||||
expect(onEvents).toHaveBeenCalledWith([{ type: 'update', path: join(root, 'logs', 'HEAD') }])
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
expect(created).toHaveLength(3)
|
||||
await subscription.unsubscribe()
|
||||
expect(replacement.close).toHaveBeenCalledTimes(1)
|
||||
}
|
||||
)
|
||||
|
||||
it('awaits the retired generation as well as its replacement', async () => {
|
||||
const subscription = subscribe()
|
||||
const old = created[1]
|
||||
old.close.mockImplementation(() => {})
|
||||
created[0].change('rename', 'logs')
|
||||
const settled = vi.fn()
|
||||
const completion = subscription.unsubscribe()
|
||||
void completion.then(settled)
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
expect(settled).not.toHaveBeenCalled()
|
||||
expect(created[2].close).toHaveBeenCalledTimes(1)
|
||||
old.emit('close')
|
||||
await completion
|
||||
expect(old.close).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('preserves failed retired teardown for unsubscribe and host recovery', async () => {
|
||||
const onError = vi.fn()
|
||||
const subscription = subscribe(onError)
|
||||
const failure = new Error('retired handle still live')
|
||||
created[1].close.mockImplementation(() => {
|
||||
throw failure
|
||||
})
|
||||
created[0].change('rename', 'logs')
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
expect(onError).toHaveBeenCalledWith(failure)
|
||||
await expect(subscription.unsubscribe()).rejects.toBe(failure)
|
||||
expect(created[2].close).toHaveBeenCalledTimes(1)
|
||||
created[1].emit('close')
|
||||
})
|
||||
|
||||
it('rebinds after terminal error even when the directory inode is unchanged', async () => {
|
||||
const subscription = subscribe()
|
||||
created[1].emit('error', new Error('terminal'))
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
expect(created).toHaveLength(3)
|
||||
await subscription.unsubscribe()
|
||||
expect(created[2].close).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('retries failed binding creation instead of recording an unwatched identity', async () => {
|
||||
const subscription = subscribe()
|
||||
statMock.mockResolvedValue(identity(2))
|
||||
statSyncMock.mockReturnValue(identity(2))
|
||||
watchMock.mockImplementationOnce(() => {
|
||||
throw new Error('missing')
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
const attempts = watchMock.mock.calls.length
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
expect(watchMock).toHaveBeenCalledTimes(attempts + 1)
|
||||
await subscription.unsubscribe()
|
||||
})
|
||||
|
||||
it('does not let an old async identity probe rebind a replacement', async () => {
|
||||
const subscription = subscribe()
|
||||
const pending = Promise.withResolvers<ReturnType<typeof identity>>()
|
||||
statMock.mockImplementation((path) =>
|
||||
path === join(root, 'logs') ? pending.promise : Promise.resolve(identity())
|
||||
)
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
created[0].change('rename', 'logs')
|
||||
pending.resolve(identity(2))
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
expect(created).toHaveLength(3)
|
||||
await subscription.unsubscribe()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,48 @@
|
||||
import type { FSWatcher } from 'node:fs'
|
||||
|
||||
export type ShallowWatcherBinding = {
|
||||
identity: string | null
|
||||
close: () => Promise<void>
|
||||
}
|
||||
|
||||
export function createShallowWatcherBinding(
|
||||
watcher: FSWatcher,
|
||||
identity: string | null,
|
||||
onTerminal: () => void,
|
||||
onError: (error: unknown) => void
|
||||
): ShallowWatcherBinding {
|
||||
const { promise, resolve, reject } = Promise.withResolvers<void>()
|
||||
let terminal = false
|
||||
let closing = false
|
||||
const finish = (): void => {
|
||||
if (terminal) {
|
||||
return
|
||||
}
|
||||
terminal = true
|
||||
resolve()
|
||||
onTerminal()
|
||||
}
|
||||
watcher.once('close', finish)
|
||||
watcher.on('error', (error) => {
|
||||
// Node closes its native handle before emitting error, without emitting close.
|
||||
finish()
|
||||
onError(error)
|
||||
})
|
||||
return {
|
||||
identity,
|
||||
close: () => {
|
||||
if (!terminal && !closing) {
|
||||
closing = true
|
||||
try {
|
||||
watcher.close()
|
||||
} catch (error) {
|
||||
// A throw alone does not prove the native handle was released.
|
||||
if (!terminal) {
|
||||
reject(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
return promise
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user