diff --git a/packages/core-backend/CHANGELOG.md b/packages/core-backend/CHANGELOG.md index edfe6a3ca8a..9f177b95f06 100644 --- a/packages/core-backend/CHANGELOG.md +++ b/packages/core-backend/CHANGELOG.md @@ -7,10 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -### Fixed +### Changed - **BREAKING:** Align `V6_DEFI_POSITION_TYPES` (and inferred `V6DeFiPositionType`) with Accounts API / Zerion wallet fungible position types: `deposit`, `loan`, `locked`, `staked`, `reward`, `wallet`, `investment` ([#9683](https://github.com/MetaMask/core/pull/9683)) +### 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 ([#9678](https://github.com/MetaMask/core/pull/9678)) + ## [7.0.0] ### Added 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 8a4709effcf..60eabc07703 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([]); @@ -438,12 +438,60 @@ describe('OHLCVService', () => { }); }); - it('should unsubscribe old channel after grace period during time-range switching', async () => { + 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 = { ...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,20 +499,204 @@ 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); }); }); + 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 not retry after destroy aborts in-flight unsubscribe', async () => { + await withService(async ({ service, mocks }) => { + let releaseUnsub!: (error?: Error) => void; + mocks.getSubscriptionsByChannel.mockReturnValue([ + { + unsubscribe: (): Promise => + 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).toHaveBeenCalledTimes(1); + expect(mocks.forceReconnection).not.toHaveBeenCalled(); + }); + }); + + 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 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'); + }); + + 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(); + }); + }); }); // =========================================================================== @@ -868,6 +1100,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(); + }); + }); }); // =========================================================================== @@ -938,7 +1210,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 +1231,147 @@ 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 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(() => { + 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(() => { + 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 +1509,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 +1519,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..c6110ae0354 100644 --- a/packages/core-backend/src/ws/ohlcv/OHLCVService.ts +++ b/packages/core-backend/src/ws/ohlcv/OHLCVService.ts @@ -14,6 +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 { projectLogger, createModuleLogger } from '../../logger.js'; import type { BackendWebSocketServiceMethodActions } from '../BackendWebSocketService-method-action-types.js'; @@ -43,6 +44,16 @@ 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 between failed WebSocket unsubscribe attempts. */ +const UNSUB_RETRY_DELAYS_MS = [1_000, 2_000, 4_000] as const; + +const unsubRetryPolicy = retry(handleAll, { + // 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]), +}); + // ============================================================================= // Types — Channel Tracking // ============================================================================= @@ -50,6 +61,7 @@ const GRACE_PERIOD_MS = 3_000; type ChannelEntry = { refCount: number; gracePeriodTimer?: ReturnType; + retryAbort?: AbortController; }; // ============================================================================= @@ -142,7 +154,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 +242,25 @@ export class OHLCVService { async #subscribeInner(channel: string): Promise { const entry = this.#channels.get(channel); - if (entry?.gracePeriodTimer) { + if (entry?.retryAbort) { + entry.retryAbort.abort(); + entry.retryAbort = undefined; + entry.refCount = 1; + + if ( + this.#messenger.call( + 'BackendWebSocketService:channelHasSubscription', + channel, + ) + ) { + await this.#flushOtherChannels(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', { @@ -241,6 +273,7 @@ export class OHLCVService { channel, ) ) { + await this.#flushOtherChannels(channel); entry.refCount += 1; log('OHLCV-WS: WS subscription still alive, bumped refCount', { channel, @@ -254,6 +287,9 @@ export class OHLCVService { entry.refCount += 1; return; } + + await this.#flushOtherChannels(channel); + try { await this.#messenger.call('BackendWebSocketService:connect'); @@ -329,9 +365,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); } @@ -339,6 +374,120 @@ 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; + } + entry.retryAbort?.abort(); + entry.retryAbort = 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 { + 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'); + } + + this.#channels.delete(channel); + log('OHLCV-WS: WS unsubscribe completed', { channel }); + } finally { + releaseLock(); + } + }, signal); + } catch { + if (signal.aborted) { + return; + } + + log('OHLCV-WS: Unsubscribe retries exhausted — forcing reconnection', { + channel, + }); + this.#channels.delete(channel); + // 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 + }); + } + } + async #performUnsubscribe(channel: string): Promise { const releaseLock = await this.#mutex.acquire(); try { @@ -354,29 +503,13 @@ 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, - ); - - for (const sub of subscriptions) { - await sub.unsubscribe(); - } - 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', - }); - } + this.#clearChannelTimers(this.#channels.get(channel) ?? { refCount: 0 }); } finally { releaseLock(); } + + this.#scheduleUnsubscribeRetry(channel); } /** @@ -391,8 +524,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 +669,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(); diff --git a/yarn.lock b/yarn.lock index 9a56be86c58..6f947ccbb81 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"