starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-18 08:03:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(runtime): prevent stale watcher revival

Browse Source
This commit is contained in:
Brennan Benson
2026-07-13 20:17:59 -07:00
parent 2cb3c9e435
commit c9f935aed7
7 changed files with 323 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/ipc/filesystem-watcher-local-unsubscribe.test.ts
+59
View File
@@ -563,6 +563,49 @@ describe('local filesystem watcher unsubscribe cleanup', () => {
expect(unsubscribeMock).toHaveBeenCalledTimes(1)
})
it('retries on a fresh generation when a listener arrives after install abort', async () => {
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
let replacementCallback: Parameters<typeof subscribeViaWatcherProcess>[1] = () => undefined
const replacementUnsubscribe = vi.fn()
vi.mocked(subscribeViaWatcherProcess)
.mockImplementationOnce(
(_dir, _callback, _options, hooks) =>
new Promise((_resolve, reject) => {
hooks?.signal?.addEventListener('abort', () => reject(new Error('subscribe aborted')), {
once: true
})
})
)
.mockImplementationOnce(async (_dir, callback) => {
replacementCallback = callback
return { unsubscribe: replacementUnsubscribe }
})
const firstSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
const replacementSender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 2 }
const firstWatch = handlers['fs:watchWorktree'](
{ sender: firstSender },
{ worktreePath: '/tmp/repo' }
) as Promise<unknown>
await vi.waitFor(() => expect(subscribeViaWatcherProcess).toHaveBeenCalledTimes(1))
handlers['fs:unwatchWorktree']({ sender: firstSender }, { worktreePath: '/tmp/repo' })
const replacementWatch = handlers['fs:watchWorktree'](
{ sender: replacementSender },
{ worktreePath: '/tmp/repo' }
) as Promise<unknown>
await Promise.all([firstWatch, replacementWatch])
expect(subscribeViaWatcherProcess).toHaveBeenCalledTimes(2)
replacementCallback(null, [{ type: 'update', path: '/tmp/repo/retry.txt' }] as never)
await vi.waitFor(() =>
expect(replacementSender.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/tmp/repo',
events: [{ kind: 'update', absolutePath: '/tmp/repo/retry.txt', isDirectory: true }]
})
)
})
it('cancels an opening local watcher for worktree deletion', async () => {
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
let resolveSubscribe: (subscription: { unsubscribe: () => void }) => void = () => {}
@@ -625,6 +668,22 @@ describe('local filesystem watcher unsubscribe cleanup', () => {
})
})
it('does not restore a local listener stopped while deletion is pending', async () => {
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
const firstUnsubscribe = vi.fn()
vi.mocked(subscribeParcelWatcher).mockResolvedValue({ unsubscribe: firstUnsubscribe } as never)
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
await handlers['fs:watchWorktree']({ sender }, { worktreePath: '/tmp/repo' })
await closeLocalWatcherForWorktreePath('/tmp/repo')
handlers['fs:unwatchWorktree']({ sender }, { worktreePath: '/tmp/repo' })
await restoreLocalWatcherAfterFailedRemoval('/tmp/repo')
expect(firstUnsubscribe).toHaveBeenCalledTimes(1)
expect(subscribeParcelWatcher).toHaveBeenCalledTimes(1)
expect(sender.send).not.toHaveBeenCalled()
})
it('cancels an opening local watcher during app shutdown', async () => {
vi.mocked(stat).mockResolvedValue({ isDirectory: () => true } as never)
let resolveSubscribe: (subscription: { unsubscribe: () => void }) => void = () => {}
src/main/ipc/filesystem-watcher.test.ts
+18
View File
@@ -423,6 +423,24 @@ describe('registerFilesystemWatcherHandlers', () => {
})
})
it('does not restore an SSH listener stopped while deletion is pending', async () => {
const firstUnwatch = vi.fn()
const watchMock = vi.fn().mockResolvedValue(firstUnwatch)
const closeWatch = vi.fn().mockResolvedValue(undefined)
getSshFilesystemProviderMock.mockReturnValue({ watch: watchMock, closeWatch })
const sender = { isDestroyed: () => false, send: vi.fn(), once: vi.fn(), id: 1 }
const args = { worktreePath: '/home/me/repo', connectionId: 'conn-1' }
await handlers['fs:watchWorktree']({ sender }, args)
await closeRemoteWatcherForWorktreePath('conn-1', '/home/me/repo')
handlers['fs:unwatchWorktree']({ sender }, args)
await restoreRemoteWatcherAfterFailedRemoval('conn-1', '/home/me/repo')
expect(closeWatch).toHaveBeenCalledWith('/home/me/repo')
expect(watchMock).toHaveBeenCalledTimes(1)
expect(sender.send).not.toHaveBeenCalled()
})
it('preserves remote event routing when acknowledged teardown rejects', async () => {
const sendOne = vi.fn()
const sendTwo = vi.fn()
src/main/ipc/filesystem-watcher.ts
+72 -9
View File
@@ -84,6 +84,10 @@ const suspendedLocalWatcherListeners = new Map<
string,
{ worktreePath: string; listeners: Map<number, WebContents> }
>()
// Why: an install cancelled by shutdown cannot be revived by a waiter that
// resumes after a later handler call reopens the watcher subsystem.
let localWatchersClosed = false
let localWatcherLifecycleGeneration = 0
const failedLocalUnsubscribes = new Map<string, unknown>()
type LocalWatcherInstallToken = {
cancelled: boolean
@@ -105,7 +109,7 @@ function addInFlightLocalInstallListener(
token: LocalWatcherInstallToken,
sender: WebContents
): void {
if (sender.isDestroyed()) {
if (sender.isDestroyed() || token.abortController.signal.aborted) {
return
}
token.listeners.set(sender.id, sender)
@@ -486,6 +490,12 @@ async function createWatcher(
// ── Subscribe / Unsubscribe ──────────────────────────────────────────
function cleanupLocalWatchersForSender(senderId: number): void {
for (const [rootKey, suspended] of suspendedLocalWatcherListeners) {
suspended.listeners.delete(senderId)
if (suspended.listeners.size === 0) {
suspendedLocalWatcherListeners.delete(rootKey)
}
}
cleanupInFlightLocalInstallsForSender(senderId)
for (const [key, watchedRoot] of watchedRoots) {
if (watchedRoot.listeners.has(senderId)) {
@@ -563,10 +573,17 @@ function addLocalWatchListener(rootKey: string, sender: WebContents): void {
registerSenderCleanup(sender)
}
async function subscribe(worktreePath: string, sender: WebContents): Promise<void> {
async function subscribe(
worktreePath: string,
sender: WebContents,
generation = localWatcherLifecycleGeneration
): Promise<void> {
if (localWatchersClosed || generation !== localWatcherLifecycleGeneration) {
return
}
const finishInstall = beginWatcherInstall(worktreePath)
try {
await subscribeWhileRemovalAllowed(worktreePath, sender)
await subscribeWhileRemovalAllowed(worktreePath, sender, generation)
} finally {
finishInstall()
}
@@ -574,8 +591,12 @@ async function subscribe(worktreePath: string, sender: WebContents): Promise<voi
async function subscribeWhileRemovalAllowed(
worktreePath: string,
sender: WebContents
sender: WebContents,
generation: number
): Promise<void> {
if (localWatchersClosed || generation !== localWatcherLifecycleGeneration) {
return
}
const { key: rootKey, path: rootPath } = localWatcherRoot(worktreePath)
if (sender.isDestroyed()) {
return
@@ -608,7 +629,8 @@ async function subscribeWhileRemovalAllowed(
const pendingInstall = pendingLocalInstallPromises.get(rootKey)
if (pendingInstall) {
const inFlight = inFlightLocalInstalls.get(rootKey)
if (inFlight) {
const canJoinInstall = inFlight && !inFlight.abortController.signal.aborted
if (canJoinInstall) {
// Why: an unwatch may cancel an install while another renderer is still
// awaiting the same root; a new live listener should keep it alive.
addInFlightLocalInstallListener(inFlight, sender)
@@ -617,6 +639,28 @@ async function subscribeWhileRemovalAllowed(
}
}
const result = await pendingInstall
if (
result === 'cancelled' &&
!canJoinInstall &&
!localWatchersClosed &&
generation === localWatcherLifecycleGeneration
) {
// Why: AbortSignal cannot be revived. Listeners arriving after physical
// cancellation wait out that generation, then own a fresh install.
if (pendingLocalInstallPromises.get(rootKey) === pendingInstall) {
pendingLocalInstallPromises.delete(rootKey)
}
const retryListeners = new Map(
capacityRetryListeners.map((listener) => [listener.id, listener])
)
retryListeners.set(sender.id, sender)
for (const listener of retryListeners.values()) {
if (!listener.isDestroyed()) {
await subscribeWhileRemovalAllowed(worktreePath, listener, generation)
}
}
return
}
if (!inFlight) {
if (result === 'installed') {
for (const listener of capacityRetryListeners) {
@@ -742,6 +786,11 @@ async function doInstallLocalWatcher(
function unsubscribe(worktreePath: string, senderId: number): void {
const { key: rootKey } = localWatcherRoot(worktreePath)
const suspended = suspendedLocalWatcherListeners.get(rootKey)
suspended?.listeners.delete(senderId)
if (suspended?.listeners.size === 0) {
suspendedLocalWatcherListeners.delete(rootKey)
}
const capacityRetry = pendingLocalCapacityRetries.get(rootKey)
if (capacityRetry) {
capacityRetry.listeners.delete(senderId)
@@ -1079,6 +1128,12 @@ function releaseRemoteWatchListener(key: string, senderId: number): void {
}
function cleanupRemoteWatchersForSender(senderId: number): void {
for (const [key, suspended] of suspendedRemoteWatcherListeners) {
suspended.listeners.delete(senderId)
if (suspended.listeners.size === 0) {
suspendedRemoteWatcherListeners.delete(key)
}
}
cleanupInFlightRemoteInstallsForSender(senderId)
for (const key of Array.from(remoteWatchers.keys())) {
releaseRemoteWatchListener(key, senderId)
@@ -1382,6 +1437,9 @@ export function registerFilesystemWatcherHandlers(): void {
}
return
}
// Why: tests and post-shutdown renderer reattachment reopen the local
// subsystem, while stale callers retain the prior generation.
localWatchersClosed = false
await subscribe(args.worktreePath, event.sender)
}
)
@@ -1391,6 +1449,11 @@ export function registerFilesystemWatcherHandlers(): void {
(_event, args: { worktreePath: string; connectionId?: string }): void => {
if (args.connectionId) {
const key = remoteWatcherKey(args.connectionId, args.worktreePath)
const suspended = suspendedRemoteWatcherListeners.get(key)
suspended?.listeners.delete(_event.sender.id)
if (suspended?.listeners.size === 0) {
suspendedRemoteWatcherListeners.delete(key)
}
const retry = pendingRemoteWatcherRetryListeners.get(key)
retry?.listeners.delete(_event.sender.id)
const retryTimer = pendingRemoteWatcherRetries.get(key)
@@ -1445,12 +1508,12 @@ export async function closeAllWatchers(): Promise<void> {
pendingRemoteWatcherRetries.clear()
pendingRemoteWatcherRetryListeners.clear()
loggedUnavailableRemoteWatchers.clear()
// Why: latch the subsystem shut and drop the dedup map so a late install that
// begins after teardown is refused instead of registering post-shutdown. Bump
// the generation so a waiter that resumes after a later reopen still recurses
// on a stale lifecycle and is refused.
// Why: latch both watcher subsystems shut so late installs cannot register
// post-shutdown. Generation bumps also reject waiters from an older lifecycle.
remoteWatchersClosed = true
remoteWatcherLifecycleGeneration += 1
localWatchersClosed = true
localWatcherLifecycleGeneration += 1
pendingRemoteInstallPromises.clear()
// Why: cancel any in-flight provider.watch() calls so their resolved
// unwatch handles are discarded instead of being installed after shutdown.
src/renderer/src/runtime/runtime-file-client.test.ts
+50
View File
@@ -1845,6 +1845,56 @@ describe('runtime file client', () => {
expect(retryPayload).toHaveBeenCalledTimes(1)
})
it('evicts a failed watcher setup before notifying retrying listeners', async () => {
const callbacks: { onResponse: (response: unknown) => void; onClose: () => void }[] = []
const unsubscribes = [vi.fn(), vi.fn()]
runtimeEnvironmentSubscribe.mockImplementation((_args, nextCallbacks) => {
callbacks.push(nextCallbacks)
return Promise.resolve({
unsubscribe: unsubscribes[callbacks.length - 1],
sendBinary: vi.fn()
})
})
const retryPayload = vi.fn()
let retryPromise: Promise<() => void> | undefined
const context = {
settings: { activeRuntimeEnvironmentId: 'env-1' },
worktreeId: 'wt-1',
worktreePath: '/remote/repo'
}
await subscribeRuntimeFileChanges(context, vi.fn(), () => {
retryPromise = subscribeRuntimeFileChanges(context, retryPayload, vi.fn())
})
callbacks[0].onResponse({
id: 'rpc-setup-failed',
ok: false,
error: { code: 'watch_failed', message: 'root unavailable' },
_meta: { runtimeId: 'remote-runtime' }
})
await retryPromise
expect(runtimeEnvironmentSubscribe).toHaveBeenCalledTimes(2)
expect(unsubscribes[0]).toHaveBeenCalledTimes(1)
callbacks[1]?.onResponse({
id: 'rpc-changed',
ok: true,
result: {
type: 'changed',
worktree: 'id:wt-1',
events: [{ kind: 'update', absolutePath: '/remote/repo/retry.txt' }]
},
_meta: { runtimeId: 'remote-runtime' }
})
expect(retryPayload).toHaveBeenCalledTimes(1)
callbacks[1]?.onResponse({
id: 'rpc-retry-end',
ok: true,
result: { type: 'end' },
_meta: { runtimeId: 'remote-runtime' }
})
})
it('shares one remote file watch subscription across listeners for the same worktree', async () => {
const firstPayload = vi.fn()
const secondPayload = vi.fn()
src/renderer/src/runtime/runtime-file-client.ts
+20 -18
View File
@@ -891,7 +891,7 @@ function createSharedRuntimeFileWatch(
handleSharedRuntimeFileWatchResponse(key, shared, worktreePath, response)
},
onError: (error) => {
notifySharedRuntimeFileWatchError(shared, new Error(error.message))
failSharedRuntimeFileWatch(key, shared, new Error(error.message))
},
onClose: () => {
if (sharedRuntimeFileWatches.get(key) === shared) {
@@ -913,11 +913,7 @@ function createSharedRuntimeFileWatch(
}
})
.catch((err) => {
if (sharedRuntimeFileWatches.get(key) === shared) {
sharedRuntimeFileWatches.delete(key)
}
shared.closed = true
notifySharedRuntimeFileWatchError(shared, err instanceof Error ? err : new Error(String(err)))
failSharedRuntimeFileWatch(key, shared, err instanceof Error ? err : new Error(String(err)))
throw err
})
return shared
@@ -949,15 +945,7 @@ function handleSharedRuntimeFileWatchResponse(
} else if (event.type === 'error') {
// Why: error listeners may synchronously retry. Evict the terminal watch
// before callbacks run so the retry cannot join a stream awaiting `end`.
if (sharedRuntimeFileWatches.get(key) === shared) {
sharedRuntimeFileWatches.delete(key)
}
shared.closed = true
const listeners = Array.from(shared.listeners)
shared.listeners.clear()
for (const listener of listeners) {
listener.onError?.(new Error(event.message))
}
failSharedRuntimeFileWatch(key, shared, new Error(event.message))
} else if (event.type === 'end') {
// Why: shared-control completes without onClose; evict and release its
// transport handle so later listeners start cleanly without retained state.
@@ -972,12 +960,26 @@ function handleSharedRuntimeFileWatchResponse(
unsubscribe?.()
}
} catch (err) {
notifySharedRuntimeFileWatchError(shared, err instanceof Error ? err : new Error(String(err)))
failSharedRuntimeFileWatch(key, shared, err instanceof Error ? err : new Error(String(err)))
}
}
function notifySharedRuntimeFileWatchError(shared: SharedRuntimeFileWatch, error: Error): void {
for (const listener of Array.from(shared.listeners)) {
function failSharedRuntimeFileWatch(
key: string,
shared: SharedRuntimeFileWatch,
error: Error
): void {
if (sharedRuntimeFileWatches.get(key) === shared) {
sharedRuntimeFileWatches.delete(key)
}
shared.closed = true
shared.remoteSubscriptionId = null
const unsubscribe = shared.unsubscribe
shared.unsubscribe = null
const listeners = Array.from(shared.listeners)
shared.listeners.clear()
unsubscribe?.()
for (const listener of listeners) {
listener.onError?.(error)
}
}
src/renderer/src/web/web-runtime-client.test.ts
+85
View File
@@ -277,6 +277,91 @@ describe('WebRuntimeClient', () => {
expect(internals.subscriptions.size).toBe(0)
})
it('does not replay a file watch stopped after its socket disconnects', async () => {
const client = new WebRuntimeClient({
v: 2,
endpoint: 'ws://127.0.0.1:6768',
deviceToken: 'token',
publicKeyB64: Buffer.alloc(32).toString('base64')
})
const initialSocket = fakeSockets[0]!
initialSocket.readyState = FakeWebSocket.OPEN
const internals = client as unknown as {
ws: FakeWebSocket | null
sharedKey: Uint8Array | null
state: string
subscriptions: Map<string, { needsReplay: boolean }>
setState(next: string): void
}
internals.sharedKey = new Uint8Array(32)
internals.state = 'connected'
const handle = await client.subscribe(
'files.watch',
{ worktree: 'wt-1' },
{ onResponse: vi.fn() }
)
initialSocket.onclose?.()
expect(Array.from(internals.subscriptions.values())[0]?.needsReplay).toBe(true)
handle.unsubscribe()
expect(internals.subscriptions.size).toBe(0)
const replacementSocket = new FakeWebSocket('ws://127.0.0.1:6768')
replacementSocket.readyState = FakeWebSocket.OPEN
internals.ws = replacementSocket
internals.sharedKey = new Uint8Array(32)
internals.setState('connected')
expect(replacementSocket.send).not.toHaveBeenCalled()
client.close()
})
it('evicts a failed file-watch setup before reconnect can replay it', async () => {
const client = new WebRuntimeClient({
v: 2,
endpoint: 'ws://127.0.0.1:6768',
deviceToken: 'token',
publicKeyB64: Buffer.alloc(32).toString('base64')
})
const initialSocket = fakeSockets[0]!
initialSocket.readyState = FakeWebSocket.OPEN
const sharedKey = new Uint8Array(32)
const onResponse = vi.fn()
const internals = client as unknown as {
ws: FakeWebSocket | null
sharedKey: Uint8Array | null
state: string
subscriptions: Map<string, unknown>
handleSocketMessage(rawData: unknown): Promise<void>
setState(next: string): void
}
internals.sharedKey = sharedKey
internals.state = 'connected'
await client.subscribe('files.watch', { worktree: 'wt-1' }, { onResponse })
const subscriptionId = Array.from(internals.subscriptions.keys())[0]!
await internals.handleSocketMessage(
encrypt(
JSON.stringify({
id: subscriptionId,
ok: false,
error: { code: 'watch_failed', message: 'root unavailable' }
}),
sharedKey
)
)
expect(onResponse).toHaveBeenCalledTimes(1)
expect(internals.subscriptions.size).toBe(0)
initialSocket.onclose?.()
const replacementSocket = new FakeWebSocket('ws://127.0.0.1:6768')
replacementSocket.readyState = FakeWebSocket.OPEN
internals.ws = replacementSocket
internals.sharedKey = sharedKey
internals.setState('connected')
expect(replacementSocket.send).not.toHaveBeenCalled()
client.close()
})
it('keeps file watches on the owning WebSocket instead of opening child clients', async () => {
const client = new WebRuntimeClient({
v: 2,
src/renderer/src/web/web-runtime-client.ts
+19 -4
View File
@@ -36,6 +36,7 @@ type SubscriptionCallbacks = {
onError?: (error: { code: string; message: string }) => void
onClose?: () => void
onTransportInterrupted?: () => void
onTransportReplayed?: () => void
}
type RuntimeSubscription = {
@@ -196,6 +197,7 @@ export class WebRuntimeClient {
)
let stopped = false
let remoteSubscriptionId: string | null = null
let transportInterrupted = false
let unwatchStarted = false
let handle: WebRuntimeSubscriptionHandle | null = null
const dropLocalSubscription = (): void => {
@@ -252,6 +254,7 @@ export class WebRuntimeClient {
const wrappedCallbacks: SubscriptionCallbacks = {
...callbacks,
onResponse: (response) => {
transportInterrupted = false
const nextSubscriptionId = getFileWatchSubscriptionId(response)
if (nextSubscriptionId) {
remoteSubscriptionId = nextSubscriptionId
@@ -282,6 +285,7 @@ export class WebRuntimeClient {
}
},
onTransportInterrupted: () => {
transportInterrupted = true
remoteSubscriptionId = null
if (!stopped) {
return
@@ -294,6 +298,9 @@ export class WebRuntimeClient {
// Why: socket close physically releases the old server subscription;
// a locally stopped watch must not be replayed on the replacement.
dropLocalSubscription()
},
onTransportReplayed: () => {
transportInterrupted = false
}
}
handle = await this.subscribeOnCurrentConnection(
@@ -311,6 +318,10 @@ export class WebRuntimeClient {
stopped = true
if (remoteSubscriptionId) {
unwatchAndDropLocalSubscription()
} else if (transportInterrupted) {
// Why: socket close already released the old server subscription;
// remove its replay record instead of reviving a locally stopped watch.
dropLocalSubscription()
}
// Why: an older server may not publish its id until ready. Retain the
// callback so a late response can still physically unwatch the root.
@@ -545,6 +556,11 @@ export class WebRuntimeClient {
const subscription = this.subscriptions.get(response.id)
if (subscription && isSubscriptionResponse(response)) {
// Why: setup failures are terminal. Evict before callbacks so a
// synchronous retry creates a fresh watch and reconnect cannot replay it.
if (response.ok === false) {
this.subscriptions.delete(response.id)
}
subscription.callbacks.onResponse(response)
if (response.ok && isEndResult(response.result)) {
this.subscriptions.delete(response.id)
@@ -715,13 +731,15 @@ export class WebRuntimeClient {
subscription.needsReplay = false
this.subscriptions.set(subscription.id, subscription)
if (
!this.sendEncrypted({
this.sendEncrypted({
id: subscription.id,
deviceToken: this.pairing.deviceToken,
method: subscription.method,
params: subscription.params
})
) {
subscription.callbacks.onTransportReplayed?.()
} else {
subscription.needsReplay = true
}
}
@@ -844,9 +862,6 @@ function isSubscriptionResponse(
if (response.ok === false) {
return true
}
if (response.ok === false) {
return true
}
const success = response as RuntimeRpcResponse<unknown> & { ok: true; streaming?: true }
return (
success.streaming === true || isEndResult(success.result) || isScrollbackResult(success.result)