diff --git a/migration/1785400000000-AddFinancialLogQueryIndex.js b/migration/1785400000000-AddFinancialLogQueryIndex.js new file mode 100644 index 0000000000..52d1d331b3 --- /dev/null +++ b/migration/1785400000000-AddFinancialLogQueryIndex.js @@ -0,0 +1,64 @@ +/** + * @typedef {import('typeorm').MigrationInterface} MigrationInterface + * @typedef {import('typeorm').QueryRunner} QueryRunner + */ + +/** + * Add a composite index on the `log` table to cut disk I/O and remove the explicit Sort step + * from the minute-interval LedgerMarkService production query. + * + * The `log` table is the largest table in the database at 1353 MB (527339 rows) and previously + * had only the primary key as an index. LedgerMarkService runs a query every minute that filters + * on `system`, `subsystem`, `severity`, `valid` and a `created` range, then orders by + * `created, id`. + * + * EXPLAIN (ANALYZE, BUFFERS) in production without this index showed a Parallel Seq Scan followed + * by an explicit Sort, reading 86270 blocks (~674 MB) from disk per call, discarding 175629 rows + * per parallel worker at the filter, and taking 86 ms with correspondingly high disk load. + * + * Column order (system, subsystem, severity, valid, created, id) is intentional: four equality + * predicates first, then the two ORDER BY columns in query order. `id` is part of the index + * rather than trailing it: the query orders by `created, id`, so an index ending at `created` + * would still need a sort step whenever several rows share a `created` value — which happens, + * since the table is written at roughly two rows per minute. + * + * What this index does NOT fix: disk I/O and Sort are addressed, but not the transferred payload. + * The query still returns up to 5001 rows with an average 5.8 KB `message` column (~47 MB per + * call), which continues to pressure the Node event loop. That is a separate open topic and + * explicitly out of scope for this migration. + * + * CREATE INDEX CONCURRENTLY is not used: migrations in this codebase run transactionally and + * boot-blockingly (see src/config/config.ts, migrationsRun gated by the SQL_MIGRATE env var). + * CREATE INDEX CONCURRENTLY is not allowed inside a transaction and would crash the migration. + * + * Lock behaviour, stated precisely: a plain CREATE INDEX holds a SHARE lock for the ENTIRE build, + * not briefly. Reads continue throughout; writes to `log` block for as long as the build runs over + * the 1353 MB table. `SET LOCAL lock_timeout` caps only how long we WAIT to acquire that lock, not + * how long we hold it. The build duration has not been measured against production data, so no + * upper bound is claimed here. This is judged acceptable because `log` takes only about 2900 rows + * per day (~2 per minute) and those writes are retried by their jobs, not lost. + * + * @class + * @implements {MigrationInterface} + */ +module.exports = class AddFinancialLogQueryIndex1785400000000 { + name = 'AddFinancialLogQueryIndex1785400000000'; + + /** + * @param {QueryRunner} queryRunner + */ + async up(queryRunner) { + await queryRunner.query(`SET LOCAL lock_timeout = '5s'`); + await queryRunner.query( + `CREATE INDEX "IDX_log_financial_query" ON "log" ("system", "subsystem", "severity", "valid", "created", "id")`, + ); + } + + /** + * @param {QueryRunner} queryRunner + */ + async down(queryRunner) { + await queryRunner.query(`SET LOCAL lock_timeout = '5s'`); + await queryRunner.query(`DROP INDEX "public"."IDX_log_financial_query"`); + } +}; diff --git a/src/subdomains/supporting/payment/entities/transaction-request.entity.ts b/src/subdomains/supporting/payment/entities/transaction-request.entity.ts index 4b1c563e75..bb6fc841a9 100644 --- a/src/subdomains/supporting/payment/entities/transaction-request.entity.ts +++ b/src/subdomains/supporting/payment/entities/transaction-request.entity.ts @@ -127,7 +127,8 @@ export class TransactionRequest extends IEntity { aktionariatResponse?: string; // tx hash of the on-chain transfer that settled this request (set by the settlement job); - // each settlement tx may complete at most one request per user + // a settlement tx may contain multiple transfer events (batch settlement), each of which + // may complete at most one request per user @Column({ length: 256, nullable: true }) settlementTxId?: string; diff --git a/src/subdomains/supporting/payment/services/transaction-request.service.ts b/src/subdomains/supporting/payment/services/transaction-request.service.ts index 9dc7560b60..85281e9866 100644 --- a/src/subdomains/supporting/payment/services/transaction-request.service.ts +++ b/src/subdomains/supporting/payment/services/transaction-request.service.ts @@ -316,13 +316,15 @@ export class TransactionRequestService { }); } - async getUsedSettlementTxIds(userId: number): Promise { + async getUsedSettlements(userId: number): Promise<{ settlementTxId: string; estimatedAmount: number }[]> { return this.transactionRequestRepo .find({ where: { user: { id: userId }, settlementTxId: Not(IsNull()) }, - select: { settlementTxId: true }, + select: { settlementTxId: true, estimatedAmount: true }, }) - .then((requests) => requests.map((r) => r.settlementTxId)); + .then((requests) => + requests.map((r) => ({ settlementTxId: r.settlementTxId, estimatedAmount: r.estimatedAmount })), + ); } async updateEstimatedAmount(id: number, estimatedAmount: number): Promise { diff --git a/src/subdomains/supporting/realunit/__tests__/realunit-job.service.spec.ts b/src/subdomains/supporting/realunit/__tests__/realunit-job.service.spec.ts index 4cb7b05246..b42d03b385 100644 --- a/src/subdomains/supporting/realunit/__tests__/realunit-job.service.spec.ts +++ b/src/subdomains/supporting/realunit/__tests__/realunit-job.service.spec.ts @@ -39,7 +39,7 @@ describe('RealUnitJobService', () => { transactionRequestService = createMock(); jest.spyOn(realunitService, 'getRealuAsset').mockResolvedValue(realuAsset); - jest.spyOn(transactionRequestService, 'getUsedSettlementTxIds').mockResolvedValue([]); + jest.spyOn(transactionRequestService, 'getUsedSettlements').mockResolvedValue([]); const module: TestingModule = await Test.createTestingModule({ imports: [TestSharedModule], @@ -115,9 +115,11 @@ describe('RealUnitJobService', () => { expect(transactionRequestService.complete).toHaveBeenCalledWith(10, '0xSettlementTx'); }); - it('should not reuse a settlement tx that already completed a quote in an earlier run', async () => { + it('should not reuse a settlement transfer that already completed a quote in an earlier run', async () => { jest.spyOn(transactionRequestService, 'getOpenBuyQuotes').mockResolvedValue([quote] as any); - jest.spyOn(transactionRequestService, 'getUsedSettlementTxIds').mockResolvedValue(['0xSettlementTx']); + jest + .spyOn(transactionRequestService, 'getUsedSettlements') + .mockResolvedValue([{ settlementTxId: '0xSettlementTx', estimatedAmount: 72.123 }]); mockHistory([settlementEvent]); await service.completeSettledQuotes(); @@ -125,6 +127,82 @@ describe('RealUnitJobService', () => { expect(transactionRequestService.complete).not.toHaveBeenCalled(); }); + it('should complete multiple quotes settled in a single batch tx', async () => { + const smallQuote = { ...quote, id: 10, estimatedAmount: 219.71 }; + const largeQuote = { ...quote, id: 11, estimatedAmount: 22047 }; + jest.spyOn(transactionRequestService, 'getOpenBuyQuotes').mockResolvedValue([smallQuote, largeQuote] as any); + mockHistory([ + { ...settlementEvent, txHash: '0xBatchTx', transfer: { ...settlementEvent.transfer, value: '219' } }, + { ...settlementEvent, txHash: '0xBatchTx', transfer: { ...settlementEvent.transfer, value: '22047' } }, + ]); + + await service.completeSettledQuotes(); + + expect(transactionRequestService.complete).toHaveBeenCalledTimes(2); + expect(transactionRequestService.complete).toHaveBeenCalledWith(10, '0xBatchTx'); + expect(transactionRequestService.complete).toHaveBeenCalledWith(11, '0xBatchTx'); + }); + + it('should complete a quote from a batch tx whose other transfer already settled an earlier request', async () => { + const largeQuote = { ...quote, id: 11, estimatedAmount: 22047 }; + jest.spyOn(transactionRequestService, 'getOpenBuyQuotes').mockResolvedValue([largeQuote] as any); + jest + .spyOn(transactionRequestService, 'getUsedSettlements') + .mockResolvedValue([{ settlementTxId: '0xBatchTx', estimatedAmount: 219.71 }]); + mockHistory([ + { ...settlementEvent, txHash: '0xBatchTx', transfer: { ...settlementEvent.transfer, value: '219' } }, + { ...settlementEvent, txHash: '0xBatchTx', transfer: { ...settlementEvent.transfer, value: '22047' } }, + ]); + + await service.completeSettledQuotes(); + + expect(transactionRequestService.complete).toHaveBeenCalledWith(11, '0xBatchTx'); + }); + + it('should not reuse a same-amount transfer within a batch tx across runs', async () => { + jest.spyOn(transactionRequestService, 'getOpenBuyQuotes').mockResolvedValue([quote] as any); + jest + .spyOn(transactionRequestService, 'getUsedSettlements') + .mockResolvedValue([{ settlementTxId: '0xBatchTx', estimatedAmount: 72.9 }]); + mockHistory([{ ...settlementEvent, txHash: '0xBatchTx' }]); + + await service.completeSettledQuotes(); + + expect(transactionRequestService.complete).not.toHaveBeenCalled(); + }); + + it('should complete a second same-amount quote when the batch tx contains two matching transfers', async () => { + jest.spyOn(transactionRequestService, 'getOpenBuyQuotes').mockResolvedValue([quote] as any); + jest + .spyOn(transactionRequestService, 'getUsedSettlements') + .mockResolvedValue([{ settlementTxId: '0xBatchTx', estimatedAmount: 72.9 }]); + mockHistory([ + { ...settlementEvent, txHash: '0xBatchTx' }, + { ...settlementEvent, txHash: '0xBatchTx', timestamp: new Date('2026-06-30T09:04:00Z') }, + ]); + + await service.completeSettledQuotes(); + + expect(transactionRequestService.complete).toHaveBeenCalledWith(10, '0xBatchTx'); + }); + + it('should complete a quote when the consumed transfer is not the first event of the batch tx', async () => { + const largeQuote = { ...quote, id: 11, estimatedAmount: 22047 }; + jest.spyOn(transactionRequestService, 'getOpenBuyQuotes').mockResolvedValue([largeQuote] as any); + jest + .spyOn(transactionRequestService, 'getUsedSettlements') + .mockResolvedValue([{ settlementTxId: '0xBatchTx', estimatedAmount: 219.71 }]); + // the consumed transfer is the second event here, so a tx-hash-only match would skip the wrong one + mockHistory([ + { ...settlementEvent, txHash: '0xBatchTx', transfer: { ...settlementEvent.transfer, value: '22047' } }, + { ...settlementEvent, txHash: '0xBatchTx', transfer: { ...settlementEvent.transfer, value: '219' } }, + ]); + + await service.completeSettledQuotes(); + + expect(transactionRequestService.complete).toHaveBeenCalledWith(11, '0xBatchTx'); + }); + it('should match the oldest unused settlement transfer', async () => { const laterEvent = { ...settlementEvent, txHash: '0xLaterTx', timestamp: new Date('2026-07-01T12:00:00Z') }; jest.spyOn(transactionRequestService, 'getOpenBuyQuotes').mockResolvedValue([quote] as any); diff --git a/src/subdomains/supporting/realunit/realunit-job.service.ts b/src/subdomains/supporting/realunit/realunit-job.service.ts index ec7da7b76c..b75475ebaf 100644 --- a/src/subdomains/supporting/realunit/realunit-job.service.ts +++ b/src/subdomains/supporting/realunit/realunit-job.service.ts @@ -27,8 +27,11 @@ export class RealUnitJobService { if (!openQuotes.length) return; const historyCache = new Map(); - // per user: settlement txs already consumed by earlier runs (persisted) or earlier in this run - const usedTxIdsByUser = new Map>(); + // per user: settlement transfers already consumed by completed requests (persisted) or earlier in this + // run. The issuer may settle multiple purchases in a single tx (one transfer event each), so consumption + // is tracked per transfer event, not per tx. The history carries no per-event id, so a consumed event is + // identified by its (tx hash, share amount) pairing and counted to also cover same-amount settlements. + const consumedByUser = new Map>(); for (const quote of openQuotes) { try { @@ -41,27 +44,18 @@ export class RealUnitJobService { historyCache.set(address, history); } - let usedTxIds = usedTxIdsByUser.get(quote.user.id); - if (!usedTxIds) { - usedTxIds = new Set(await this.transactionRequestService.getUsedSettlementTxIds(quote.user.id)); - usedTxIdsByUser.set(quote.user.id, usedTxIds); + let consumed = consumedByUser.get(quote.user.id); + if (!consumed) { + consumed = await this.getConsumedSettlements(quote.user.id); + consumedByUser.set(quote.user.id, consumed); } - // quotes are ordered oldest-first, so match the oldest unused settlement transfer - const settlement = history - .filter( - (e) => - e.transfer && - !usedTxIds.has(e.txHash) && - Util.equalsIgnoreCase(e.transfer.to, address) && - Number(e.transfer.value) === expectedShares && - e.timestamp >= quote.created, - ) - .sort((a, b) => a.timestamp.getTime() - b.timestamp.getTime()) - .at(0); + // quotes are ordered oldest-first, so match the oldest unconsumed settlement transfer + const settlement = this.findUnconsumedSettlement(history, consumed, address, expectedShares, quote.created); if (!settlement) continue; - usedTxIds.add(settlement.txHash); + const key = this.settlementKey(settlement.txHash, expectedShares); + consumed.set(key, (consumed.get(key) ?? 0) + 1); await this.transactionRequestService.complete(quote.id, settlement.txHash); this.logger.info( @@ -83,4 +77,52 @@ export class RealUnitJobService { async reconcilePendingTransfers(): Promise { await this.realunitService.reconcilePendingTransfers(); } + + // --- HELPER METHODS --- // + + private async getConsumedSettlements(userId: number): Promise> { + const settlements = await this.transactionRequestService.getUsedSettlements(userId); + + const consumed = new Map(); + for (const settlement of settlements) { + const key = this.settlementKey(settlement.settlementTxId, Math.floor(settlement.estimatedAmount)); + consumed.set(key, (consumed.get(key) ?? 0) + 1); + } + + return consumed; + } + + // Walks all incoming transfers oldest-first and treats the first n events of each (tx hash, share amount) + // pairing as consumed, where n is the number of settlements already recorded for that pairing — so a batch + // settlement tx can complete one request per contained transfer event, but never the same event twice. + private findUnconsumedSettlement( + history: HistoryEventDto[], + consumed: Map, + address: string, + expectedShares: number, + minTimestamp: Date, + ): HistoryEventDto | undefined { + const incomingTransfers = Util.sort( + history.filter((e) => e.transfer && Util.equalsIgnoreCase(e.transfer.to, address)), + 'timestamp', + ); + + const seen = new Map(); + + for (const event of incomingTransfers) { + const shares = Number(event.transfer.value); + const key = this.settlementKey(event.txHash, shares); + const position = (seen.get(key) ?? 0) + 1; + seen.set(key, position); + + if (position <= (consumed.get(key) ?? 0)) continue; + if (shares === expectedShares && event.timestamp >= minTimestamp) return event; + } + + return undefined; + } + + private settlementKey(txHash: string, shares: number): string { + return `${txHash.toLowerCase()}|${shares}`; + } }