diff --git a/src/shared/services/process.service.ts b/src/shared/services/process.service.ts index 6d1046662e..2c17e4fefd 100644 --- a/src/shared/services/process.service.ts +++ b/src/shared/services/process.service.ts @@ -22,6 +22,7 @@ export enum Process { LIQUIDITY_MANAGEMENT_CHECK_BALANCES = 'LiquidityManagementCheckBalances', MONITORING = 'Monitoring', MONITOR_CONNECTION_POOL = 'MonitorConnectionPool', + MONITOR_EVENT_LOOP = 'MonitorEventLoop', UPDATE_STATISTIC = 'UpdateStatistic', KYC = 'Kyc', KYC_IDENT_REVIEW = 'KycIdentReview', diff --git a/src/shared/utils/__tests__/queue-handler.spec.ts b/src/shared/utils/__tests__/queue-handler.spec.ts new file mode 100644 index 0000000000..5db7053a75 --- /dev/null +++ b/src/shared/utils/__tests__/queue-handler.spec.ts @@ -0,0 +1,58 @@ +import { QueueHandler } from 'src/shared/utils/queue-handler'; + +describe('QueueHandler', () => { + it('runs queued items and returns their results', async () => { + const queue = new QueueHandler(1000, undefined, 1); + + await expect(queue.handle(async () => 42)).resolves.toBe(42); + + queue.stop(); + }); + + it('does not execute items whose queue timeout fired while they were still waiting', async () => { + const queue = new QueueHandler(100, undefined, 1); + const ran: number[] = []; + + const first = queue.handle(async () => { + ran.push(1); + await new Promise((resolve) => setTimeout(resolve, 250)); + }); + const second = queue.handle(async () => { + ran.push(2); + }); + + await expect(first).rejects.toThrow('Queue timeout'); + await expect(second).rejects.toThrow('Queue timeout'); + + // let the first action finish and the queue drain — the second must have been discarded + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(ran).toEqual([1]); + + queue.stop(); + }); + + it('frees the worker slot via item timeout when an action never settles', async () => { + const queue = new QueueHandler(undefined, 50, 1); + + const hanging = queue.handle(() => new Promise(() => undefined)); + await expect(hanging).rejects.toThrow(); + + // slot must be free again: a follow-up item still runs + await expect(queue.handle(async () => 'ok')).resolves.toBe('ok'); + + queue.stop(); + }); + + it('rejects when the action throws synchronously and frees the worker slot for the next item', async () => { + const queue = new QueueHandler(1000, undefined, 1); + + const throwing = queue.handle((): Promise => { + throw new Error('boom'); + }); + await expect(throwing).rejects.toThrow('boom'); + + await expect(queue.handle(async () => 'ok')).resolves.toBe('ok'); + + queue.stop(); + }); +}); diff --git a/src/shared/utils/queue-handler.ts b/src/shared/utils/queue-handler.ts index 31384e6363..09d69caa93 100644 --- a/src/shared/utils/queue-handler.ts +++ b/src/shared/utils/queue-handler.ts @@ -7,16 +7,20 @@ class QueueItem { private resolve: (value: T | PromiseLike) => void; private reject: (e: Error) => void; + private settled = false; + constructor( private readonly action: () => Promise, timeout?: number, ) { this.promise = new Promise((resolve, reject) => { this.resolve = (v) => { + this.settled = true; if (this.timeout) clearTimeout(this.timeout); resolve(v); }; this.reject = (e) => { + this.settled = true; if (this.timeout) clearTimeout(this.timeout); reject(e); }; @@ -24,12 +28,19 @@ class QueueItem { if (timeout) this.timeout = setTimeout(() => this.reject(new Error('Queue timeout')), timeout); } + get isSettled(): boolean { + return this.settled; + } + public wait(): Promise { return this.promise; } public async doWork(timeout: number) { - const promise = timeout ? Util.timeout(this.action(), timeout) : this.action(); + // Defer the call so a synchronous throw inside the action becomes a rejection + // instead of escaping doWork and leaving the item unsettled forever. + const action = Promise.resolve().then(() => this.action()); + const promise = timeout ? Util.timeout(action, timeout) : action; await promise.then(this.resolve).catch(this.reject); } @@ -81,7 +92,11 @@ export class QueueHandler { while (this.isRunning) { try { if (this.queue.length > 0 && this.workParallelCounter < this.maxWorkParallel) { - const work = this.queue.shift().doWork(this.itemTimeout); + const item = this.queue.shift(); + // already settled (queue timeout while waiting): the caller is gone, don't run the action + if (item.isSettled) continue; + + const work = item.doWork(this.itemTimeout); this.workParallelCounter++; void work.finally(() => this.workParallelCounter--); diff --git a/src/subdomains/core/monitoring/monitor-event-loop.service.ts b/src/subdomains/core/monitoring/monitor-event-loop.service.ts new file mode 100644 index 0000000000..59713154e3 --- /dev/null +++ b/src/subdomains/core/monitoring/monitor-event-loop.service.ts @@ -0,0 +1,35 @@ +import { Injectable, OnModuleDestroy } from '@nestjs/common'; +import { CronExpression } from '@nestjs/schedule'; +import { monitorEventLoopDelay } from 'perf_hooks'; +import { DfxLogger } from 'src/shared/services/dfx-logger'; +import { Process } from 'src/shared/services/process.service'; +import { DfxCron } from 'src/shared/utils/cron'; + +@Injectable() +export class MonitorEventLoopService implements OnModuleDestroy { + private readonly logger = new DfxLogger(MonitorEventLoopService); + + private readonly histogram = monitorEventLoopDelay({ resolution: 20 }); + + constructor() { + this.histogram.enable(); + } + + // Disable the histogram so its sampling timer does not continue after module teardown. + onModuleDestroy(): void { + this.histogram.disable(); + } + + @DfxCron(CronExpression.EVERY_10_SECONDS, { process: Process.MONITOR_EVENT_LOOP }) + monitorEventLoop(): void { + const toMs = (ns: number) => Math.round(ns / 1e6); + + this.logger.info( + `EventLoop delay: mean ${toMs(this.histogram.mean)}ms / p95 ${toMs( + this.histogram.percentile(95), + )}ms / max ${toMs(this.histogram.max)}ms`, + ); + + this.histogram.reset(); + } +} diff --git a/src/subdomains/core/monitoring/monitoring.module.ts b/src/subdomains/core/monitoring/monitoring.module.ts index 8de29db84d..d3ae5d7ac0 100644 --- a/src/subdomains/core/monitoring/monitoring.module.ts +++ b/src/subdomains/core/monitoring/monitoring.module.ts @@ -14,6 +14,7 @@ import { FiatPayInModule } from 'src/subdomains/supporting/fiat-payin/fiat-payin import { NotificationModule } from 'src/subdomains/supporting/notification/notification.module'; import { PricingModule } from 'src/subdomains/supporting/pricing/pricing.module'; import { MonitorConnectionPoolService } from './monitor-connection-pool.service'; +import { MonitorEventLoopService } from './monitor-event-loop.service'; import { HealthController } from './health.controller'; import { MonitoringController } from './monitoring.controller'; import { MonitoringService } from './monitoring.service'; @@ -53,6 +54,7 @@ import { SystemStateSnapshotRepository } from './system-state-snapshot.repositor SystemStateSnapshotRepository, MonitoringService, MonitorConnectionPoolService, + MonitorEventLoopService, NodeBalanceObserver, NodeHealthObserver, PaymentObserver, diff --git a/src/subdomains/generic/gs/__tests__/gs.service.spec.ts b/src/subdomains/generic/gs/__tests__/gs.service.spec.ts index 1495773b6f..3b7b05fce8 100644 --- a/src/subdomains/generic/gs/__tests__/gs.service.spec.ts +++ b/src/subdomains/generic/gs/__tests__/gs.service.spec.ts @@ -3,8 +3,9 @@ import { createMock } from '@golevelup/ts-jest'; import { DataSource } from 'typeorm'; import { UserRole } from 'src/shared/auth/user-role.enum'; import { DfxLogger } from 'src/shared/services/dfx-logger'; +import { QueueHandler } from 'src/shared/utils/queue-handler'; import { GsService } from '../gs.service'; -import { DbQueryDto } from 'src/subdomains/generic/gs/dto/db-query.dto'; +import { DbQueryDto, DbReturnData } from 'src/subdomains/generic/gs/dto/db-query.dto'; import { assertDebugAllowlistInvariants, DebugAllowedColumns, @@ -106,6 +107,10 @@ describe('GsService', () => { service = buildGsService(kycDocumentService, dataSource); }); + afterEach(() => { + service['exportQueue'].stop(); + }); + // Helper that captures the SQL string and the bound-parameter array passed to the data // source. Tests use it to assert what reached Postgres — not just whether the call // succeeded, but that user input flowed through parameters rather than the SQL string. @@ -3365,6 +3370,7 @@ describe('GsService', () => { ); const realService = buildGsService(realKycDocumentService, createMock()); + realService['exportQueue'].stop(); // this test never dispatches through the queue const userData = personalUser(1); // Two disjoint select paths force an empty common prefix, so getAllUserDocuments (user + spider) runs. @@ -3588,3 +3594,98 @@ describe('DebugQueryDto - ValidationPipe layer', () => { expect(constraintNames(errors)).toContain('arrayMaxSize'); }); }); + +// Typed bridge to GsService internals the export-queue tests need (same pattern as +// asKycFileBlobs above: a narrow, documented cast instead of `any`). +type GsServiceInternals = { + exportQueue: QueueHandler; + executeDbData: GsService['getDbData']; + executeExtendedDbData: GsService['getExtendedDbData']; + getRawDbData: (query: DbQueryDto) => Promise[]>; + transformResultArray: (data: Record[], table: string, role: UserRole) => DbReturnData; +}; + +function internals(service: GsService): GsServiceInternals { + return service as unknown as GsServiceInternals; +} + +function exportQuery(overrides: Partial = {}): DbQueryDto { + return plainToInstance(DbQueryDto, { table: 'user', updatedSince: '2026-01-01', ...overrides }); +} + +describe('GS export queue', () => { + let service: GsService; + + beforeEach(() => { + service = buildGsService(createMock(), createMock()); + }); + + afterEach(() => { + internals(service).exportQueue.stop(); + }); + + it('processes at most 2 exports concurrently and preserves per-call results', async () => { + let active = 0; + let peak = 0; + const gates: (() => void)[] = []; + + jest.spyOn(internals(service), 'executeDbData').mockImplementation(async (query) => { + active++; + peak = Math.max(peak, active); + await new Promise((resolve) => gates.push(resolve)); + active--; + return { keys: ['min'], values: [[query.min]] }; + }); + + const calls = [1, 2, 3, 4].map((min) => service.getDbData(exportQuery({ min }), UserRole.ADMIN)); + + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(peak).toBe(2); + expect(active).toBe(2); + + gates.splice(0).forEach((release) => release()); + await new Promise((resolve) => setTimeout(resolve, 100)); + gates.splice(0).forEach((release) => release()); + + const results = await Promise.all(calls); + expect(results.map((r) => r.values[0][0])).toEqual([1, 2, 3, 4]); + expect(peak).toBe(2); + }); + + it('routes custom exports through the same queue', async () => { + const handleSpy = jest.spyOn(internals(service).exportQueue, 'handle'); + jest.spyOn(internals(service), 'executeExtendedDbData').mockResolvedValue({ keys: [], values: [] }); + + await service.getExtendedDbData(exportQuery({ table: 'bank_tx' }), UserRole.ADMIN); + + expect(handleSpy).toHaveBeenCalledTimes(1); + }); + + describe('maxLine default cap', () => { + let received: DbQueryDto[]; + + beforeEach(() => { + received = []; + jest.spyOn(internals(service), 'getRawDbData').mockImplementation(async (query) => { + received.push(query); + return []; + }); + jest.spyOn(internals(service), 'transformResultArray').mockReturnValue({ keys: [], values: [] }); + }); + + it('applies the default cap when maxLine is absent', async () => { + await service.getDbData(exportQuery(), UserRole.ADMIN); + expect(received[0].maxLine).toBe(10000); + }); + + it('applies the default cap when maxLine is null', async () => { + await service.getDbData(exportQuery({ maxLine: null }), UserRole.ADMIN); + expect(received[0].maxLine).toBe(10000); + }); + + it('keeps an explicitly provided maxLine', async () => { + await service.getDbData(exportQuery({ maxLine: 500 }), UserRole.ADMIN); + expect(received[0].maxLine).toBe(500); + }); + }); +}); diff --git a/src/subdomains/generic/gs/gs.controller.ts b/src/subdomains/generic/gs/gs.controller.ts index 468896f265..52e2134dd1 100644 --- a/src/subdomains/generic/gs/gs.controller.ts +++ b/src/subdomains/generic/gs/gs.controller.ts @@ -47,7 +47,13 @@ export class GsController { this.logAndCheckTrigger(query, jwt); - return this.gsService.getExtendedDbData(query, jwt.role); + try { + return await this.gsService.getExtendedDbData(query, jwt.role); + } catch (e) { + const { table, identifier } = this.sanitizeLogFields(query); + this.logger.verbose(`Custom DB data call for ${table} in ${identifier} failed:`, e); + throw new BadRequestException(e.message); + } } @Get('support') diff --git a/src/subdomains/generic/gs/gs.service.ts b/src/subdomains/generic/gs/gs.service.ts index d5bf66acd6..c2cff7622e 100644 --- a/src/subdomains/generic/gs/gs.service.ts +++ b/src/subdomains/generic/gs/gs.service.ts @@ -1,6 +1,7 @@ import { BadRequestException, forwardRef, Inject, Injectable, NotFoundException } from '@nestjs/common'; import { UserRole } from 'src/shared/auth/user-role.enum'; import { DfxLogger } from 'src/shared/services/dfx-logger'; +import { QueueHandler } from 'src/shared/utils/queue-handler'; import { Util } from 'src/shared/utils/util'; import { BuyCryptoService } from 'src/subdomains/core/buy-crypto/process/services/buy-crypto.service'; import { BuyService } from 'src/subdomains/core/buy-crypto/routes/buy/buy.service'; @@ -77,6 +78,14 @@ interface DebugQueryEmitCtx { export class GsService { private readonly logger = new DfxLogger(GsService); + // Sheet exports are latency-tolerant batch consumers: cap their concurrency so sync + // bursts cannot monopolize the process at the expense of interactive requests. The item + // timeout frees a worker slot even if a query never settles (e.g. a dead connection). + private readonly exportQueue = new QueueHandler(240_000, 240_000, 2); + + // applied when a request specifies no maxLine — unbounded exports must be requested explicitly + private static readonly DEFAULT_MAX_LINE = 10000; + constructor( private readonly userDataService: UserDataService, private readonly userService: UserService, @@ -102,6 +111,13 @@ export class GsService { ) {} async getDbData(query: DbQueryDto, role: UserRole): Promise { + return this.exportQueue.handle(() => this.executeDbData(query, role)); + } + + private async executeDbData(query: DbQueryDto, role: UserRole): Promise { + const cappedByDefault = query.maxLine == null; + if (cappedByDefault) query.maxLine = GsService.DEFAULT_MAX_LINE; + const additionalSelect = Array.from( new Set([ ...(query.select?.filter((s) => s.includes('-') && !s.includes('documents')).map((s) => s.split('-')[0]) || []), @@ -131,6 +147,8 @@ export class GsService { }), ); + this.warnIfCapped(cappedByDefault && data.length >= GsService.DEFAULT_MAX_LINE, query); + const runTime = Util.round((Date.now() - startTime) / 1000, 1); if (runTime > 3) { @@ -168,14 +186,34 @@ export class GsService { } async getExtendedDbData(query: DbQueryBaseDto, role: UserRole): Promise { + return this.exportQueue.handle(() => this.executeExtendedDbData(query, role)); + } + + private async executeExtendedDbData(query: DbQueryBaseDto, role: UserRole): Promise { + const cappedByDefault = query.maxLine == null; + if (cappedByDefault) query.maxLine = GsService.DEFAULT_MAX_LINE; + switch (query.table) { case 'bank_tx': { - const data = await this.getExtendedBankTxData(query); + const { data, capReached } = await this.getExtendedBankTxData(query); + this.warnIfCapped(cappedByDefault && capReached, query); return this.transformResultArray(data, query.table, role); } } } + private warnIfCapped(hitDefaultCap: boolean, query: DbQueryBaseDto): void { + if (hitDefaultCap) + this.logger.warn( + `GS export for ${ + query.identifier ? Util.sanitizeLogValue(query.identifier, 64) : 'missing' + } hit the default maxLine cap (${GsService.DEFAULT_MAX_LINE}) on table ${Util.sanitizeLogValue( + query.table, + 64, + )} — rows beyond the cap were not returned`, + ); + } + async getSupportData(query: SupportDataQuery): Promise { const userData = await this.getUserData(query); if (!userData) throw new NotFoundException('User data not found'); @@ -821,7 +859,9 @@ export class GsService { } } - private async getExtendedBankTxData(dbQuery: DbQueryBaseDto): Promise { + private async getExtendedBankTxData( + dbQuery: DbQueryBaseDto, + ): Promise<{ data: Record[]; capReached: boolean }> { const select = dbQuery.select ? dbQuery.select.map((e) => dbQuery.table + '.' + e).join(',') : dbQuery.table; const buyCryptoData = await this.dataSource @@ -837,7 +877,7 @@ export class GsService { .andWhere('bank_tx.updated >= :updated', { updated: dbQuery.updatedSince }) .andWhere('bank_tx.type = :type', { type: BankTxType.BUY_CRYPTO }) .orderBy('bank_tx.id', dbQuery.sorting) - .take(dbQuery.maxLine) + .limit(dbQuery.maxLine) .getRawMany() .catch((e: Error) => { throw new BadRequestException(e.message); @@ -856,7 +896,7 @@ export class GsService { .andWhere('bank_tx.updated >= :updated', { updated: dbQuery.updatedSince }) .andWhere('bank_tx.type = :type', { type: BankTxType.BUY_FIAT }) .orderBy('bank_tx.id', dbQuery.sorting) - .take(dbQuery.maxLine) + .limit(dbQuery.maxLine) .getRawMany() .catch((e: Error) => { throw new BadRequestException(e.message); @@ -878,17 +918,21 @@ export class GsService { fiat: BankTxType.BUY_FIAT, }) .orderBy('bank_tx.id', dbQuery.sorting) - .take(dbQuery.maxLine) + .limit(dbQuery.maxLine) .getRawMany() .catch((e: Error) => { throw new BadRequestException(e.message); }); - return Util.sort( - buyCryptoData.concat(buyFiatData, bankTxRestData), - dbQuery.select ? 'id' : 'bank_tx_id', - dbQuery.sorting, - ); + return { + data: Util.sort( + buyCryptoData.concat(buyFiatData, bankTxRestData), + dbQuery.select ? 'id' : 'bank_tx_id', + dbQuery.sorting, + ), + // each leg is capped individually — only a full leg means rows were actually cut off + capReached: [buyCryptoData, buyFiatData, bankTxRestData].some((d) => d.length >= dbQuery.maxLine), + }; } private filterSelectDocumentColumn(select: string[]): string[] {