Files
orca/src/relay/fs-handler.test.ts
T
Neil 7104056984 fix(watcher): route relay watch-root capacity refusals off the fast ladder (#17950)
* fix(ssh): stop two unrecoverable relay refusal loops

A relay refusal that is a pure function of state the client cannot change was
being retried forever, on two different paths.

- pty.openClient: a superseded owner proof is refuted evidence, not a transient
  fault. The client kept re-presenting the identical proof, so every reconnect
  reproduced the same refusal until the relay was redeployed (#12895, #12931).
  It is now dropped exactly as a stale lease already is, and the claim re-asked
  without it.
- fs.watch: the relay's watch-root capacity refusal was classified 'unavailable'
  and retried at 1 Hz per root for 60s, re-armed indefinitely. A folder
  workspace with more repos than the cap turns that into a permanent install
  storm scaled by the excess root count (#11196). It is now its own 'capacity'
  result that goes straight to the existing dormant backoff, mirroring what the
  local watcher path already does.

* fix(watcher): route relay watch-root capacity refusals off the fast ladder

A full watch-root cap is a decision, not a fault, so a 1 Hz reinstall per refused
root only bills the relay the load that keeps the cap busy (#11196). Capacity
refusals now go straight to the dormant backoff.

The relay side no longer refuses on a slot it is about to hand back: an over-cap
caused by roots still unsubscribing waits once on the teardowns settling — the
release event, mirroring WatcherSupervisorCapacityWait — before it answers. A
parked waiter is excluded from the accounting so it cannot take a slot from the
root already reclaiming one.

Drops the SSH owner-recovery half of this branch. Its premise — that a -32043
SUPERSEDED refusal is permanent — is false: the refusal fires only while the
incumbent is 'active', and assertPtyConsumerOwnerRecovery explicitly admits the
identical lower-generation proof once the incumbent flips to 'disconnected'
(relay-pty-consumer-owner-displacement.test.ts proves it). The remedy could not
work either: the proofless re-ask routes into refuseHeldPtyConsumerOwner, which
is declared `: never` and, with sameClient true by construction, always throws.
It would have traded one refusal loop for another, minus the checkpoints and
minus the proof that resumes the claim once the relay reaps the incumbent.

* fix(i18n): restore the activity-options key the rebase dropped

* fix(i18n): union en.json with main so the rebase cannot drop keys
2026-09-02 15:14:21 -07:00

821 lines
28 KiB
TypeScript

import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'
import { FsHandler } from './fs-handler'
import { MAX_TEXT_FILE_SIZE } from './fs-handler-utils'
import { RelayContext } from './context'
import type { RelayDispatcher } from './dispatcher'
import * as fs from 'node:fs/promises'
import * as path from 'node:path'
import { mkdtempSync, writeFileSync, mkdirSync, symlinkSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { subscribeWithInProcessWatcher } from '../main/ipc/parcel-watcher-in-process-fallback'
import { createMockDispatcher } from './relay-fs-test-dispatcher'
const { mockSubscribe } = vi.hoisted(() => ({
mockSubscribe: vi.fn()
}))
vi.mock('@parcel/watcher', () => ({
subscribe: mockSubscribe
}))
function statIdentity(stats: {
dev?: number
ino?: number
nlink?: number
size?: number
mtimeMs?: number
}) {
return `${stats.dev}:${stats.ino}:${stats.nlink ?? 'unknown'}:${stats.size}:${stats.mtimeMs}`
}
describe('FsHandler', () => {
let dispatcher: ReturnType<typeof createMockDispatcher>
let handler: FsHandler
let tmpDir: string
beforeEach(() => {
mockSubscribe.mockReset()
mockSubscribe.mockResolvedValue({ unsubscribe: vi.fn() })
tmpDir = mkdtempSync(path.join(tmpdir(), 'relay-fs-'))
dispatcher = createMockDispatcher()
const ctx = new RelayContext()
handler = new FsHandler(dispatcher as unknown as RelayDispatcher, ctx, {
dispose: vi.fn(),
forgetRoot: vi.fn(),
subscribe: subscribeWithInProcessWatcher
})
})
afterEach(async () => {
handler.dispose()
await fs.rm(tmpDir, { recursive: true, force: true })
})
it('registers all expected handlers', () => {
const methods = Array.from(dispatcher._requestHandlers.keys())
expect(methods).toContain('fs.readDir')
expect(methods).toContain('fs.readFile')
expect(methods).toContain('fs.tempDir')
expect(methods).toContain('fs.writeFile')
expect(methods).toContain('fs.stat')
expect(methods).toContain('fs.deletePath')
expect(methods).toContain('fs.createFile')
expect(methods).toContain('fs.createDir')
expect(methods).toContain('fs.createDirNoClobber')
expect(methods).toContain('fs.rename')
expect(methods).toContain('fs.renameNoClobber')
expect(methods).toContain('fs.copy')
expect(methods).toContain('fs.realpath')
expect(methods).toContain('fs.search')
expect(methods).toContain('fs.listFiles')
expect(methods).toContain('fs.workspaceSpaceScan')
expect(methods).toContain('fs.watch')
expect(methods).toContain('fs.unwatchAndWait')
const notifMethods = Array.from(dispatcher._notificationHandlers.keys())
expect(notifMethods).toContain('fs.unwatch')
})
it('tempDir returns the relay host temp directory', async () => {
await expect(dispatcher.callRequest('fs.tempDir')).resolves.toBe(tmpdir())
})
it('readDir returns entries directories-first in natural name order', async () => {
mkdirSync(path.join(tmpDir, 'subdir'))
for (const name of ['file.txt', '100 - b.txt', '99 - a.txt', '9 - c.txt']) {
writeFileSync(path.join(tmpDir, name), 'x')
}
const result = (await dispatcher.callRequest('fs.readDir', { dirPath: tmpDir })) as {
name: string
isDirectory: boolean
}[]
expect(result[0]).toMatchObject({ name: 'subdir', isDirectory: true })
expect(result.slice(1).map((e) => e.name)).toEqual([
'9 - c.txt',
'99 - a.txt',
'100 - b.txt',
'file.txt'
])
})
it('readDir reports symlinked directories as directories', async () => {
const targetDir = path.join(tmpDir, 'external-models')
const linkPath = path.join(tmpDir, 'Model')
mkdirSync(targetDir)
symlinkSync(targetDir, linkPath, process.platform === 'win32' ? 'junction' : 'dir')
const result = (await dispatcher.callRequest('fs.readDir', { dirPath: tmpDir })) as {
name: string
isDirectory: boolean
isSymlink: boolean
}[]
expect(result.find((e) => e.name === 'Model')).toEqual({
name: 'Model',
isDirectory: true,
isSymlink: true
})
})
it('readFile returns text content for text files', async () => {
const filePath = path.join(tmpDir, 'test.txt')
writeFileSync(filePath, 'hello world')
const result = (await dispatcher.callRequest('fs.readFile', { filePath })) as {
content: string
isBinary: boolean
}
expect(result.content).toBe('hello world')
expect(result.isBinary).toBe(false)
})
it('readFile returns text files larger than the old 5MB guard', async () => {
const filePath = path.join(tmpDir, 'large.json')
const content = 'a'.repeat(6 * 1024 * 1024)
writeFileSync(filePath, content)
const result = (await dispatcher.callRequest('fs.readFile', { filePath })) as {
content: string
isBinary: boolean
}
expect(result.content).toBe(content)
expect(result.isBinary).toBe(false)
})
it('readFile returns binary marker for large unknown binary files', async () => {
const filePath = path.join(tmpDir, 'archive.bin')
const content = Buffer.alloc(6 * 1024 * 1024, 0x61)
content[0] = 0x00
writeFileSync(filePath, content)
const result = (await dispatcher.callRequest('fs.readFile', { filePath })) as {
content: string
isBinary: boolean
}
expect(result.content).toBe('')
expect(result.isBinary).toBe(true)
})
it('readFile returns base64 for image files', async () => {
const filePath = path.join(tmpDir, 'test.png')
writeFileSync(filePath, Buffer.from([0x89, 0x50, 0x4e, 0x47]))
const result = (await dispatcher.callRequest('fs.readFile', { filePath })) as {
content: string
isBinary: boolean
isImage: boolean
mimeType: string
}
expect(result.isBinary).toBe(true)
expect(result.isImage).toBe(true)
expect(result.mimeType).toBe('image/png')
expect(result.content).toBeTruthy()
})
it('readFile throws for files exceeding size limit', async () => {
const filePath = path.join(tmpDir, 'huge.txt')
writeFileSync(filePath, Buffer.alloc(11 * 1024 * 1024, 'a'))
await expect(dispatcher.callRequest('fs.readFile', { filePath })).rejects.toThrow(
'File too large'
)
})
it('writeFile creates/overwrites file content', async () => {
const filePath = path.join(tmpDir, 'write-test.txt')
await dispatcher.callRequest('fs.writeFile', { filePath, content: 'new content' })
const content = await fs.readFile(filePath, 'utf-8')
expect(content).toBe('new content')
})
it('readTerminalArtifact reads through a verified artifact handle', async () => {
const filePath = path.join(tmpDir, 'artifact-read.json')
writeFileSync(filePath, '{"ok":true}')
const stats = await fs.stat(filePath)
const result = (await dispatcher.callRequest('fs.readTerminalArtifact', {
filePath,
expectedRealPath: await fs.realpath(filePath),
expectedStatIdentity: statIdentity(stats),
maxBytes: 512 * 1024
})) as { content: string; isBinary: boolean }
expect(result).toEqual({ content: '{"ok":true}', isBinary: false })
})
it('readTerminalArtifact treats SVG artifacts as editable text', async () => {
const filePath = path.join(tmpDir, 'artifact.svg')
writeFileSync(filePath, '<svg><text>ok</text></svg>')
const stats = await fs.stat(filePath)
const result = (await dispatcher.callRequest('fs.readTerminalArtifact', {
filePath,
expectedRealPath: await fs.realpath(filePath),
expectedStatIdentity: statIdentity(stats),
maxBytes: 512 * 1024
})) as { content: string; isBinary: boolean; isImage?: boolean }
expect(result).toEqual({ content: '<svg><text>ok</text></svg>', isBinary: false })
})
it('readTerminalArtifact rejects content beyond the requested byte limit', async () => {
const filePath = path.join(tmpDir, 'artifact-read-too-large.txt')
writeFileSync(filePath, 'abcdef')
await expect(
dispatcher.callRequest('fs.readTerminalArtifact', {
filePath,
expectedRealPath: await fs.realpath(filePath),
maxBytes: 5
})
).rejects.toThrow('file_too_large')
})
it('writeTerminalArtifact writes through a verified artifact handle', async () => {
const filePath = path.join(tmpDir, 'artifact-write.json')
writeFileSync(filePath, '{"ok":true}')
const stats = await fs.stat(filePath)
const result = (await dispatcher.callRequest('fs.writeTerminalArtifact', {
filePath,
content: '{"ok":false}',
expectedRealPath: await fs.realpath(filePath),
expectedStatIdentity: statIdentity(stats),
maxBytes: 512 * 1024
})) as { stat: { type: string; size: number } }
await expect(fs.readFile(filePath, 'utf-8')).resolves.toBe('{"ok":false}')
expect(result.stat).toMatchObject({ type: 'file', size: 12 })
})
it.skipIf(process.platform === 'win32')(
'writeTerminalArtifact preserves executable mode across the atomic rename',
async () => {
const filePath = path.join(tmpDir, 'artifact-executable.sh')
writeFileSync(filePath, '#!/bin/sh\necho ok\n')
await fs.chmod(filePath, 0o755)
const stats = await fs.stat(filePath)
await dispatcher.callRequest('fs.writeTerminalArtifact', {
filePath,
content: '#!/bin/sh\necho changed\n',
expectedRealPath: await fs.realpath(filePath),
expectedStatIdentity: statIdentity(stats),
maxBytes: 512 * 1024
})
expect((await fs.stat(filePath)).mode & 0o777).toBe(0o755)
}
)
it('writeTerminalArtifact rejects oversized existing content before writing', async () => {
const filePath = path.join(tmpDir, 'artifact-write-too-large.txt')
writeFileSync(filePath, 'abcdef')
await expect(
dispatcher.callRequest('fs.writeTerminalArtifact', {
filePath,
content: 'ok',
expectedRealPath: await fs.realpath(filePath),
maxBytes: 5
})
).rejects.toThrow('file_too_large')
await expect(fs.readFile(filePath, 'utf-8')).resolves.toBe('abcdef')
})
it('writeTerminalArtifact clamps client-supplied maxBytes to the text-file cap', async () => {
const filePath = path.join(tmpDir, 'artifact-write-clamp.txt')
writeFileSync(filePath, 'abcdef')
await expect(
dispatcher.callRequest('fs.writeTerminalArtifact', {
filePath,
content: 'a'.repeat(MAX_TEXT_FILE_SIZE + 1),
expectedRealPath: await fs.realpath(filePath),
maxBytes: Number.MAX_SAFE_INTEGER
})
).rejects.toThrow('file_too_large')
await expect(fs.readFile(filePath, 'utf-8')).resolves.toBe('abcdef')
})
it('writeTerminalArtifact rejects a retargeted symlink before writing outside temp', async () => {
const filePath = path.join(tmpDir, 'artifact-link.json')
const outsidePath = path.join(tmpDir, 'outside.json')
writeFileSync(filePath, '{"ok":true}')
writeFileSync(outsidePath, '{"secret":true}')
const stats = await fs.stat(filePath)
const expectedRealPath = await fs.realpath(filePath)
await fs.rm(filePath)
symlinkSync(outsidePath, filePath)
await expect(
dispatcher.callRequest('fs.writeTerminalArtifact', {
filePath,
content: '{"ok":false}',
expectedRealPath,
expectedStatIdentity: statIdentity(stats),
maxBytes: 512 * 1024
})
).rejects.toThrow('terminal_file_grant_stale')
await expect(fs.readFile(outsidePath, 'utf-8')).resolves.toBe('{"secret":true}')
})
it('writeTerminalArtifact rejects hard-linked files before writing', async () => {
const outsidePath = path.join(tmpDir, 'outside-hardlink.json')
const filePath = path.join(tmpDir, 'artifact-hardlink.json')
writeFileSync(outsidePath, '{"secret":true}')
await fs.link(outsidePath, filePath)
const stats = await fs.stat(filePath)
await expect(
dispatcher.callRequest('fs.writeTerminalArtifact', {
filePath,
content: '{"ok":false}',
expectedRealPath: await fs.realpath(filePath),
expectedStatIdentity: statIdentity(stats),
maxBytes: 512 * 1024
})
).rejects.toThrow('terminal_file_grant_stale')
await expect(fs.readFile(outsidePath, 'utf-8')).resolves.toBe('{"secret":true}')
})
it('stat returns file metadata', async () => {
const filePath = path.join(tmpDir, 'stat-test.txt')
writeFileSync(filePath, 'test')
const result = (await dispatcher.callRequest('fs.stat', { filePath })) as {
size: number
type: string
mtime: number
}
expect(result.type).toBe('file')
expect(result.size).toBe(4)
expect(typeof result.mtime).toBe('number')
})
it('stat returns directory type for directories', async () => {
const result = (await dispatcher.callRequest('fs.stat', { filePath: tmpDir })) as {
type: string
}
expect(result.type).toBe('directory')
})
it('stat returns directory type for symlinked directories', async () => {
const targetDir = path.join(tmpDir, 'external-models')
const linkPath = path.join(tmpDir, 'Model')
mkdirSync(targetDir)
symlinkSync(targetDir, linkPath, process.platform === 'win32' ? 'junction' : 'dir')
const result = (await dispatcher.callRequest('fs.stat', { filePath: linkPath })) as {
type: string
}
expect(result.type).toBe('directory')
})
it('lstat returns symlink type without following links', async () => {
const targetFile = path.join(tmpDir, 'target.txt')
const linkPath = path.join(tmpDir, 'link.txt')
writeFileSync(targetFile, 'target')
symlinkSync(targetFile, linkPath)
const result = (await dispatcher.callRequest('fs.lstat', { filePath: linkPath })) as {
type: string
}
expect(result.type).toBe('symlink')
})
it('workspaceSpaceScan returns bounded top-level size details', async () => {
mkdirSync(path.join(tmpDir, 'node_modules'))
writeFileSync(path.join(tmpDir, 'node_modules', 'pkg.js'), Buffer.alloc(512))
writeFileSync(path.join(tmpDir, 'file.log'), Buffer.alloc(128))
const result = (await dispatcher.callRequest(
'fs.workspaceSpaceScan',
{ rootPath: tmpDir },
{ isStale: () => false }
)) as {
sizeBytes: number
topLevelItems: { name: string; sizeBytes: number }[]
}
expect(result.sizeBytes).toBeGreaterThanOrEqual(640)
expect(result.topLevelItems.map((item) => item.name)).toContain('node_modules')
expect(result.topLevelItems.map((item) => item.name)).toContain('file.log')
})
it('deletePath removes files', async () => {
const filePath = path.join(tmpDir, 'to-delete.txt')
writeFileSync(filePath, 'bye')
await dispatcher.callRequest('fs.deletePath', { targetPath: filePath })
await expect(fs.access(filePath)).rejects.toThrow()
})
it('holds the relay watcher fence through recursive directory deletion', async () => {
const directoryPath = path.join(tmpDir, 'watched-orphan')
mkdirSync(directoryPath)
const unsubscribe = vi.fn()
mockSubscribe.mockResolvedValue({ unsubscribe })
await dispatcher.callRequest(
'fs.watch',
{ rootPath: directoryPath, watchId: 77 },
{ clientId: 3, isStale: () => false }
)
await dispatcher.callRequest('fs.deletePath', { targetPath: directoryPath, recursive: true })
expect(unsubscribe).toHaveBeenCalledTimes(1)
expect(dispatcher.notifyClient).toHaveBeenCalledWith(3, 'fs.watchFailed', {
rootPath: directoryPath,
watchId: 77,
message: 'Remote worktree is being removed'
})
await expect(fs.access(directoryPath)).rejects.toThrow()
})
it('createFile creates an empty file with parent dirs', async () => {
const filePath = path.join(tmpDir, 'deep', 'nested', 'file.txt')
await dispatcher.callRequest('fs.createFile', { filePath })
const content = await fs.readFile(filePath, 'utf-8')
expect(content).toBe('')
})
it('createDir creates directories recursively', async () => {
const dirPath = path.join(tmpDir, 'a', 'b', 'c')
await dispatcher.callRequest('fs.createDir', { dirPath })
const stats = await fs.stat(dirPath)
expect(stats.isDirectory()).toBe(true)
})
it('createDirNoClobber fails when the directory already exists', async () => {
const dirPath = path.join(tmpDir, 'existing')
mkdirSync(dirPath)
await expect(dispatcher.callRequest('fs.createDirNoClobber', { dirPath })).rejects.toThrow()
})
it('rename moves files', async () => {
const oldPath = path.join(tmpDir, 'old.txt')
const newPath = path.join(tmpDir, 'new.txt')
writeFileSync(oldPath, 'content')
await dispatcher.callRequest('fs.rename', { oldPath, newPath })
await expect(fs.access(oldPath)).rejects.toThrow()
const content = await fs.readFile(newPath, 'utf-8')
expect(content).toBe('content')
})
it('rename preserves raw fs.rename overwrite semantics', async () => {
const oldPath = path.join(tmpDir, 'old.txt')
const newPath = path.join(tmpDir, 'existing.txt')
writeFileSync(oldPath, 'new')
writeFileSync(newPath, 'keep')
await dispatcher.callRequest('fs.rename', { oldPath, newPath })
expect(await fs.readFile(newPath, 'utf-8')).toBe('new')
await expect(fs.access(oldPath)).rejects.toThrow()
})
it('renameNoClobber moves files when destination is available', async () => {
const oldPath = path.join(tmpDir, 'old.txt')
const newPath = path.join(tmpDir, 'new.txt')
writeFileSync(oldPath, 'content')
await dispatcher.callRequest('fs.renameNoClobber', { oldPath, newPath })
await expect(fs.access(oldPath)).rejects.toThrow()
expect(await fs.readFile(newPath, 'utf-8')).toBe('content')
})
it('renameNoClobber does not overwrite an existing destination', async () => {
const oldPath = path.join(tmpDir, 'old.txt')
const newPath = path.join(tmpDir, 'existing.txt')
writeFileSync(oldPath, 'new')
writeFileSync(newPath, 'keep')
await expect(
dispatcher.callRequest('fs.renameNoClobber', { oldPath, newPath })
).rejects.toThrow()
expect(await fs.readFile(newPath, 'utf-8')).toBe('keep')
expect(await fs.readFile(oldPath, 'utf-8')).toBe('new')
})
it('copy duplicates files', async () => {
const src = path.join(tmpDir, 'src.txt')
const dst = path.join(tmpDir, 'dst.txt')
writeFileSync(src, 'original')
await dispatcher.callRequest('fs.copy', { source: src, destination: dst })
const content = await fs.readFile(dst, 'utf-8')
expect(content).toBe('original')
})
it('copy does not overwrite an existing destination', async () => {
const src = path.join(tmpDir, 'src.txt')
const dst = path.join(tmpDir, 'dst.txt')
writeFileSync(src, 'original')
writeFileSync(dst, 'existing')
await expect(
dispatcher.callRequest('fs.copy', { source: src, destination: dst })
).rejects.toThrow('EEXIST')
const content = await fs.readFile(dst, 'utf-8')
expect(content).toBe('existing')
})
it('realpath resolves symlinks', async () => {
const realFile = path.join(tmpDir, 'real.txt')
const linkPath = path.join(tmpDir, 'link.txt')
writeFileSync(realFile, 'real')
symlinkSync(realFile, linkPath)
const result = (await dispatcher.callRequest('fs.realpath', { filePath: linkPath })) as string
// On macOS, /var is a symlink to /private/var, so resolve both to compare
expect(result).toBe(await fs.realpath(realFile))
})
it('does not let stale pending watch remove newer replacement watch', async () => {
const firstUnsubscribe = vi.fn()
const secondUnsubscribe = vi.fn()
let resolveFirst!: () => void
mockSubscribe
.mockReturnValueOnce(
new Promise((resolve) => {
resolveFirst = () => resolve({ unsubscribe: firstUnsubscribe })
})
)
.mockResolvedValueOnce({ unsubscribe: secondUnsubscribe })
const firstWatch = dispatcher.callRequest(
'fs.watch',
{ rootPath: tmpDir },
{ isStale: () => true }
)
while (mockSubscribe.mock.calls.length === 0) {
await Promise.resolve()
}
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }, { isStale: () => false })
resolveFirst()
await firstWatch
expect(firstUnsubscribe).toHaveBeenCalled()
expect(secondUnsubscribe).not.toHaveBeenCalled()
dispatcher.callNotification('fs.unwatch', { rootPath: tmpDir })
expect(secondUnsubscribe).toHaveBeenCalled()
})
it('unsubscribes an active stale watch before replacing it', async () => {
const firstUnsubscribe = vi.fn()
const secondUnsubscribe = vi.fn()
mockSubscribe
.mockResolvedValueOnce({ unsubscribe: firstUnsubscribe })
.mockResolvedValueOnce({ unsubscribe: secondUnsubscribe })
let stale = false
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }, { isStale: () => stale })
stale = true
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }, { isStale: () => false })
expect(firstUnsubscribe).toHaveBeenCalled()
expect(secondUnsubscribe).not.toHaveBeenCalled()
dispatcher.callNotification('fs.unwatch', { rootPath: tmpDir })
expect(secondUnsubscribe).toHaveBeenCalled()
})
it('replaces a stale watch for the same root before enforcing the watch cap', async () => {
const firstUnsubscribe = vi.fn()
const replacementUnsubscribe = vi.fn()
mockSubscribe
.mockResolvedValueOnce({ unsubscribe: firstUnsubscribe })
.mockResolvedValueOnce({ unsubscribe: replacementUnsubscribe })
let stale = false
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }, { isStale: () => stale })
for (let index = 0; index < 19; index++) {
await dispatcher.callRequest('fs.watch', {
rootPath: path.join(tmpDir, `watched-${index}`)
})
}
stale = true
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }, { isStale: () => false })
expect(firstUnsubscribe).toHaveBeenCalled()
expect(replacementUnsubscribe).not.toHaveBeenCalled()
})
it('removes stale watches for any root before enforcing the watch cap', async () => {
const staleUnsubscribe = vi.fn()
mockSubscribe
.mockResolvedValueOnce({ unsubscribe: staleUnsubscribe })
.mockResolvedValue({ unsubscribe: vi.fn() })
let stale = false
await dispatcher.callRequest(
'fs.watch',
{ rootPath: path.join(tmpDir, 'stale-root') },
{
isStale: () => stale
}
)
for (let index = 0; index < 19; index += 1) {
await dispatcher.callRequest('fs.watch', {
rootPath: path.join(tmpDir, `watched-${index}`)
})
}
stale = true
await expect(
dispatcher.callRequest('fs.watch', { rootPath: path.join(tmpDir, 'new-root') })
).resolves.toBeUndefined()
expect(staleUnsubscribe).toHaveBeenCalledTimes(1)
})
it('keeps a shared watch alive until every client unwatches it', async () => {
const unsubscribe = vi.fn()
mockSubscribe.mockResolvedValue({ unsubscribe })
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }, { isStale: () => false })
await dispatcher.callRequest(
'fs.watch',
{ rootPath: tmpDir },
{
clientId: 2,
isStale: () => false
}
)
dispatcher.callNotification(
'fs.unwatch',
{ rootPath: tmpDir },
{
clientId: 1,
isStale: () => false
}
)
expect(unsubscribe).not.toHaveBeenCalled()
dispatcher.callNotification(
'fs.unwatch',
{ rootPath: tmpDir },
{
clientId: 2,
isStale: () => false
}
)
expect(unsubscribe).toHaveBeenCalledTimes(1)
})
it('settles acknowledged unwatch only after native unsubscribe completes', async () => {
let resolveUnsubscribe: () => void = () => {}
const unsubscribe = vi.fn(
() =>
new Promise<void>((resolve) => {
resolveUnsubscribe = resolve
})
)
mockSubscribe.mockResolvedValue({ unsubscribe })
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir })
let settled = false
const unwatch = dispatcher.callRequest('fs.unwatchAndWait', { rootPath: tmpDir }).then(() => {
settled = true
})
await vi.waitFor(() => expect(unsubscribe).toHaveBeenCalledTimes(1))
expect(settled).toBe(false)
resolveUnsubscribe()
await unwatch
expect(settled).toBe(true)
})
it('waits for in-flight native setup before acknowledging teardown', async () => {
handler.dispose()
let resolveSubscribe: (value: { unsubscribe: () => Promise<void> }) => void = () => {}
const unsubscribe = vi.fn(async () => undefined)
const subscribe = vi.fn(
() =>
new Promise<{ unsubscribe: () => Promise<void> }>((resolve) => {
resolveSubscribe = resolve
})
)
handler = new FsHandler(dispatcher as unknown as RelayDispatcher, new RelayContext(), {
dispose: vi.fn(),
forgetRoot: vi.fn(),
subscribe
})
const watch = dispatcher.callRequest('fs.watch', { rootPath: tmpDir })
let unwatchSettled = false
const unwatch = dispatcher.callRequest('fs.unwatchAndWait', { rootPath: tmpDir }).then(() => {
unwatchSettled = true
})
await vi.waitFor(() => expect(subscribe).toHaveBeenCalledTimes(1))
expect(unwatchSettled).toBe(false)
resolveSubscribe({ unsubscribe })
await Promise.all([watch, unwatch])
expect(unsubscribe).toHaveBeenCalledTimes(1)
})
it('joins a physical unsubscribe already started by the notification path', async () => {
let resolveUnsubscribe: () => void = () => {}
const unsubscribe = vi.fn(
() =>
new Promise<void>((resolve) => {
resolveUnsubscribe = resolve
})
)
mockSubscribe.mockResolvedValue({ unsubscribe })
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir })
dispatcher.callNotification('fs.unwatch', { rootPath: tmpDir })
let settled = false
const joined = dispatcher.callRequest('fs.unwatchAndWait', { rootPath: tmpDir }).then(() => {
settled = true
})
await Promise.resolve()
expect(settled).toBe(false)
resolveUnsubscribe()
await joined
})
it('retains a failed native unsubscribe slot until acknowledged retry succeeds', async () => {
const unsubscribe = vi
.fn()
.mockRejectedValueOnce(new Error('native handle still active'))
.mockResolvedValueOnce(undefined)
mockSubscribe.mockResolvedValue({ unsubscribe })
await dispatcher.callRequest('fs.watch', { rootPath: tmpDir })
await expect(dispatcher.callRequest('fs.unwatchAndWait', { rootPath: tmpDir })).rejects.toThrow(
'native handle still active'
)
await expect(
dispatcher.callRequest('fs.unwatchAndWait', { rootPath: tmpDir })
).resolves.toBeUndefined()
expect(unsubscribe).toHaveBeenCalledTimes(2)
await expect(dispatcher.callRequest('fs.watch', { rootPath: tmpDir })).resolves.toBeUndefined()
expect(mockSubscribe).toHaveBeenCalledTimes(2)
})
it('allows a shared watch attach even when the root watch cap is full', async () => {
mockSubscribe.mockResolvedValue({ unsubscribe: vi.fn() })
for (let index = 0; index < 20; index += 1) {
const dir = path.join(tmpDir, `watched-${index}`)
await fs.mkdir(dir)
await dispatcher.callRequest(
'fs.watch',
{ rootPath: dir },
{
clientId: index + 1,
isStale: () => false
}
)
}
await expect(
dispatcher.callRequest(
'fs.watch',
{ rootPath: path.join(tmpDir, 'watched-0') },
{
clientId: 99,
isStale: () => false
}
)
).resolves.toBeUndefined()
})
it('releases a client watch when the dispatcher detaches that client', async () => {
const unsubscribe = vi.fn()
mockSubscribe.mockResolvedValue({ unsubscribe })
await dispatcher.callRequest(
'fs.watch',
{ rootPath: tmpDir },
{
clientId: 7,
isStale: () => false
}
)
dispatcher.detachClient(7)
expect(unsubscribe).toHaveBeenCalledTimes(1)
})
})