From 710405698472c3d9618c3114c7f15088be4d72cc Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:14:21 -0700 Subject: [PATCH] fix(watcher): route relay watch-root capacity refusals off the fast ladder (#17950) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 --- src/main/ipc/filesystem-watcher-handlers.ts | 7 + .../ipc/filesystem-watcher-lifecycle-state.ts | 5 +- ...filesystem-watcher-remote-capacity.test.ts | 91 +++++++++++++ .../filesystem-watcher-remote-controller.ts | 3 +- .../ipc/filesystem-watcher-remote-dormant.ts | 4 +- .../ipc/filesystem-watcher-remote-install.ts | 5 + ...ilesystem-watcher-remote-provider-rearm.ts | 5 + .../ipc/filesystem-watcher-remote-removal.ts | 5 +- .../ipc/filesystem-watcher-remote-retry.ts | 6 + src/relay/fs-handler.test.ts | 114 +---------------- src/relay/relay-filesystem-watch-registry.ts | 18 ++- src/relay/relay-fs-test-dispatcher.ts | 120 ++++++++++++++++++ src/relay/relay-watch-root-capacity-gate.ts | 86 +++++++++++++ src/relay/relay-watch-root-capacity.test.ts | 107 ++++++++++++++++ src/relay/relay-watcher-root-capacity.ts | 20 ++- src/relay/relay-watcher-teardown-tracker.ts | 11 ++ src/shared/watch-root-capacity-refusal.ts | 9 ++ 17 files changed, 487 insertions(+), 129 deletions(-) create mode 100644 src/main/ipc/filesystem-watcher-remote-capacity.test.ts create mode 100644 src/relay/relay-fs-test-dispatcher.ts create mode 100644 src/relay/relay-watch-root-capacity-gate.ts create mode 100644 src/relay/relay-watch-root-capacity.test.ts create mode 100644 src/shared/watch-root-capacity-refusal.ts diff --git a/src/main/ipc/filesystem-watcher-handlers.ts b/src/main/ipc/filesystem-watcher-handlers.ts index 7a2b6230abd..bac74f63589 100644 --- a/src/main/ipc/filesystem-watcher-handlers.ts +++ b/src/main/ipc/filesystem-watcher-handlers.ts @@ -10,6 +10,7 @@ import { import { installRemoteWatcher, reinstallRemoteWatchersForConnection, + scheduleDormantRemoteWatcherRearm, scheduleRemoteWatcherRetry } from './filesystem-watcher-remote-controller' import { rememberDesiredRemoteWatcher } from './filesystem-watcher-remote-desired' @@ -41,6 +42,12 @@ export function registerFilesystemWatcherHandlers(): void { args.connectionId, args.worktreePath ) + if (result === 'capacity') { + // Why straight to the dormant backoff: the cap is full until some other root is released, + // which a 1 Hz reinstall cannot bring about — it only adds relay load per refused root. + scheduleDormantRemoteWatcherRearm(args.connectionId, args.worktreePath) + return + } if (result === 'unavailable') { if (!watcherLifecycleState.loggedUnavailableRemoteWatchers.has(key)) { watcherLifecycleState.loggedUnavailableRemoteWatchers.add(key) diff --git a/src/main/ipc/filesystem-watcher-lifecycle-state.ts b/src/main/ipc/filesystem-watcher-lifecycle-state.ts index 548d958f90c..7d6a120c0a7 100644 --- a/src/main/ipc/filesystem-watcher-lifecycle-state.ts +++ b/src/main/ipc/filesystem-watcher-lifecycle-state.ts @@ -36,7 +36,10 @@ export type RemoteWatcherState = { batch: RemoteWatcherEventBatch } -export type RemoteWatcherInstallResult = 'installed' | 'unavailable' | 'cancelled' +// Why 'capacity' is not 'unavailable': the relay refused because its watch-root cap is full, which is +// a decision, not a fault. The 1 Hz unavailable retry cannot change that answer, and a folder +// workspace whose repo count exceeds the cap turns it into a permanent per-root storm (#11196). +export type RemoteWatcherInstallResult = 'installed' | 'unavailable' | 'capacity' | 'cancelled' export type RemoteWatcherResyncState = { lastSentAt: number diff --git a/src/main/ipc/filesystem-watcher-remote-capacity.test.ts b/src/main/ipc/filesystem-watcher-remote-capacity.test.ts new file mode 100644 index 00000000000..97a8f82bf33 --- /dev/null +++ b/src/main/ipc/filesystem-watcher-remote-capacity.test.ts @@ -0,0 +1,91 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { handleMock, getSshFilesystemProviderMock } = vi.hoisted(() => ({ + handleMock: vi.fn(), + getSshFilesystemProviderMock: vi.fn() +})) + +vi.mock('electron', () => ({ + ipcMain: { handle: handleMock } +})) + +vi.mock('fs/promises', () => ({ stat: vi.fn() })) +vi.mock('@parcel/watcher', () => ({ subscribe: vi.fn() })) +vi.mock('./filesystem-watcher-wsl', () => ({ createWslWatcher: vi.fn() })) +vi.mock('../providers/ssh-filesystem-dispatch', () => ({ + getSshFilesystemProvider: getSshFilesystemProviderMock, + onSshFilesystemProviderRegistered: () => () => {} +})) + +import { WATCH_ROOT_CAPACITY_REFUSAL_MESSAGE } from '../../shared/watch-root-capacity-refusal' +import { closeAllWatchers, registerFilesystemWatcherHandlers } from './filesystem-watcher' +import { watcherLifecycleState } from './filesystem-watcher-lifecycle-state' +import { getRemoteWatcherKey } from './filesystem-watcher-paths' + +type HandlerMap = Record unknown> + +describe('remote filesystem watcher capacity refusals', () => { + const handlers: HandlerMap = {} + + beforeEach(async () => { + handleMock.mockReset() + getSshFilesystemProviderMock.mockReset() + for (const key of Object.keys(handlers)) { + delete handlers[key] + } + handleMock.mockImplementation((channel, handler) => { + handlers[channel] = handler + }) + registerFilesystemWatcherHandlers() + await closeAllWatchers() + }) + + afterEach(async () => { + for (const dormant of watcherLifecycleState.dormantRemoteWatchers.values()) { + clearTimeout(dormant.timer) + } + watcherLifecycleState.dormantRemoteWatchers.clear() + await closeAllWatchers() + vi.useRealTimers() + }) + + // A folder workspace with more repos than the relay's watch-root cap leaves every excess root + // permanently refused; the 1 Hz unavailable ladder then bills the relay one install per root per + // second, which is the load that pinned it (#11196). + it('does not retry a relay watch-root capacity refusal on the fast ladder', async () => { + vi.useFakeTimers() + const watchMock = vi.fn(async () => { + throw new Error(WATCH_ROOT_CAPACITY_REFUSAL_MESSAGE) + }) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 } + const args = { worktreePath: '/home/me/repos/one', connectionId: 'conn-capacity' } + + await handlers['fs:watchWorktree']({ sender }, args) + const key = getRemoteWatcherKey(args.connectionId, args.worktreePath) + + expect(watchMock).toHaveBeenCalledTimes(1) + expect(watcherLifecycleState.pendingRemoteWatcherRetries.has(key)).toBe(false) + expect(watcherLifecycleState.dormantRemoteWatchers.has(key)).toBe(true) + + await vi.advanceTimersByTimeAsync(10_000) + expect(watchMock).toHaveBeenCalledTimes(1) + }) + + it('still retries an ordinary unavailable install on the fast ladder', async () => { + vi.useFakeTimers() + const watchMock = vi.fn(async () => { + throw new Error('Relay channel lost') + }) + getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock }) + const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 } + const args = { worktreePath: '/home/me/repos/two', connectionId: 'conn-unavailable' } + + await handlers['fs:watchWorktree']({ sender }, args) + const key = getRemoteWatcherKey(args.connectionId, args.worktreePath) + + expect(watcherLifecycleState.pendingRemoteWatcherRetries.has(key)).toBe(true) + await vi.advanceTimersByTimeAsync(2_500) + expect(watchMock.mock.calls.length).toBeGreaterThan(1) + }) +}) diff --git a/src/main/ipc/filesystem-watcher-remote-controller.ts b/src/main/ipc/filesystem-watcher-remote-controller.ts index 75bc4aa75d5..1b6953cd292 100644 --- a/src/main/ipc/filesystem-watcher-remote-controller.ts +++ b/src/main/ipc/filesystem-watcher-remote-controller.ts @@ -63,7 +63,8 @@ export function reinstallRemoteWatchersForConnection(connectionId: string): void reinstallRemoteWatchersForConnectionCore(connectionId, { install: installRemoteWatcher, requestResync: requestRemoteWatcherResync, - scheduleRetry: scheduleRemoteWatcherRetry + scheduleRetry: scheduleRemoteWatcherRetry, + scheduleDormant: scheduleDormantRemoteWatcherRearm }) } diff --git a/src/main/ipc/filesystem-watcher-remote-dormant.ts b/src/main/ipc/filesystem-watcher-remote-dormant.ts index 741753b42e5..7c168114f51 100644 --- a/src/main/ipc/filesystem-watcher-remote-dormant.ts +++ b/src/main/ipc/filesystem-watcher-remote-dormant.ts @@ -97,8 +97,8 @@ async function rearmDormantRemoteWatcher( worktreePath, listeners.filter((_, index) => results[index] === 'installed') ) - // Why: 'cancelled' means shutdown or the last listener left, so only 'unavailable' stays dormant. - if (results.some((result) => result === 'unavailable')) { + // Why: 'cancelled' means shutdown or the last listener left, so only a refusal stays dormant. + if (results.some((result) => result === 'unavailable' || result === 'capacity')) { scheduleDormantRemoteWatcherRearmCore( connectionId, worktreePath, diff --git a/src/main/ipc/filesystem-watcher-remote-install.ts b/src/main/ipc/filesystem-watcher-remote-install.ts index c069c8dc16b..3ab5babbd91 100644 --- a/src/main/ipc/filesystem-watcher-remote-install.ts +++ b/src/main/ipc/filesystem-watcher-remote-install.ts @@ -1,5 +1,6 @@ import type { WebContents } from 'electron' import type { FsChangedPayload } from '../../shared/filesystem-entry-types' +import { isWatchRootCapacityRefusal } from '../../shared/watch-root-capacity-refusal' import { WATCH_BATCH_MAX_WAIT_MS, WATCH_BATCH_TRAILING_MS @@ -202,6 +203,10 @@ async function doInstallRemoteWatcher( if (cancelToken.cancelled || cancelToken.abortController.signal.aborted) { return 'cancelled' } + if (isWatchRootCapacityRefusal(err)) { + console.warn(`[filesystem-watcher] relay watch-root capacity reached for ${key}`) + return 'capacity' + } console.warn(`[filesystem-watcher] SSH watcher unavailable for ${key}:`, err) return 'unavailable' } finally { diff --git a/src/main/ipc/filesystem-watcher-remote-provider-rearm.ts b/src/main/ipc/filesystem-watcher-remote-provider-rearm.ts index 6b0ea28ef9b..3cc657c4086 100644 --- a/src/main/ipc/filesystem-watcher-remote-provider-rearm.ts +++ b/src/main/ipc/filesystem-watcher-remote-provider-rearm.ts @@ -29,6 +29,7 @@ export function reinstallRemoteWatchersForConnectionCore( install: InstallRemoteWatcher requestResync: RequestRemoteWatcherResync scheduleRetry: ScheduleRemoteWatcherRetry + scheduleDormant: (connectionId: string, worktreePath: string) => void } ): void { if (watcherLifecycleState.remoteWatchersClosed) { @@ -90,6 +91,10 @@ export function reinstallRemoteWatchersForConnectionCore( desired.worktreePath, listeners.filter((_, index) => results[index] === 'installed') ) + if (results.some((result) => result === 'capacity')) { + dependencies.scheduleDormant(desired.connectionId, desired.worktreePath) + return + } if (results.some((result) => result === 'unavailable')) { for (const listener of listeners) { dependencies.scheduleRetry( diff --git a/src/main/ipc/filesystem-watcher-remote-removal.ts b/src/main/ipc/filesystem-watcher-remote-removal.ts index 61424dc1db9..0e9c741851a 100644 --- a/src/main/ipc/filesystem-watcher-remote-removal.ts +++ b/src/main/ipc/filesystem-watcher-remote-removal.ts @@ -9,6 +9,7 @@ import { } from './filesystem-watcher-listener-lifecycle' import { installRemoteWatcher, + scheduleDormantRemoteWatcherRearm, scheduleRemoteWatcherRetry } from './filesystem-watcher-remote-controller' @@ -75,7 +76,9 @@ export async function restoreRemoteWatcherAfterFailedRemoval( continue } const result = await installRemoteWatcher(sender, connectionId, worktreePath) - if (result === 'unavailable') { + if (result === 'capacity') { + scheduleDormantRemoteWatcherRearm(connectionId, worktreePath) + } else if (result === 'unavailable') { scheduleRemoteWatcherRetry(sender, connectionId, worktreePath) } sender.send('fs:changed', { diff --git a/src/main/ipc/filesystem-watcher-remote-retry.ts b/src/main/ipc/filesystem-watcher-remote-retry.ts index 0151ecf4419..d73f9ed4db4 100644 --- a/src/main/ipc/filesystem-watcher-remote-retry.ts +++ b/src/main/ipc/filesystem-watcher-remote-retry.ts @@ -100,6 +100,12 @@ export function scheduleRemoteWatcherRetryCore( listeners.filter((_, index) => results[index] === 'installed') ) } + // Why capacity leaves the fast window: the relay is refusing on a full watch-root cap, and a + // 1 Hz reinstall per refused root is exactly the load that keeps the cap busy (#11196). + if (results.some((result) => result === 'capacity')) { + dependencies.scheduleDormant(connectionId, worktreePath) + return + } // Why: don't re-arm on 'cancelled' (renderer stopped watching) — it would fire a stale overflow when the 60s window expires. if (results.some((result) => result === 'unavailable')) { for (const listener of listeners) { diff --git a/src/relay/fs-handler.test.ts b/src/relay/fs-handler.test.ts index 9dc4812ef98..fcea45c898c 100644 --- a/src/relay/fs-handler.test.ts +++ b/src/relay/fs-handler.test.ts @@ -8,6 +8,7 @@ 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() @@ -17,91 +18,6 @@ vi.mock('@parcel/watcher', () => ({ subscribe: mockSubscribe })) -function createMockDispatcher() { - const requestHandlers = new Map< - string, - ( - params: Record, - context?: { clientId: number; isStale: () => boolean } - ) => Promise - >() - const notificationHandlers = new Map< - string, - ( - params: Record, - context?: { clientId: number; isStale: () => boolean } - ) => void - >() - const detachListeners = new Set<(clientId: number) => void>() - const notifications: { method: string; params?: Record }[] = [] - - return { - onRequest: vi.fn( - ( - method: string, - handler: ( - params: Record, - context?: { clientId: number; isStale: () => boolean } - ) => Promise - ) => { - requestHandlers.set(method, handler) - } - ), - onNotification: vi.fn( - ( - method: string, - handler: ( - params: Record, - context?: { clientId: number; isStale: () => boolean } - ) => void - ) => { - notificationHandlers.set(method, handler) - } - ), - notify: vi.fn((method: string, params?: Record) => { - notifications.push({ method, params }) - }), - notifyClient: vi.fn(), - onClientDetached: vi.fn((listener: (clientId: number) => void) => { - detachListeners.add(listener) - return () => detachListeners.delete(listener) - }), - _requestHandlers: requestHandlers, - _notificationHandlers: notificationHandlers, - _notifications: notifications, - async callRequest( - method: string, - params: Record = {}, - context?: { clientId?: number; isStale: () => boolean } - ) { - const handler = requestHandlers.get(method) - if (!handler) { - throw new Error(`No handler for ${method}`) - } - return handler(params, { - clientId: context?.clientId ?? 1, - isStale: context?.isStale ?? (() => false) - }) - }, - callNotification( - method: string, - params: Record = {}, - context?: { clientId: number; isStale: () => boolean } - ) { - const handler = notificationHandlers.get(method) - if (!handler) { - throw new Error(`No handler for ${method}`) - } - handler(params, context ?? { clientId: 1, isStale: () => false }) - }, - detachClient(clientId: number) { - for (const listener of detachListeners) { - listener(clientId) - } - } - } -} - function statIdentity(stats: { dev?: number ino?: number @@ -837,34 +753,6 @@ describe('FsHandler', () => { await joined }) - it('blocks replacement watches behind physical unsubscribe and counts the pending slot', async () => { - let resolveUnsubscribe: () => void = () => {} - const unsubscribe = vi.fn( - () => - new Promise((resolve) => { - resolveUnsubscribe = resolve - }) - ) - mockSubscribe.mockResolvedValue({ unsubscribe }) - await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }) - dispatcher.callNotification('fs.unwatch', { rootPath: tmpDir }) - - const replacement = dispatcher.callRequest('fs.watch', { rootPath: tmpDir }) - for (let index = 0; index < 19; index += 1) { - await dispatcher.callRequest('fs.watch', { - rootPath: path.join(tmpDir, `pending-cap-${index}`) - }) - } - await expect( - dispatcher.callRequest('fs.watch', { rootPath: path.join(tmpDir, 'over-pending-cap') }) - ).rejects.toThrow('Maximum number of file watchers reached') - expect(mockSubscribe).toHaveBeenCalledTimes(20) - - resolveUnsubscribe() - await replacement - expect(mockSubscribe).toHaveBeenCalledTimes(21) - }) - it('retains a failed native unsubscribe slot until acknowledged retry succeeds', async () => { const unsubscribe = vi .fn() diff --git a/src/relay/relay-filesystem-watch-registry.ts b/src/relay/relay-filesystem-watch-registry.ts index de583b2f50a..14f5d5b2ee8 100644 --- a/src/relay/relay-filesystem-watch-registry.ts +++ b/src/relay/relay-filesystem-watch-registry.ts @@ -16,7 +16,7 @@ import { type RelayWatcherTeardownState } from './relay-watcher-teardown-tracker' import { emitRelayWatcherTerminalFailure } from './relay-watcher-terminal-notifier' -import { assertRelayWatcherRootCapacity } from './relay-watcher-root-capacity' +import { RelayWatchRootCapacityGate } from './relay-watch-root-capacity-gate' import { normalizeRuntimePathForComparison } from '../shared/cross-platform-path' import { trackRelayWatcherSetup, @@ -33,6 +33,11 @@ const RELAY_WATCH_OPTIONS = buildParcelWatcherIgnoreOptions(WATCHER_IGNORE_DIRS) export class RelayFilesystemWatchRegistry { private readonly watches = new Map() private readonly pendingSetups = new Map() + private readonly capacityGate = new RelayWatchRootCapacityGate( + this.watches, + this.pendingSetups, + () => this.teardownTracker + ) private readonly teardownTracker: RelayWatcherTeardownTracker private readonly removalFence: RelayWatcherRemovalFence @@ -95,6 +100,10 @@ export class RelayFilesystemWatchRegistry { if (rootTeardown) { await rootTeardown } + const capacityRelease = this.capacityGate.release(rootKey, context?.signal) + if (capacityRelease) { + await capacityRelease + } const clientId = context?.clientId ?? 0 const isStale = context?.isStale ?? (() => false) const existing = this.watches.get(rootKey) @@ -107,12 +116,7 @@ export class RelayFilesystemWatchRegistry { return } - assertRelayWatcherRootCapacity( - this.watches.keys(), - this.pendingSetups.keys(), - this.teardownTracker.rootPaths(), - rootKey - ) + this.capacityGate.assert(rootKey) const state = createRelayWatcherState(rootKey, rootPath, clientId, isStale, watchId) this.watches.set(rootKey, state) diff --git a/src/relay/relay-fs-test-dispatcher.ts b/src/relay/relay-fs-test-dispatcher.ts new file mode 100644 index 00000000000..17d0e20bd19 --- /dev/null +++ b/src/relay/relay-fs-test-dispatcher.ts @@ -0,0 +1,120 @@ +import { vi, type Mock } from 'vitest' + +type MockRequestHandler = ( + params: Record, + context?: { clientId: number; isStale: () => boolean } +) => Promise +type MockNotificationHandler = ( + params: Record, + context?: { clientId: number; isStale: () => boolean } +) => void +type MockCallContext = { clientId?: number; isStale: () => boolean } + +// Explicit rather than inferred: vi.fn()'s inferred type is not nameable across project boundaries. +export type MockRelayFsDispatcher = { + onRequest: Mock + onNotification: Mock + notify: Mock + notifyClient: Mock + onClientDetached: Mock + _requestHandlers: Map + _notificationHandlers: Map + _notifications: { method: string; params?: Record }[] + callRequest: ( + method: string, + params?: Record, + context?: MockCallContext + ) => Promise + callNotification: ( + method: string, + params?: Record, + context?: { clientId: number; isStale: () => boolean } + ) => void + detachClient: (clientId: number) => void +} + +/** Records handlers and notifications so a test can drive FsHandler without a real transport. */ +export function createMockDispatcher(): MockRelayFsDispatcher { + const requestHandlers = new Map< + string, + ( + params: Record, + context?: { clientId: number; isStale: () => boolean } + ) => Promise + >() + const notificationHandlers = new Map< + string, + ( + params: Record, + context?: { clientId: number; isStale: () => boolean } + ) => void + >() + const detachListeners = new Set<(clientId: number) => void>() + const notifications: { method: string; params?: Record }[] = [] + + return { + onRequest: vi.fn( + ( + method: string, + handler: ( + params: Record, + context?: { clientId: number; isStale: () => boolean } + ) => Promise + ) => { + requestHandlers.set(method, handler) + } + ), + onNotification: vi.fn( + ( + method: string, + handler: ( + params: Record, + context?: { clientId: number; isStale: () => boolean } + ) => void + ) => { + notificationHandlers.set(method, handler) + } + ), + notify: vi.fn((method: string, params?: Record) => { + notifications.push({ method, params }) + }), + notifyClient: vi.fn(), + onClientDetached: vi.fn((listener: (clientId: number) => void) => { + detachListeners.add(listener) + return () => detachListeners.delete(listener) + }), + _requestHandlers: requestHandlers, + _notificationHandlers: notificationHandlers, + _notifications: notifications, + async callRequest( + method: string, + params: Record = {}, + context?: { clientId?: number; isStale: () => boolean } + ) { + const handler = requestHandlers.get(method) + if (!handler) { + throw new Error(`No handler for ${method}`) + } + return handler(params, { + clientId: context?.clientId ?? 1, + isStale: context?.isStale ?? (() => false) + }) + }, + callNotification( + method: string, + params: Record = {}, + context?: { clientId: number; isStale: () => boolean } + ) { + const handler = notificationHandlers.get(method) + if (!handler) { + throw new Error(`No handler for ${method}`) + } + handler(params, context ?? { clientId: 1, isStale: () => false }) + }, + detachClient(clientId: number) { + for (const listener of detachListeners) { + listener(clientId) + } + } + } +} diff --git a/src/relay/relay-watch-root-capacity-gate.ts b/src/relay/relay-watch-root-capacity-gate.ts new file mode 100644 index 00000000000..4e430332be2 --- /dev/null +++ b/src/relay/relay-watch-root-capacity-gate.ts @@ -0,0 +1,86 @@ +import { + assertRelayWatcherRootCapacity, + exceedsRelayWatcherRootCapacity +} from './relay-watcher-root-capacity' + +type RelayWatchRootTeardowns = { + rootPaths: () => string[] + /** Resolves when every teardown in flight has settled, or undefined when none is. */ + settlePending: () => Promise | undefined +} + +/** + * Decides whether a prospective watch root fits, and waits out an over-cap that only unsubscribing + * roots are causing. + * + * Why waiting beats refusing: a reconnect tears the old roots down as it installs the new ones, so + * the cap is briefly full of slots already promised back. The client answers a capacity refusal + * with a 60s-to-30min dormancy that no release event can shorten, so refusing on a transient + * overlap costs half an hour of blindness. Mirrors WatcherSupervisorCapacityWait. + */ +export class RelayWatchRootCapacityGate { + // Why tracked: a root parked on the wait has been granted nothing, so counting its setup entry + // would let it hold a slot away from the root already reclaiming one. + private readonly waiting = new Set() + + constructor( + private readonly activeRoots: ReadonlyMap, + private readonly setupRoots: ReadonlyMap, + // Thunk: the registry builds its teardown tracker after this field initializes. + private readonly teardowns: () => RelayWatchRootTeardowns + ) {} + + assert(rootKey: string): void { + assertRelayWatcherRootCapacity( + this.activeRoots.keys(), + this.claimedSetupRoots(rootKey), + this.teardowns().rootPaths(), + rootKey + ) + } + + /** + * The wait to hold before {@link assert}, or undefined when there is nothing to wait for. + * + * Undefined rather than a resolved promise so an install that already fits stays synchronous — + * a suspension here would let a concurrent watch of the same root join the setup, not the watch. + */ + release(rootKey: string, signal?: AbortSignal): Promise | undefined { + if ( + !exceedsRelayWatcherRootCapacity( + this.activeRoots.keys(), + this.claimedSetupRoots(rootKey), + this.teardowns().rootPaths(), + rootKey + ) + ) { + return undefined + } + const released = this.teardowns().settlePending() + if (!released) { + return undefined + } + this.waiting.add(rootKey) + // Once, and never past the caller: a genuinely full cap must still reach the refusal that sends + // the client dormant, and an unsubscribe that never settles must not park the request with it. + return (signal ? Promise.race([released, abortSignalSettled(signal)]) : released).finally( + () => { + this.waiting.delete(rootKey) + } + ) + } + + /** Setup roots that currently hold a slot — a parked capacity waiter holds none. */ + private claimedSetupRoots(rootKey: string): string[] { + return [...this.setupRoots.keys()].filter((key) => key === rootKey || !this.waiting.has(key)) + } +} + +/** Resolves (never rejects) when the request is abandoned, so a race can drop out of a wait. */ +function abortSignalSettled(signal: AbortSignal): Promise { + return signal.aborted + ? Promise.resolve() + : new Promise((resolve) => + signal.addEventListener('abort', () => resolve(), { once: true }) + ) +} diff --git a/src/relay/relay-watch-root-capacity.test.ts b/src/relay/relay-watch-root-capacity.test.ts new file mode 100644 index 00000000000..de5a5fc2b4b --- /dev/null +++ b/src/relay/relay-watch-root-capacity.test.ts @@ -0,0 +1,107 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import * as fs from 'node:fs/promises' +import * as path from 'node:path' +import { mkdtempSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { RelayContext } from './context' +import type { RelayDispatcher } from './dispatcher' +import { FsHandler } from './fs-handler' +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 +})) + +describe('relay watch-root capacity', () => { + let dispatcher: ReturnType + let handler: FsHandler + let tmpDir: string + + beforeEach(() => { + mockSubscribe.mockReset() + mockSubscribe.mockResolvedValue({ unsubscribe: vi.fn() }) + tmpDir = mkdtempSync(path.join(tmpdir(), 'relay-fs-cap-')) + dispatcher = createMockDispatcher() + handler = new FsHandler(dispatcher as unknown as RelayDispatcher, new RelayContext(), { + dispose: vi.fn(), + forgetRoot: vi.fn(), + subscribe: subscribeWithInProcessWatcher + }) + }) + + afterEach(async () => { + handler.dispose() + await fs.rm(tmpDir, { recursive: true, force: true }) + }) + + it('blocks replacement watches behind physical unsubscribe and counts the pending slot', async () => { + let resolveUnsubscribe: () => void = () => {} + const unsubscribe = vi.fn( + () => + new Promise((resolve) => { + resolveUnsubscribe = resolve + }) + ) + mockSubscribe.mockResolvedValue({ unsubscribe }) + await dispatcher.callRequest('fs.watch', { rootPath: tmpDir }) + dispatcher.callNotification('fs.unwatch', { rootPath: tmpDir }) + + const replacement = dispatcher.callRequest('fs.watch', { rootPath: tmpDir }) + for (let index = 0; index < 19; index += 1) { + await dispatcher.callRequest('fs.watch', { + rootPath: path.join(tmpDir, `pending-cap-${index}`) + }) + } + // The replacement claims the slot the teardown releases, so this cap is genuinely full: the + // request waits for the release event and is still refused once it has happened. + const overCap = dispatcher + .callRequest('fs.watch', { rootPath: path.join(tmpDir, 'over-pending-cap') }) + .then( + () => null, + (error: Error) => error + ) + expect(mockSubscribe).toHaveBeenCalledTimes(20) + + resolveUnsubscribe() + await replacement + expect(await overCap).toMatchObject({ message: 'Maximum number of file watchers reached' }) + expect(mockSubscribe).toHaveBeenCalledTimes(21) + }) + + it('waits out a teardown that frees a slot instead of refusing on it', async () => { + let resolveUnsubscribe: () => void = () => {} + mockSubscribe.mockResolvedValue({ + unsubscribe: vi.fn( + () => + new Promise((resolve) => { + resolveUnsubscribe = resolve + }) + ) + }) + for (let index = 0; index < 20; index += 1) { + await dispatcher.callRequest('fs.watch', { rootPath: path.join(tmpDir, `full-${index}`) }) + } + dispatcher.callNotification('fs.unwatch', { rootPath: path.join(tmpDir, 'full-0') }) + + // Why not a refusal: the slot is already promised back, and the client answers a capacity + // refusal with a 60s-to-30min dormancy that no release event can shorten. + let settled = false + const fresh = dispatcher + .callRequest('fs.watch', { rootPath: path.join(tmpDir, 'fresh') }) + .then(() => { + settled = true + }) + await Promise.resolve() + expect(settled).toBe(false) + expect(mockSubscribe).toHaveBeenCalledTimes(20) + + resolveUnsubscribe() + await fresh + expect(mockSubscribe).toHaveBeenCalledTimes(21) + }) +}) diff --git a/src/relay/relay-watcher-root-capacity.ts b/src/relay/relay-watcher-root-capacity.ts index 025462e7bd3..35178166f7c 100644 --- a/src/relay/relay-watcher-root-capacity.ts +++ b/src/relay/relay-watcher-root-capacity.ts @@ -1,14 +1,26 @@ +import { WATCH_ROOT_CAPACITY_REFUSAL_MESSAGE } from '../shared/watch-root-capacity-refusal' + const MAX_RELAY_WATCH_ROOTS = 20 +// Why teardown roots count: a root still unsubscribing owns its native handles until it settles. +export function exceedsRelayWatcherRootCapacity( + activeRoots: Iterable, + pendingRoots: Iterable, + teardownRoots: Iterable, + prospectiveRoot: string +): boolean { + const physicalRoots = new Set([...activeRoots, ...pendingRoots, ...teardownRoots]) + physicalRoots.add(prospectiveRoot) + return physicalRoots.size > MAX_RELAY_WATCH_ROOTS +} + export function assertRelayWatcherRootCapacity( activeRoots: Iterable, pendingRoots: Iterable, teardownRoots: Iterable, prospectiveRoot: string ): void { - const physicalRoots = new Set([...activeRoots, ...pendingRoots, ...teardownRoots]) - physicalRoots.add(prospectiveRoot) - if (physicalRoots.size > MAX_RELAY_WATCH_ROOTS) { - throw new Error('Maximum number of file watchers reached') + if (exceedsRelayWatcherRootCapacity(activeRoots, pendingRoots, teardownRoots, prospectiveRoot)) { + throw new Error(WATCH_ROOT_CAPACITY_REFUSAL_MESSAGE) } } diff --git a/src/relay/relay-watcher-teardown-tracker.ts b/src/relay/relay-watcher-teardown-tracker.ts index 91b1460054d..785e4dd16ea 100644 --- a/src/relay/relay-watcher-teardown-tracker.ts +++ b/src/relay/relay-watcher-teardown-tracker.ts @@ -98,6 +98,17 @@ export class RelayWatcherTeardownTracker { rootPaths(): string[] { return [...this.pending.keys(), ...this.failed.keys()] } + + /** + * The capacity-release event: resolves once every teardown in flight right now has settled. + * + * `undefined` when nothing is unsubscribing, which is the only honest answer to "could a slot + * still come back?" — a failed teardown keeps its handles and releases nothing. + */ + settlePending(): Promise | undefined { + const inFlight = [...this.pending.values()] + return inFlight.length === 0 ? undefined : Promise.allSettled(inFlight).then(() => undefined) + } } function callUnsubscribe(subscription: WatcherProcessSubscription): Promise { diff --git a/src/shared/watch-root-capacity-refusal.ts b/src/shared/watch-root-capacity-refusal.ts new file mode 100644 index 00000000000..234fecb4ef7 --- /dev/null +++ b/src/shared/watch-root-capacity-refusal.ts @@ -0,0 +1,9 @@ +// Why a shared string rather than an error code: the refusal crosses the relay wire as a JSON-RPC +// error message, and relays deploy independently of clients. Both sides must spell it the same way, +// and a client that does not recognise it simply falls back to the ordinary unavailable handling. +export const WATCH_ROOT_CAPACITY_REFUSAL_MESSAGE = 'Maximum number of file watchers reached' + +export function isWatchRootCapacityRefusal(error: unknown): boolean { + const message = (error as { message?: unknown } | null | undefined)?.message + return typeof message === 'string' && message.includes(WATCH_ROOT_CAPACITY_REFUSAL_MESSAGE) +}