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.
This commit is contained in:
2026-09-12 15:37:56 +03:00
parent b1906c8e3c
commit 8ec172f48e
6 changed files with 621 additions and 61 deletions
@@ -1,4 +1,7 @@
import { InstrumentMetadataService } from './instrument-metadata.service'; 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', () => { describe('InstrumentMetadataService', () => {
const tbank = { const tbank = {
@@ -8,6 +11,13 @@ describe('InstrumentMetadataService', () => {
const logger = { event: jest.fn() }; const logger = { event: jest.fn() };
let service: InstrumentMetadataService; let service: InstrumentMetadataService;
function rememberExpired(key: string, value: TbankInstrument): void {
const cache = (
service as unknown as { metadata: LRUCache<string, TbankInstrument> }
).metadata;
cache.set(key, value, { ttl: 1, start: performance.now() - 1000 });
}
beforeEach(() => { beforeEach(() => {
jest.clearAllMocks(); jest.clearAllMocks();
tbank.getInstrumentByFigi.mockReset(); tbank.getInstrumentByFigi.mockReset();
@@ -58,4 +68,223 @@ describe('InstrumentMetadataService', () => {
expect(tbank.getInstrumentByUid).toHaveBeenCalledTimes(1); 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();
});
}); });
+51 -19
View File
@@ -1,5 +1,6 @@
import { Injectable } from '@nestjs/common'; import { Injectable, ServiceUnavailableException } from '@nestjs/common';
import { LRUCache } from 'lru-cache'; import { LRUCache } from 'lru-cache';
import { createHash } from 'node:crypto';
import { StructuredLogger } from '../common/logging/structured-logger.service'; import { StructuredLogger } from '../common/logging/structured-logger.service';
import { TbankClient, type TbankInstrument } from './tbank.client'; import { TbankClient, type TbankInstrument } from './tbank.client';
@@ -15,6 +16,7 @@ export class InstrumentMetadataService {
>({ >({
max: maxCachedInstruments, max: maxCachedInstruments,
ttl: metadataForMs, ttl: metadataForMs,
noDeleteOnStaleGet: true,
}); });
private readonly pending = new Map<string, Promise<TbankInstrument | null>>(); private readonly pending = new Map<string, Promise<TbankInstrument | null>>();
@@ -27,51 +29,79 @@ export class InstrumentMetadataService {
token: string, token: string,
instrumentUid: string | null | undefined, instrumentUid: string | null | undefined,
figi: string | null | undefined, figi: string | null | undefined,
signal?: AbortSignal,
): Promise<TbankInstrument | null> { ): Promise<TbankInstrument | null> {
const uid = instrumentUid?.trim(); const uid = instrumentUid?.trim();
let byUid: TbankInstrument | null = null;
if (uid) { if (uid) {
const byUid = await this.lookup(`uid:${uid}`, () => byUid = await this.lookup(
this.tbank.getInstrumentByUid(token, uid), 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(); const instrumentFigi = figi?.trim();
return instrumentFigi const byFigi = instrumentFigi
? this.lookup(`figi:${instrumentFigi}`, () => ? await this.lookup(
this.tbank.getInstrumentByFigi(token, instrumentFigi), token,
`figi:${instrumentFigi}`,
() => this.tbank.getInstrumentByFigi(token, instrumentFigi),
signal,
) )
: null; : null;
return byFigi ?? byUid;
} }
private async lookup( private async lookup(
token: string,
key: string, key: string,
fetch: () => Promise<TbankInstrument | null>, fetch: () => Promise<TbankInstrument | null>,
signal?: AbortSignal,
): Promise<TbankInstrument | null> { ): Promise<TbankInstrument | null> {
const cached = this.metadata.get(key); const cached = this.metadata.get(key);
if (cached !== undefined) if (cached !== undefined)
return cached === missingInstrument ? null : cached; return cached === missingInstrument ? null : cached;
const pending = this.pending.get(key); if (signal?.aborted) return this.lastKnown(key);
if (pending) return pending; const owner = createHash('sha256').update(token).digest('hex');
const request = this.fetch(key, fetch); const requestKey = `${owner}:${key}`;
this.pending.set(key, request); let pending = this.pending.get(requestKey);
try { if (!pending) {
return await request; pending = this.fetch(key, fetch, this.lastKnown(key)).finally(() => {
} finally { this.pending.delete(requestKey);
this.pending.delete(key); });
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( private async fetch(
key: string, key: string,
request: () => Promise<TbankInstrument | null>, request: () => Promise<TbankInstrument | null>,
previous: TbankInstrument | null,
): Promise<TbankInstrument | null> { ): Promise<TbankInstrument | null> {
try { try {
const value = await request(); 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', { 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) { } catch (error) {
this.logger.event('warn', 'instrument-metadata.failed', { this.logger.event('warn', 'instrument-metadata.failed', {
code: code:
@@ -80,7 +110,9 @@ export class InstrumentMetadataService {
'UNKNOWN') 'UNKNOWN')
: 'UNKNOWN', : 'UNKNOWN',
}); });
return null; if (error instanceof Error && error.message === 'INVALID_TBANK_TOKEN')
throw error;
return error instanceof ServiceUnavailableException ? previous : null;
} }
} }
} }
+74
View File
@@ -184,6 +184,80 @@ describe('PortfolioService', () => {
expect(tbank.getInstrumentByFigi).not.toHaveBeenCalled(); 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 () => { it('uses fresh snapshots, falls back only for temporary errors, and clears them', async () => {
await service.get('user', 'account'); await service.get('user', 'account');
await service.get('user', 'account'); await service.get('user', 'account');
+34 -6
View File
@@ -20,6 +20,8 @@ import { TbankClient, TbankMoney } from './tbank.client';
const freshForMs = 30_000; const freshForMs = 30_000;
const fallbackForMs = 300_000; const fallbackForMs = 300_000;
const positionsBatchSize = 5;
const metadataTimeoutMs = 30_000;
const types: Record<string, string> = { const types: Record<string, string> = {
share: 'STOCK', share: 'STOCK',
bond: 'BOND', bond: 'BOND',
@@ -119,7 +121,7 @@ export class PortfolioService {
this.snapshots.set(key, { this.snapshots.set(key, {
generation, generation,
snapshot, snapshot,
storedAt: Date.now(), storedAt: Date.parse(snapshot.lastSuccessfulAt),
}); });
return snapshot; return snapshot;
} catch (error) { } catch (error) {
@@ -151,9 +153,29 @@ export class PortfolioService {
}[], }[],
money: TbankMoney[], money: TbankMoney[],
): Promise<PortfolioDto> { ): Promise<PortfolioDto> {
const positions = await Promise.all( const lastSuccessfulAt = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
upstream.map((position) => this.position(token, position)), 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<string, ReturnType<typeof decimal>>(); const cash = new Map<string, ReturnType<typeof decimal>>();
for (const value of money) { for (const value of money) {
const currency = value.currency?.trim().toUpperCase(); const currency = value.currency?.trim().toUpperCase();
@@ -169,7 +191,7 @@ export class PortfolioService {
(position): position is PortfolioPositionDto => position !== null, (position): position is PortfolioPositionDto => position !== null,
); );
return { return {
lastSuccessfulAt: new Date().toISOString().replace(/\.\d{3}Z$/, 'Z'), lastSuccessfulAt,
isStale: false, isStale: false,
totalsByCurrency: totalsByCurrency(upstream, money), totalsByCurrency: totalsByCurrency(upstream, money),
positions: normalizedPositions, positions: normalizedPositions,
@@ -181,6 +203,7 @@ export class PortfolioService {
private async position( private async position(
token: string, token: string,
position: Parameters<PortfolioService['normalize']>[1][number], position: Parameters<PortfolioService['normalize']>[1][number],
signal: AbortSignal,
): Promise<PortfolioPositionDto | null> { ): Promise<PortfolioPositionDto | null> {
const quantity = decimal(position.quantity ?? {}); const quantity = decimal(position.quantity ?? {});
if (isZero(quantity)) return null; if (isZero(quantity)) return null;
@@ -206,7 +229,12 @@ export class PortfolioService {
), ),
} }
: null; : null;
const metadata = await this.metadata.resolve(token, instrumentUid, figi); const metadata = await this.metadata.resolve(
token,
instrumentUid,
figi,
signal,
);
return { return {
figi, figi,
instrumentUid, instrumentUid,
+185 -1
View File
@@ -1,4 +1,4 @@
import { ServiceUnavailableException } from '@nestjs/common'; import { NotFoundException, ServiceUnavailableException } from '@nestjs/common';
import { status } from '@grpc/grpc-js'; import { status } from '@grpc/grpc-js';
import { TbankClient } from './tbank.client'; import { TbankClient } from './tbank.client';
@@ -120,3 +120,187 @@ describe('TbankClient operations', () => {
jest.useRealTimers(); 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<void, [string, string, Record<string, unknown>]>;
};
beforeEach(() => {
jest.useFakeTimers();
requestRpc = jest.fn();
queue = {
size: 0,
run: jest.fn((operation: () => Promise<unknown>) => operation()),
};
logger = {
event: jest.fn<void, [string, string, Record<string, unknown>]>(),
};
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);
});
});
+48 -35
View File
@@ -259,26 +259,14 @@ export class TbankClient {
token: string, token: string,
figi: string, figi: string,
): Promise<TbankInstrument | null> { ): Promise<TbankInstrument | null> {
return this.call(token, this.instruments, 'getInstrumentBy', { return this.getInstrument(token, 'INSTRUMENT_ID_TYPE_FIGI', figi);
idType: 'INSTRUMENT_ID_TYPE_FIGI',
id: figi,
}).then(
(response: { instrument?: TbankInstrument }) =>
response.instrument ?? null,
);
} }
getInstrumentByUid( getInstrumentByUid(
token: string, token: string,
instrumentUid: string, instrumentUid: string,
): Promise<TbankInstrument | null> { ): Promise<TbankInstrument | null> {
return this.call(token, this.instruments, 'getInstrumentBy', { return this.getInstrument(token, 'INSTRUMENT_ID_TYPE_UID', instrumentUid);
idType: 'INSTRUMENT_ID_TYPE_UID',
id: instrumentUid,
}).then(
(response: { instrument?: TbankInstrument }) =>
response.instrument ?? null,
);
} }
getDividends( getDividends(
@@ -435,24 +423,39 @@ export class TbankClient {
).then((response) => response.candles ?? []); ).then((response) => response.candles ?? []);
} }
private async getInstrument(
token: string,
idType: 'INSTRUMENT_ID_TYPE_UID' | 'INSTRUMENT_ID_TYPE_FIGI',
id: string,
): Promise<TbankInstrument | null> {
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<Response>( private async call<Response>(
token: string, token: string,
client: TbankClientService, client: TbankClientService,
method: string, method: string,
request: object, request: object,
): Promise<Response> { ): Promise<Response> {
if (this.queue.size >= 50) { const metadataCall = method === 'getInstrumentBy';
this.logger.event('error', 'tbank.rpc.failed', { const maxAttempts = metadataCall ? 3 : 2;
operation: method, for (let attempts = 1; attempts <= maxAttempts; attempts += 1) {
code: 'TBANK_UNAVAILABLE',
attempts: 0,
});
throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
}
for (let attempts = 1; attempts <= 2; attempts += 1) {
const startedAt = performance.now(); const startedAt = performance.now();
const queueFull = this.queue.size >= 50;
try { try {
if (queueFull)
throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
const response = await this.queue.run(() => const response = await this.queue.run(() =>
this.request<Response>(token, client, method, request), this.request<Response>(token, client, method, request),
); );
@@ -464,32 +467,42 @@ export class TbankClient {
return response; return response;
} catch (error) { } catch (error) {
const grpcCode = (error as ServiceError).code; const grpcCode = (error as ServiceError).code;
const retryable = [ const failureContext = queueFull
status.UNAVAILABLE, ? { code: 'TBANK_UNAVAILABLE' }
status.DEADLINE_EXCEEDED, : { grpcCode };
status.RESOURCE_EXHAUSTED, const retryable = queueFull
].includes(grpcCode); ? metadataCall
if (retryable && attempts < 2) { : [
status.UNAVAILABLE,
status.DEADLINE_EXCEEDED,
status.RESOURCE_EXHAUSTED,
].includes(grpcCode);
if (retryable && attempts < maxAttempts) {
this.logger.event('warn', 'tbank.rpc.retry', { this.logger.event('warn', 'tbank.rpc.retry', {
operation: method, operation: method,
grpcCode, ...failureContext,
attempts, attempts,
durationMs: Math.round(performance.now() - startedAt), durationMs: Math.round(performance.now() - startedAt),
}); });
await new Promise((resolve) => setTimeout(resolve, 1000)); await new Promise((resolve) =>
setTimeout(resolve, metadataCall ? attempts * 2000 : 1000),
);
continue; continue;
} }
this.logger.event('error', 'tbank.rpc.failed', { this.logger.event('error', 'tbank.rpc.failed', {
operation: method, operation: method,
grpcCode, ...failureContext,
attempts, attempts: queueFull && !metadataCall ? 0 : attempts,
durationMs: Math.round(performance.now() - startedAt), durationMs: Math.round(performance.now() - startedAt),
}); });
if ( if (
[status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(grpcCode) [status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(grpcCode)
) )
throw new Error('INVALID_TBANK_TOKEN'); 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 NotFoundException();
throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }); throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
} }