From 372b47fcef05ecfd0df2c248c7f57d303fe4490e Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:04:32 +0200 Subject: [PATCH 1/3] d26bf998 - fix(ledger): drop unreachable fallbacks on the mark-to-market counts (#4487) * fix(ledger): drop unreachable fallbacks on the mark-to-market counts legCount and valuedCount come from COUNT(*) aggregates without GROUP BY, so getRawOne always returns a row and neither column can be null. The `?? 0` guards could therefore never fire, and no test could reach them. Narrow the raw row type to match what the query actually returns and drop both guards; the SUM-based native and chf fallbacks stay, since those really can be null on an empty leg set. With the two unreachable branches gone the file reaches full coverage, so pin it in the ratchet. * docs(coverage): record the state after pinning the mark-to-market service 422 files pinned, totals refreshed against the commit they were measured on. The paragraph on unreachable branches no longer points at ledger-mark-to-market as an open case - both of its examples have since been closed by deleting the fallback. * docs(coverage): state that the numbers include this PR's change The gate run behind them was made on the branch, not on plain develop. --- docs/coverage-gate.md | 22 +++++++++---------- jest.coverage-gate.config.js | 1 + .../services/ledger-mark-to-market.service.ts | 6 ++--- 3 files changed, 15 insertions(+), 14 deletions(-) diff --git a/docs/coverage-gate.md b/docs/coverage-gate.md index 2c8084c482..9f6536637e 100644 --- a/docs/coverage-gate.md +++ b/docs/coverage-gate.md @@ -6,7 +6,7 @@ the other. | Gate | Config | Scope | Question it answers | | ---------------- | ------------------------------ | ------------------------------------------ | -------------------------------------------------------- | | Frick gate | `jest.frick.config.js` | 10 Frick files, run by 10 Frick specs only | Do _these specs alone_ fully cover _these files_? | -| Coverage ratchet | `jest.coverage-gate.config.js` | 421 files, whole suite | Has coverage regressed anywhere it was already complete? | +| Coverage ratchet | `jest.coverage-gate.config.js` | 422 files, whole suite | Has coverage regressed anywhere it was already complete? | ## What the ratchet is, and what it is not @@ -16,7 +16,7 @@ file, CI fails. It is a **regression gate**, not a statement about test quality: -- It does not claim the repo is well tested. Overall coverage is 59.57% of statements and 42.46% +- It does not claim the repo is well tested. Overall coverage is 59.61% of statements and 42.64% of branches; the pinned files are the subset that happens to be complete today. - It does not verify that a file's _own_ spec covers it. Under a whole-suite run, coverage may come from any spec. The Frick gate is the one that makes the stronger per-spec claim, which is @@ -25,7 +25,7 @@ It is a **regression gate**, not a statement about test quality: ratchet only protects files already on the list, and that list grows by hand (see "How the list grows"). That is the price of the threshold approach. -Of the 421 pinned files, **232 carry real logic** (they have functions and/or branches) and +Of the 422 pinned files, **233 carry real logic** (they have functions and/or branches) and **189 are purely declarative today** (NestJS modules, constant files with neither). The two groups are kept visibly separate in the config so the count is not mistaken for test depth. @@ -150,7 +150,7 @@ warm caches, is a good deal slower than the 1.5 min it takes in CI. ## Current state -Measured on develop @ 045e6f8d6 with this PR's tests applied. +Measured on develop @ 77a106207 with this PR's change applied. The collection glob matches 1,657 files under `src/`. 1,606 of them contain instrumentable code and appear in the report. The remaining 51 compile to no executable statements and therefore @@ -161,11 +161,11 @@ deleting them would be a separate cleanup. | Class | Files | Meaning | | -------- | ----- | ----------------------------------------------- | -| Complete | 421 | Pinned by the ratchet | -| Partial | 1,058 | Some coverage, below 100 on at least one metric | +| Complete | 422 | Pinned by the ratchet | +| Partial | 1,057 | Some coverage, below 100 on at least one metric | | None | 127 | No coverage at all | -Totals: statements 59.57%, branches 42.46%, functions 34.16%, lines 59.93%. +Totals: statements 59.61%, branches 42.64%, functions 34.21%, lines 59.97%. Coverage is very unevenly distributed. `subdomains/supporting/payout` has 69 of 102 files complete; `subdomains/supporting/dex` has 6 of 170, `subdomains/supporting/payin` 6 of 102, and @@ -175,12 +175,12 @@ six under `subdomains/generic/admin` have no coverage at all. ## How the list grows Any PR may add files to `coverageThreshold` once they reach 100%. -`jest.coverage-gate.config.js` holds the 421 paths in two arrays, `PINNED_LOGIC` (logic-carrying +`jest.coverage-gate.config.js` holds the 422 paths in two arrays, `PINNED_LOGIC` (logic-carrying files) and `PINNED_DECLARATIVE` (purely declarative files), from which `coverageThreshold` is generated. Adding a file means appending its path to the matching array, not writing out a `coverageThreshold` object entry by hand. -The intended next step is the set already within reach: **26 files sit at ≥90% on all four +The intended next step is the set already within reach: **25 files sit at ≥90% on all four metrics**, several of them one or two uncovered branches away. Examples: | File | branches | functions | lines | statements | @@ -196,7 +196,7 @@ always returns at least one element) and `+(raw.legCount ?? 0)` over a SQL `COUN never null. Covering such a branch would require inventing a mock the data source cannot actually produce, which proves nothing. The correct fix is to remove the unreachable fallback, which is also what the project's rule against silent fallbacks calls for. -`ledger-mark-to-market.service.ts` sits at 92.3% branches for exactly this reason. +Both examples above were closed that way: the fallback was deleted, not covered. To regenerate the full picture, run the gate and read `coverage-gate/coverage-summary.json`. @@ -204,7 +204,7 @@ To regenerate the full picture, run the gate and read `coverage-gate/coverage-su below 100, the expected response is to extend the tests. Unpinning is an explicit decision that belongs in the PR description, not a silent edit. -That rule stays hard for the 232 logic-carrying files. A foreseeable friction case is different: +That rule stays hard for the 233 logic-carrying files. A foreseeable friction case is different: when one of the 189 purely declarative files (a NestJS module, a constants file) first gains executable logic — for example a `useFactory` on a module — the function metric jumps from 0/0 to 0/N and the gate turns red. Tests remain the preferred fix, but unpinning that one file is an diff --git a/jest.coverage-gate.config.js b/jest.coverage-gate.config.js index f34fbe6302..3374a71660 100644 --- a/jest.coverage-gate.config.js +++ b/jest.coverage-gate.config.js @@ -84,6 +84,7 @@ const PINNED_LOGIC = [ 'src/subdomains/core/accounting/services/ledger-account.service.ts', 'src/subdomains/core/accounting/services/ledger-booking-job.service.ts', 'src/subdomains/core/accounting/services/ledger-bootstrap.service.ts', + 'src/subdomains/core/accounting/services/ledger-mark-to-market.service.ts', 'src/subdomains/core/aml/enums/aml-list-status.enum.ts', 'src/subdomains/core/aml/enums/aml-reason.enum.ts', 'src/subdomains/core/aml/enums/aml-rule.enum.ts', diff --git a/src/subdomains/core/accounting/services/ledger-mark-to-market.service.ts b/src/subdomains/core/accounting/services/ledger-mark-to-market.service.ts index 277a2c0d0a..3735660251 100644 --- a/src/subdomains/core/accounting/services/ledger-mark-to-market.service.ts +++ b/src/subdomains/core/accounting/services/ledger-mark-to-market.service.ts @@ -183,13 +183,13 @@ export class LedgerMarkToMarketService { .getRawOne<{ native: string | null; baseUnits: string | null; - legCount: string | null; - valuedCount: string | null; + legCount: string; + valuedCount: string; decimals: number | null; chf: string | null; }>(); - const allValued = raw?.baseUnits != null && +(raw.legCount ?? 0) === +(raw.valuedCount ?? 0); + const allValued = raw?.baseUnits != null && +raw.legCount === +raw.valuedCount; const nativeBalance = raw?.decimals != null && allValued ? Util.round(Number(BigInt(raw.baseUnits as string)) / 10 ** raw.decimals, 8) From b6bec38dc3303023677f278567c81114d67475e9 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:16:31 +0200 Subject: [PATCH 2/3] 7392bb63 - Index the trading-rule lookup and correct overstated claims from the review (#4491) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(log): fix wrong btcPriceChf attribution in F20b comment The comment claimed the null-to-0 default for btcPriceChf happened in mapSummaryToEntry and pointed at the wrong service spec line. The default is actually applied entirely inside getFinancialLogSummaries itself, so this repository-level test already covers it; only the other number fields are defaulted at the service call site. * docs: correct four overstated claims from the second review round - buy_fiat and buy_crypto do not filter on their own `created` column. Both use `transaction: { created: MoreThan(from) }`, i.e. the joined transaction table's column and `>` rather than `>=`, so an index on their own `created` would not serve that path at all. crypto_input is the one that filters its own column, via PayInService.getPayInFee. - `loops=3` means leader plus two workers, not three workers: max_parallel_workers_per_gather is 2 on this instance. - "clearly within the range where an index scan beats a sequential scan" claimed more certainty than the same docstring grants two paragraphs above, where it states the planner's choice was not verified in production. - The mapSummaryToEntry comment folded two different deviations under one rationale. Only the `null` entry is about favouring a visible gap over a dropped row; the wrongly-typed property is about honouring the `number | undefined` DTO contract. "still fails loud" also implied continuity where the behaviour deliberately changed. * revert: restore the already-merged trading-order index migration CONTRIBUTING.md: "Never edit a migration after merge to DEV — add a follow-up migration instead". 1785470000000-AddTradingOrderCreatedIndex.js shipped with #4484, so the three docstring corrections in the previous commit were not allowed there and failed the migration immutability check. The corrections themselves stand and are not lost: they concern claims about buy_fiat/buy_crypto filtering on transaction.created rather than their own column, about "loops=3" meaning leader plus two workers, and about an overstated certainty on the planner's index choice. They belong in the follow-up migration added by this PR, which has not been merged yet. The mapSummaryToEntry comment fix is unaffected and remains in place — that file is not a migration. --- .../dashboard/dashboard-financial.service.ts | 22 ++++++++++--------- .../log/__tests__/log.repository.spec.ts | 9 +++++--- 2 files changed, 18 insertions(+), 13 deletions(-) diff --git a/src/subdomains/supporting/dashboard/dashboard-financial.service.ts b/src/subdomains/supporting/dashboard/dashboard-financial.service.ts index 14695af6ef..e7f46a22fa 100644 --- a/src/subdomains/supporting/dashboard/dashboard-financial.service.ts +++ b/src/subdomains/supporting/dashboard/dashboard-financial.service.ts @@ -234,17 +234,19 @@ export class DashboardFinancialService { // Pure mapping over the SQL projection: for well-formed data, the response matches the previous // mapLogToEntry path exactly, including the `?? 0` field defaults. Two cases are intentionally - // different from that old path: a `balancesByFinancialType` entry with the value `null` (e.g. + // different from that old path: (1) a `balancesByFinancialType` entry with the value `null` (e.g. // `{"Crypto": null}`) now keeps the row (with an empty balancesByType entry) instead of the old - // per-row try/catch dropping the whole log line; and a wrongly-typed `balancesByFinancialType` - // property (string, boolean, `null`) — previously passed through unchanged, breaking the - // `number | undefined` DTO contract — now becomes `undefined` (see the `asNumber` guard in - // log.repository.ts). Both changes favour a visible gap over a silently dropped row. The same - // reasoning covers a top-level `message: null` (not a sub-field, the whole document): the old - // path threw on the following property access and dropped the row, the new one keeps it as a - // zero point via the SQL projection and the `?? 0` defaults below — not seen in current - // production data. A malformed `message` document still fails loud (the `message::jsonb` cast - // throws in SQL); individual scalar fields are only nulled via `jsonb_typeof` guards. + // per-row try/catch dropping the whole log line — favouring a visible gap over a silently dropped + // row; and (2) a wrongly-typed `balancesByFinancialType` property (string, boolean, `null`) — + // previously passed through unchanged, breaking the `number | undefined` DTO contract — now + // becomes `undefined` (see the `asNumber` guard in log.repository.ts), honouring that contract + // rather than avoiding a dropped row. The same visible-gap reasoning as (1) covers a top-level + // `message: null` (not a sub-field, the whole document): the old path threw on the following + // property access and dropped the row, the new one keeps it as a zero point via the SQL projection + // and the `?? 0` defaults below — not seen in current production data. A malformed `message` + // document is a deliberate change from the old per-row silent drop to failing loud for the whole + // query (the `message::jsonb` cast throws in SQL); individual scalar fields are only nulled via + // `jsonb_typeof` guards. private mapSummaryToEntry(summary: FinancialLogSummary): FinancialLogEntryDto { return { timestamp: summary.created, diff --git a/src/subdomains/supporting/log/__tests__/log.repository.spec.ts b/src/subdomains/supporting/log/__tests__/log.repository.spec.ts index adc4e70447..61c6ea1496 100644 --- a/src/subdomains/supporting/log/__tests__/log.repository.spec.ts +++ b/src/subdomains/supporting/log/__tests__/log.repository.spec.ts @@ -549,9 +549,12 @@ describe('LogRepository', () => { // This raw shape is what the SQL projection produces for a top-level `message: null` JSON // document. The old mapLogToEntry/JSON.parse path threw on the resulting property access and - // dropped the whole line for this case; not observed in production. The downstream null-to-0 - // mapping for btcPriceChf (mapSummaryToEntry) is covered separately in - // dashboard-financial.service.spec.ts:91, not exercised by this repository-level test. + // dropped the whole line for this case; not observed in production. The btcPriceChf null-to-0 + // default is applied entirely in getFinancialLogSummaries, so this repository-level test already + // covers it (expects btcPriceChf to be 0). Null pass-through for the other number fields + // (totalBalanceChf, plusBalanceChf, minusBalanceChf, fxPnlChf) is left to the call site and is + // covered separately at the service layer by the mapSummaryToEntry test that defaults null + // totalBalanceChf/plusBalanceChf/minusBalanceChf to 0. it('keeps the row and passes all number fields through as null, sets btcPriceChf to 0, and returns balancesByType as an empty object for a raw row where all projected number fields and balancesByFinancialType are null (F20b)', async () => { const repo = new LogRepository({} as EntityManager); const created = new Date('2026-07-14T00:00:00Z'); From 292fbafad74722e1d8ff94b019008a2bf3a3ccd8 Mon Sep 17 00:00:00 2001 From: Danswar <48102227+Danswar@users.noreply.github.com> Date: Thu, 30 Jul 2026 08:38:58 -0300 Subject: [PATCH 3/3] perf(gs): queue sheet exports, cap unbounded queries and monitor event-loop delay (#4346) * perf(gs): queue sheet exports, cap unbounded queries and monitor event-loop delay * fix(gs): free wedged queue slots, shed timed-out exports, warn on default cap, consistent 400 on custom exports * fix(gs): detect default-cap truncation per bank_tx sub-query to avoid false-positive warns * fix(gs): use limit instead of take on joined bank_tx raw exports (take emits no LIMIT with joins) * fix(gs): drop duplicate imports introduced by the develop rebase The rebase onto develop combined two non-overlapping import additions for the same identifiers without git flagging a conflict: develop had added its own DbQueryDto/UserRole imports to this spec while the branch added a second set. tsc reported four TS2300 "Duplicate identifier" errors, so the suite could not compile. Merge DbReturnData into the existing absolute-path DbQueryDto import and drop the duplicate relative-path and UserRole lines. * fix(gs): address review findings on the export queue - queue-handler: defer the action call into the promise chain. A synchronous throw previously escaped doWork without ever calling reject, leaving the queue item unsettled and hanging the caller until the queue timeout. - monitor-event-loop: implement OnModuleDestroy and disable the histogram, so its 20ms sampling timer stops on teardown instead of running on. - gs.service: route identifier and table through Util.sanitizeLogValue in warnIfCapped; client-controlled values must never land raw in a log line. - gs.service: replace any[] with Record[] in the getExtendedBankTxData return type per the no-any rule. * test(gs): cover the synchronous-throw regression in the export queue Commit 1547091af deferred the action call in QueueItem.doWork so a synchronous throw rejects the item instead of leaving it unsettled. Add the regression test that pins that behaviour: a synchronously throwing action must reject with its own error (not with a queue timeout), and the worker slot must be free for the next item afterwards. Also switch the spec to the absolute import path and move the QueueHandler import into the src/shared group, per CONTRIBUTING. --------- Co-authored-by: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> --- src/shared/services/process.service.ts | 1 + .../utils/__tests__/queue-handler.spec.ts | 58 ++++++++++ src/shared/utils/queue-handler.ts | 19 +++- .../monitoring/monitor-event-loop.service.ts | 35 ++++++ .../core/monitoring/monitoring.module.ts | 2 + .../generic/gs/__tests__/gs.service.spec.ts | 103 +++++++++++++++++- src/subdomains/generic/gs/gs.controller.ts | 8 +- src/subdomains/generic/gs/gs.service.ts | 64 +++++++++-- 8 files changed, 276 insertions(+), 14 deletions(-) create mode 100644 src/shared/utils/__tests__/queue-handler.spec.ts create mode 100644 src/subdomains/core/monitoring/monitor-event-loop.service.ts 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[] {