From 8ec172f48eefa2237dc0c16919fbb2e726b39a65 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Sat, 12 Sep 2026 15:37:56 +0300 Subject: [PATCH] fix: recover portfolio instrument metadata on backend Bound metadata concurrency and waiting, retry transient lookups, and retain known metadata during failures. Isolate in-flight lookups by token and preserve financial observation time. --- src/tbank/instrument-metadata.service.spec.ts | 229 ++++++++++++++++++ src/tbank/instrument-metadata.service.ts | 70 ++++-- src/tbank/portfolio.service.spec.ts | 74 ++++++ src/tbank/portfolio.service.ts | 40 ++- src/tbank/tbank.client.spec.ts | 186 +++++++++++++- src/tbank/tbank.client.ts | 83 ++++--- 6 files changed, 621 insertions(+), 61 deletions(-) diff --git a/src/tbank/instrument-metadata.service.spec.ts b/src/tbank/instrument-metadata.service.spec.ts index 4cc5848..769a43d 100644 --- a/src/tbank/instrument-metadata.service.spec.ts +++ b/src/tbank/instrument-metadata.service.spec.ts @@ -1,4 +1,7 @@ import { InstrumentMetadataService } from './instrument-metadata.service'; +import { ServiceUnavailableException } from '@nestjs/common'; +import type { LRUCache } from 'lru-cache'; +import type { TbankInstrument } from './tbank.client'; describe('InstrumentMetadataService', () => { const tbank = { @@ -8,6 +11,13 @@ describe('InstrumentMetadataService', () => { const logger = { event: jest.fn() }; let service: InstrumentMetadataService; + function rememberExpired(key: string, value: TbankInstrument): void { + const cache = ( + service as unknown as { metadata: LRUCache } + ).metadata; + cache.set(key, value, { ttl: 1, start: performance.now() - 1000 }); + } + beforeEach(() => { jest.clearAllMocks(); tbank.getInstrumentByFigi.mockReset(); @@ -58,4 +68,223 @@ describe('InstrumentMetadataService', () => { expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(1); }); + + it('tries FIGI when UID metadata has no usable name or ticker', async () => { + tbank.getInstrumentByUid.mockResolvedValue({ name: ' ', ticker: '' }); + tbank.getInstrumentByFigi.mockResolvedValue({ name: 'Облигация' }); + await expect( + service.resolve('token', 'uid-1', 'figi-1'), + ).resolves.toMatchObject({ name: 'Облигация' }); + }); + + it('retains expired successful metadata through temporary failures and refreshes it later', async () => { + const known = { + name: 'Облигация', + ticker: 'BOND', + brand: { logoName: 'bond.png' }, + }; + tbank.getInstrumentByUid.mockResolvedValueOnce(known); + await service.resolve('token', 'uid-1', null); + rememberExpired('uid:uid-1', known); + tbank.getInstrumentByUid.mockRejectedValue( + new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }), + ); + await expect(service.resolve('token', 'uid-1', null)).resolves.toEqual( + known, + ); + await expect(service.resolve('token', 'uid-1', null)).resolves.toEqual( + known, + ); + tbank.getInstrumentByUid.mockResolvedValue({ + ...known, + name: 'Новое название', + }); + await expect( + service.resolve('token', 'uid-1', null), + ).resolves.toMatchObject({ name: 'Новое название' }); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(4); + }); + + it('does not mask an invalid token as missing metadata or retry by FIGI', async () => { + const error = new Error('INVALID_TBANK_TOKEN'); + tbank.getInstrumentByUid.mockRejectedValue(error); + await expect(service.resolve('token', 'uid-1', 'figi-1')).rejects.toBe( + error, + ); + expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); + }); + + it('keeps known metadata through an incomplete refresh without renewing its freshness', async () => { + const known = { name: 'Облигация', ticker: 'BOND' }; + rememberExpired('uid:uid-1', known); + tbank.getInstrumentByUid + .mockResolvedValueOnce({ name: ' ', ticker: '', isin: 'RU0000000001' }) + .mockRejectedValueOnce( + new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }), + ); + + await expect(service.resolve('token', 'uid-1', 'figi-1')).resolves.toEqual( + known, + ); + await expect(service.resolve('token', 'uid-1', 'figi-1')).resolves.toEqual( + known, + ); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(2); + expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); + }); + + it('keeps a known FIGI name after partial UID metadata and does not cache the partial response', async () => { + const known = { name: 'Облигация', ticker: 'BOND' }; + rememberExpired('figi:figi-1', known); + tbank.getInstrumentByUid + .mockResolvedValueOnce({ isin: 'RU0000000001' }) + .mockResolvedValueOnce({ name: 'Новое название' }); + tbank.getInstrumentByFigi.mockRejectedValue( + new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }), + ); + + await expect(service.resolve('token', 'uid-1', 'figi-1')).resolves.toEqual( + known, + ); + await expect(service.resolve('token', 'uid-1', 'figi-1')).resolves.toEqual({ + name: 'Новое название', + }); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(2); + expect(tbank.getInstrumentByFigi).toHaveBeenCalledTimes(1); + }); + + it('returns cold partial fields without caching them', async () => { + const partial = { isin: 'RU0000000001' }; + tbank.getInstrumentByUid.mockResolvedValue(partial); + + await expect(service.resolve('token', 'uid-1', null)).resolves.toEqual( + partial, + ); + await expect(service.resolve('token', 'uid-1', null)).resolves.toEqual( + partial, + ); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(2); + }); + + it('replaces an expired name with authoritative confirmed absence', async () => { + rememberExpired('uid:uid-1', { name: 'Старое название' }); + tbank.getInstrumentByUid.mockResolvedValue(null); + + await expect(service.resolve('token', 'uid-1', null)).resolves.toBeNull(); + await expect(service.resolve('token', 'uid-1', null)).resolves.toBeNull(); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(1); + }); + + it('isolates in-flight credentials while sharing successful public metadata', async () => { + const error = new Error('INVALID_TBANK_TOKEN'); + const known = { name: 'Облигация' }; + let rejectInvalid!: (reason: Error) => void; + tbank.getInstrumentByUid.mockImplementation((token: string) => + token === 'invalid-token' + ? new Promise((_resolve, reject) => { + rejectInvalid = reject; + }) + : Promise.resolve(known), + ); + + const invalid = expect( + service.resolve('invalid-token', 'uid-1', null), + ).rejects.toBe(error); + const valid = service.resolve('valid-token', 'uid-1', null); + rejectInvalid(error); + + await invalid; + await expect(valid).resolves.toEqual(known); + await expect( + service.resolve('another-token', 'uid-1', null), + ).resolves.toEqual(known); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(2); + }); + + it('returns cached or nullable metadata without starting requests after abort', async () => { + const known = { name: 'Облигация' }; + rememberExpired('figi:figi-1', known); + const controller = new AbortController(); + controller.abort(); + + await expect( + service.resolve('token', 'uid-1', 'figi-1', controller.signal), + ).resolves.toEqual(known); + await expect( + service.resolve('token', 'uid-2', 'figi-2', controller.signal), + ).resolves.toBeNull(); + expect(tbank.getInstrumentByUid).not.toHaveBeenCalled(); + expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); + }); + + it('returns expired metadata promptly when the waiting caller aborts', async () => { + const known = { name: 'Облигация' }; + rememberExpired('uid:uid-1', known); + tbank.getInstrumentByUid.mockReturnValue(new Promise(() => {})); + const controller = new AbortController(); + const result = service.resolve( + 'token', + 'uid-1', + 'figi-1', + controller.signal, + ); + + controller.abort(); + + await expect(result).resolves.toEqual(known); + expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); + }); + + it('keeps a shared fetch alive after one caller aborts and caches its result', async () => { + const known = { name: 'Облигация' }; + let finish!: (value: TbankInstrument) => void; + tbank.getInstrumentByUid.mockReturnValue( + new Promise((resolve) => { + finish = resolve; + }), + ); + const controller = new AbortController(); + const aborted = service.resolve( + 'token', + 'uid-1', + 'figi-1', + controller.signal, + ); + controller.abort(); + await expect(aborted).resolves.toBeNull(); + + const waiting = service.resolve('token', 'uid-1', null); + finish(known); + + await expect(waiting).resolves.toEqual(known); + await expect( + service.resolve('another-token', 'uid-1', null), + ).resolves.toEqual(known); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(1); + expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); + }); + + it('does not start FIGI lookup after an aborted UID lookup fails later', async () => { + let fail!: (reason: Error) => void; + tbank.getInstrumentByUid.mockReturnValue( + new Promise((_resolve, reject) => { + fail = reject; + }), + ); + const controller = new AbortController(); + const result = service.resolve( + 'token', + 'uid-1', + 'figi-1', + controller.signal, + ); + controller.abort(); + await expect(result).resolves.toBeNull(); + + fail(new Error('INVALID_TBANK_TOKEN')); + await Promise.resolve(); + await Promise.resolve(); + + expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); + }); }); diff --git a/src/tbank/instrument-metadata.service.ts b/src/tbank/instrument-metadata.service.ts index 21eb0af..53a8db8 100644 --- a/src/tbank/instrument-metadata.service.ts +++ b/src/tbank/instrument-metadata.service.ts @@ -1,5 +1,6 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, ServiceUnavailableException } from '@nestjs/common'; import { LRUCache } from 'lru-cache'; +import { createHash } from 'node:crypto'; import { StructuredLogger } from '../common/logging/structured-logger.service'; import { TbankClient, type TbankInstrument } from './tbank.client'; @@ -15,6 +16,7 @@ export class InstrumentMetadataService { >({ max: maxCachedInstruments, ttl: metadataForMs, + noDeleteOnStaleGet: true, }); private readonly pending = new Map>(); @@ -27,51 +29,79 @@ export class InstrumentMetadataService { token: string, instrumentUid: string | null | undefined, figi: string | null | undefined, + signal?: AbortSignal, ): Promise { const uid = instrumentUid?.trim(); + let byUid: TbankInstrument | null = null; if (uid) { - const byUid = await this.lookup(`uid:${uid}`, () => - this.tbank.getInstrumentByUid(token, uid), + byUid = await this.lookup( + token, + `uid:${uid}`, + () => this.tbank.getInstrumentByUid(token, uid), + signal, ); - if (byUid) return byUid; + if (byUid?.name?.trim() || byUid?.ticker?.trim()) return byUid; } const instrumentFigi = figi?.trim(); - return instrumentFigi - ? this.lookup(`figi:${instrumentFigi}`, () => - this.tbank.getInstrumentByFigi(token, instrumentFigi), + const byFigi = instrumentFigi + ? await this.lookup( + token, + `figi:${instrumentFigi}`, + () => this.tbank.getInstrumentByFigi(token, instrumentFigi), + signal, ) : null; + return byFigi ?? byUid; } private async lookup( + token: string, key: string, fetch: () => Promise, + signal?: AbortSignal, ): Promise { const cached = this.metadata.get(key); if (cached !== undefined) return cached === missingInstrument ? null : cached; - const pending = this.pending.get(key); - if (pending) return pending; - const request = this.fetch(key, fetch); - this.pending.set(key, request); - try { - return await request; - } finally { - this.pending.delete(key); + if (signal?.aborted) return this.lastKnown(key); + const owner = createHash('sha256').update(token).digest('hex'); + const requestKey = `${owner}:${key}`; + let pending = this.pending.get(requestKey); + if (!pending) { + pending = this.fetch(key, fetch, this.lastKnown(key)).finally(() => { + this.pending.delete(requestKey); + }); + this.pending.set(requestKey, pending); } + if (!signal) return pending; + return new Promise((resolve, reject) => { + const onAbort = () => resolve(this.lastKnown(key)); + signal.addEventListener('abort', onAbort, { once: true }); + void pending.then(resolve, reject).finally(() => { + signal.removeEventListener('abort', onAbort); + }); + }); + } + + private lastKnown(key: string): TbankInstrument | null { + const value = this.metadata.peek(key, { allowStale: true }); + return value === missingInstrument ? null : (value ?? null); } private async fetch( key: string, request: () => Promise, + previous: TbankInstrument | null, ): Promise { try { const value = await request(); - this.metadata.set(key, value ?? missingInstrument); + const found = Boolean(value?.name?.trim() || value?.ticker?.trim()); + if (value === null || found) + this.metadata.set(key, value ?? missingInstrument); this.logger.event('debug', 'instrument-metadata.resolved', { - found: Boolean(value?.name?.trim() || value?.ticker?.trim()), + found, }); - return value; + return found || value === null ? value : (previous ?? value); } catch (error) { this.logger.event('warn', 'instrument-metadata.failed', { code: @@ -80,7 +110,9 @@ export class InstrumentMetadataService { 'UNKNOWN') : 'UNKNOWN', }); - return null; + if (error instanceof Error && error.message === 'INVALID_TBANK_TOKEN') + throw error; + return error instanceof ServiceUnavailableException ? previous : null; } } } diff --git a/src/tbank/portfolio.service.spec.ts b/src/tbank/portfolio.service.spec.ts index 2cd56dc..8b77529 100644 --- a/src/tbank/portfolio.service.spec.ts +++ b/src/tbank/portfolio.service.spec.ts @@ -184,6 +184,80 @@ describe('PortfolioService', () => { expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); }); + it('loads a large mixed portfolio without overflowing metadata requests or losing names', async () => { + const positions = Array.from({ length: 60 }, (_, index) => ({ + figi: `BATCH-${index}`, + instrumentType: index % 2 ? 'bond' : 'share', + quantity: { units: 2 }, + currentPrice: { currency: 'RUB', units: 10 }, + })); + tbank.getPortfolio.mockResolvedValue({ positions }); + let active = 0; + tbank.getInstrumentByFigi.mockImplementation( + async (_token: string, figi: string) => { + active += 1; + try { + if (active > 50) + throw new ServiceUnavailableException({ + code: 'TBANK_UNAVAILABLE', + }); + await Promise.resolve(); + return { + name: `Название ${figi}`, + ticker: figi, + isin: `ISIN-${figi}`, + brand: { logoName: `${figi}.png` }, + }; + } finally { + active -= 1; + } + }, + ); + const result = await service.get('user', 'account'); + expect(result.positions).toMatchObject( + positions.map((position, index) => ({ + displayName: `Название BATCH-${index}`, + type: position.instrumentType === 'share' ? 'STOCK' : 'BOND', + ticker: `BATCH-${index}`, + isin: `ISIN-BATCH-${index}`, + logoName: `BATCH-${index}.png`, + quantity: '2', + currentValue: { currency: 'RUB', value: '20' }, + })), + ); + }); + + it('bounds total metadata waiting and timestamps the financial response before enrichment', async () => { + jest.useFakeTimers({ now: new Date('2026-09-12T12:00:00Z') }); + try { + tbank.getPortfolio.mockResolvedValue({ + positions: Array.from({ length: 60 }, (_, index) => ({ + figi: `BUDGET-${index}`, + instrumentUid: `budget-uid-${index}`, + instrumentType: index % 2 ? 'bond' : 'share', + quantity: { units: 2 }, + currentPrice: { currency: 'RUB', units: 10 }, + })), + }); + tbank.getInstrumentByUid.mockImplementation(() => new Promise(() => {})); + const pending = service.get('user', 'account'); + await jest.advanceTimersByTimeAsync(0); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(5); + await jest.advanceTimersByTimeAsync(30_000); + const result = await pending; + expect(result.lastSuccessfulAt).toBe('2026-09-12T12:00:00Z'); + expect(result.positions).toHaveLength(60); + expect(result.positions[59]).toMatchObject({ + displayName: 'BUDGET-59', + currentValue: { currency: 'RUB', value: '20' }, + }); + expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(5); + expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); + } finally { + jest.useRealTimers(); + } + }); + it('uses fresh snapshots, falls back only for temporary errors, and clears them', async () => { await service.get('user', 'account'); await service.get('user', 'account'); diff --git a/src/tbank/portfolio.service.ts b/src/tbank/portfolio.service.ts index eb8f449..f2de97d 100644 --- a/src/tbank/portfolio.service.ts +++ b/src/tbank/portfolio.service.ts @@ -20,6 +20,8 @@ import { TbankClient, TbankMoney } from './tbank.client'; const freshForMs = 30_000; const fallbackForMs = 300_000; +const positionsBatchSize = 5; +const metadataTimeoutMs = 30_000; const types: Record = { share: 'STOCK', bond: 'BOND', @@ -119,7 +121,7 @@ export class PortfolioService { this.snapshots.set(key, { generation, snapshot, - storedAt: Date.now(), + storedAt: Date.parse(snapshot.lastSuccessfulAt), }); return snapshot; } catch (error) { @@ -151,9 +153,29 @@ export class PortfolioService { }[], money: TbankMoney[], ): Promise { - const positions = await Promise.all( - upstream.map((position) => this.position(token, position)), - ); + const lastSuccessfulAt = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z'); + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), metadataTimeoutMs); + const positions: (PortfolioPositionDto | null)[] = []; + try { + for ( + let offset = 0; + offset < upstream.length; + offset += positionsBatchSize + ) { + positions.push( + ...(await Promise.all( + upstream + .slice(offset, offset + positionsBatchSize) + .map((position) => + this.position(token, position, controller.signal), + ), + )), + ); + } + } finally { + clearTimeout(timeout); + } const cash = new Map>(); for (const value of money) { const currency = value.currency?.trim().toUpperCase(); @@ -169,7 +191,7 @@ export class PortfolioService { (position): position is PortfolioPositionDto => position !== null, ); return { - lastSuccessfulAt: new Date().toISOString().replace(/\.\d{3}Z$/, 'Z'), + lastSuccessfulAt, isStale: false, totalsByCurrency: totalsByCurrency(upstream, money), positions: normalizedPositions, @@ -181,6 +203,7 @@ export class PortfolioService { private async position( token: string, position: Parameters[1][number], + signal: AbortSignal, ): Promise { const quantity = decimal(position.quantity ?? {}); if (isZero(quantity)) return null; @@ -206,7 +229,12 @@ export class PortfolioService { ), } : null; - const metadata = await this.metadata.resolve(token, instrumentUid, figi); + const metadata = await this.metadata.resolve( + token, + instrumentUid, + figi, + signal, + ); return { figi, instrumentUid, diff --git a/src/tbank/tbank.client.spec.ts b/src/tbank/tbank.client.spec.ts index f3c864e..5e74bb9 100644 --- a/src/tbank/tbank.client.spec.ts +++ b/src/tbank/tbank.client.spec.ts @@ -1,4 +1,4 @@ -import { ServiceUnavailableException } from '@nestjs/common'; +import { NotFoundException, ServiceUnavailableException } from '@nestjs/common'; import { status } from '@grpc/grpc-js'; import { TbankClient } from './tbank.client'; @@ -120,3 +120,187 @@ describe('TbankClient operations', () => { jest.useRealTimers(); }); }); + +describe('TbankClient metadata retries', () => { + const instrument = { name: 'Облигация', ticker: 'BOND' }; + let client: TbankClient; + let requestRpc: jest.Mock; + let queue: { size: number; run: jest.Mock }; + let logger: { + event: jest.Mock]>; + }; + + beforeEach(() => { + jest.useFakeTimers(); + requestRpc = jest.fn(); + queue = { + size: 0, + run: jest.fn((operation: () => Promise) => operation()), + }; + logger = { + event: jest.fn]>(), + }; + client = Object.assign( + Object.create(TbankClient.prototype) as TbankClient, + { + request: requestRpc, + queue, + logger, + }, + ); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it.each([ + status.UNAVAILABLE, + status.DEADLINE_EXCEEDED, + status.RESOURCE_EXHAUSTED, + ])( + 'recovers metadata after transient gRPC status %s with 2s/4s pauses', + async (code) => { + requestRpc + .mockRejectedValueOnce({ code }) + .mockRejectedValueOnce({ code }) + .mockResolvedValueOnce({ instrument }); + + const result = client.getInstrumentByUid('token', 'uid'); + await jest.advanceTimersByTimeAsync(1999); + expect(requestRpc).toHaveBeenCalledTimes(1); + await jest.advanceTimersByTimeAsync(1); + expect(requestRpc).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(3999); + expect(requestRpc).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(1); + + await expect(result).resolves.toEqual(instrument); + expect(queue.run).toHaveBeenCalledTimes(3); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + it('rechecks a full queue before admitting a metadata retry', async () => { + queue.size = 50; + requestRpc.mockResolvedValue({ instrument }); + + const result = client.getInstrumentByFigi('token', 'figi'); + await jest.advanceTimersByTimeAsync(1999); + expect(queue.run).not.toHaveBeenCalled(); + queue.size = 49; + await jest.advanceTimersByTimeAsync(1); + + await expect(result).resolves.toEqual(instrument); + expect(queue.run).toHaveBeenCalledTimes(1); + expect(logger.event).toHaveBeenCalledWith( + 'warn', + 'tbank.rpc.retry', + expect.objectContaining({ code: 'TBANK_UNAVAILABLE', attempts: 1 }), + ); + expect(logger.event.mock.calls[0][2]).not.toHaveProperty('grpcCode'); + }); + + it('does not bypass queue admission after an upstream failure', async () => { + requestRpc + .mockRejectedValueOnce({ code: status.UNAVAILABLE }) + .mockResolvedValueOnce({ instrument }); + + const result = client.getInstrumentByUid('token', 'uid'); + queue.size = 50; + await jest.advanceTimersByTimeAsync(2000); + expect(queue.run).toHaveBeenCalledTimes(1); + queue.size = 0; + await jest.advanceTimersByTimeAsync(4000); + + await expect(result).resolves.toEqual(instrument); + expect(queue.run).toHaveBeenCalledTimes(2); + }); + + it('stops retrying a saturated queue after three admission attempts', async () => { + queue.size = 50; + const result = expect( + client.getInstrumentByUid('token', 'uid'), + ).rejects.toBeInstanceOf(ServiceUnavailableException); + + await jest.advanceTimersByTimeAsync(6000); + await result; + + expect(queue.run).not.toHaveBeenCalled(); + expect(logger.event).toHaveBeenCalledTimes(3); + expect(logger.event).toHaveBeenLastCalledWith( + 'error', + 'tbank.rpc.failed', + expect.objectContaining({ code: 'TBANK_UNAVAILABLE', attempts: 3 }), + ); + expect(jest.getTimerCount()).toBe(0); + }); + + it('stops retrying an unavailable upstream after three RPCs', async () => { + requestRpc.mockRejectedValue({ code: status.UNAVAILABLE }); + const result = expect( + client.getInstrumentByUid('token', 'uid'), + ).rejects.toBeInstanceOf(ServiceUnavailableException); + + await jest.advanceTimersByTimeAsync(6000); + await result; + + expect(queue.run).toHaveBeenCalledTimes(3); + expect(jest.getTimerCount()).toBe(0); + }); + + it.each([status.UNAUTHENTICATED, status.PERMISSION_DENIED])( + 'does not retry or mask metadata authentication status %s', + async (code) => { + requestRpc.mockRejectedValue({ code }); + + await expect(client.getInstrumentByUid('token', 'uid')).rejects.toThrow( + 'INVALID_TBANK_TOKEN', + ); + expect(requestRpc).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + it.each(['getInstrumentByUid', 'getInstrumentByFigi'] as const)( + 'returns confirmed absence from %s without retrying', + async (method) => { + requestRpc.mockRejectedValue({ code: status.NOT_FOUND }); + + await expect(client[method]('token', 'instrument')).resolves.toBeNull(); + expect(requestRpc).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }, + ); + + it('keeps confirmed absence from ShareBy as a not-found error', async () => { + requestRpc.mockRejectedValue({ code: status.NOT_FOUND }); + + await expect(client.getShare('token', 'MISSING')).rejects.toBeInstanceOf( + NotFoundException, + ); + expect(requestRpc).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }); + + it('does not retry invalid metadata arguments', async () => { + requestRpc.mockRejectedValue({ code: status.INVALID_ARGUMENT }); + + await expect( + client.getInstrumentByUid('token', 'uid'), + ).rejects.toBeInstanceOf(ServiceUnavailableException); + expect(requestRpc).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }); + + it('keeps other RPCs failing immediately when the queue is full', async () => { + queue.size = 50; + + await expect( + client.getPortfolio('token', 'account'), + ).rejects.toBeInstanceOf(ServiceUnavailableException); + + expect(queue.run).not.toHaveBeenCalled(); + expect(jest.getTimerCount()).toBe(0); + }); +}); diff --git a/src/tbank/tbank.client.ts b/src/tbank/tbank.client.ts index df670de..7b3e18b 100644 --- a/src/tbank/tbank.client.ts +++ b/src/tbank/tbank.client.ts @@ -259,26 +259,14 @@ export class TbankClient { token: string, figi: string, ): Promise { - return this.call(token, this.instruments, 'getInstrumentBy', { - idType: 'INSTRUMENT_ID_TYPE_FIGI', - id: figi, - }).then( - (response: { instrument?: TbankInstrument }) => - response.instrument ?? null, - ); + return this.getInstrument(token, 'INSTRUMENT_ID_TYPE_FIGI', figi); } getInstrumentByUid( token: string, instrumentUid: string, ): Promise { - return this.call(token, this.instruments, 'getInstrumentBy', { - idType: 'INSTRUMENT_ID_TYPE_UID', - id: instrumentUid, - }).then( - (response: { instrument?: TbankInstrument }) => - response.instrument ?? null, - ); + return this.getInstrument(token, 'INSTRUMENT_ID_TYPE_UID', instrumentUid); } getDividends( @@ -435,24 +423,39 @@ export class TbankClient { ).then((response) => response.candles ?? []); } + private async getInstrument( + token: string, + idType: 'INSTRUMENT_ID_TYPE_UID' | 'INSTRUMENT_ID_TYPE_FIGI', + id: string, + ): Promise { + try { + const response = await this.call<{ instrument?: TbankInstrument }>( + token, + this.instruments, + 'getInstrumentBy', + { idType, id }, + ); + return response.instrument ?? null; + } catch (error) { + if (error instanceof NotFoundException) return null; + throw error; + } + } + private async call( token: string, client: TbankClientService, method: string, request: object, ): Promise { - if (this.queue.size >= 50) { - this.logger.event('error', 'tbank.rpc.failed', { - operation: method, - code: 'TBANK_UNAVAILABLE', - attempts: 0, - }); - throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }); - } - - for (let attempts = 1; attempts <= 2; attempts += 1) { + const metadataCall = method === 'getInstrumentBy'; + const maxAttempts = metadataCall ? 3 : 2; + for (let attempts = 1; attempts <= maxAttempts; attempts += 1) { const startedAt = performance.now(); + const queueFull = this.queue.size >= 50; try { + if (queueFull) + throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }); const response = await this.queue.run(() => this.request(token, client, method, request), ); @@ -464,32 +467,42 @@ export class TbankClient { return response; } catch (error) { const grpcCode = (error as ServiceError).code; - const retryable = [ - status.UNAVAILABLE, - status.DEADLINE_EXCEEDED, - status.RESOURCE_EXHAUSTED, - ].includes(grpcCode); - if (retryable && attempts < 2) { + const failureContext = queueFull + ? { code: 'TBANK_UNAVAILABLE' } + : { grpcCode }; + const retryable = queueFull + ? metadataCall + : [ + status.UNAVAILABLE, + status.DEADLINE_EXCEEDED, + status.RESOURCE_EXHAUSTED, + ].includes(grpcCode); + if (retryable && attempts < maxAttempts) { this.logger.event('warn', 'tbank.rpc.retry', { operation: method, - grpcCode, + ...failureContext, attempts, durationMs: Math.round(performance.now() - startedAt), }); - await new Promise((resolve) => setTimeout(resolve, 1000)); + await new Promise((resolve) => + setTimeout(resolve, metadataCall ? attempts * 2000 : 1000), + ); continue; } this.logger.event('error', 'tbank.rpc.failed', { operation: method, - grpcCode, - attempts, + ...failureContext, + attempts: queueFull && !metadataCall ? 0 : attempts, durationMs: Math.round(performance.now() - startedAt), }); if ( [status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(grpcCode) ) throw new Error('INVALID_TBANK_TOKEN'); - if (grpcCode === status.NOT_FOUND && method === 'shareBy') + if ( + grpcCode === status.NOT_FOUND && + (method === 'shareBy' || metadataCall) + ) throw new NotFoundException(); throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }); }