From f270feb649b66891c429c2ede6f985eec0ab7bb4 Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Tue, 28 Jul 2026 14:57:17 +0200 Subject: [PATCH 1/8] chore: improve unsub failure and retry logic --- packages/core-backend/CHANGELOG.md | 4 + .../src/ws/ohlcv/OHLCVService.test.ts | 94 ++++++++-- .../core-backend/src/ws/ohlcv/OHLCVService.ts | 164 +++++++++++++++--- 3 files changed, 230 insertions(+), 32 deletions(-) diff --git a/packages/core-backend/CHANGELOG.md b/packages/core-backend/CHANGELOG.md index d1f1345bb79..be830f17181 100644 --- a/packages/core-backend/CHANGELOG.md +++ b/packages/core-backend/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- `OHLCVService` now flushes grace-period channels when subscribing to a different asset/interval, retries failed WebSocket unsubscribes with backoff before forcing reconnection, and only removes channel tracking after a successful unsubscribe ([#TBD](https://github.com/MetaMask/core/pull/TBD)) + ## [7.0.0] ### Added diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts index 8a4709effcf..27a3b22cc57 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts @@ -438,12 +438,14 @@ describe('OHLCVService', () => { }); }); - it('should unsubscribe old channel after grace period during time-range switching', async () => { + it('should flush the old channel immediately when subscribing to a different interval', async () => { const opts1m = SUB_OPTS; const opts1h: OHLCVSubscriptionOptions = { ...SUB_OPTS, interval: '1h', }; + const channel1h = + 'market-data.v1.eip155:8453/erc20:0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913.1h.usd'; await withService(async ({ service, mocks }) => { const mockUnsub = jest.fn().mockResolvedValue(undefined); @@ -451,17 +453,23 @@ describe('OHLCVService', () => { { unsubscribe: mockUnsub }, ]); - // Subscribe 1m → unsubscribe → subscribe 1h await service.subscribe(opts1m); await service.unsubscribe(opts1m); await service.subscribe(opts1h); - // 1m is in grace period, not yet unsubscribed - expect(mockUnsub).not.toHaveBeenCalled(); + expect(mockUnsub).toHaveBeenCalledTimes(1); + expect(mocks.getSubscriptionsByChannel).toHaveBeenCalledWith( + EXPECTED_CHANNEL, + ); + expect(mocks.subscribe).toHaveBeenCalledWith( + expect.objectContaining({ + channels: [channel1h], + }), + ); - // Grace period expires — old channel cleaned up jest.advanceTimersByTime(3000); await completeAsyncOperations(); + expect(mockUnsub).toHaveBeenCalledTimes(1); }); }); @@ -938,7 +946,7 @@ describe('OHLCVService', () => { // =========================================================================== describe('error paths', () => { - it('should publish subscriptionError when unsubscribe fails', async () => { + it('should publish subscriptionError and retry when unsubscribe fails', async () => { await withService(async ({ service, mocks, messenger }) => { mocks.getSubscriptionsByChannel.mockImplementation(() => { throw new Error('ws gone'); @@ -959,6 +967,73 @@ describe('OHLCVService', () => { operation: 'unsubscribe', }); expect(mocks.forceReconnection).not.toHaveBeenCalled(); + + jest.advanceTimersByTime(1000); + await completeAsyncOperations(); + expect(mocks.getSubscriptionsByChannel).toHaveBeenCalledTimes(2); + + jest.advanceTimersByTime(2000); + await completeAsyncOperations(); + expect(mocks.getSubscriptionsByChannel).toHaveBeenCalledTimes(3); + + jest.advanceTimersByTime(4000); + await completeAsyncOperations(); + expect(mocks.getSubscriptionsByChannel).toHaveBeenCalledTimes(4); + expect(mocks.forceReconnection).toHaveBeenCalledTimes(1); + }); + }); + + it('should cancel unsubscribe retry when the same channel is subscribed again', async () => { + await withService(async ({ service, mocks }) => { + mocks.getSubscriptionsByChannel.mockImplementation(() => { + throw new Error('ws gone'); + }); + mocks.channelHasSubscription.mockReturnValue(true); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await completeAsyncOperations(); + + mocks.subscribe.mockClear(); + mocks.connect.mockClear(); + await service.subscribe(SUB_OPTS); + + expect(mocks.connect).not.toHaveBeenCalled(); + expect(mocks.subscribe).not.toHaveBeenCalled(); + + jest.advanceTimersByTime(7000); + await completeAsyncOperations(); + expect(mocks.forceReconnection).not.toHaveBeenCalled(); + }); + }); + + it('should delete failed-cleanup channels on reconnect', async () => { + await withService(async ({ service, mocks, rootMessenger }) => { + mocks.getSubscriptionsByChannel.mockImplementation(() => { + throw new Error('ws gone'); + }); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await completeAsyncOperations(); + + mocks.subscribe.mockClear(); + rootMessenger.publish( + 'BackendWebSocketService:connectionStateChanged', + { + ...BASE_CONNECTION_INFO, + state: WebSocketState.CONNECTED, + connectedAt: Date.now(), + reconnectAttempts: 1, + }, + ); + await completeAsyncOperations(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); }); }); @@ -1096,7 +1171,7 @@ describe('OHLCVService', () => { // =========================================================================== describe('destroy', () => { - it('should clear grace-period timers and remove channel callback', async () => { + it('should clear grace-period and retry timers and remove channel callback', async () => { await withService(async ({ service, mocks }) => { const mockUnsub = jest.fn(); mocks.getSubscriptionsByChannel.mockReturnValue([ @@ -1106,14 +1181,13 @@ describe('OHLCVService', () => { await service.subscribe(SUB_OPTS); await service.unsubscribe(SUB_OPTS); - // Grace timer is running — destroy should clear it service.destroy(); - jest.advanceTimersByTime(5000); + jest.advanceTimersByTime(10000); await completeAsyncOperations(); - // Timer was cleared so the actual unsubscribe should NOT have fired expect(mockUnsub).not.toHaveBeenCalled(); + expect(mocks.forceReconnection).not.toHaveBeenCalled(); expect(mocks.removeChannelCallback).toHaveBeenCalledWith( 'system-notifications.v1.market-data.v1', diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index 1ccb5b9c45c..13b0eb9cd4e 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -43,6 +43,12 @@ const SYSTEM_NOTIFICATIONS_CHANNEL = `system-notifications.v1.${SUBSCRIPTION_NAM /** Delay before actually unsubscribing from a channel after refCount reaches 0. */ const GRACE_PERIOD_MS = 3_000; +/** Backoff delays for retrying failed WebSocket unsubscribes. */ +const UNSUB_RETRY_DELAYS_MS = [1_000, 2_000, 4_000] as const; + +/** Max unsubscribe retries before forcing a WebSocket reconnection. */ +const MAX_UNSUB_RETRIES = 3; + // ============================================================================= // Types — Channel Tracking // ============================================================================= @@ -50,6 +56,8 @@ const GRACE_PERIOD_MS = 3_000; type ChannelEntry = { refCount: number; gracePeriodTimer?: ReturnType; + retryTimer?: ReturnType; + retryCount?: number; }; // ============================================================================= @@ -142,7 +150,9 @@ export type OHLCVServiceMessenger = Messenger< * * Features: * - Reference counting: multiple UI consumers share one WebSocket subscription - * - Grace-period unsubscribe: avoids rapid unsub/resub during navigation + * - Grace-period unsubscribe: reuses same-channel subs on rapid back navigation + * - Grace-period flush: immediately unsubscribes other channels on navigation + * - Unsubscribe retry: retries failed unsubs with backoff before force reconnect * - Idempotency: duplicate subscribe calls for the same channel are no-ops * - Reconnect resilience: resubscribes all active channels on reconnect * - Chain-status forwarding: listens to system-notifications for chain up/down @@ -228,7 +238,25 @@ export class OHLCVService { async #subscribeInner(channel: string): Promise { const entry = this.#channels.get(channel); - if (entry?.gracePeriodTimer) { + if (entry?.retryTimer) { + clearTimeout(entry.retryTimer); + entry.retryTimer = undefined; + entry.retryCount = 0; + entry.refCount = 1; + + if ( + this.#messenger.call( + 'BackendWebSocketService:channelHasSubscription', + channel, + ) + ) { + log('OHLCV-WS: Cancelled unsubscribe retry — reusing WS subscription', { + channel, + }); + return; + } + // WS subscription was lost — fall through to recreate it. + } else if (entry?.gracePeriodTimer) { clearTimeout(entry.gracePeriodTimer); entry.gracePeriodTimer = undefined; log('OHLCV-WS: Cancelled grace-period unsubscribe', { @@ -254,6 +282,9 @@ export class OHLCVService { entry.refCount += 1; return; } + + await this.#flushOtherChannels(channel); + try { await this.#messenger.call('BackendWebSocketService:connect'); @@ -339,6 +370,105 @@ export class OHLCVService { // Private — WebSocket Subscription Helpers // ============================================================================= + /** + * Immediately unsubscribe other channels in grace or failed-cleanup state. + * Called while the subscribe mutex is held before opening a new channel. + * + * @param exceptChannel - Channel being subscribed; excluded from flush. + */ + async #flushOtherChannels(exceptChannel: string): Promise { + for (const [channel, channelEntry] of this.#channels.entries()) { + if (channel === exceptChannel || channelEntry.refCount > 0) { + continue; + } + + this.#clearChannelTimers(channelEntry); + log('OHLCV-WS: Flushing grace-period channel before new subscribe', { + flushedChannel: channel, + newChannel: exceptChannel, + }); + + const success = await this.#unsubscribeChannelOnServer(channel); + if (success) { + this.#channels.delete(channel); + } else { + this.#scheduleUnsubscribeRetry(channel); + } + } + } + + #clearChannelTimers(entry: ChannelEntry): void { + if (entry.gracePeriodTimer) { + clearTimeout(entry.gracePeriodTimer); + entry.gracePeriodTimer = undefined; + } + if (entry.retryTimer) { + clearTimeout(entry.retryTimer); + entry.retryTimer = undefined; + } + } + + async #unsubscribeChannelOnServer(channel: string): Promise { + try { + const subscriptions = this.#messenger.call( + 'BackendWebSocketService:getSubscriptionsByChannel', + channel, + ); + + for (const sub of subscriptions) { + await sub.unsubscribe(); + } + return true; + } catch (error) { + log('OHLCV-WS: Unsubscription failed', { channel, error }); + this.#messenger.publish('OHLCVService:subscriptionError', { + channel, + error: String(error), + operation: 'unsubscribe', + }); + return false; + } + } + + #scheduleUnsubscribeRetry(channel: string): void { + let entry = this.#channels.get(channel); + if (!entry) { + entry = { refCount: 0 }; + this.#channels.set(channel, entry); + } + + const retryCount = (entry.retryCount ?? 0) + 1; + entry.retryCount = retryCount; + + if (retryCount > MAX_UNSUB_RETRIES) { + log('OHLCV-WS: Unsubscribe retries exhausted — forcing reconnection', { + channel, + }); + this.#clearChannelTimers(entry); + this.#channels.delete(channel); + this.#messenger + .call('BackendWebSocketService:forceReconnection') + .catch(() => { + // no-op + }); + return; + } + + const delayMs = UNSUB_RETRY_DELAYS_MS[retryCount - 1]; + log('OHLCV-WS: Scheduling unsubscribe retry', { + channel, + retryCount, + delayMs, + }); + + entry.retryTimer = setTimeout(() => { + entry.retryTimer = undefined; + this.#performUnsubscribe(channel).catch(() => { + // no-op + }); + }, delayMs); + } + async #performUnsubscribe(channel: string): Promise { const releaseLock = await this.#mutex.acquire(); try { @@ -354,25 +484,15 @@ export class OHLCVService { log('OHLCV-WS: Grace period expired — performing actual WS unsubscribe', { channel, }); - this.#channels.delete(channel); - try { - const subscriptions = this.#messenger.call( - 'BackendWebSocketService:getSubscriptionsByChannel', - channel, - ); + this.#clearChannelTimers(entry ?? { refCount: 0 }); - for (const sub of subscriptions) { - await sub.unsubscribe(); - } + const success = await this.#unsubscribeChannelOnServer(channel); + if (success) { + this.#channels.delete(channel); log('OHLCV-WS: WS unsubscribe completed', { channel }); - } catch (error) { - log('OHLCV-WS: Unsubscription failed', { channel, error }); - this.#messenger.publish('OHLCVService:subscriptionError', { - channel, - error: String(error), - operation: 'unsubscribe', - }); + } else { + this.#scheduleUnsubscribeRetry(channel); } } finally { releaseLock(); @@ -391,8 +511,10 @@ export class OHLCVService { count: channelCount, }); - for (const [channel, entry] of this.#channels.entries()) { + for (const [channel, entry] of [...this.#channels.entries()]) { if (entry.refCount === 0) { + this.#clearChannelTimers(entry); + this.#channels.delete(channel); continue; } @@ -534,9 +656,7 @@ export class OHLCVService { */ destroy(): void { for (const entry of this.#channels.values()) { - if (entry.gracePeriodTimer) { - clearTimeout(entry.gracePeriodTimer); - } + this.#clearChannelTimers(entry); } this.#channels.clear(); this.#chainsUp.clear(); From fb84a1ef8fa465941dbadea04ec584cdd806f40b Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Tue, 28 Jul 2026 15:03:07 +0200 Subject: [PATCH 2/8] chore: lint and changelog --- packages/core-backend/CHANGELOG.md | 2 +- packages/core-backend/src/ws/ohlcv/OHLCVService.ts | 12 +++++++----- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/packages/core-backend/CHANGELOG.md b/packages/core-backend/CHANGELOG.md index be830f17181..c2884089caf 100644 --- a/packages/core-backend/CHANGELOG.md +++ b/packages/core-backend/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- `OHLCVService` now flushes grace-period channels when subscribing to a different asset/interval, retries failed WebSocket unsubscribes with backoff before forcing reconnection, and only removes channel tracking after a successful unsubscribe ([#TBD](https://github.com/MetaMask/core/pull/TBD)) +- `OHLCVService` now flushes grace-period channels when subscribing to a different asset/interval, retries failed WebSocket unsubscribes with backoff before forcing reconnection, and only removes channel tracking after a successful unsubscribe ([#9678](https://github.com/MetaMask/core/pull/9678)) ## [7.0.0] diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index 13b0eb9cd4e..7a3f6a9e53f 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -437,14 +437,16 @@ export class OHLCVService { this.#channels.set(channel, entry); } - const retryCount = (entry.retryCount ?? 0) + 1; - entry.retryCount = retryCount; + const channelEntry = entry; + + const retryCount = (channelEntry.retryCount ?? 0) + 1; + channelEntry.retryCount = retryCount; if (retryCount > MAX_UNSUB_RETRIES) { log('OHLCV-WS: Unsubscribe retries exhausted — forcing reconnection', { channel, }); - this.#clearChannelTimers(entry); + this.#clearChannelTimers(channelEntry); this.#channels.delete(channel); this.#messenger .call('BackendWebSocketService:forceReconnection') @@ -461,8 +463,8 @@ export class OHLCVService { delayMs, }); - entry.retryTimer = setTimeout(() => { - entry.retryTimer = undefined; + channelEntry.retryTimer = setTimeout(() => { + channelEntry.retryTimer = undefined; this.#performUnsubscribe(channel).catch(() => { // no-op }); From f2b5f30453472e7f7aa16aa3da201b283abb3ef5 Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Tue, 28 Jul 2026 15:17:52 +0200 Subject: [PATCH 3/8] chore: ut --- .../src/ws/ohlcv/OHLCVService.test.ts | 243 +++++++++++++++++- .../core-backend/src/ws/ohlcv/OHLCVService.ts | 23 +- 2 files changed, 251 insertions(+), 15 deletions(-) diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts index 27a3b22cc57..ffd26479c83 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts @@ -84,7 +84,7 @@ const getMessenger = (): { }); const mockConnect = jest.fn(); - const mockForceReconnection = jest.fn(); + const mockForceReconnection = jest.fn().mockResolvedValue(undefined); const mockSubscribe = jest.fn(); const mockChannelHasSubscription = jest.fn().mockReturnValue(false); const mockGetSubscriptionsByChannel = jest.fn().mockReturnValue([]); @@ -473,6 +473,181 @@ describe('OHLCVService', () => { expect(mockUnsub).toHaveBeenCalledTimes(1); }); }); + it('should schedule unsubscribe retry when flush fails during channel switch', async () => { + const opts1h: OHLCVSubscriptionOptions = { + ...SUB_OPTS, + interval: '1h', + }; + + await withService(async ({ service, mocks, messenger }) => { + const errorListener = jest.fn(); + messenger.subscribe('OHLCVService:subscriptionError', errorListener); + + mocks.getSubscriptionsByChannel.mockImplementation((channel: string) => { + if (channel === EXPECTED_CHANNEL) { + throw new Error('flush unsub failed'); + } + return [{ unsubscribe: jest.fn().mockResolvedValue(undefined) }]; + }); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + await service.subscribe(opts1h); + + expect(errorListener).toHaveBeenCalledWith({ + channel: EXPECTED_CHANNEL, + error: expect.stringContaining('flush unsub failed'), + operation: 'unsubscribe', + }); + + jest.advanceTimersByTime(1000); + await completeAsyncOperations(); + + expect(mocks.getSubscriptionsByChannel.mock.calls.length).toBeGreaterThan( + 1, + ); + expect(mocks.forceReconnection).not.toHaveBeenCalled(); + }); + }); + + it('should skip grace-period unsubscribe when a new subscriber arrives first', async () => { + await withService(async ({ service, mocks }) => { + const mockUnsub = jest.fn().mockResolvedValue(undefined); + mocks.getSubscriptionsByChannel.mockReturnValue([ + { unsubscribe: mockUnsub }, + ]); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + let releaseConnect!: () => void; + mocks.connect.mockImplementation( + () => + new Promise((resolve) => { + releaseConnect = resolve; + }), + ); + mocks.channelHasSubscription.mockReturnValue(true); + + const subscribePromise = service.subscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await flushPromises(); + + releaseConnect(); + mocks.connect.mockResolvedValue(undefined); + await subscribePromise; + await completeAsyncOperations(); + + expect(mockUnsub).not.toHaveBeenCalled(); + }); + }); + }); + + // =========================================================================== + // Unsubscribe retry + // =========================================================================== + + describe('unsubscribe retry', () => { + it('should succeed on a later retry without forcing reconnection', async () => { + await withService(async ({ service, mocks }) => { + const mockUnsub = jest + .fn() + .mockRejectedValueOnce(new Error('transient')) + .mockResolvedValue(undefined); + + mocks.getSubscriptionsByChannel.mockReturnValue([ + { unsubscribe: mockUnsub }, + ]); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await completeAsyncOperations(); + + jest.advanceTimersByTime(1000); + await completeAsyncOperations(); + + expect(mockUnsub).toHaveBeenCalledTimes(2); + expect(mocks.forceReconnection).not.toHaveBeenCalled(); + }); + }); + + it('should recreate channel tracking when unsubscribe fails after tracking was cleared', async () => { + await withService(async ({ service, mocks }) => { + let releaseUnsub!: (error?: Error) => void; + mocks.getSubscriptionsByChannel.mockReturnValue([ + { + unsubscribe: () => + new Promise((_resolve, reject) => { + releaseUnsub = reject; + }), + }, + ]); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await flushPromises(); + + service.destroy(); + releaseUnsub(new Error('ws gone')); + await completeAsyncOperations(); + + jest.advanceTimersByTime(1000); + await completeAsyncOperations(); + + expect(mocks.getSubscriptionsByChannel.mock.calls.length).toBeGreaterThan( + 1, + ); + }); + }); + + it('should force reconnection when unsubscribe retries are exhausted', async () => { + await withService(async ({ service, mocks }) => { + mocks.getSubscriptionsByChannel.mockImplementation(() => { + throw new Error('ws gone'); + }); + mocks.forceReconnection.mockRejectedValue(new Error('reconnect failed')); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await completeAsyncOperations(); + jest.advanceTimersByTime(1000); + await completeAsyncOperations(); + jest.advanceTimersByTime(2000); + await completeAsyncOperations(); + jest.advanceTimersByTime(4000); + await completeAsyncOperations(); + + expect(mocks.forceReconnection).toHaveBeenCalledTimes(1); + }); + }); + + it('should clear retry timers when destroy is called during retry backoff', async () => { + await withService(async ({ service, mocks }) => { + mocks.getSubscriptionsByChannel.mockImplementation(() => { + throw new Error('ws gone'); + }); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await completeAsyncOperations(); + + service.destroy(); + + jest.advanceTimersByTime(10000); + await completeAsyncOperations(); + + expect(mocks.forceReconnection).not.toHaveBeenCalled(); + }); + }); }); // =========================================================================== @@ -876,6 +1051,46 @@ describe('OHLCVService', () => { ); }); }); + + it('should not publish chainStatusChanged down when no chains are tracked', async () => { + await withService(async ({ messenger, rootMessenger }) => { + const statusListener = jest.fn(); + messenger.subscribe('OHLCVService:chainStatusChanged', statusListener); + + rootMessenger.publish( + 'BackendWebSocketService:connectionStateChanged', + { + ...BASE_CONNECTION_INFO, + state: WebSocketState.DISCONNECTED, + connectedAt: undefined, + reconnectAttempts: 0, + }, + ); + await completeAsyncOperations(); + + expect(statusListener).not.toHaveBeenCalled(); + }); + }); + + it('should ignore non-terminal connection states', async () => { + await withService(async ({ service, mocks, rootMessenger }) => { + await service.subscribe(SUB_OPTS); + mocks.connect.mockClear(); + mocks.subscribe.mockClear(); + + rootMessenger.publish( + 'BackendWebSocketService:connectionStateChanged', + { + ...BASE_CONNECTION_INFO, + state: WebSocketState.CONNECTING, + }, + ); + await completeAsyncOperations(); + + expect(mocks.connect).not.toHaveBeenCalled(); + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + }); }); // =========================================================================== @@ -1009,6 +1224,32 @@ describe('OHLCVService', () => { }); }); + it('should recreate WS subscription when retry is cancelled but server subscription is gone', async () => { + await withService(async ({ service, mocks }) => { + mocks.getSubscriptionsByChannel.mockImplementation(() => { + throw new Error('ws gone'); + }); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await completeAsyncOperations(); + + mocks.channelHasSubscription.mockReturnValue(false); + mocks.getSubscriptionsByChannel.mockReturnValue([ + { unsubscribe: jest.fn().mockResolvedValue(undefined) }, + ]); + mocks.subscribe.mockClear(); + mocks.connect.mockClear(); + + await service.subscribe(SUB_OPTS); + + expect(mocks.connect).toHaveBeenCalledTimes(1); + expect(mocks.subscribe).toHaveBeenCalledTimes(1); + }); + }); + it('should delete failed-cleanup channels on reconnect', async () => { await withService(async ({ service, mocks, rootMessenger }) => { mocks.getSubscriptionsByChannel.mockImplementation(() => { diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index 7a3f6a9e53f..61f242e7485 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -360,9 +360,8 @@ export class OHLCVService { entry.gracePeriodTimer = setTimeout(() => { entry.gracePeriodTimer = undefined; - this.#performUnsubscribe(channel).catch(() => { - // no-op - }); + // eslint-disable-next-line @typescript-eslint/no-floating-promises + this.#performUnsubscribe(channel); }, GRACE_PERIOD_MS); } @@ -431,13 +430,8 @@ export class OHLCVService { } #scheduleUnsubscribeRetry(channel: string): void { - let entry = this.#channels.get(channel); - if (!entry) { - entry = { refCount: 0 }; - this.#channels.set(channel, entry); - } - - const channelEntry = entry; + const channelEntry = this.#channels.get(channel) ?? { refCount: 0 }; + this.#channels.set(channel, channelEntry); const retryCount = (channelEntry.retryCount ?? 0) + 1; channelEntry.retryCount = retryCount; @@ -465,9 +459,8 @@ export class OHLCVService { channelEntry.retryTimer = setTimeout(() => { channelEntry.retryTimer = undefined; - this.#performUnsubscribe(channel).catch(() => { - // no-op - }); + // eslint-disable-next-line @typescript-eslint/no-floating-promises + this.#performUnsubscribe(channel); }, delayMs); } @@ -487,7 +480,9 @@ export class OHLCVService { channel, }); - this.#clearChannelTimers(entry ?? { refCount: 0 }); + this.#clearChannelTimers( + this.#channels.get(channel) ?? { refCount: 0 }, + ); const success = await this.#unsubscribeChannelOnServer(channel); if (success) { From ae60f4e5c2b8655c5d3ef72da71e55e528294127 Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Tue, 28 Jul 2026 15:28:51 +0200 Subject: [PATCH 4/8] chore: lint --- .../src/ws/ohlcv/OHLCVService.test.ts | 32 +++++++++++-------- .../core-backend/src/ws/ohlcv/OHLCVService.ts | 4 +-- 2 files changed, 19 insertions(+), 17 deletions(-) diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts index ffd26479c83..2269b03ed7c 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts @@ -483,12 +483,14 @@ describe('OHLCVService', () => { const errorListener = jest.fn(); messenger.subscribe('OHLCVService:subscriptionError', errorListener); - mocks.getSubscriptionsByChannel.mockImplementation((channel: string) => { - if (channel === EXPECTED_CHANNEL) { - throw new Error('flush unsub failed'); - } - return [{ unsubscribe: jest.fn().mockResolvedValue(undefined) }]; - }); + mocks.getSubscriptionsByChannel.mockImplementation( + (channel: string) => { + if (channel === EXPECTED_CHANNEL) { + throw new Error('flush unsub failed'); + } + return [{ unsubscribe: jest.fn().mockResolvedValue(undefined) }]; + }, + ); await service.subscribe(SUB_OPTS); await service.unsubscribe(SUB_OPTS); @@ -503,9 +505,9 @@ describe('OHLCVService', () => { jest.advanceTimersByTime(1000); await completeAsyncOperations(); - expect(mocks.getSubscriptionsByChannel.mock.calls.length).toBeGreaterThan( - 1, - ); + expect( + mocks.getSubscriptionsByChannel.mock.calls.length, + ).toBeGreaterThan(1); expect(mocks.forceReconnection).not.toHaveBeenCalled(); }); }); @@ -579,7 +581,7 @@ describe('OHLCVService', () => { let releaseUnsub!: (error?: Error) => void; mocks.getSubscriptionsByChannel.mockReturnValue([ { - unsubscribe: () => + unsubscribe: (): Promise => new Promise((_resolve, reject) => { releaseUnsub = reject; }), @@ -599,9 +601,9 @@ describe('OHLCVService', () => { jest.advanceTimersByTime(1000); await completeAsyncOperations(); - expect(mocks.getSubscriptionsByChannel.mock.calls.length).toBeGreaterThan( - 1, - ); + expect( + mocks.getSubscriptionsByChannel.mock.calls.length, + ).toBeGreaterThan(1); }); }); @@ -610,7 +612,9 @@ describe('OHLCVService', () => { mocks.getSubscriptionsByChannel.mockImplementation(() => { throw new Error('ws gone'); }); - mocks.forceReconnection.mockRejectedValue(new Error('reconnect failed')); + mocks.forceReconnection.mockRejectedValue( + new Error('reconnect failed'), + ); await service.subscribe(SUB_OPTS); await service.unsubscribe(SUB_OPTS); diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index 61f242e7485..8647c10994b 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -480,9 +480,7 @@ export class OHLCVService { channel, }); - this.#clearChannelTimers( - this.#channels.get(channel) ?? { refCount: 0 }, - ); + this.#clearChannelTimers(this.#channels.get(channel) ?? { refCount: 0 }); const success = await this.#unsubscribeChannelOnServer(channel); if (success) { From 2107b2f421aa5fa6bc9adad4d3dc538f455b7633 Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Tue, 28 Jul 2026 16:04:10 +0200 Subject: [PATCH 5/8] chore: fix flush unused sub on return to same token --- .../src/ws/ohlcv/OHLCVService.test.ts | 94 +++++++++++++++++++ .../core-backend/src/ws/ohlcv/OHLCVService.ts | 2 + 2 files changed, 96 insertions(+) diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts index 2269b03ed7c..2989420a3b6 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts @@ -438,6 +438,52 @@ describe('OHLCVService', () => { }); }); + it('should flush other grace channels when re-subscribing to the same channel during grace', async () => { + const opts1h: OHLCVSubscriptionOptions = { + ...SUB_OPTS, + interval: '1h', + }; + const channel1h = + 'market-data.v1.eip155:8453/erc20:0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913.1h.usd'; + + await withService(async ({ service, mocks }) => { + const mockUnsub1m = jest.fn().mockResolvedValue(undefined); + const mockUnsub1h = jest.fn().mockResolvedValue(undefined); + + mocks.getSubscriptionsByChannel.mockImplementation( + (channel: string) => { + if (channel === EXPECTED_CHANNEL) { + throw new Error('flush unsub failed'); + } + if (channel === channel1h) { + return [{ unsubscribe: mockUnsub1h }]; + } + return [{ unsubscribe: mockUnsub1m }]; + }, + ); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + await service.subscribe(opts1h); + await service.unsubscribe(opts1h); + + mocks.getSubscriptionsByChannel.mockClear(); + mockUnsub1h.mockClear(); + mocks.channelHasSubscription.mockReturnValue(true); + mocks.connect.mockClear(); + mocks.subscribe.mockClear(); + + await service.subscribe(opts1h); + + expect(mocks.getSubscriptionsByChannel).toHaveBeenCalledWith( + EXPECTED_CHANNEL, + ); + expect(mockUnsub1h).not.toHaveBeenCalled(); + expect(mocks.connect).not.toHaveBeenCalled(); + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + }); + it('should flush the old channel immediately when subscribing to a different interval', async () => { const opts1m = SUB_OPTS; const opts1h: OHLCVSubscriptionOptions = { @@ -1228,6 +1274,54 @@ describe('OHLCVService', () => { }); }); + it('should flush other grace channels when reusing a channel pending unsubscribe retry', async () => { + const opts1h: OHLCVSubscriptionOptions = { + ...SUB_OPTS, + interval: '1h', + }; + const channel1h = + 'market-data.v1.eip155:8453/erc20:0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913.1h.usd'; + + await withService(async ({ service, mocks }) => { + const mockUnsub1h = jest.fn().mockResolvedValue(undefined); + + mocks.getSubscriptionsByChannel.mockImplementation(() => { + throw new Error('ws gone'); + }); + + await service.subscribe(SUB_OPTS); + await service.unsubscribe(SUB_OPTS); + + jest.advanceTimersByTime(3000); + await completeAsyncOperations(); + + mocks.getSubscriptionsByChannel.mockImplementation( + (channel: string) => { + if (channel === channel1h) { + return [{ unsubscribe: mockUnsub1h }]; + } + throw new Error('ws gone'); + }, + ); + + await service.subscribe(opts1h); + await service.unsubscribe(opts1h); + + mocks.getSubscriptionsByChannel.mockClear(); + mockUnsub1h.mockClear(); + mocks.channelHasSubscription.mockReturnValue(true); + mocks.connect.mockClear(); + mocks.subscribe.mockClear(); + + await service.subscribe(SUB_OPTS); + + expect(mocks.getSubscriptionsByChannel).toHaveBeenCalledWith(channel1h); + expect(mockUnsub1h).toHaveBeenCalledTimes(1); + expect(mocks.connect).not.toHaveBeenCalled(); + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + }); + it('should recreate WS subscription when retry is cancelled but server subscription is gone', async () => { await withService(async ({ service, mocks }) => { mocks.getSubscriptionsByChannel.mockImplementation(() => { diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index 8647c10994b..79966356db3 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -250,6 +250,7 @@ export class OHLCVService { channel, ) ) { + await this.#flushOtherChannels(channel); log('OHLCV-WS: Cancelled unsubscribe retry — reusing WS subscription', { channel, }); @@ -269,6 +270,7 @@ export class OHLCVService { channel, ) ) { + await this.#flushOtherChannels(channel); entry.refCount += 1; log('OHLCV-WS: WS subscription still alive, bumped refCount', { channel, From 74a7e01cf6cd0ab9054151e4455866da8a70d7ce Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Wed, 29 Jul 2026 14:09:53 +0200 Subject: [PATCH 6/8] feat: se Cockatiel for OHLCV unsubscribe retries --- packages/core-backend/package.json | 1 + .../src/ws/ohlcv/OHLCVService.test.ts | 9 +- .../core-backend/src/ws/ohlcv/OHLCVService.ts | 100 +++++++++++------- yarn.lock | 1 + 4 files changed, 65 insertions(+), 46 deletions(-) diff --git a/packages/core-backend/package.json b/packages/core-backend/package.json index a565b4708f6..181e977284b 100644 --- a/packages/core-backend/package.json +++ b/packages/core-backend/package.json @@ -64,6 +64,7 @@ "@metamask/utils": "^11.11.0", "@tanstack/query-core": "^5.62.16", "async-mutex": "^0.5.0", + "cockatiel": "^3.1.2", "uuid": "^8.3.2" }, "devDependencies": { diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts index 2989420a3b6..60eabc07703 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.test.ts @@ -622,7 +622,7 @@ describe('OHLCVService', () => { }); }); - it('should recreate channel tracking when unsubscribe fails after tracking was cleared', async () => { + it('should not retry after destroy aborts in-flight unsubscribe', async () => { await withService(async ({ service, mocks }) => { let releaseUnsub!: (error?: Error) => void; mocks.getSubscriptionsByChannel.mockReturnValue([ @@ -647,9 +647,8 @@ describe('OHLCVService', () => { jest.advanceTimersByTime(1000); await completeAsyncOperations(); - expect( - mocks.getSubscriptionsByChannel.mock.calls.length, - ).toBeGreaterThan(1); + expect(mocks.getSubscriptionsByChannel).toHaveBeenCalledTimes(1); + expect(mocks.forceReconnection).not.toHaveBeenCalled(); }); }); @@ -678,7 +677,7 @@ describe('OHLCVService', () => { }); }); - it('should clear retry timers when destroy is called during retry backoff', async () => { + it('should abort in-flight unsubscribe retry when destroy is called during backoff', async () => { await withService(async ({ service, mocks }) => { mocks.getSubscriptionsByChannel.mockImplementation(() => { throw new Error('ws gone'); diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index 79966356db3..b931a940a20 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -14,6 +14,11 @@ import type { } from '@metamask/controller-utils'; import type { Messenger } from '@metamask/messenger'; import { Mutex } from 'async-mutex'; +import { + handleAll, + IterableBackoff, + retry, +} from 'cockatiel'; import { projectLogger, createModuleLogger } from '../../logger.js'; import type { BackendWebSocketServiceMethodActions } from '../BackendWebSocketService-method-action-types.js'; @@ -46,8 +51,10 @@ const GRACE_PERIOD_MS = 3_000; /** Backoff delays for retrying failed WebSocket unsubscribes. */ const UNSUB_RETRY_DELAYS_MS = [1_000, 2_000, 4_000] as const; -/** Max unsubscribe retries before forcing a WebSocket reconnection. */ -const MAX_UNSUB_RETRIES = 3; +const unsubRetryPolicy = retry(handleAll, { + maxAttempts: 3, + backoff: new IterableBackoff([...UNSUB_RETRY_DELAYS_MS]), +}); // ============================================================================= // Types — Channel Tracking @@ -56,8 +63,7 @@ const MAX_UNSUB_RETRIES = 3; type ChannelEntry = { refCount: number; gracePeriodTimer?: ReturnType; - retryTimer?: ReturnType; - retryCount?: number; + retryAbort?: AbortController; }; // ============================================================================= @@ -238,10 +244,9 @@ export class OHLCVService { async #subscribeInner(channel: string): Promise { const entry = this.#channels.get(channel); - if (entry?.retryTimer) { - clearTimeout(entry.retryTimer); - entry.retryTimer = undefined; - entry.retryCount = 0; + if (entry?.retryAbort) { + entry.retryAbort.abort(); + entry.retryAbort = undefined; entry.refCount = 1; if ( @@ -403,10 +408,8 @@ export class OHLCVService { clearTimeout(entry.gracePeriodTimer); entry.gracePeriodTimer = undefined; } - if (entry.retryTimer) { - clearTimeout(entry.retryTimer); - entry.retryTimer = undefined; - } + entry.retryAbort?.abort(); + entry.retryAbort = undefined; } async #unsubscribeChannelOnServer(channel: string): Promise { @@ -432,38 +435,59 @@ export class OHLCVService { } #scheduleUnsubscribeRetry(channel: string): void { - const channelEntry = this.#channels.get(channel) ?? { refCount: 0 }; - this.#channels.set(channel, channelEntry); + const entry = this.#channels.get(channel) ?? { refCount: 0 }; + entry.retryAbort?.abort(); + entry.retryAbort = new AbortController(); + this.#channels.set(channel, entry); + + const { signal } = entry.retryAbort; + + // eslint-disable-next-line @typescript-eslint/no-floating-promises + this.#runUnsubRetryLoop(channel, signal); + } + + async #runUnsubRetryLoop( + channel: string, + signal: AbortSignal, + ): Promise { + try { + await unsubRetryPolicy.execute(async () => { + const releaseLock = await this.#mutex.acquire(); + try { + const current = this.#channels.get(channel); + if (current && current.refCount > 0) { + return; + } + + const success = await this.#unsubscribeChannelOnServer(channel); + if (!success) { + throw new Error('unsubscribe failed'); + } - const retryCount = (channelEntry.retryCount ?? 0) + 1; - channelEntry.retryCount = retryCount; + this.#channels.delete(channel); + log('OHLCV-WS: WS unsubscribe completed', { channel }); + } finally { + releaseLock(); + } + }, signal); + } catch { + if (signal.aborted) { + return; + } - if (retryCount > MAX_UNSUB_RETRIES) { log('OHLCV-WS: Unsubscribe retries exhausted — forcing reconnection', { channel, }); - this.#clearChannelTimers(channelEntry); this.#channels.delete(channel); - this.#messenger + // Last resort: reconnects the shared BackendWebSocketService instance + // (AccountActivityService and other consumers share this connection). + // They resubscribe on CONNECTED; OHLCV only resubscribes refCount > 0. + await this.#messenger .call('BackendWebSocketService:forceReconnection') .catch(() => { // no-op }); - return; } - - const delayMs = UNSUB_RETRY_DELAYS_MS[retryCount - 1]; - log('OHLCV-WS: Scheduling unsubscribe retry', { - channel, - retryCount, - delayMs, - }); - - channelEntry.retryTimer = setTimeout(() => { - channelEntry.retryTimer = undefined; - // eslint-disable-next-line @typescript-eslint/no-floating-promises - this.#performUnsubscribe(channel); - }, delayMs); } async #performUnsubscribe(channel: string): Promise { @@ -483,17 +507,11 @@ export class OHLCVService { }); this.#clearChannelTimers(this.#channels.get(channel) ?? { refCount: 0 }); - - const success = await this.#unsubscribeChannelOnServer(channel); - if (success) { - this.#channels.delete(channel); - log('OHLCV-WS: WS unsubscribe completed', { channel }); - } else { - this.#scheduleUnsubscribeRetry(channel); - } } finally { releaseLock(); } + + this.#scheduleUnsubscribeRetry(channel); } /** diff --git a/yarn.lock b/yarn.lock index ced16eb9d50..3556fa920b7 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6586,6 +6586,7 @@ __metadata: "@ts-bridge/cli": "npm:^0.6.4" "@types/jest": "npm:^30.0.0" async-mutex: "npm:^0.5.0" + cockatiel: "npm:^3.1.2" deepmerge: "npm:^4.2.2" jest: "npm:^30.4.2" jest-environment-jsdom: "npm:^30.4.1" From a2e47cf2d3560640c5ab1635ec00363d1e658a44 Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Wed, 29 Jul 2026 14:33:49 +0200 Subject: [PATCH 7/8] feat: derive OHLCV unsub maxAttempts from backoff delays --- packages/core-backend/src/ws/ohlcv/OHLCVService.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index b931a940a20..01ef27e7c2c 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -48,11 +48,13 @@ const SYSTEM_NOTIFICATIONS_CHANNEL = `system-notifications.v1.${SUBSCRIPTION_NAM /** Delay before actually unsubscribing from a channel after refCount reaches 0. */ const GRACE_PERIOD_MS = 3_000; -/** Backoff delays for retrying failed WebSocket unsubscribes. */ +/** Backoff delays between failed WebSocket unsubscribe attempts. */ const UNSUB_RETRY_DELAYS_MS = [1_000, 2_000, 4_000] as const; const unsubRetryPolicy = retry(handleAll, { - maxAttempts: 3, + // Cockatiel stops retrying once the failure index reaches `maxAttempts`, so + // `length` here yields one initial attempt plus three delayed retries (4 total). + maxAttempts: UNSUB_RETRY_DELAYS_MS.length, backoff: new IterableBackoff([...UNSUB_RETRY_DELAYS_MS]), }); From 54e16dd9ab40347dd4e40304cf2eed8e1a5d0cea Mon Sep 17 00:00:00 2001 From: sahar-fehri Date: Wed, 29 Jul 2026 14:47:41 +0200 Subject: [PATCH 8/8] feat: lint --- packages/core-backend/src/ws/ohlcv/OHLCVService.ts | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts index 01ef27e7c2c..c6110ae0354 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -14,11 +14,7 @@ import type { } from '@metamask/controller-utils'; import type { Messenger } from '@metamask/messenger'; import { Mutex } from 'async-mutex'; -import { - handleAll, - IterableBackoff, - retry, -} from 'cockatiel'; +import { handleAll, IterableBackoff, retry } from 'cockatiel'; import { projectLogger, createModuleLogger } from '../../logger.js'; import type { BackendWebSocketServiceMethodActions } from '../BackendWebSocketService-method-action-types.js';