diff --git a/config/ts-nocheck-baseline.txt b/config/ts-nocheck-baseline.txt index 7b102bdf419..787e9ffe89a 100644 --- a/config/ts-nocheck-baseline.txt +++ b/config/ts-nocheck-baseline.txt @@ -163,7 +163,6 @@ src/main/runtime/runtime-file-commands-revoke-terminal-file-grants-for-client.ts src/main/runtime/runtime-file-commands-search-local-runtime-files.ts src/main/runtime/runtime-file-commands-search-remote-quick-open-file-paths.ts src/main/runtime/runtime-file-commands-search-runtime-files.ts -src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.ts src/main/runtime/runtime-file-commands-terminal-artifact-access.ts src/main/runtime/runtime-file-commands-terminal-file-paths.ts src/main/runtime/runtime-file-commands-write-file-explorer-file.ts diff --git a/src/main/runtime/orca-runtime-files-ssh-rearm.test.ts b/src/main/runtime/orca-runtime-files-ssh-rearm.test.ts index 4403d61b63d..add68726e79 100644 --- a/src/main/runtime/orca-runtime-files-ssh-rearm.test.ts +++ b/src/main/runtime/orca-runtime-files-ssh-rearm.test.ts @@ -1,5 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { FsChangeEvent } from '../../shared/filesystem-entry-types' +import { createSshDisposalError } from '../ssh/ssh-channel-multiplexer' +import { createSshFileExplorerWatchTestConnection } from './ssh-file-explorer-watch-test-connection' const { resolveAuthorizedPathMock, @@ -47,6 +49,8 @@ import { const ROOT_PATH = '/home/me/repo' const CONNECTION_ID = 'conn-1' const OVERFLOW_EVENTS: FsChangeEvent[] = [{ kind: 'overflow', absolutePath: ROOT_PATH }] +const connections: ReturnType[] = [] +const releases: (() => Promise)[] = [] /** Drive the provider-registration hook the way a relay reconnect would. */ function emitProviderRegistered(connectionId: string): void { @@ -80,29 +84,168 @@ describe('remote file-explorer watch re-arm', () => { }) afterEach(async () => { + for (const { mux } of connections) { + mux.dispose() + } + for (const release of releases.splice(0)) { + await release() + } await awaitRuntimeFileWatcherUnsubscribes() _resetRuntimeFileWatcherLeasesForTests() + for (const { provider } of connections.splice(0)) { + provider.dispose() + } }) + function connect() { + const connection = createSshFileExplorerWatchTestConnection(ROOT_PATH, CONNECTION_ID) + connections.push(connection) + return connection + } + + async function startInitialWatch() { + const first = connect() + getSshFilesystemProviderMock.mockReturnValue(first.provider) + const watch = vi.spyOn(first.provider, 'watch') + const commands = createRuntimeFileCommands() + const onEvents = vi.fn() + const controller = new AbortController() + const onTerminalError = vi.fn(() => controller.abort()) + const setup = commands.watchFileExplorer( + 'id:wt-1', + onEvents, + onTerminalError, + controller.signal + ) + await vi.waitFor(() => expect(first.countRequests('fs.watch')).toBe(1)) + const finish = async () => { + first.settleWatch() + const release = await setup + releases.push(release) + return release + } + return { first, watch, onEvents, onTerminalError, controller, setup, finish } + } + + it.each(['event', 'terminal error'] as const)( + 'fences the initial provider %s while replacement setup is pending', + async (stage) => { + const { first, onEvents, onTerminalError, controller, finish } = await startInitialWatch() + const release = await finish() + const next = connect() + getSshFilesystemProviderMock.mockReturnValue(next.provider) + emitProviderRegistered(CONNECTION_ID) + await vi.waitFor(() => expect(next.countRequests('fs.watch')).toBe(1)) + if (stage === 'event') { + first.emitChange() + } else { + first.failWatch('late initial failure') + } + expect(onEvents).not.toHaveBeenCalled() + expect(onTerminalError).not.toHaveBeenCalled() + expect(controller.signal.aborted).toBe(false) + next.settleWatch() + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledExactlyOnceWith(OVERFLOW_EVENTS)) + next.emitChange() + expect(onEvents).toHaveBeenCalledTimes(2) + await release() + expect(next.countRequests('fs.unwatch')).toBe(1) + } + ) + + it('fences initial callbacks after same-provider replacement and unsubscribe', async () => { + const { first, watch, onEvents, onTerminalError, finish } = await startInitialWatch() + const release = await finish() + const initialEvent = watch.mock.calls[0]?.[1] + const initialError = watch.mock.calls[0]?.[2]?.onTerminalError + emitProviderRegistered(CONNECTION_ID) + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledExactlyOnceWith(OVERFLOW_EVENTS)) + initialEvent?.([{ kind: 'update', absolutePath: `${ROOT_PATH}/stale.ts` }]) + initialError?.(new Error('old generation failed')) + expect(onEvents).toHaveBeenCalledOnce() + expect(onTerminalError).not.toHaveBeenCalled() + first.emitChange() + expect(onEvents).toHaveBeenCalledTimes(2) + await release() + initialEvent?.([{ kind: 'update', absolutePath: `${ROOT_PATH}/closed.ts` }]) + initialError?.(new Error('closed initial watch failed')) + expect(onEvents).toHaveBeenCalledTimes(2) + expect(onTerminalError).not.toHaveBeenCalled() + expect(first.countRequests('fs.unwatch')).toBe(1) + }) + + it('delivers current initial events during setup and genuine current failures after setup', async () => { + const { first, onEvents, onTerminalError, controller, finish } = await startInitialWatch() + first.emitChange() + expect(onEvents).toHaveBeenCalledOnce() + await finish() + first.emitChange() + expect(onEvents).toHaveBeenCalledTimes(2) + first.failWatch('current initial failure') + expect(onTerminalError).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ message: 'current initial failure' }) + ) + expect(controller.signal.aborted).toBe(true) + }) + + it('keeps a completed initial watch armed after typed connection loss', async () => { + const { watch, onEvents, onTerminalError, controller, finish } = await startInitialWatch() + await finish() + watch.mock.calls[0]?.[2]?.onTerminalError?.(createSshDisposalError('connection_lost')) + expect(onTerminalError).not.toHaveBeenCalled() + expect(controller.signal.aborted).toBe(false) + const next = connect() + getSshFilesystemProviderMock.mockReturnValue(next.provider) + emitProviderRegistered(CONNECTION_ID) + await vi.waitFor(() => expect(next.countRequests('fs.watch')).toBe(1)) + next.settleWatch() + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledExactlyOnceWith(OVERFLOW_EVENTS)) + }) + + it.each(['connection loss', 'cancellation'] as const)( + 'rejects initial setup on %s and fences callbacks after setup ends', + async (reason) => { + const { first, watch, onEvents, onTerminalError, controller, setup } = + await startInitialWatch() + if (reason === 'connection loss') { + first.mux.dispose('connection_lost') + await expect(setup).rejects.toMatchObject({ code: 'CONNECTION_LOST' }) + } else { + controller.abort() + await expect(setup).rejects.toMatchObject({ name: 'AbortError' }) + } + watch.mock.calls[0]?.[1]([{ kind: 'update', absolutePath: `${ROOT_PATH}/late.ts` }]) + watch.mock.calls[0]?.[2]?.onTerminalError?.(new Error('late setup failure')) + expect(onEvents).not.toHaveBeenCalled() + expect(onTerminalError).not.toHaveBeenCalled() + expect(providerRegistrationListeners.size).toBe(0) + } + ) + it('reinstalls and resyncs when the connection re-registers its provider', async () => { // Why: dispose() on transport loss stops the registration without firing onTerminalError, so // nothing else tells this watch it died. const firstUnwatch = vi.fn() const secondUnwatch = vi.fn() - const watch = vi.fn().mockResolvedValueOnce(firstUnwatch).mockResolvedValueOnce(secondUnwatch) - getSshFilesystemProviderMock.mockReturnValue({ watch }) + const firstProvider = { watch: vi.fn().mockResolvedValue(firstUnwatch) } + const secondProvider = { watch: vi.fn().mockResolvedValue(secondUnwatch) } + getSshFilesystemProviderMock.mockReturnValue(firstProvider) const commands = createRuntimeFileCommands() const onEvents = vi.fn() - await commands.watchFileExplorer('id:wt-1', onEvents) - expect(watch).toHaveBeenCalledTimes(1) + const unsubscribe = await commands.watchFileExplorer('id:wt-1', onEvents) + expect(firstProvider.watch).toHaveBeenCalledOnce() + getSshFilesystemProviderMock.mockReturnValue(secondProvider) emitProviderRegistered(CONNECTION_ID) - await vi.waitFor(() => expect(watch).toHaveBeenCalledTimes(2)) + await vi.waitFor(() => expect(secondProvider.watch).toHaveBeenCalledOnce()) expect(onEvents).toHaveBeenCalledWith(OVERFLOW_EVENTS) // The dead transport's handle must not be closed against the fresh registration. expect(firstUnwatch).not.toHaveBeenCalled() + await unsubscribe() + expect(secondUnwatch).toHaveBeenCalledOnce() + expect(firstUnwatch).not.toHaveBeenCalled() }) it('ignores registrations for other connections', async () => { diff --git a/src/main/runtime/orca-runtime-files-watch-host-scope.test.ts b/src/main/runtime/orca-runtime-files-watch-host-scope.test.ts index 0cf0f19acbb..e9ebb7ae124 100644 --- a/src/main/runtime/orca-runtime-files-watch-host-scope.test.ts +++ b/src/main/runtime/orca-runtime-files-watch-host-scope.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it, vi } from 'vitest' import { EventEmitter } from 'node:events' +import type { FsChangeEvent } from '../../shared/filesystem-entry-types' import { + getSshFilesystemProviderMock, resolveAuthorizedPathMock, statMock, watchInWatcherProcessMock, @@ -102,7 +104,9 @@ describe('RuntimeFileCommands', () => { it('keeps SSH runtime watches on the remote filesystem provider', async () => { const remoteDispose = vi.fn() - const providerWatch = vi.fn(() => remoteDispose) + const providerWatch = vi.fn( + (_rootPath: string, _callback: (events: FsChangeEvent[]) => void) => remoteDispose + ) vi.mocked(getSshFilesystemProvider).mockReturnValue({ watch: providerWatch } as never) const { commands, store } = createRuntimeFileCommands({ path: '/remote/repo' }) store.getRepo.mockReturnValue({ connectionId: 'ssh-1' }) @@ -110,10 +114,15 @@ describe('RuntimeFileCommands', () => { const unsubscribe = await commands.watchFileExplorer('id:wt-1', onEvents) - expect(providerWatch).toHaveBeenCalledWith('/remote/repo', onEvents, { + expect(providerWatch).toHaveBeenCalledWith('/remote/repo', expect.any(Function), { signal: undefined, onTerminalError: expect.any(Function) }) + const callback = providerWatch.mock.calls[0]?.[1] + callback?.([{ kind: 'update', absolutePath: '/remote/repo/current.ts' }]) + expect(onEvents).toHaveBeenCalledWith([ + { kind: 'update', absolutePath: '/remote/repo/current.ts' } + ]) expect(watchInWatcherProcessMock).not.toHaveBeenCalled() await unsubscribe() expect(remoteDispose).toHaveBeenCalledTimes(1) @@ -149,11 +158,10 @@ describe('RuntimeFileCommands', () => { it('scopes same-path runtime watcher teardown to its SSH execution host', async () => { const firstDispose = vi.fn() const secondDispose = vi.fn() - vi.mocked(getSshFilesystemProvider).mockImplementation( - (connectionId) => - ({ - watch: vi.fn(() => (connectionId === 'ssh-1' ? firstDispose : secondDispose)) - }) as never + const firstProvider = { watch: vi.fn(() => firstDispose) } + const secondProvider = { watch: vi.fn(() => secondDispose) } + getSshFilesystemProviderMock.mockImplementation((connectionId) => + connectionId === 'ssh-1' ? firstProvider : secondProvider ) const first = createRuntimeFileCommands({ path: '/same/repo' }) const second = createRuntimeFileCommands({ path: '/same/repo' }) diff --git a/src/main/runtime/orca-runtime-files-watch.test.ts b/src/main/runtime/orca-runtime-files-watch.test.ts index 7de40e51df5..9fa150ba9a7 100644 --- a/src/main/runtime/orca-runtime-files-watch.test.ts +++ b/src/main/runtime/orca-runtime-files-watch.test.ts @@ -4,6 +4,7 @@ import type * as Fs from 'node:fs' import type * as FsPromises from 'node:fs/promises' import type * as FilesystemAuth from '../ipc/filesystem-auth' import type { FsChangeEvent } from '../../shared/filesystem-entry-types' +import type { IFilesystemProvider } from '../providers/types' import { WatcherProcessFailure } from '../ipc/parcel-watcher-process-failure' import { acquireWatcherRemovalGate } from '../ipc/watcher-removal-gate' @@ -543,38 +544,62 @@ describe('RuntimeFileCommands file watching', () => { expect(drained).toBe(true) }) - it('forwards the abort signal into SSH-backed file explorer watches', async () => { - const watch = vi.fn(async () => () => {}) - getSshFilesystemProviderMock.mockReturnValue({ watch }) - const store = { getRepo: vi.fn(() => ({ connectionId: 'ssh-1' })) } - const commands = new RuntimeFileCommands({ - getRuntimeId: () => 'runtime-1', - requireStore: () => store, - resolveWorktreeSelector: vi.fn(async () => ({ - id: 'wt-1', - repoId: 'repo-1', - path: '/remote/repo' - })), - resolveRuntimeFileTarget: vi.fn(async () => ({ - worktree: { + it.each(['abort', 'unsubscribe'] as const)( + 'forwards the SSH watch signal and fences terminal errors after %s', + async (stop) => { + const dispose = vi.fn() + const watch = vi.fn().mockResolvedValue(dispose) + getSshFilesystemProviderMock.mockReturnValue({ watch }) + const store = { getRepo: vi.fn(() => ({ connectionId: 'ssh-1' })) } + const commands = new RuntimeFileCommands({ + getRuntimeId: () => 'runtime-1', + requireStore: () => store, + resolveWorktreeSelector: vi.fn(async () => ({ id: 'wt-1', repoId: 'repo-1', path: '/remote/repo' - }, - executionHostId: 'ssh:ssh-1' - })), - resolveRuntimeGitTarget: vi.fn(), - openFile: vi.fn() - } as never) - const controller = new AbortController() - const onTerminalError = vi.fn() + })), + resolveRuntimeFileTarget: vi.fn(async () => ({ + worktree: { + id: 'wt-1', + repoId: 'repo-1', + path: '/remote/repo' + }, + executionHostId: 'ssh:ssh-1' + })), + resolveRuntimeGitTarget: vi.fn(), + openFile: vi.fn() + } as never) + const controller = new AbortController() + const onTerminalError = vi.fn() - await commands.watchFileExplorer('id:wt-1', vi.fn(), onTerminalError, controller.signal) + const unsubscribe = await commands.watchFileExplorer( + 'id:wt-1', + vi.fn(), + onTerminalError, + controller.signal + ) - expect(watch).toHaveBeenCalledWith('/remote/repo', expect.any(Function), { - signal: controller.signal, - onTerminalError - }) - expect(watchInWatcherProcessMock).not.toHaveBeenCalled() - }) + expect(watch).toHaveBeenCalledWith('/remote/repo', expect.any(Function), { + signal: controller.signal, + onTerminalError: expect.any(Function) + }) + const options = watch.mock.calls[0]?.[2] + expect(options?.signal).toBe(controller.signal) + const currentError = new Error('current remote watch failed') + options?.onTerminalError?.(currentError) + expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(currentError) + if (stop === 'abort') { + controller.abort() + } else { + await unsubscribe() + expect(controller.signal.aborted).toBe(false) + } + options?.onTerminalError?.(new Error('late remote watch failure')) + expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(currentError) + await unsubscribe() + expect(dispose).toHaveBeenCalledOnce() + expect(watchInWatcherProcessMock).not.toHaveBeenCalled() + } + ) }) diff --git a/src/main/runtime/runtime-file-commands-assert-remote-terminal-file-grant-path-still-canonical.ts b/src/main/runtime/runtime-file-commands-assert-remote-terminal-file-grant-path-still-canonical.ts index 9a6af2a82b4..c4d7231ddfe 100644 --- a/src/main/runtime/runtime-file-commands-assert-remote-terminal-file-grant-path-still-canonical.ts +++ b/src/main/runtime/runtime-file-commands-assert-remote-terminal-file-grant-path-still-canonical.ts @@ -100,21 +100,47 @@ export class RuntimeFileCommandsWithAssertRemoteTerminalFileGrantPathStillCanoni if (!route.provider) { throw new Error(SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE) } - // Why: the RPC layer already threads AbortSignal for local watches; SSH must cancel the remote fs.watch, not wait it out. - const close = await route.provider.watch(target.path, callback, { - signal, - onTerminalError - }) - const rearm = armSshFileExplorerWatchRearm({ - runtimeId: this.host.getRuntimeId(), - connectionId: route.connectionId, - rootPath: target.path, - callback, - onTerminalError, - signal, - initialUnwatch: close - }) - return { unsubscribe: rearm.unsubscribe, rootPaths: [target.path] } + const provider = route.provider + const isCurrentInitialProvider = (): boolean => + !signal?.aborted && provider === getSshFilesystemProvider(route.connectionId) + let initialCallbacks = { + callback: (events: FsChangeEvent[]) => { + if (isCurrentInitialProvider()) { + callback(events) + } + }, + onTerminalError: (error: Error) => { + if (isCurrentInitialProvider()) { + onTerminalError(error) + } + } + } + try { + // Initial callbacks join the rearm generation after setup succeeds. + const close = await provider.watch( + target.path, + (events) => initialCallbacks.callback(events), + { signal, onTerminalError: (error) => initialCallbacks.onTerminalError(error) } + ) + const rearm = armSshFileExplorerWatchRearm({ + runtimeId: this.host.getRuntimeId(), + connectionId: route.connectionId, + rootPath: target.path, + callback, + onTerminalError, + signal, + initialUnwatch: close, + initialProvider: provider + }) + initialCallbacks = rearm.initialCallbacks + return { unsubscribe: rearm.unsubscribe, rootPaths: [target.path] } + } catch (error) { + initialCallbacks = { + callback: () => undefined, + onTerminalError: () => undefined + } + throw error + } } const rootPath = await resolveAuthorizedPath(target.path, this.host.requireStore()) diff --git a/src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.test.ts b/src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.test.ts new file mode 100644 index 00000000000..9f2650af743 --- /dev/null +++ b/src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.test.ts @@ -0,0 +1,421 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { IFilesystemProvider } from '../providers/types' +import { + createSshDisposalError, + SSH_MUX_REQUEST_TIMEOUT_CODE +} from '../ssh/ssh-channel-multiplexer' +import { createSshFileExplorerWatchTestConnection as createWatchConnection } from './ssh-file-explorer-watch-test-connection' +import { armSshFileExplorerWatchRearm } from './runtime-file-commands-ssh-file-watcher-rearm' + +type WatchProvider = Pick + +const { getProvider, registrationListeners, rearms } = vi.hoisted(() => ({ + getProvider: vi.fn<(connectionId: string) => WatchProvider | undefined>(), + registrationListeners: new Set<(connectionId: string) => void>(), + rearms: new Map void>>() +})) + +vi.mock('../providers/ssh-filesystem-dispatch', () => ({ + getSshFilesystemProvider: getProvider, + onSshFilesystemProviderRegistered: (listener: (connectionId: string) => void) => { + registrationListeners.add(listener) + return () => registrationListeners.delete(listener) + } +})) + +vi.mock('./runtime-file-commands-mobile-file-list-limit', () => ({ + runtimeWatcherReleaseKey: (runtimeId: string, connectionId: string, rootPath: string) => + `${runtimeId}:${connectionId}:${rootPath}`, + sshFileExplorerWatchRearms: rearms +})) + +function registerProvider(provider: WatchProvider): void { + getProvider.mockReturnValue(provider) + for (const listener of registrationListeners) { + listener('ssh-1') + } +} + +function pendingWatch() { + let resolve: (unwatch: () => void) => void = () => undefined + let reject: (error: Error) => void = () => undefined + const promise = new Promise<() => void>((resolvePromise, rejectPromise) => { + resolve = resolvePromise + reject = rejectPromise + }) + return { watch: vi.fn(() => promise), resolve, reject } +} + +describe('SSH file explorer watcher rearm', () => { + let unsubscribe: (() => Promise) | undefined + const connections: ReturnType[] = [] + + beforeEach(() => { + getProvider.mockReset() + registrationListeners.clear() + rearms.clear() + }) + + afterEach(async () => { + for (const { mux } of connections) { + mux.dispose() + } + await unsubscribe?.() + unsubscribe = undefined + for (const { provider } of connections.splice(0)) { + provider.dispose() + } + expect(registrationListeners.size).toBe(0) + expect(rearms.size).toBe(0) + }) + + function install( + initialProvider: WatchProvider = { watch: vi.fn() }, + initialUnwatch: () => void = vi.fn() + ) { + if (!getProvider.getMockImplementation()) { + getProvider.mockReturnValue(initialProvider) + } + const onEvents = vi.fn() + const controller = new AbortController() + const onTerminalError = vi.fn((_error: Error) => controller.abort()) + const rearm = armSshFileExplorerWatchRearm({ + runtimeId: 'runtime-1', + connectionId: 'ssh-1', + rootPath: '/remote/repo', + callback: onEvents, + onTerminalError, + signal: controller.signal, + initialUnwatch, + initialProvider + }) + unsubscribe = rearm.unsubscribe + return { onEvents, onTerminalError, initialUnwatch, controller } + } + + it('ignores a superseded setup failure and installs the current provider', async () => { + const { onEvents, onTerminalError, controller } = install() + const old = pendingWatch() + registerProvider(old) + await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce()) + + const currentUnwatch = vi.fn() + const current = { watch: vi.fn(async () => currentUnwatch) } + registerProvider(current) + old.reject(new Error('previous transport disconnected')) + + await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce()) + expect(onTerminalError).not.toHaveBeenCalled() + expect(controller.signal.aborted).toBe(false) + expect(onEvents).toHaveBeenCalledExactlyOnceWith([ + { kind: 'overflow', absolutePath: '/remote/repo' } + ]) + await unsubscribe?.() + expect(currentUnwatch).toHaveBeenCalledOnce() + }) + + it('replaces an initial watch whose provider changed before rearm was installed', async () => { + const original = pendingWatch() + const initialUnwatch = vi.fn() + const initialSetup = original.watch('/remote/repo', vi.fn()) + const currentUnwatch = vi.fn() + const current = { watch: vi.fn(async () => currentUnwatch) } + registerProvider(current) + original.resolve(initialUnwatch) + await initialSetup + const { onEvents, onTerminalError } = install(original, initialUnwatch) + + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce()) + expect(current.watch).toHaveBeenCalledOnce() + expect(initialUnwatch).not.toHaveBeenCalled() + expect(onTerminalError).not.toHaveBeenCalled() + await unsubscribe?.() + expect(currentUnwatch).toHaveBeenCalledOnce() + expect(initialUnwatch).not.toHaveBeenCalled() + }) + + it('coalesces registrations that arrive before replacement setup starts', async () => { + const { onEvents } = install() + const old = { watch: vi.fn(async () => vi.fn()) } + const current = { watch: vi.fn(async () => vi.fn()) } + registerProvider(old) + registerProvider(current) + + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce()) + expect(old.watch).not.toHaveBeenCalled() + expect(current.watch).toHaveBeenCalledOnce() + }) + + it('closes a superseded successful setup without publishing its refresh', async () => { + const { onEvents, onTerminalError } = install() + const old = pendingWatch() + registerProvider(old) + await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce()) + const currentUnwatch = vi.fn() + const current = { watch: vi.fn(async () => currentUnwatch) } + registerProvider(current) + const obsoleteUnwatch = vi.fn() + old.resolve(obsoleteUnwatch) + + await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce()) + expect(obsoleteUnwatch).toHaveBeenCalledOnce() + expect(onTerminalError).not.toHaveBeenCalled() + expect(onEvents).toHaveBeenCalledOnce() + await unsubscribe?.() + expect(currentUnwatch).toHaveBeenCalledOnce() + }) + + it('reports a genuine current-provider setup failure', async () => { + const { onTerminalError, onEvents } = install() + const error = new Error('current host refused watch') + registerProvider({ + watch: vi.fn(async () => { + throw error + }) + }) + + await vi.waitFor(() => expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(error)) + expect(onEvents).not.toHaveBeenCalled() + }) + + it.each(['setup', 'terminal callback'] as const)( + 'waits for registration after a current-provider %s loses its connection', + async (stage) => { + const { onTerminalError, onEvents, controller } = install() + const lost = pendingWatch() + registerProvider(lost) + await vi.waitFor(() => expect(lost.watch).toHaveBeenCalledOnce()) + const error = createSshDisposalError('connection_lost') + if (stage === 'setup') { + lost.reject(error) + } else { + lost.resolve(vi.fn()) + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce()) + lost.watch.mock.calls[0]?.[2]?.onTerminalError?.(error) + } + await new Promise((resolve) => setImmediate(resolve)) + expect(onTerminalError).not.toHaveBeenCalled() + expect(controller.signal.aborted).toBe(false) + + const currentUnwatch = vi.fn() + const current = { watch: vi.fn(async () => currentUnwatch) } + registerProvider(current) + await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce()) + expect(onEvents).toHaveBeenLastCalledWith([ + { kind: 'overflow', absolutePath: '/remote/repo' } + ]) + await unsubscribe?.() + expect(currentUnwatch).toHaveBeenCalledOnce() + } + ) + + it.each([ + ['shutdown', createSshDisposalError('shutdown')], + [ + 'timeout', + Object.assign(new Error('watch timed out'), { code: SSH_MUX_REQUEST_TIMEOUT_CODE }) + ], + ['uncoded connection message', new Error('CONNECTION_LOST')] + ])('reports a current-provider %s setup failure', async (_label, error) => { + const { onTerminalError } = install() + registerProvider({ + watch: vi.fn(async () => { + throw error + }) + }) + await vi.waitFor(() => expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(error)) + }) + + it.each(['active', 'canceled'] as const)( + 'keeps an %s owner correct through actual multiplexer loss and later registration', + async (owner) => { + const connect = () => { + const connection = createWatchConnection() + connections.push(connection) + return connection + } + const first = connect() + getProvider.mockReturnValue(first.provider) + const initialSetup = first.provider.watch('/remote/repo', vi.fn()) + first.settleWatch() + const { onEvents, onTerminalError, controller } = install(first.provider, await initialSetup) + first.mux.dispose('connection_lost') + first.provider.dispose() + const next = connect() + registerProvider(next.provider) + await vi.waitFor(() => expect(next.countRequests('fs.watch')).toBe(1)) + next.mux.dispose('connection_lost') + await new Promise((resolve) => setImmediate(resolve)) + expect(getProvider('ssh-1')).toBe(next.provider) + expect(onTerminalError).not.toHaveBeenCalled() + expect(controller.signal.aborted).toBe(false) + expect(onEvents).not.toHaveBeenCalled() + if (owner === 'canceled') { + await unsubscribe?.() + } + + next.provider.dispose() + const current = connect() + registerProvider(current.provider) + if (owner === 'canceled') { + await new Promise((resolve) => setImmediate(resolve)) + expect(current.countRequests('fs.watch')).toBe(0) + return + } + await vi.waitFor(() => expect(current.countRequests('fs.watch')).toBe(1)) + current.settleWatch() + await vi.waitFor(() => + expect(onEvents).toHaveBeenCalledExactlyOnceWith([ + { kind: 'overflow', absolutePath: '/remote/repo' } + ]) + ) + current.emitChange() + expect(onEvents).toHaveBeenLastCalledWith([ + { kind: 'update', absolutePath: '/remote/repo/current.ts' } + ]) + await unsubscribe?.() + expect(current.countRequests('fs.unwatch')).toBe(1) + current.emitChange() + expect(onEvents).toHaveBeenCalledTimes(2) + } + ) + + it.each(['unregistered', 'replaced'] as const)( + 'ignores a setup failure when its provider is %s between rejection handlers', + async (change) => { + const { onTerminalError, onEvents, controller } = install() + const old = pendingWatch() + registerProvider(old) + await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce()) + const current = { watch: vi.fn(async () => vi.fn()) } + const checkedWhileCurrent = vi.fn(() => { + queueMicrotask(() => { + if (change === 'replaced') { + registerProvider(current) + } else { + getProvider.mockReturnValue(undefined) + } + }) + return old + }) + getProvider.mockImplementationOnce(checkedWhileCurrent) + old.reject(new Error('previous setup failed')) + await new Promise((resolve) => setImmediate(resolve)) + + expect(checkedWhileCurrent).toHaveBeenCalledOnce() + expect(onTerminalError).not.toHaveBeenCalled() + expect(controller.signal.aborted).toBe(false) + if (change === 'replaced') { + expect(current.watch).toHaveBeenCalledOnce() + expect(onEvents).toHaveBeenCalledExactlyOnceWith([ + { kind: 'overflow', absolutePath: '/remote/repo' } + ]) + } else { + expect(onEvents).not.toHaveBeenCalled() + } + } + ) + + it('ignores late terminal callbacks from a superseded provider', async () => { + const { onTerminalError, controller } = install() + const old = pendingWatch() + registerProvider(old) + await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce()) + const current = { watch: vi.fn(async () => vi.fn()) } + registerProvider(current) + old.watch.mock.calls[0]?.[2]?.onTerminalError?.(new Error('previous watcher stopped')) + old.resolve(vi.fn()) + + await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce()) + expect(onTerminalError).not.toHaveBeenCalled() + expect(controller.signal.aborted).toBe(false) + }) + + it('reports a terminal callback from the current provider', async () => { + const { onTerminalError, onEvents } = install() + const current = pendingWatch() + registerProvider(current) + await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce()) + current.resolve(vi.fn()) + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce()) + const error = new Error('current watcher stopped') + current.watch.mock.calls[0]?.[2]?.onTerminalError?.(error) + + expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(error) + }) + + it('ignores late events from a superseded provider and delivers current events', async () => { + const { onEvents } = install() + const old = pendingWatch() + registerProvider(old) + await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce()) + const current = pendingWatch() + registerProvider(current) + old.watch.mock.calls[0]?.[1]([{ kind: 'update', absolutePath: '/remote/repo/old.ts' }]) + old.resolve(vi.fn()) + await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce()) + current.resolve(vi.fn()) + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce()) + const currentEvents = [{ kind: 'update' as const, absolutePath: '/remote/repo/current.ts' }] + current.watch.mock.calls[0]?.[1](currentEvents) + + expect(onEvents).toHaveBeenCalledTimes(2) + expect(onEvents).toHaveBeenLastCalledWith(currentEvents) + }) + + it('releases the previous subscriber when the same provider is registered again', async () => { + const { onEvents } = install() + const callbacks = new Set[1]>() + const provider = { + watch: vi.fn(async (_root, callback) => { + callbacks.add(callback) + return () => { + callbacks.delete(callback) + } + }) + } + registerProvider(provider) + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce()) + registerProvider(provider) + await vi.waitFor(() => expect(onEvents).toHaveBeenCalledTimes(2)) + + expect(callbacks.size).toBe(1) + const events = [{ kind: 'update' as const, absolutePath: '/remote/repo/current.ts' }] + for (const callback of callbacks) { + callback(events) + } + expect(onEvents).toHaveBeenCalledTimes(3) + expect(onEvents).toHaveBeenLastCalledWith(events) + await unsubscribe?.() + expect(callbacks.size).toBe(0) + }) + + it('does not report a setup failure after unsubscribe starts', async () => { + const { onTerminalError, onEvents, initialUnwatch } = install() + const pending = pendingWatch() + registerProvider(pending) + await vi.waitFor(() => expect(pending.watch).toHaveBeenCalledOnce()) + const closed = unsubscribe?.() + pending.reject(new Error('setup canceled during shutdown')) + + await closed + expect(initialUnwatch).toHaveBeenCalledOnce() + expect(onTerminalError).not.toHaveBeenCalled() + expect(onEvents).not.toHaveBeenCalled() + }) + + it('closes a replacement that finishes after unsubscribe starts', async () => { + const { onTerminalError, onEvents } = install() + const pending = pendingWatch() + registerProvider(pending) + await vi.waitFor(() => expect(pending.watch).toHaveBeenCalledOnce()) + const closed = unsubscribe?.() + const lateUnwatch = vi.fn() + pending.resolve(lateUnwatch) + + await closed + expect(lateUnwatch).toHaveBeenCalledOnce() + expect(onTerminalError).not.toHaveBeenCalled() + expect(onEvents).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.ts b/src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.ts index 3e74fc0e61d..0ff89703493 100644 --- a/src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.ts +++ b/src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.ts @@ -1,5 +1,6 @@ -// @ts-nocheck -- mechanically split declarations. import type { FsChangeEvent } from '../../shared/filesystem-entry-types' +import type { IFilesystemProvider } from '../providers/types' +import { toRemoteRuntimeClientErrorLike } from '../../shared/remote-runtime-client-error-classification' import { runtimeWatcherReleaseKey, sshFileExplorerWatchRearms @@ -17,40 +18,99 @@ export function armSshFileExplorerWatchRearm(args: { onTerminalError: (error: Error) => void signal?: AbortSignal initialUnwatch: () => void -}): { unsubscribe: () => Promise } { + initialProvider: Pick +}): { + initialCallbacks: { + callback: (events: FsChangeEvent[]) => void + onTerminalError: (error: Error) => void + } + unsubscribe: () => Promise +} { const key = runtimeWatcherReleaseKey(args.runtimeId, args.connectionId, args.rootPath) let currentUnwatch = args.initialUnwatch + let currentProvider = args.initialProvider let stopped = false let reinstalling: Promise | null = null + let providerGeneration = 0 - const reinstall = async (): Promise => { - const provider = getSshFilesystemProvider(args.connectionId) - if (stopped || !provider) { + const reportTerminalError = (error: unknown): void => { + // Connection loss leaves the established stream armed for the next provider. + if (toRemoteRuntimeClientErrorLike(error).code === 'CONNECTION_LOST') { return } - // Why: the old handle is scoped to the dead transport; closing it here would only risk - // unwatching the root we just re-registered on the new one. - const nextUnwatch = await provider.watch(args.rootPath, args.callback, { - signal: args.signal, - onTerminalError: args.onTerminalError - }) - if (stopped) { + args.onTerminalError(error instanceof Error ? error : new Error(String(error))) + } + + const reinstall = async ( + generation: number, + provider: Pick | undefined + ): Promise => { + if (stopped || generation !== providerGeneration || !provider) { + return + } + const isCurrent = (): boolean => + !stopped && + !args.signal?.aborted && + generation === providerGeneration && + provider === getSshFilesystemProvider(args.connectionId) + let nextUnwatch: () => void + try { + nextUnwatch = await provider.watch( + args.rootPath, + (events) => { + if (isCurrent()) { + args.callback(events) + } + }, + { + signal: args.signal, + onTerminalError: (error) => { + if (isCurrent()) { + reportTerminalError(error) + } + } + } + ) + } catch (error) { + if (!isCurrent()) { + return + } + throw error + } + if (!isCurrent()) { nextUnwatch() return } + // Why: a dead transport's handle must never unwatch the replacement transport's root. + if (currentProvider === provider) { + currentUnwatch() + } currentUnwatch = nextUnwatch + currentProvider = provider args.callback([{ kind: 'overflow', absolutePath: args.rootPath }]) } - const unsubscribeRearm = onSshFilesystemProviderRegistered((registeredId) => { - if (registeredId !== args.connectionId || stopped) { + const scheduleReinstall = (): void => { + if (stopped) { return } - // Why: reconnect storms can register repeatedly; chain so a second one can't double-install. + const generation = ++providerGeneration + let attemptProvider: Pick | undefined + // Why: obsolete reconnect attempts cannot terminate the stream or publish a stale refresh. const attempt = (reinstalling ?? Promise.resolve()) - .then(reinstall) + .then(() => { + attemptProvider = getSshFilesystemProvider(args.connectionId) + return reinstall(generation, attemptProvider) + }) .catch((error: unknown) => { - args.onTerminalError(error instanceof Error ? error : new Error(String(error))) + if ( + !stopped && + !args.signal?.aborted && + generation === providerGeneration && + attemptProvider === getSshFilesystemProvider(args.connectionId) + ) { + reportTerminalError(error) + } }) .finally(() => { if (reinstalling === attempt) { @@ -58,6 +118,11 @@ export function armSshFileExplorerWatchRearm(args: { } }) reinstalling = attempt + } + const unsubscribeRearm = onSshFilesystemProviderRegistered((registeredId) => { + if (registeredId === args.connectionId) { + scheduleReinstall() + } }) const stop = (): void => { @@ -72,8 +137,28 @@ export function armSshFileExplorerWatchRearm(args: { const rearms = sshFileExplorerWatchRearms.get(key) ?? new Set<() => void>() rearms.add(stop) sshFileExplorerWatchRearms.set(key, rearms) + if (getSshFilesystemProvider(args.connectionId) !== args.initialProvider) { + scheduleReinstall() + } + const isInitialCurrent = (): boolean => + !stopped && + !args.signal?.aborted && + providerGeneration === 0 && + args.initialProvider === getSshFilesystemProvider(args.connectionId) return { + initialCallbacks: { + callback: (events) => { + if (isInitialCurrent()) { + args.callback(events) + } + }, + onTerminalError: (error) => { + if (isInitialCurrent()) { + reportTerminalError(error) + } + } + }, unsubscribe: () => { stop() const close = async (): Promise => currentUnwatch() diff --git a/src/main/runtime/ssh-file-explorer-watch-test-connection.ts b/src/main/runtime/ssh-file-explorer-watch-test-connection.ts new file mode 100644 index 00000000000..9521c175a8c --- /dev/null +++ b/src/main/runtime/ssh-file-explorer-watch-test-connection.ts @@ -0,0 +1,65 @@ +import { SshFilesystemProvider } from '../providers/ssh-filesystem-provider' +import { SshChannelMultiplexer, type MultiplexerTransport } from '../ssh/ssh-channel-multiplexer' +import { + encodeJsonRpcFrame, + HEADER_LENGTH, + MessageType, + parseJsonRpcMessage, + type JsonRpcMessage +} from '../ssh/relay-protocol' + +export function createSshFileExplorerWatchTestConnection( + rootPath = '/remote/repo', + connectionId = 'ssh-1' +) { + const written: Buffer[] = [] + let receive: (data: Buffer) => void = () => undefined + let sequence = 1 + const transport: MultiplexerTransport = { + write: (data) => { + written.push(data) + }, + onData: (callback) => { + receive = callback + }, + onClose: () => undefined + } + const mux = new SshChannelMultiplexer(transport) + const provider = new SshFilesystemProvider(connectionId, mux) + const messages = (): JsonRpcMessage[] => + written + .filter((frame) => frame[0] === MessageType.Regular) + .map((frame) => parseJsonRpcMessage(frame.subarray(HEADER_LENGTH))) + const send = (message: JsonRpcMessage): void => { + receive(encodeJsonRpcFrame(message, sequence++, 0)) + } + const watchRequest = () => { + const request = messages().find( + (message) => 'method' in message && message.method === 'fs.watch' + ) + if (!request || !('id' in request) || !('method' in request)) { + throw new Error('No pending fs.watch request') + } + return request + } + return { + mux, + provider, + countRequests: (method: string) => + messages().filter((message) => 'method' in message && message.method === method).length, + settleWatch: () => send({ jsonrpc: '2.0', id: watchRequest().id, result: null }), + emitChange: (absolutePath = `${rootPath}/current.ts`) => + send({ + jsonrpc: '2.0', + method: 'fs.changed', + params: { events: [{ kind: 'update', absolutePath }] } + }), + failWatch: (message: string) => { + const watchId = watchRequest().params?.watchId + if (typeof watchId !== 'number') { + throw new Error('No remote watch id') + } + send({ jsonrpc: '2.0', method: 'fs.watchFailed', params: { rootPath, watchId, message } }) + } + } +}