From c9f935aed70a9680aff35fc97d14581f567e4ad5 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 13 Jul 2026 20:02:14 -0700 Subject: [PATCH] fix(runtime): prevent stale watcher revival --- ...lesystem-watcher-local-unsubscribe.test.ts | 59 +++++++++++++ src/main/ipc/filesystem-watcher.test.ts | 18 ++++ src/main/ipc/filesystem-watcher.ts | 81 ++++++++++++++++-- .../src/runtime/runtime-file-client.test.ts | 50 +++++++++++ .../src/runtime/runtime-file-client.ts | 38 +++++---- .../src/web/web-runtime-client.test.ts | 85 +++++++++++++++++++ src/renderer/src/web/web-runtime-client.ts | 23 ++++- 7 files changed, 323 insertions(+), 31 deletions(-) diff --git a/src/main/ipc/filesystem-watcher-local-unsubscribe.test.ts b/src/main/ipc/filesystem-watcher-local-unsubscribe.test.ts index 3b81356f9b8..ab8cfb8b782 100644 --- a/src/main/ipc/filesystem-watcher-local-unsubscribe.test.ts +++ b/src/main/ipc/filesystem-watcher-local-unsubscribe.test.ts @@ -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[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 + 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 + + 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 = () => {} diff --git a/src/main/ipc/filesystem-watcher.test.ts b/src/main/ipc/filesystem-watcher.test.ts index ffbc90db2ba..4d757e8c8d0 100644 --- a/src/main/ipc/filesystem-watcher.test.ts +++ b/src/main/ipc/filesystem-watcher.test.ts @@ -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() diff --git a/src/main/ipc/filesystem-watcher.ts b/src/main/ipc/filesystem-watcher.ts index d51b2e3f712..6e5876e9add 100644 --- a/src/main/ipc/filesystem-watcher.ts +++ b/src/main/ipc/filesystem-watcher.ts @@ -84,6 +84,10 @@ const suspendedLocalWatcherListeners = new Map< string, { worktreePath: string; listeners: Map } >() +// 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() 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 { +async function subscribe( + worktreePath: string, + sender: WebContents, + generation = localWatcherLifecycleGeneration +): Promise { + 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 { + 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 { 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. diff --git a/src/renderer/src/runtime/runtime-file-client.test.ts b/src/renderer/src/runtime/runtime-file-client.test.ts index 52086493559..6b4cbd268bd 100644 --- a/src/renderer/src/runtime/runtime-file-client.test.ts +++ b/src/renderer/src/runtime/runtime-file-client.test.ts @@ -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() diff --git a/src/renderer/src/runtime/runtime-file-client.ts b/src/renderer/src/runtime/runtime-file-client.ts index 6033ce58efe..a84d4382e5a 100644 --- a/src/renderer/src/runtime/runtime-file-client.ts +++ b/src/renderer/src/runtime/runtime-file-client.ts @@ -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) } } diff --git a/src/renderer/src/web/web-runtime-client.test.ts b/src/renderer/src/web/web-runtime-client.test.ts index d9ce441e9be..c96bcbe9b68 100644 --- a/src/renderer/src/web/web-runtime-client.test.ts +++ b/src/renderer/src/web/web-runtime-client.test.ts @@ -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 + 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 + handleSocketMessage(rawData: unknown): Promise + 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, diff --git a/src/renderer/src/web/web-runtime-client.ts b/src/renderer/src/web/web-runtime-client.ts index 7bb99ffd46c..022fb1456b9 100644 --- a/src/renderer/src/web/web-runtime-client.ts +++ b/src/renderer/src/web/web-runtime-client.ts @@ -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 & { ok: true; streaming?: true } return ( success.streaming === true || isEndResult(success.result) || isScrollbackResult(success.result)