perf(gs): queue sheet exports, cap unbounded queries and monitor event-loop delay

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',

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<number> => {
+      throw new Error('boom');
+    });
+    await expect(throwing).rejects.toThrow('boom');
+
+    await expect(queue.handle(async () => 'ok')).resolves.toBe('ok');
+
+    queue.stop();
+  });
+});

src/shared/utils/queue-handler.ts

@@ -7,29 +7,40 @@ class QueueItem<T> {
   private resolve: (value: T | PromiseLike<T>) => void;
   private reject: (e: Error) => void;
 
+  private settled = false;
+
   constructor(
     private readonly action: () => Promise<T>,
     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);
       };
     });
     if (timeout) this.timeout = setTimeout(() => this.reject(new Error('Queue timeout')), timeout);
   }
 
+  get isSettled(): boolean {
+    return this.settled;
+  }
+
   public wait(): Promise<T> {
     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--);

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();
+  }
+}

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,

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<DataSource>());
+      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<Record<string, unknown>[]>;
+  transformResultArray: (data: Record<string, unknown>[], table: string, role: UserRole) => DbReturnData;
+};
+
+function internals(service: GsService): GsServiceInternals {
+  return service as unknown as GsServiceInternals;
+}
+
+function exportQuery(overrides: Partial<DbQueryDto> = {}): DbQueryDto {
+  return plainToInstance(DbQueryDto, { table: 'user', updatedSince: '2026-01-01', ...overrides });
+}
+
+describe('GS export queue', () => {
+  let service: GsService;
+
+  beforeEach(() => {
+    service = buildGsService(createMock<KycDocumentService>(), createMock<DataSource>());
+  });
+
+  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<void>((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);
+    });
+  });
+});

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')