mirror of
https://github.com/stablyai/orca.git
synced 2026-09-22 00:02:31 +00:00
fix(ssh): re-arm remote file watches when the provider reconnects (#10445)
* fix(ssh): re-arm remote file watches when the provider reconnects Remote file changes stopped being detected over SSH until the file was reopened. The watch pipeline itself was fine — nothing ever re-established the subscription after the transport it was made on went away. Two paths left an editor tab permanently stale: - A reconnect kills the relay's watch registrations, and the previous provider's unwatch handle belongs to the dead transport. - A connect slower than the 60s retry window made installRemoteWatcher give up for good; first deploy to a new host far exceeds that. Neither recovered, because installRemoteWatcher is only reachable from the fs:watchWorktree handler, the retry timer, and the removal-restore path, and the renderer only issues a watch for newly added targets. Reopening the file just re-read it — the watcher stayed dead. Give the layer that owns the transport the job of re-arming: registerSshFilesystemProvider now notifies subscribers, which covers both establish and reconnect since registerProviders runs on both. The watcher keeps the intent to watch in a registry that outlives any single connection, and on registration drops the stale entry (installRemoteWatcher treats an existing entry as installed and would otherwise hand back a watcher that can never fire), reinstalls, and emits overflow so consumers resync the gap. Intent is dropped on unwatch and sender destroy so a closed tab is never resurrected. Verified over SSH to a Rocky Linux 10 host: after the reconnect that previously killed it, a remote append lands in the editor in ~2s with a live remote watcher process, and a second edit in ~1.5s. * fix(ssh): drop watch intent when a remote worktree is removed A removed worktree kept its entry in the intent registry, so a reconnect landing before the renderer's unwatch would re-watch a deleted path — 60s of retries against the host and then a bogus overflow. Also covers two reinstall cases: several senders on one connection must collapse onto a single relay watch (and all of them resync), and a destroyed renderer must not be reinstalled. * fix(ssh): resync when a reconnect's watch only lands on a retry The reinstall emitted the overflow only for listeners whose first install returned 'installed'. If that attempt failed — relay-watcher.js still spawning on the fresh transport, a transient fs.watch rejection — the 1s retry restored the watch but never signalled the gap, so everything that changed while the transport was down stayed invisible: the STA-2525 symptom reappearing inside the fix for it. Thread the resync intent through the retry record so the overflow lands when the retry does. All pre-existing callers default to false, so the watch/terminal-error/restore paths are unchanged. * test(ssh): cover the resync merge when a fresh watch claims the retry slot The `resyncOnInstall ||=` merge was uncovered: removing it left every existing test green. It is load-bearing — if a second renderer joins the reinstall's failing install and reaches the retry slot first with resync=false, the whole chain stays false and the retry restores the watch without ever signalling the gap. New sibling file rather than an append: filesystem-watcher.test.ts is ~10 counted lines from the 800-line lint cap, and AGENTS.md forbids a max-lines disable.
This commit is contained in:
@@ -24,7 +24,8 @@ vi.mock('./filesystem-watcher-wsl', () => ({
|
||||
}))
|
||||
|
||||
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
|
||||
getSshFilesystemProvider: vi.fn()
|
||||
getSshFilesystemProvider: vi.fn(),
|
||||
onSshFilesystemProviderRegistered: () => () => {}
|
||||
}))
|
||||
|
||||
import { closeAllWatchers, registerFilesystemWatcherHandlers } from './filesystem-watcher'
|
||||
|
||||
@@ -32,7 +32,8 @@ vi.mock('./parcel-watcher-process', async (importOriginal) => {
|
||||
})
|
||||
|
||||
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
|
||||
getSshFilesystemProvider: vi.fn()
|
||||
getSshFilesystemProvider: vi.fn(),
|
||||
onSshFilesystemProviderRegistered: () => () => {}
|
||||
}))
|
||||
|
||||
import {
|
||||
|
||||
@@ -16,7 +16,8 @@ vi.mock('./parcel-watcher-process', () => ({
|
||||
}))
|
||||
vi.mock('./filesystem-watcher-wsl', () => ({ createWslWatcher: vi.fn() }))
|
||||
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
|
||||
getSshFilesystemProvider: vi.fn()
|
||||
getSshFilesystemProvider: vi.fn(),
|
||||
onSshFilesystemProviderRegistered: () => () => {}
|
||||
}))
|
||||
|
||||
import { closeAllWatchers, registerFilesystemWatcherHandlers } from './filesystem-watcher'
|
||||
|
||||
@@ -13,7 +13,8 @@ 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
|
||||
getSshFilesystemProvider: getSshFilesystemProviderMock,
|
||||
onSshFilesystemProviderRegistered: () => () => {}
|
||||
}))
|
||||
|
||||
import { closeAllWatchers, registerFilesystemWatcherHandlers } from './filesystem-watcher'
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { handleMock, getSshFilesystemProviderMock, providerRegistrationListeners } = vi.hoisted(
|
||||
() => ({
|
||||
handleMock: vi.fn(),
|
||||
getSshFilesystemProviderMock: vi.fn(),
|
||||
providerRegistrationListeners: new Set<(connectionId: string) => void>()
|
||||
})
|
||||
)
|
||||
|
||||
/** Drive the provider-registration hook the way a relay establish/reconnect would. */
|
||||
function emitProviderRegistered(connectionId: string): void {
|
||||
for (const listener of providerRegistrationListeners) {
|
||||
listener(connectionId)
|
||||
}
|
||||
}
|
||||
|
||||
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: (listener: (connectionId: string) => void) => {
|
||||
providerRegistrationListeners.add(listener)
|
||||
return () => providerRegistrationListeners.delete(listener)
|
||||
}
|
||||
}))
|
||||
|
||||
import { closeAllWatchers, registerFilesystemWatcherHandlers } from './filesystem-watcher'
|
||||
|
||||
type HandlerMap = Record<string, (_event: unknown, args: unknown) => Promise<unknown> | unknown>
|
||||
|
||||
describe('remote filesystem watcher re-arm', () => {
|
||||
const handlers: HandlerMap = {}
|
||||
|
||||
beforeEach(async () => {
|
||||
vi.useRealTimers()
|
||||
handleMock.mockReset()
|
||||
getSshFilesystemProviderMock.mockReset()
|
||||
for (const key of Object.keys(handlers)) {
|
||||
delete handlers[key]
|
||||
}
|
||||
handleMock.mockImplementation((channel, handler) => {
|
||||
handlers[channel] = handler
|
||||
})
|
||||
registerFilesystemWatcherHandlers()
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('still resyncs when a fresh watch beat the failed reinstall to the retry slot', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const senderOne = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
const senderTwo = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 }
|
||||
const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: vi.fn().mockResolvedValue(vi.fn()) })
|
||||
|
||||
await handlers['fs:watchWorktree']({ sender: senderOne }, args)
|
||||
|
||||
// Hold the reinstall's fs.watch open so a second renderer joins it and claims the retry slot first.
|
||||
let failReinstall: (error: Error) => void = () => {}
|
||||
const heldWatch = new Promise<never>((_resolve, reject) => {
|
||||
failReinstall = reject
|
||||
})
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: vi.fn().mockReturnValue(heldWatch) })
|
||||
senderOne.send.mockClear()
|
||||
emitProviderRegistered('conn-1')
|
||||
|
||||
const joinedWatch = handlers['fs:watchWorktree']({ sender: senderTwo }, args)
|
||||
const retryWatchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
failReinstall(new Error('relay not ready'))
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: retryWatchMock })
|
||||
await joinedWatch
|
||||
await vi.advanceTimersByTimeAsync(1_000)
|
||||
|
||||
// senderOne's watch really died with the old transport, so its resync must survive the merge.
|
||||
expect(retryWatchMock).toHaveBeenCalledTimes(1)
|
||||
expect(senderOne.send).toHaveBeenCalledWith('fs:changed', {
|
||||
worktreePath: '/home/me/repo',
|
||||
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }]
|
||||
})
|
||||
|
||||
warnSpy.mockRestore()
|
||||
await closeAllWatchers()
|
||||
vi.useRealTimers()
|
||||
})
|
||||
})
|
||||
@@ -23,7 +23,8 @@ vi.mock('./filesystem-watcher-wsl', () => ({
|
||||
}))
|
||||
|
||||
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
|
||||
getSshFilesystemProvider: vi.fn()
|
||||
getSshFilesystemProvider: vi.fn(),
|
||||
onSshFilesystemProviderRegistered: () => () => {}
|
||||
}))
|
||||
|
||||
import { closeAllWatchers, registerFilesystemWatcherHandlers } from './filesystem-watcher'
|
||||
|
||||
@@ -1,9 +1,19 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { handleMock, getSshFilesystemProviderMock } = vi.hoisted(() => ({
|
||||
handleMock: vi.fn(),
|
||||
getSshFilesystemProviderMock: vi.fn()
|
||||
}))
|
||||
const { handleMock, getSshFilesystemProviderMock, providerRegistrationListeners } = vi.hoisted(
|
||||
() => ({
|
||||
handleMock: vi.fn(),
|
||||
getSshFilesystemProviderMock: vi.fn(),
|
||||
providerRegistrationListeners: new Set<(connectionId: string) => void>()
|
||||
})
|
||||
)
|
||||
|
||||
/** Drive the provider-registration hook the way a relay establish/reconnect would. */
|
||||
function emitProviderRegistered(connectionId: string): void {
|
||||
for (const listener of providerRegistrationListeners) {
|
||||
listener(connectionId)
|
||||
}
|
||||
}
|
||||
|
||||
vi.mock('electron', () => ({
|
||||
ipcMain: {
|
||||
@@ -24,12 +34,17 @@ vi.mock('./filesystem-watcher-wsl', () => ({
|
||||
}))
|
||||
|
||||
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
|
||||
getSshFilesystemProvider: getSshFilesystemProviderMock
|
||||
getSshFilesystemProvider: getSshFilesystemProviderMock,
|
||||
onSshFilesystemProviderRegistered: (listener: (connectionId: string) => void) => {
|
||||
providerRegistrationListeners.add(listener)
|
||||
return () => providerRegistrationListeners.delete(listener)
|
||||
}
|
||||
}))
|
||||
|
||||
import {
|
||||
closeAllWatchers,
|
||||
closeRemoteWatcherForWorktreePath,
|
||||
forgetRemoteWatcherRemovalSnapshot,
|
||||
registerFilesystemWatcherHandlers,
|
||||
restoreRemoteWatcherAfterFailedRemoval
|
||||
} from './filesystem-watcher'
|
||||
@@ -327,6 +342,195 @@ describe('registerFilesystemWatcherHandlers', () => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('reinstalls an SSH worktree watch when the provider is re-registered after a reconnect', async () => {
|
||||
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
const staleUnwatch = vi.fn()
|
||||
const watchMock = vi.fn().mockResolvedValue(staleUnwatch)
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock })
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
expect(watchMock).toHaveBeenCalledTimes(1)
|
||||
|
||||
// The reconnect replaces the provider; the watch made on the dead transport can never fire again.
|
||||
emitProviderRegistered('conn-1')
|
||||
await vi.waitFor(() => expect(watchMock).toHaveBeenCalledTimes(2))
|
||||
expect(staleUnwatch).toHaveBeenCalledTimes(1)
|
||||
|
||||
// Events missed while the watch was down are unrecoverable, so consumers are told to resync.
|
||||
await vi.waitFor(() =>
|
||||
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
|
||||
worktreePath: '/home/me/repo',
|
||||
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }]
|
||||
})
|
||||
)
|
||||
|
||||
const reinstalledEvents = watchMock.mock.calls[1][1] as (events: unknown[]) => void
|
||||
reinstalledEvents([{ kind: 'update', absolutePath: '/home/me/repo/file.ts' }])
|
||||
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
|
||||
worktreePath: '/home/me/repo',
|
||||
events: [{ kind: 'update', absolutePath: '/home/me/repo/file.ts' }]
|
||||
})
|
||||
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('re-arms an SSH watch whose first install found no provider yet', async () => {
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
// A connect slower than the retry window leaves the renderer subscribed with nothing installed.
|
||||
getSshFilesystemProviderMock.mockReturnValue(undefined)
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
|
||||
const watchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock })
|
||||
emitProviderRegistered('conn-1')
|
||||
|
||||
await vi.waitFor(() => expect(watchMock).toHaveBeenCalledTimes(1))
|
||||
warnSpy.mockRestore()
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('resyncs after a reconnect whose reinstall only succeeded on a retry', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: vi.fn().mockResolvedValue(vi.fn()) })
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
|
||||
// The relay is back, but its first fs.watch on the fresh transport still fails.
|
||||
const retryWatchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
getSshFilesystemProviderMock
|
||||
.mockReturnValueOnce({ watch: vi.fn().mockRejectedValue(new Error('relay not ready')) })
|
||||
.mockReturnValue({ watch: retryWatchMock })
|
||||
sender.send.mockClear()
|
||||
emitProviderRegistered('conn-1')
|
||||
await vi.advanceTimersByTimeAsync(1_000)
|
||||
|
||||
expect(retryWatchMock).toHaveBeenCalledTimes(1)
|
||||
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
|
||||
worktreePath: '/home/me/repo',
|
||||
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }]
|
||||
})
|
||||
|
||||
warnSpy.mockRestore()
|
||||
await closeAllWatchers()
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('does not resurrect an SSH watch the renderer already unwatched', async () => {
|
||||
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
const watchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock })
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
handlers['fs:unwatchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
|
||||
emitProviderRegistered('conn-1')
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
|
||||
expect(watchMock).toHaveBeenCalledTimes(1)
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('leaves watches on other connections untouched when one provider re-registers', async () => {
|
||||
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
const watchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock })
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
|
||||
emitProviderRegistered('conn-2')
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
|
||||
expect(watchMock).toHaveBeenCalledTimes(1)
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('reinstalls one shared watch when several senders share a re-registered connection', async () => {
|
||||
const senderOne = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
const senderTwo = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 }
|
||||
const watchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock })
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender: senderOne },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender: senderTwo },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
expect(watchMock).toHaveBeenCalledTimes(1)
|
||||
|
||||
// Per-listener reinstall must still collapse onto one relay watch, and every listener resyncs.
|
||||
emitProviderRegistered('conn-1')
|
||||
await vi.waitFor(() => expect(senderTwo.send).toHaveBeenCalled())
|
||||
expect(watchMock).toHaveBeenCalledTimes(2)
|
||||
for (const sender of [senderOne, senderTwo]) {
|
||||
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
|
||||
worktreePath: '/home/me/repo',
|
||||
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }]
|
||||
})
|
||||
}
|
||||
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('does not reinstall an SSH watch for a renderer that was destroyed', async () => {
|
||||
let destroyed = false
|
||||
const destroyHandlers: (() => void)[] = []
|
||||
const sender = {
|
||||
isDestroyed: () => destroyed,
|
||||
send: vi.fn(),
|
||||
once: vi.fn((_event: string, handler: () => void) => {
|
||||
destroyHandlers.push(handler)
|
||||
}),
|
||||
id: 1
|
||||
}
|
||||
const watchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock })
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
expect(destroyHandlers).toHaveLength(1)
|
||||
|
||||
destroyed = true
|
||||
for (const handler of destroyHandlers) {
|
||||
handler()
|
||||
}
|
||||
|
||||
emitProviderRegistered('conn-1')
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
|
||||
expect(watchMock).toHaveBeenCalledTimes(1)
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('shares SSH worktree watchers across renderer senders until the last unwatch', async () => {
|
||||
const sendOne = vi.fn()
|
||||
const sendTwo = vi.fn()
|
||||
@@ -423,6 +627,29 @@ describe('registerFilesystemWatcherHandlers', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('does not re-arm an SSH watch for a worktree that was successfully deleted', async () => {
|
||||
const watchMock = vi.fn().mockResolvedValue(vi.fn())
|
||||
const closeWatch = vi.fn().mockResolvedValue(undefined)
|
||||
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock, closeWatch })
|
||||
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
|
||||
|
||||
await handlers['fs:watchWorktree'](
|
||||
{ sender },
|
||||
{ worktreePath: '/home/me/repo', connectionId: 'conn-1' }
|
||||
)
|
||||
await closeRemoteWatcherForWorktreePath('conn-1', '/home/me/repo')
|
||||
forgetRemoteWatcherRemovalSnapshot('conn-1', '/home/me/repo')
|
||||
|
||||
// A reconnect can land before the renderer's unwatch; the path no longer exists on the host.
|
||||
emitProviderRegistered('conn-1')
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
|
||||
expect(watchMock).toHaveBeenCalledTimes(1)
|
||||
expect(sender.send).not.toHaveBeenCalled()
|
||||
await closeAllWatchers()
|
||||
})
|
||||
|
||||
it('does not restore an SSH listener stopped while deletion is pending', async () => {
|
||||
const firstUnwatch = vi.fn()
|
||||
const watchMock = vi.fn().mockResolvedValue(firstUnwatch)
|
||||
|
||||
@@ -11,7 +11,10 @@ import {
|
||||
import { isWslPath } from '../wsl'
|
||||
import { createWslWatcher } from './filesystem-watcher-wsl'
|
||||
import type { WatchedRoot } from './filesystem-watcher-wsl'
|
||||
import { getSshFilesystemProvider } from '../providers/ssh-filesystem-dispatch'
|
||||
import {
|
||||
getSshFilesystemProvider,
|
||||
onSshFilesystemProviderRegistered
|
||||
} from '../providers/ssh-filesystem-dispatch'
|
||||
import { MAX_BATCHED_WATCHER_EVENTS, queueWatcherEvents } from './filesystem-watcher-event-batch'
|
||||
import { disposeWatcherProcess, subscribeViaWatcherProcess } from './parcel-watcher-process'
|
||||
import { isWatcherProcessFailure } from './parcel-watcher-process-failure'
|
||||
@@ -889,11 +892,19 @@ const suspendedRemoteWatcherListeners = new Map<
|
||||
string,
|
||||
{ connectionId: string; worktreePath: string; listeners: Map<number, WebContents> }
|
||||
>()
|
||||
// Why: the renderer subscribes once per target and never re-issues, so the intent to watch has to
|
||||
// outlive any single connection — an install that failed or died with a dropped transport is
|
||||
// re-armed from here when a provider appears. Without it a reconnect (or a connect slower than the
|
||||
// retry window) leaves the watch dead until the app restarts.
|
||||
const desiredRemoteWatchers = new Map<
|
||||
string,
|
||||
{ connectionId: string; worktreePath: string; listeners: Map<number, WebContents> }
|
||||
>()
|
||||
const loggedUnavailableRemoteWatchers = new Set<string>()
|
||||
const pendingRemoteWatcherRetries = new Map<string, ReturnType<typeof setTimeout>>()
|
||||
const pendingRemoteWatcherRetryListeners = new Map<
|
||||
string,
|
||||
{ listeners: Map<number, WebContents>; startedAt: number }
|
||||
{ listeners: Map<number, WebContents>; startedAt: number; resyncOnInstall: boolean }
|
||||
>()
|
||||
// Why: last-listener cleanup aborts relay setup; late success is unwatched rather than installed after the renderer stopped watching.
|
||||
const inFlightRemoteInstalls = new Map<string, RemoteWatcherInstallToken>()
|
||||
@@ -903,6 +914,7 @@ const pendingRemoteInstallPromises = new Map<string, Promise<RemoteWatcherInstal
|
||||
let remoteWatchersClosed = false
|
||||
// Why: closeAllWatchers bumps this so a joiner that awaited across shutdown+reopen is refused (the latch alone can't tell it from a fresh call).
|
||||
let remoteWatcherLifecycleGeneration = 0
|
||||
let unsubscribeFromProviderRegistrations: (() => void) | null = null
|
||||
const REMOTE_WATCH_RETRY_MS = 1_000
|
||||
const REMOTE_WATCH_RETRY_TIMEOUT_MS = 60_000
|
||||
|
||||
@@ -979,7 +991,11 @@ export function forgetRemoteWatcherRemovalSnapshot(
|
||||
connectionId: string,
|
||||
worktreePath: string
|
||||
): void {
|
||||
suspendedRemoteWatcherListeners.delete(remoteWatcherKey(connectionId, worktreePath))
|
||||
const key = remoteWatcherKey(connectionId, worktreePath)
|
||||
suspendedRemoteWatcherListeners.delete(key)
|
||||
// Why: the worktree is gone — keeping the intent lets a reconnect landing before the renderer's
|
||||
// unwatch re-watch a deleted path (60s of retries against the host, then a bogus overflow).
|
||||
desiredRemoteWatchers.delete(key)
|
||||
}
|
||||
|
||||
function addInFlightRemoteInstallListener(
|
||||
@@ -1050,6 +1066,9 @@ function releaseRemoteWatchListener(key: string, senderId: number): void {
|
||||
}
|
||||
|
||||
function cleanupRemoteWatchersForSender(senderId: number): void {
|
||||
for (const key of Array.from(desiredRemoteWatchers.keys())) {
|
||||
forgetDesiredRemoteWatcher(key, senderId)
|
||||
}
|
||||
for (const [key, suspended] of suspendedRemoteWatcherListeners) {
|
||||
suspended.listeners.delete(senderId)
|
||||
if (suspended.listeners.size === 0) {
|
||||
@@ -1249,7 +1268,10 @@ function scheduleRemoteWatcherRetry(
|
||||
sender: WebContents,
|
||||
connectionId: string,
|
||||
worktreePath: string,
|
||||
startedAt = Date.now()
|
||||
startedAt = Date.now(),
|
||||
// Why: a retry that replaces a watch which was already live owes the renderer an overflow once it
|
||||
// lands — the events lost while it was down are otherwise never signalled.
|
||||
resyncOnInstall = false
|
||||
): void {
|
||||
const key = remoteWatcherKey(connectionId, worktreePath)
|
||||
const existingRetry = pendingRemoteWatcherRetryListeners.get(key)
|
||||
@@ -1257,12 +1279,14 @@ function scheduleRemoteWatcherRetry(
|
||||
if (!sender.isDestroyed()) {
|
||||
existingRetry.listeners.set(sender.id, sender)
|
||||
}
|
||||
existingRetry.resyncOnInstall ||= resyncOnInstall
|
||||
return
|
||||
}
|
||||
|
||||
const retry = {
|
||||
listeners: new Map(sender.isDestroyed() ? [] : [[sender.id, sender]]),
|
||||
startedAt
|
||||
startedAt,
|
||||
resyncOnInstall
|
||||
}
|
||||
pendingRemoteWatcherRetryListeners.set(key, retry)
|
||||
|
||||
@@ -1296,10 +1320,26 @@ function scheduleRemoteWatcherRetry(
|
||||
listeners.map((listener) => installRemoteWatcher(listener, connectionId, worktreePath))
|
||||
)
|
||||
.then((results) => {
|
||||
if (retry.resyncOnInstall) {
|
||||
for (const [index, listener] of listeners.entries()) {
|
||||
if (results[index] === 'installed' && !listener.isDestroyed()) {
|
||||
listener.send('fs:changed', {
|
||||
worktreePath,
|
||||
events: [{ kind: 'overflow', absolutePath: worktreePath }]
|
||||
} satisfies FsChangedPayload)
|
||||
}
|
||||
}
|
||||
}
|
||||
// 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) {
|
||||
scheduleRemoteWatcherRetry(listener, connectionId, worktreePath, retry.startedAt)
|
||||
scheduleRemoteWatcherRetry(
|
||||
listener,
|
||||
connectionId,
|
||||
worktreePath,
|
||||
retry.startedAt,
|
||||
retry.resyncOnInstall
|
||||
)
|
||||
}
|
||||
}
|
||||
})
|
||||
@@ -1308,7 +1348,13 @@ function scheduleRemoteWatcherRetry(
|
||||
return
|
||||
}
|
||||
for (const listener of listeners) {
|
||||
scheduleRemoteWatcherRetry(listener, connectionId, worktreePath, retry.startedAt)
|
||||
scheduleRemoteWatcherRetry(
|
||||
listener,
|
||||
connectionId,
|
||||
worktreePath,
|
||||
retry.startedAt,
|
||||
retry.resyncOnInstall
|
||||
)
|
||||
}
|
||||
})
|
||||
}, REMOTE_WATCH_RETRY_MS)
|
||||
@@ -1318,6 +1364,13 @@ function scheduleRemoteWatcherRetry(
|
||||
// ── Public API ───────────────────────────────────────────────────────
|
||||
|
||||
export function registerFilesystemWatcherHandlers(): void {
|
||||
// Why: re-registration replaces the handler set, so drop the previous subscription instead of
|
||||
// stacking a second re-arm on every provider registration.
|
||||
unsubscribeFromProviderRegistrations?.()
|
||||
unsubscribeFromProviderRegistrations = onSshFilesystemProviderRegistered(
|
||||
reinstallRemoteWatchersForConnection
|
||||
)
|
||||
|
||||
ipcMain.handle(
|
||||
'fs:watchWorktree',
|
||||
async (event, args: { worktreePath: string; connectionId?: string }): Promise<void> => {
|
||||
@@ -1325,6 +1378,9 @@ export function registerFilesystemWatcherHandlers(): void {
|
||||
// Why: a real new watch reopens the subsystem after closeAllWatchers latched it shut (also resets tests between cases).
|
||||
remoteWatchersClosed = false
|
||||
const key = remoteWatcherKey(args.connectionId, args.worktreePath)
|
||||
// Why: record intent before the install so a provider registering mid-flight (or long after
|
||||
// this attempt gives up) can still re-arm this listener.
|
||||
rememberDesiredRemoteWatcher(args.connectionId, args.worktreePath, event.sender)
|
||||
const result = await installRemoteWatcher(
|
||||
event.sender,
|
||||
args.connectionId,
|
||||
@@ -1353,6 +1409,9 @@ export function registerFilesystemWatcherHandlers(): void {
|
||||
(_event, args: { worktreePath: string; connectionId?: string }): void => {
|
||||
if (args.connectionId) {
|
||||
const key = remoteWatcherKey(args.connectionId, args.worktreePath)
|
||||
// Why: the caller stopped watching on purpose — drop the intent or a later provider
|
||||
// registration would resurrect a watch nobody asked for.
|
||||
forgetDesiredRemoteWatcher(key, _event.sender.id)
|
||||
const suspended = suspendedRemoteWatcherListeners.get(key)
|
||||
suspended?.listeners.delete(_event.sender.id)
|
||||
if (suspended?.listeners.size === 0) {
|
||||
@@ -1386,8 +1445,133 @@ function remoteWatcherKey(connectionId: string, worktreePath: string): string {
|
||||
return JSON.stringify([connectionId, normalizeRuntimePathForComparison(worktreePath)])
|
||||
}
|
||||
|
||||
function rememberDesiredRemoteWatcher(
|
||||
connectionId: string,
|
||||
worktreePath: string,
|
||||
sender: WebContents
|
||||
): void {
|
||||
if (sender.isDestroyed()) {
|
||||
return
|
||||
}
|
||||
const key = remoteWatcherKey(connectionId, worktreePath)
|
||||
const desired = desiredRemoteWatchers.get(key) ?? {
|
||||
connectionId,
|
||||
worktreePath,
|
||||
listeners: new Map<number, WebContents>()
|
||||
}
|
||||
desired.listeners.set(sender.id, sender)
|
||||
desiredRemoteWatchers.set(key, desired)
|
||||
registerSenderCleanup(sender)
|
||||
}
|
||||
|
||||
function forgetDesiredRemoteWatcher(key: string, senderId: number): void {
|
||||
const desired = desiredRemoteWatchers.get(key)
|
||||
if (!desired) {
|
||||
return
|
||||
}
|
||||
desired.listeners.delete(senderId)
|
||||
if (desired.listeners.size === 0) {
|
||||
desiredRemoteWatchers.delete(key)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Rebuild remote watches for a connection whose filesystem provider was just (re)registered.
|
||||
*
|
||||
* Why: the relay's watch registrations die with the transport they were made on, and the previous
|
||||
* provider's unwatch handle is scoped to that dead transport. Reinstalling is the only way the
|
||||
* subscription comes back, and consumers get an overflow so they resync whatever changed while the
|
||||
* watch was down.
|
||||
*/
|
||||
function reinstallRemoteWatchersForConnection(connectionId: string): void {
|
||||
if (remoteWatchersClosed) {
|
||||
return
|
||||
}
|
||||
for (const [key, desired] of Array.from(desiredRemoteWatchers)) {
|
||||
if (desired.connectionId !== connectionId) {
|
||||
continue
|
||||
}
|
||||
for (const [senderId, sender] of Array.from(desired.listeners)) {
|
||||
if (sender.isDestroyed()) {
|
||||
desired.listeners.delete(senderId)
|
||||
}
|
||||
}
|
||||
if (desired.listeners.size === 0) {
|
||||
desiredRemoteWatchers.delete(key)
|
||||
continue
|
||||
}
|
||||
|
||||
// Why: drop the entry the dead transport left behind first — installRemoteWatcher treats an
|
||||
// existing entry as already-installed and would hand back a watcher that can never fire again.
|
||||
const stale = remoteWatchers.get(key)
|
||||
if (stale) {
|
||||
remoteWatchers.delete(key)
|
||||
try {
|
||||
stale.unwatch()
|
||||
} catch {
|
||||
// Why: the handle belongs to the replaced transport; failing to close it is expected.
|
||||
}
|
||||
}
|
||||
const retryTimer = pendingRemoteWatcherRetries.get(key)
|
||||
if (retryTimer) {
|
||||
clearTimeout(retryTimer)
|
||||
pendingRemoteWatcherRetries.delete(key)
|
||||
pendingRemoteWatcherRetryListeners.delete(key)
|
||||
}
|
||||
loggedUnavailableRemoteWatchers.delete(key)
|
||||
|
||||
const listeners = Array.from(desired.listeners.values())
|
||||
void Promise.all(
|
||||
listeners.map((listener) =>
|
||||
installRemoteWatcher(listener, desired.connectionId, desired.worktreePath)
|
||||
)
|
||||
)
|
||||
.then((results) => {
|
||||
for (const [index, listener] of listeners.entries()) {
|
||||
if (results[index] !== 'installed' || listener.isDestroyed()) {
|
||||
continue
|
||||
}
|
||||
// Why: events between the transport dropping and this reinstall are gone for good;
|
||||
// overflow is the existing "resync, I can't tell you what changed" signal.
|
||||
listener.send('fs:changed', {
|
||||
worktreePath: desired.worktreePath,
|
||||
events: [{ kind: 'overflow', absolutePath: desired.worktreePath }]
|
||||
} satisfies FsChangedPayload)
|
||||
}
|
||||
if (results.some((result) => result === 'unavailable')) {
|
||||
for (const listener of listeners) {
|
||||
scheduleRemoteWatcherRetry(
|
||||
listener,
|
||||
desired.connectionId,
|
||||
desired.worktreePath,
|
||||
Date.now(),
|
||||
true
|
||||
)
|
||||
}
|
||||
}
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
if (isWatcherRemovalInProgressError(error)) {
|
||||
return
|
||||
}
|
||||
for (const listener of listeners) {
|
||||
scheduleRemoteWatcherRetry(
|
||||
listener,
|
||||
desired.connectionId,
|
||||
desired.worktreePath,
|
||||
Date.now(),
|
||||
true
|
||||
)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/** Tear down all watchers on app shutdown. */
|
||||
export async function closeAllWatchers(): Promise<void> {
|
||||
// Why: drop the intent with the rest of the state, but keep the provider-registration
|
||||
// subscription — a new fs:watchWorktree reopens the subsystem and still needs the re-arm hook.
|
||||
desiredRemoteWatchers.clear()
|
||||
senderCleanupRegistered.clear()
|
||||
unwatchableRoots.clear()
|
||||
suspendedLocalWatcherListeners.clear()
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import {
|
||||
getSshFilesystemProvider,
|
||||
onSshFilesystemProviderRegistered,
|
||||
registerSshFilesystemProvider,
|
||||
unregisterSshFilesystemProvider
|
||||
} from './ssh-filesystem-dispatch'
|
||||
import type { IFilesystemProvider } from './types'
|
||||
|
||||
const provider = {} as IFilesystemProvider
|
||||
|
||||
describe('onSshFilesystemProviderRegistered', () => {
|
||||
it('notifies subscribers on every registration, including a reconnect replacing the provider', () => {
|
||||
const listener = vi.fn()
|
||||
const unsubscribe = onSshFilesystemProviderRegistered(listener)
|
||||
|
||||
registerSshFilesystemProvider('conn-1', provider)
|
||||
registerSshFilesystemProvider('conn-1', {} as IFilesystemProvider)
|
||||
|
||||
expect(listener).toHaveBeenCalledTimes(2)
|
||||
expect(listener).toHaveBeenNthCalledWith(1, 'conn-1')
|
||||
expect(listener).toHaveBeenNthCalledWith(2, 'conn-1')
|
||||
|
||||
unsubscribe()
|
||||
unregisterSshFilesystemProvider('conn-1')
|
||||
})
|
||||
|
||||
it('exposes the new provider to subscribers while they are being notified', () => {
|
||||
let seen: IFilesystemProvider | undefined
|
||||
const unsubscribe = onSshFilesystemProviderRegistered((connectionId) => {
|
||||
seen = getSshFilesystemProvider(connectionId)
|
||||
})
|
||||
|
||||
registerSshFilesystemProvider('conn-2', provider)
|
||||
|
||||
expect(seen).toBe(provider)
|
||||
unsubscribe()
|
||||
unregisterSshFilesystemProvider('conn-2')
|
||||
})
|
||||
|
||||
it('stops notifying after unsubscribe', () => {
|
||||
const listener = vi.fn()
|
||||
onSshFilesystemProviderRegistered(listener)()
|
||||
|
||||
registerSshFilesystemProvider('conn-3', provider)
|
||||
|
||||
expect(listener).not.toHaveBeenCalled()
|
||||
unregisterSshFilesystemProvider('conn-3')
|
||||
})
|
||||
|
||||
it('keeps registration working when a subscriber throws', () => {
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
const healthy = vi.fn()
|
||||
const unsubscribeThrower = onSshFilesystemProviderRegistered(() => {
|
||||
throw new Error('subscriber blew up')
|
||||
})
|
||||
const unsubscribeHealthy = onSshFilesystemProviderRegistered(healthy)
|
||||
|
||||
expect(() => registerSshFilesystemProvider('conn-4', provider)).not.toThrow()
|
||||
expect(getSshFilesystemProvider('conn-4')).toBe(provider)
|
||||
expect(healthy).toHaveBeenCalledWith('conn-4')
|
||||
|
||||
unsubscribeThrower()
|
||||
unsubscribeHealthy()
|
||||
unregisterSshFilesystemProvider('conn-4')
|
||||
warnSpy.mockRestore()
|
||||
})
|
||||
})
|
||||
@@ -5,11 +5,32 @@ const sshProviders = new Map<string, IFilesystemProvider>()
|
||||
export const SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE =
|
||||
'Remote connection dropped. Click Reconnect on the SSH target before retrying.'
|
||||
|
||||
// Why: a reconnect builds a fresh provider, so anything holding remote state tied to the old
|
||||
// transport (file watches) needs a signal to rebuild it — nothing else marks that boundary.
|
||||
const registrationListeners = new Set<(connectionId: string) => void>()
|
||||
|
||||
export function onSshFilesystemProviderRegistered(
|
||||
listener: (connectionId: string) => void
|
||||
): () => void {
|
||||
registrationListeners.add(listener)
|
||||
return () => {
|
||||
registrationListeners.delete(listener)
|
||||
}
|
||||
}
|
||||
|
||||
export function registerSshFilesystemProvider(
|
||||
connectionId: string,
|
||||
provider: IFilesystemProvider
|
||||
): void {
|
||||
sshProviders.set(connectionId, provider)
|
||||
for (const listener of registrationListeners) {
|
||||
try {
|
||||
listener(connectionId)
|
||||
} catch (error) {
|
||||
// Why: relay establish must not fail because a subscriber threw.
|
||||
console.warn('[ssh-filesystem] provider registration listener failed:', error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function unregisterSshFilesystemProvider(connectionId: string): void {
|
||||
|
||||
Reference in New Issue
Block a user