starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-30 00:27:39 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
dc3aa36ad349f0da4e43bed9ddd8221aa60bda33
orca/src/main/ipc/parcel-watcher-process-entry.test.ts
T
Neil e7c5263a93 perf(git): make local Git metadata observation event-driven (no scheduler) (#16404) 
* perf(git): make local Git metadata observation event-driven

Replaces the recurring per-repo metadata scan with native filesystem events on
macOS, Linux, and Windows. Polling is retained purely as a fallback.

- Narrow @parcel/watcher stream over <common>/worktrees, extended from macOS to
  Linux and Windows, with the Windows backend pinned explicitly.
- New shallow watcher mode over the allowlisted primary metadata leaves. It
  watches the containing directory rather than each file, so Git's atomic
  write-and-rename does not orphan the binding.
- Selected upstream refs stay on the existing bounded stat poll.

Verified on real hosts rather than in principle:

- Windows: `git worktree remove` and `git worktree prune` both succeed while the
  narrow stream holds the directory. The historical concern that an open handle
  would block prune does not reproduce.
- Linux: inotify costs one instance per event loop, not one per watch, so the
  watch budget is not a constraint.
- macOS/Linux/Windows: shallow events survive repeated commit, checkout,
  config, and pack-refs cycles.

Failure handling, each reproduced before being fixed:

- fs.watch binds an inode and reports nothing once that inode is replaced, with
  no error. Directory bindings are re-checked on a bounded cadence and rebound.
- A host whose notification path is dead accepts registrations and stays mute
  forever. Observed on a macOS machine whose fseventsd had grown to ~15GB and
  saturated a core. A one-shot delivery probe now fails the shallow subscribe on
  such a host so it falls back to polling instead of showing stale metadata.

This change stands alone on main and does not depend on the metadata poll
scheduler.

* test(git-watch): hold reserved inodes across root replacements

Linux returns a released reservation to the free list, so the second
replacement could land back on the first replacement's inode and look
unchanged to reconciliation. Verified on ext4: releasing yields inodes
[N, N+43, N+43] while holding yields [N, N+43, N+44]. macOS never
recycles, which is why this only failed on CI.

* refactor(git-watch): share one single-flight helper between watcher fallbacks

Both fallbacks tracked their in-flight promise with the same self-comparison
on settle, duplicated verbatim. Hoisting it removes a subtle invariant that
was being hand-maintained in two places.

* fix(git-watch): close the silent-staleness paths in primary metadata

Two independent reviews converged on the same root cause: nothing bounded
how long primary metadata could stay wrong once the shallow watcher stopped
reporting. Four distinct paths led there.

- A terminal watch error arriving while the status-ref poll was still starting
  left the repo with status-ref coverage only. handleWatcherError ran its
  teardown against nulls, then the in-flight poll installed itself, and the
  fallback guard mistook it for coverage and discarded the fallback. Primary
  metadata was then never observed again. The guard no longer treats status-ref
  polling as primary coverage, and startup re-checks watcher liveness after its
  awaits.

- Nothing re-read the six primary files while the watcher was nominally live.
  A lossy notification path, a dropped batch, or inotify queue overflow raises
  no error, so the error-driven fallback never fired and the inode rebind sweep
  does not detect loss. A 15-tick backstop re-stats them, turning permanent
  staleness into one tick. Measured cost is ~0.2 stats/s/repo against the 3/s
  the old poll cost.

- Reconciliation treated any late-observed entry create as a root replacement,
  so an ordinary  tore down a healthy stream ~30s later and
  opened a deaf window. It now also requires the root itself to be recreated.

- The shallow watcher recorded directory identity from a stat issued after
  binding, so a replacement in that gap pinned the dead inode's watcher to the
  new identity and the sweep would never rebind. Identity is now read first,
  which errs toward a harmless extra rebind.

Test helper: replacing the worktrees root frees several inodes at once, so
reserving one still let the recreated root reuse its own. It now verifies the
inode actually changed. Confirmed on ext4, where three holds were needed.
2026-08-25 01:22:07 -07:00

598 lines
22 KiB
TypeScript
Raw Blame History

import { EventEmitter } from 'node:events'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
const { detectShallowWatchDeliveryMock, statMock, subscribeMock, watchMock, writeFileSyncMock } =
vi.hoisted(() => ({
detectShallowWatchDeliveryMock: vi.fn(),
statMock: vi.fn(),
subscribeMock: vi.fn(),
watchMock: vi.fn(),
writeFileSyncMock: vi.fn()
}))
vi.mock('node:fs', () => ({
mkdtempSync: vi.fn(() => '/tmp/orca-watcher-canary-test'),
rmSync: vi.fn(),
watch: watchMock,
writeFileSync: writeFileSyncMock
}))
vi.mock('node:fs/promises', () => ({ stat: statMock }))
vi.mock('./shallow-watch-delivery-probe', () => ({
detectShallowWatchDelivery: detectShallowWatchDeliveryMock
}))
vi.mock('@parcel/watcher', () => ({ subscribe: subscribeMock }))
describe('parcel watcher process canary', () => {
let originalMessageListeners: ReturnType<typeof process.listeners>
let originalExitListeners: ReturnType<typeof process.listeners>
let originalDisconnectListeners: ReturnType<typeof process.listeners>
let originalSend: typeof process.send
beforeEach(() => {
vi.useFakeTimers()
vi.resetModules()
subscribeMock.mockReset()
statMock.mockReset()
watchMock.mockReset()
writeFileSyncMock.mockReset()
originalMessageListeners = process.listeners('message')
originalExitListeners = process.listeners('exit')
originalDisconnectListeners = process.listeners('disconnect')
originalSend = process.send
})
afterEach(() => {
for (const listener of process.listeners('message')) {
if (!originalMessageListeners.includes(listener)) {
process.off('message', listener)
}
}
for (const listener of process.listeners('exit')) {
if (!originalExitListeners.includes(listener)) {
process.off('exit', listener)
}
}
for (const listener of process.listeners('disconnect')) {
if (!originalDisconnectListeners.includes(listener)) {
process.off('disconnect', listener)
}
}
process.send = originalSend
vi.useRealTimers()
vi.restoreAllMocks()
})
it('fails the shallow subscribe when the host never delivers fs.watch events', async () => {
// A real macOS 26.3.1 machine registers the watch and stays mute forever.
// Reporting failure is what lets the caller fall back to polling instead of
// showing indefinitely stale Git metadata.
detectShallowWatchDeliveryMock.mockResolvedValue(false)
watchMock.mockImplementation(() => {
const watcher = new EventEmitter() as EventEmitter & { close: () => void }
watcher.close = () => watcher.emit('close')
return watcher
})
const sendMock = vi.fn()
process.send = sendMock as typeof process.send
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', {
op: 'subscribe',
id: 7,
dir: '/repo/.git',
opts: { mode: 'shallow', include: ['HEAD'] }
})
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith(
expect.objectContaining({ op: 'subscribe-failed', id: 7 })
)
expect(watchMock).not.toHaveBeenCalledWith('/repo/.git', expect.anything(), expect.anything())
})
it('hosts shallow subscriptions without invoking the recursive native watcher', async () => {
detectShallowWatchDeliveryMock.mockResolvedValue(true)
const watcherCallbacks = new Map<
string,
(eventType: string, fileName: string | Buffer | null) => void
>()
watchMock.mockImplementation(
(
path: string,
_options: unknown,
callback: (eventType: string, fileName: string | Buffer | null) => void
) => {
watcherCallbacks.set(path, callback)
const watcher = new EventEmitter() as EventEmitter & { close: () => void }
watcher.close = () => watcher.emit('close')
return watcher
}
)
const sendMock = vi.fn()
process.send = sendMock as typeof process.send
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', {
op: 'subscribe',
id: 1,
dir: '/repo/.git',
opts: { mode: 'shallow', include: ['HEAD'] }
})
await vi.advanceTimersByTimeAsync(0)
expect(subscribeMock).toHaveBeenCalledTimes(1)
watcherCallbacks.get('/repo/.git')?.('change', 'HEAD')
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith(
{
op: 'events',
id: 1,
events: [{ type: 'update', path: '/repo/.git/HEAD' }]
},
expect.any(Function)
)
})
it('does not restart while a native subscription is still crawling', async () => {
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockReturnValueOnce(new Promise(() => undefined))
const exitSpy = vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never)
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/large-repo', opts: {} })
await vi.advanceTimersByTimeAsync(30_000)
expect(writeFileSyncMock).not.toHaveBeenCalled()
expect(exitSpy).not.toHaveBeenCalled()
})
it('invalidates an outstanding probe when another subscription starts crawling', async () => {
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockReturnValueOnce(new Promise(() => undefined))
const exitSpy = vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never)
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/repo', opts: {} })
await vi.advanceTimersByTimeAsync(0)
await vi.advanceTimersByTimeAsync(10_000)
expect(writeFileSyncMock).toHaveBeenCalledTimes(1)
process.emit('message', { op: 'subscribe', id: 2, dir: '/large-repo', opts: {} })
await vi.advanceTimersByTimeAsync(20_000)
expect(writeFileSyncMock).toHaveBeenCalledTimes(1)
expect(exitSpy).not.toHaveBeenCalled()
})
it('still restarts when unsubscribe deadlocks while another root remains live', async () => {
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockResolvedValueOnce({ unsubscribe: vi.fn(() => new Promise(() => undefined)) })
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
const exitSpy = vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never)
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/repo-a', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 2, dir: '/repo-b', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'unsubscribe', id: 1 })
await vi.advanceTimersByTimeAsync(30_000)
expect(writeFileSyncMock).toHaveBeenCalledTimes(3)
expect(exitSpy).toHaveBeenCalledWith(2)
})
it('drains active crawls before starting teardown deadlock detection', async () => {
let finishSecondCrawl:
| ((subscription: { unsubscribe: () => Promise<void> }) => void)
| undefined
const deadlockedUnsubscribe = vi.fn(() => new Promise<void>(() => undefined))
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockResolvedValueOnce({ unsubscribe: deadlockedUnsubscribe })
.mockReturnValueOnce(
new Promise((resolve) => {
finishSecondCrawl = resolve
})
)
const exitSpy = vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never)
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/repo-a', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 2, dir: '/repo-b', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'unsubscribe', id: 1 })
await vi.advanceTimersByTimeAsync(30_000)
expect(deadlockedUnsubscribe).not.toHaveBeenCalled()
expect(writeFileSyncMock).not.toHaveBeenCalled()
finishSecondCrawl?.({ unsubscribe: vi.fn().mockResolvedValue(undefined) })
await vi.advanceTimersByTimeAsync(0)
await vi.advanceTimersByTimeAsync(30_000)
expect(deadlockedUnsubscribe).toHaveBeenCalledTimes(1)
expect(exitSpy).toHaveBeenCalledWith(2)
})
it('keeps later crawls queued behind teardown deadlock detection', async () => {
const deadlockedUnsubscribe = vi.fn(() => new Promise<void>(() => undefined))
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockResolvedValueOnce({ unsubscribe: deadlockedUnsubscribe })
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
const exitSpy = vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never)
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/repo-a', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 2, dir: '/repo-b', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'unsubscribe', id: 1 })
await vi.advanceTimersByTimeAsync(0)
expect(deadlockedUnsubscribe).toHaveBeenCalledTimes(1)
process.emit('message', { op: 'subscribe', id: 3, dir: '/repo-c', opts: {} })
await vi.advanceTimersByTimeAsync(30_000)
expect(subscribeMock).toHaveBeenCalledTimes(3)
expect(exitSpy).toHaveBeenCalledWith(2)
})
it('reserves teardown order while its subscription is still crawling', async () => {
let finishCrawl: ((subscription: { unsubscribe: () => Promise<void> }) => void) | undefined
const deadlockedUnsubscribe = vi.fn(() => new Promise<void>(() => undefined))
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockReturnValueOnce(
new Promise((resolve) => {
finishCrawl = resolve
})
)
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
const exitSpy = vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never)
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/live-repo', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 2, dir: '/crawling-repo', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'unsubscribe', id: 2 })
process.emit('message', { op: 'subscribe', id: 3, dir: '/later-repo', opts: {} })
await vi.advanceTimersByTimeAsync(30_000)
expect(writeFileSyncMock).not.toHaveBeenCalled()
finishCrawl?.({ unsubscribe: deadlockedUnsubscribe })
await vi.advanceTimersByTimeAsync(0)
await vi.advanceTimersByTimeAsync(30_000)
expect(deadlockedUnsubscribe).toHaveBeenCalledTimes(1)
expect(subscribeMock).toHaveBeenCalledTimes(3)
expect(exitSpy).toHaveBeenCalledWith(2)
})
it('removes a cancelled queued crawl before it reaches Parcel', async () => {
let finishActiveCrawl:
| ((subscription: { unsubscribe: () => Promise<void> }) => void)
| undefined
subscribeMock.mockResolvedValueOnce({ unsubscribe: vi.fn() }).mockReturnValueOnce(
new Promise((resolve) => {
finishActiveCrawl = resolve
})
)
const sendMock = vi.fn()
process.send = sendMock
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/active', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 2, dir: '/queued', opts: {} })
process.emit('message', { op: 'cancel-subscribe', id: 2 })
// Why: non-restart cancel reuses async unsubscribe teardown; drain the
// exclusive lifecycle queue after the active crawl finishes.
finishActiveCrawl?.({ unsubscribe: vi.fn().mockResolvedValue(undefined) })
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 2 })
expect(subscribeMock).toHaveBeenCalledTimes(2)
expect(sendMock).not.toHaveBeenCalledWith({ op: 'subscribe-started', id: 2 })
expect(sendMock).not.toHaveBeenCalledWith({ op: 'subscribed', id: 2 })
})
it('unsubscribes a late cancel after the crawl already finished', async () => {
const unsubscribe = vi.fn().mockResolvedValue(undefined)
subscribeMock.mockResolvedValueOnce({ unsubscribe: vi.fn() }).mockResolvedValueOnce({
unsubscribe
})
const sendMock = vi.fn()
process.send = sendMock
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/finished', opts: {} })
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith({ op: 'subscribed', id: 1 })
process.emit('message', { op: 'cancel-subscribe', id: 1 })
await vi.advanceTimersByTimeAsync(0)
expect(unsubscribe).toHaveBeenCalledTimes(1)
expect(sendMock).toHaveBeenCalledWith({ op: 'unsubscribed', id: 1 })
expect(sendMock).not.toHaveBeenCalledWith({ op: 'cancel-requires-restart', id: 1 })
})
it('reports native unsubscribe rejection without acknowledging handle release', async () => {
const unsubscribeError = new Error('native handle still active')
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockResolvedValueOnce({ unsubscribe: vi.fn().mockRejectedValue(unsubscribeError) })
const sendMock = vi.fn()
process.send = sendMock
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/repo', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'unsubscribe', id: 1 })
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith({
op: 'unsubscribe-failed',
id: 1,
message: 'native handle still active'
})
expect(sendMock).not.toHaveBeenCalledWith({ op: 'unsubscribed', id: 1 })
})
it('asks the host to restart when an active crawl is cancelled', async () => {
let finishCrawl: ((subscription: { unsubscribe: () => Promise<void> }) => void) | undefined
subscribeMock.mockResolvedValueOnce({ unsubscribe: vi.fn() }).mockReturnValueOnce(
new Promise((resolve) => {
finishCrawl = resolve
})
)
const sendMock = vi.fn()
process.send = sendMock
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/active', opts: {} })
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'cancel-subscribe', id: 1 })
expect(sendMock).toHaveBeenCalledWith({ op: 'cancel-requires-restart', id: 1 })
finishCrawl?.({ unsubscribe: vi.fn().mockResolvedValue(undefined) })
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).not.toHaveBeenCalledWith({ op: 'subscribed', id: 1 })
})
it('still restarts after consecutive missed events once every subscription is live', async () => {
subscribeMock.mockResolvedValue({ unsubscribe: vi.fn() })
const exitSpy = vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never)
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/repo', opts: {} })
await vi.advanceTimersByTimeAsync(0)
await vi.advanceTimersByTimeAsync(30_000)
expect(writeFileSyncMock).toHaveBeenCalledTimes(3)
expect(exitSpy).toHaveBeenCalledWith(2)
})
it('collapses and enriches runtime batches before they cross child IPC', async () => {
let callback:
| ((err: Error | null, events: { type: string; path: string }[]) => void)
| undefined
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockImplementationOnce(async (_dir, nextCallback) => {
callback = nextCallback
return { unsubscribe: vi.fn() }
})
statMock.mockImplementation(async (eventPath: string) => ({
isDirectory: () => eventPath.endsWith('/dir')
}))
const sendMock = vi.fn()
process.send = sendMock
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', {
op: 'subscribe',
id: 1,
dir: '/repo',
opts: {},
delivery: { includeDirectoryMetadata: true, maxEventsPerBatch: 2 }
})
await vi.advanceTimersByTimeAsync(0)
callback?.(null, [
{ type: 'update', path: '/repo/file.txt' },
{ type: 'create', path: '/repo/dir' }
])
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith(
{
op: 'events',
id: 1,
events: [
{ type: 'update', path: '/repo/file.txt', isDirectory: false },
{ type: 'create', path: '/repo/dir', isDirectory: true }
]
},
expect.any(Function)
)
sendMock.mockClear()
callback?.(null, [
{ type: 'update', path: '/repo/a' },
{ type: 'update', path: '/repo/b' },
{ type: 'update', path: '/repo/c' }
])
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith({ op: 'overflow', id: 1 }, expect.any(Function))
expect(statMock).toHaveBeenCalledTimes(2)
})
it('reports an FSEvents overflow and keeps delivering later events', async () => {
let callback:
| ((err: Error | null, events: { type: string; path: string }[]) => void)
| undefined
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockImplementationOnce(async (_dir, nextCallback) => {
callback = nextCallback
return { unsubscribe: vi.fn() }
})
const sendMock = vi.fn()
process.send = sendMock
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'subscribe', id: 1, dir: '/repo', opts: {} })
await vi.advanceTimersByTimeAsync(0)
callback?.(
new Error('Events were dropped by the FSEvents client. File system must be re-scanned.'),
[]
)
callback?.(null, [{ type: 'update', path: '/repo/after-overflow.txt' }])
await vi.advanceTimersByTimeAsync(0)
expect(sendMock).toHaveBeenCalledWith({
op: 'watch-error',
id: 1,
message: 'Events were dropped by the FSEvents client. File system must be re-scanned.'
})
expect(sendMock).toHaveBeenCalledWith(
{
op: 'events',
id: 1,
events: [{ type: 'update', path: '/repo/after-overflow.txt' }]
},
expect.any(Function)
)
})
it('bounds pending event batches while child IPC reports backpressure', async () => {
let callback:
| ((err: Error | null, events: { type: string; path: string }[]) => void)
| undefined
let releaseFirstEvent: (() => void) | undefined
const sentEventPaths: string[] = []
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockImplementationOnce(async (_dir, nextCallback) => {
callback = nextCallback
return { unsubscribe: vi.fn() }
})
process.send = vi.fn((message, onSent) => {
if (message.op !== 'events') {
return true
}
sentEventPaths.push(message.events[0]?.path ?? '')
if (!releaseFirstEvent) {
releaseFirstEvent = onSent
return false
}
return true
}) as typeof process.send
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', {
op: 'subscribe',
id: 1,
dir: '/repo',
opts: {},
delivery: { includeDirectoryMetadata: true, maxEventsPerBatch: 2 }
})
await vi.advanceTimersByTimeAsync(0)
callback?.(null, [{ type: 'update', path: '/repo/first.txt' }])
callback?.(null, [{ type: 'update', path: '/repo/second.txt' }])
callback?.(null, [{ type: 'update', path: '/repo/third.txt' }])
callback?.(null, [{ type: 'update', path: '/repo/fourth.txt' }])
await vi.advanceTimersByTimeAsync(0)
expect(sentEventPaths).toEqual(['/repo/first.txt'])
releaseFirstEvent?.()
await vi.advanceTimersByTimeAsync(0)
expect(sentEventPaths).toEqual(['/repo/first.txt'])
expect(process.send).toHaveBeenCalledWith({ op: 'overflow', id: 1 }, expect.any(Function))
expect(statMock).toHaveBeenCalledTimes(1)
})
it('drops queued event work when the subscription is removed', async () => {
let callback:
| ((err: Error | null, events: { type: string; path: string }[]) => void)
| undefined
let releaseFirstEvent: (() => void) | undefined
const unsubscribe = vi.fn().mockResolvedValue(undefined)
subscribeMock
.mockResolvedValueOnce({ unsubscribe: vi.fn() })
.mockImplementationOnce(async (_dir, nextCallback) => {
callback = nextCallback
return { unsubscribe }
})
process.send = vi.fn((message, onSent) => {
if (message.op === 'events' && !releaseFirstEvent) {
releaseFirstEvent = onSent
return false
}
return true
}) as typeof process.send
await import('./parcel-watcher-process-entry')
await vi.advanceTimersByTimeAsync(0)
process.emit('message', {
op: 'subscribe',
id: 1,
dir: '/repo',
opts: {},
delivery: { includeDirectoryMetadata: true, maxEventsPerBatch: 2 }
})
await vi.advanceTimersByTimeAsync(0)
callback?.(null, [{ type: 'update', path: '/repo/first.txt' }])
callback?.(null, [{ type: 'update', path: '/repo/queued.txt' }])
await vi.advanceTimersByTimeAsync(0)
process.emit('message', { op: 'unsubscribe', id: 1 })
await vi.advanceTimersByTimeAsync(0)
releaseFirstEvent?.()
await vi.advanceTimersByTimeAsync(0)
expect(unsubscribe).toHaveBeenCalledTimes(1)
expect(statMock).toHaveBeenCalledTimes(1)
expect(process.send).not.toHaveBeenCalledWith(
expect.objectContaining({
op: 'events',
events: [expect.objectContaining({ path: '/repo/queued.txt' })]
}),
expect.any(Function)
)
})
})
Reference in New Issue View Git Blame Copy Permalink