From afefec79e1308a0aab2b2937286af6981a09c6f7 Mon Sep 17 00:00:00 2001 From: Zainab <295377647+zainabbaba31-source@users.noreply.github.com> Date: Thu, 24 Sep 2026 21:36:06 +0100 Subject: [PATCH] feat(payments,credits): cap confirmation retries, persist reconciliation history, alert on usage spikes Caps scheduled confirmation polling per payment. A payment that reaches PAYMENT_CONFIRMATION_MAX_ATTEMPTS or PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK is escalated to MANUAL_REVIEW with a reason naming the cap and is excluded from later passes without another provider call. The provider-error streak cap is the one deliberate exception to the existing rule that a provider outage must not mass-flag payments, because a long bounded streak on a single payment is a different signal than one bad run; an alert is logged when either cap trips. Persists every reconciliation pass in a new reconciliation_runs table (with migration, entity, and registration) so drift can be answered historically, recording the summary counters, duration, and outcome, with failures recorded rather than masking the original error. Exposes the history at GET /payments/admin/reconciliation-runs with a bounded limit. Adds threshold-based credit-usage spike detection after a usage event is charged. It alerts only with enough history, a meaningful absolute total, and a multiple of the rolling baseline; the alert is logged, counted on /metrics, and emailed best-effort, and can never roll back or fail the charge. Isolates the wallet balance read: it now runs inside a transaction that takes the same pessimistic row lock the funding path takes, and the ledger aggregate runs through that same transaction manager. Documents the confirmed READ COMMITTED isolation guarantee and the writers it does not cover. Closes #1801 Closes #1800 Closes #1799 Closes #1798 --- backend/.env.example | 27 +- backend/src/common/metrics.service.ts | 5 + .../src/credits/metered-usage.service.spec.ts | 132 ++++++- backend/src/credits/metered-usage.service.ts | 244 +++++++++++- ...003000000-CreateReconciliationRunsTable.ts | 41 ++ .../reconciliation-run-response.dto.spec.ts | 36 ++ .../dto/reconciliation-run-response.dto.ts | 47 +++ .../entities/reconciliation-run.entity.ts | 64 +++ .../payments/payment-confirmation.service.ts | 11 +- .../src/payments/payments-admin.controller.ts | 23 ++ backend/src/payments/payments.module.ts | 8 +- .../payments/reconciliation.service.spec.ts | 236 ++++++++++- .../src/payments/reconciliation.service.ts | 370 +++++++++++++++++- backend/src/wallets/wallets.service.spec.ts | 134 ++++++- backend/src/wallets/wallets.service.ts | 53 ++- 15 files changed, 1372 insertions(+), 59 deletions(-) create mode 100644 backend/src/database/migrations/1790003000000-CreateReconciliationRunsTable.ts create mode 100644 backend/src/payments/dto/reconciliation-run-response.dto.spec.ts create mode 100644 backend/src/payments/dto/reconciliation-run-response.dto.ts create mode 100644 backend/src/payments/entities/reconciliation-run.entity.ts diff --git a/backend/.env.example b/backend/.env.example index f655b283..30089e31 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -100,8 +100,22 @@ PAYMENT_RECONCILE_BACKOFF_MAX_MINUTES=60 PAYMENT_RECONCILE_MAX_BATCH=500 # A payment unresolved this long escalates to MANUAL_REVIEW — but only on a # pass where the provider was actually reachable; a provider outage never -# by itself causes escalation (see ReconciliationService). +# by itself causes escalation (see ReconciliationService). The two hard caps +# below are independent safety bounds: they stop further provider polling and +# hand the payment to a human as soon as either cap is reached, even before +# the age threshold. If a provider-error attempt hits the total-attempt cap +# first, the next pass escalates it without making another provider call. +# The provider-error-streak cap deliberately overrides the outage safeguard +# only after a long, bounded per-payment streak; one bad run still does not +# mass-flag payments. PAYMENT_MANUAL_REVIEW_AFTER_HOURS=24 +# Total scheduled confirmation polls for one payment before it is escalated +# to MANUAL_REVIEW and excluded from future reconciliation selection. +PAYMENT_CONFIRMATION_MAX_ATTEMPTS=20 +# Consecutive provider-unreachable polls for one payment before it is +# escalated to MANUAL_REVIEW, even when the provider-error attempt is the one +# that trips the cap. +PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK=10 # When the manual-review queue depth exceeds this the reconciliation cron # logs a WARN, emails SUPPORT_EMAIL, and the depth stays visible on the # Prometheus /metrics endpoint (managehub_manual_review_queue_depth) for @@ -153,6 +167,17 @@ CREDITS_SETTLEMENT_MAX_PAYOUT_ATTEMPTS=5 # Upper bound on how many CONFIRMED payments one pass of the credit sweep # (top-ups and payment revenue splits) will apply. CREDITS_PAYMENT_SWEEP_MAX_BATCH=200 +# Usage-spike anomaly detection. After a new usage event is charged, the +# service compares the tenant's current rolling window total with the +# average of comparable prior windows. It alerts only when there are at +# least MIN_SAMPLES prior events, the window is at least MULTIPLIER times +# that baseline, and the window total is at least MIN_AMOUNT minor units. +# The absolute floor prevents tiny tenants from alerting on rounding; SMTP +# delivery is best-effort and never rolls back the charge. +CREDITS_USAGE_SPIKE_WINDOW_MINUTES=60 +CREDITS_USAGE_SPIKE_MULTIPLIER=5.0 +CREDITS_USAGE_SPIKE_MIN_AMOUNT=10000 +CREDITS_USAGE_SPIKE_MIN_SAMPLES=3 # Data retention (issue BE-144) # Set to false to stop the daily retention job entirely. diff --git a/backend/src/common/metrics.service.ts b/backend/src/common/metrics.service.ts index 877442f6..fe9ddbe5 100644 --- a/backend/src/common/metrics.service.ts +++ b/backend/src/common/metrics.service.ts @@ -42,6 +42,11 @@ export class MetricsService { this.incrementCounter('managehub_settlement_payout_failures_total', {}); } + /** Counts a detected tenant credit-usage anomaly by metered resource. */ + recordCreditUsageSpike(resource: string): void { + this.incrementCounter('managehub_credit_usage_spikes_total', { resource }); + } + renderPrometheus(): string { const lines: string[] = []; this.appendCounters(lines); diff --git a/backend/src/credits/metered-usage.service.spec.ts b/backend/src/credits/metered-usage.service.spec.ts index e3ddcec5..778559f8 100644 --- a/backend/src/credits/metered-usage.service.spec.ts +++ b/backend/src/credits/metered-usage.service.spec.ts @@ -4,27 +4,37 @@ import { InsufficientCreditException, LedgerService } from './ledger.service'; import { MeteredUsageService } from './metered-usage.service'; import { LedgerAccountKind } from './enums/ledger-account-kind.enum'; import { MeteredResource } from './enums/metered-resource.enum'; +import { createTransport } from 'nodemailer'; import { createLedgerHarness, fakeConfigService, } from './testing/in-memory-ledger'; -function build() { +jest.mock('nodemailer', () => ({ + createTransport: jest.fn(), +})); + +function build(configValues: Record = {}) { const harness = createLedgerHarness(); + const config = fakeConfigService({ + CREDITS_DEFAULT_CURRENCY: 'USD', + CREDITS_DEFAULT_OVERDRAFT_LIMIT: 0, + ...configValues, + }); const ledger = new LedgerService( harness.accounts as any, harness.transactions as any, harness.entries as any, ); - const credits = new CreditsService( - ledger, - fakeConfigService({ - CREDITS_DEFAULT_CURRENCY: 'USD', - CREDITS_DEFAULT_OVERDRAFT_LIMIT: 0, - }), + const credits = new CreditsService(ledger, config); + const metrics = { recordCreditUsageSpike: jest.fn() }; + const usage = new MeteredUsageService( + harness.usageEvents as any, + credits, + config as any, + metrics as any, ); - const usage = new MeteredUsageService(harness.usageEvents as any, credits); - return { harness, ledger, credits, usage }; + return { harness, ledger, credits, usage, config, metrics }; } async function fund(credits: CreditsService, amount: number) { @@ -45,6 +55,28 @@ const meterReading = { }; describe('MeteredUsageService', () => { + beforeEach(() => { + (createTransport as jest.Mock).mockReset(); + }); + + function seedPriorUsage(harness: any, amounts: number[]) { + const now = Date.now(); + amounts.forEach((amount, index) => { + harness.usageEvents.rows.push({ + id: `prior-${index}`, + userId: 'user-1', + resource: MeteredResource.RESOURCE_MINUTES, + units: 1, + unitPrice: amount, + amount, + currency: 'USD', + usageReference: `prior-${index}`, + ledgerTransactionId: `prior-transaction-${index}`, + createdAt: new Date(now - (70 + index * 10) * 60_000), + }); + }); + } + it('prices the reading and charges it against the credit balance', async () => { const { usage, credits, harness } = build(); await fund(credits, 1000); @@ -163,4 +195,86 @@ describe('MeteredUsageService', () => { expect((await credits.getBalance('user-1')).balance).toBe(1000 - 250); expect(await usage.listForUser('user-1')).toHaveLength(3); }); + + it('does not alert when the current window is below the spike threshold', async () => { + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_WINDOW_MINUTES: 60, + CREDITS_USAGE_SPIKE_MULTIPLIER: 5, + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + }); + seedPriorUsage(harness, [100, 100, 100]); + await fund(credits, 5000); + + await usage.recordUsage({ + ...meterReading, + units: 80, + usageReference: 'below-spike', + }); + + expect(metrics.recordCreditUsageSpike).not.toHaveBeenCalled(); + }); + + it('does not alert without the minimum number of prior samples', async () => { + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + }); + seedPriorUsage(harness, [10, 10]); + await fund(credits, 5000); + + await usage.recordUsage({ + ...meterReading, + units: 200, + usageReference: 'insufficient-history', + }); + + expect(metrics.recordCreditUsageSpike).not.toHaveBeenCalled(); + }); + + it('increments the usage-spike counter when a sufficiently large spike is detected', async () => { + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + }); + seedPriorUsage(harness, [10, 10, 10]); + await fund(credits, 5000); + + await usage.recordUsage({ + ...meterReading, + units: 200, + usageReference: 'large-spike', + }); + + expect(metrics.recordCreditUsageSpike).toHaveBeenCalledWith( + MeteredResource.RESOURCE_MINUTES, + ); + }); + + it('does not let a failing SMTP alert break usage recording', async () => { + const sendMail = jest.fn().mockRejectedValue(new Error('smtp unavailable')); + (createTransport as jest.Mock).mockReturnValue({ sendMail }); + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + SMTP_HOST: 'smtp.example.test', + SMTP_USER: 'smtp-user', + SMTP_PASS: 'smtp-pass', + SMTP_FROM_EMAIL: 'alerts@example.test', + SUPPORT_EMAIL: 'support@example.test', + }); + seedPriorUsage(harness, [10, 10, 10]); + await fund(credits, 5000); + + await expect( + usage.recordUsage({ + ...meterReading, + units: 200, + usageReference: 'smtp-failure-spike', + }), + ).resolves.toMatchObject({ charged: true }); + expect(metrics.recordCreditUsageSpike).toHaveBeenCalled(); + expect(sendMail).toHaveBeenCalledTimes(1); + expect((await credits.getBalance('user-1')).balance).toBe(4000); + }); }); diff --git a/backend/src/credits/metered-usage.service.ts b/backend/src/credits/metered-usage.service.ts index 05e45d70..22c8fa14 100644 --- a/backend/src/credits/metered-usage.service.ts +++ b/backend/src/credits/metered-usage.service.ts @@ -1,9 +1,14 @@ -import { BadRequestException, Injectable } from '@nestjs/common'; +import { BadRequestException, Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; import { InjectRepository } from '@nestjs/typeorm'; -import { Repository } from 'typeorm'; +import { MoreThanOrEqual, Repository } from 'typeorm'; import { MeteredUsageEvent } from './entities/metered-usage-event.entity'; import { MeteredResource } from './enums/metered-resource.enum'; import { CreditsService } from './credits.service'; +import { MetricsService } from '../common/metrics.service'; +import { withRequestId } from '../common/request-context'; +import { createTransport } from 'nodemailer'; +import Handlebars from 'handlebars'; const POSTGRES_UNIQUE_VIOLATION = '23505'; @@ -46,10 +51,14 @@ export interface RecordUsageResult { */ @Injectable() export class MeteredUsageService { + private readonly logger = new Logger(MeteredUsageService.name); + constructor( @InjectRepository(MeteredUsageEvent) private readonly usageRepository: Repository, private readonly credits: CreditsService, + private readonly config: ConfigService, + private readonly metrics: MetricsService, ) {} async recordUsage(input: RecordUsageInput): Promise { @@ -95,8 +104,9 @@ export class MeteredUsageService { actorId: input.actorId ?? null, }); + let event: MeteredUsageEvent; try { - const event = await this.usageRepository.save( + event = await this.usageRepository.save( this.usageRepository.create({ userId: input.userId, resource: input.resource, @@ -108,7 +118,6 @@ export class MeteredUsageService { ledgerTransactionId: charge.transaction.id, }), ); - return { event, charged: charge.posted }; } catch (error) { // Two deliveries of the same meter reading raced. The ledger already // deduped the charge, so the loser just reports the winner's event. @@ -122,6 +131,233 @@ export class MeteredUsageService { } throw error; } + + // Detection is deliberately after both durable writes. A detector, + // metrics, or email failure is observational only and must never undo + // the charge or make the caller retry a successfully recorded event. + if (charge.posted) { + try { + await this.detectUsageSpike(event); + } catch (error) { + this.logger.warn( + withRequestId( + `Unable to evaluate credit usage spike for ${input.userId}: ${ + error instanceof Error ? error.message : String(error) + }`, + ), + ); + } + } + + return { event, charged: charge.posted }; + } + + /** + * Detects a tenant-level usage spike after the new event is durable. + * + * The comparison is deliberately tenant-wide rather than resource-wide: + * an integration bug can move several metered resources at once, and a + * shared tenant total catches that pattern. The current window is the + * configured number of minutes (default 60) ending at the new event and + * includes that event. The baseline is the average total of the immediately + * preceding equal-length windows that contain at least one prior event; + * requiring at + * least CREDITS_USAGE_SPIKE_MIN_SAMPLES prior events (default 3) keeps a new + * tenant from producing a meaningless percentage. The absolute minimum + * amount (default 10,000 minor units) suppresses alerts for tiny tenants + * where normal rounding or one small event would otherwise look like a + * large ratio. A spike therefore needs + * all three signals: enough history, a meaningful absolute total, and a + * total at least CREDITS_USAGE_SPIKE_MULTIPLIER (default 5.0) times the + * baseline. + * + * This method is observational. Its caller catches every failure after + * the charge, and the SMTP implementation below is independently + * best-effort, so neither an alert problem nor a provider outage can roll + * back usage accounting. + */ + private async detectUsageSpike(event: MeteredUsageEvent): Promise { + const windowMinutes = this.getUsageSpikeWindowMinutes(); + const multiplier = this.getUsageSpikeMultiplier(); + const minimumAmount = this.getUsageSpikeMinimumAmount(); + const minimumSamples = this.getUsageSpikeMinimumSamples(); + const windowMs = windowMinutes * 60_000; + const baselineWindowCount = Math.max(3, minimumSamples); + const eventTime = + event.createdAt instanceof Date + ? new Date(event.createdAt.getTime()) + : new Date(); + const currentWindowStart = eventTime.getTime() - windowMs; + const historyStart = currentWindowStart - windowMs * baselineWindowCount; + const history = await this.usageRepository.find({ + where: { + userId: event.userId, + createdAt: MoreThanOrEqual(new Date(historyStart)), + }, + order: { createdAt: 'ASC' }, + }); + + const relevantHistory = (history ?? []).filter((candidate) => { + const timestamp = candidate.createdAt.getTime(); + return timestamp >= historyStart && timestamp <= eventTime.getTime(); + }); + const priorEvents = relevantHistory.filter( + (candidate) => candidate.createdAt.getTime() < currentWindowStart, + ); + if (priorEvents.length < minimumSamples) { + return; + } + + const priorWindowTotals = new Map(); + for (const priorEvent of priorEvents) { + const windowIndex = Math.floor( + (currentWindowStart - priorEvent.createdAt.getTime()) / windowMs, + ); + if (windowIndex >= 0 && windowIndex < baselineWindowCount) { + priorWindowTotals.set( + windowIndex, + (priorWindowTotals.get(windowIndex) ?? 0) + priorEvent.amount, + ); + } + } + const observedPriorWindowTotals = [...priorWindowTotals.values()]; + if (observedPriorWindowTotals.length === 0) { + return; + } + const baselineAverage = + observedPriorWindowTotals.reduce((sum, total) => sum + total, 0) / + observedPriorWindowTotals.length; + + const currentEvents = relevantHistory.filter( + (candidate) => candidate.createdAt.getTime() >= currentWindowStart, + ); + if (!currentEvents.some((candidate) => candidate.id === event.id)) { + currentEvents.push(event); + } + const windowTotal = currentEvents.reduce( + (sum, candidate) => sum + candidate.amount, + 0, + ); + if ( + windowTotal < minimumAmount || + windowTotal < baselineAverage * multiplier + ) { + return; + } + + const eventAmount = event.amount; + this.logger.warn( + withRequestId( + `ALERT: credit usage spike tenantId=${event.userId} ` + + `resource=${event.resource} eventAmount=${eventAmount} ` + + `windowTotal=${windowTotal} baselineAverage=${baselineAverage.toFixed(2)} ` + + `multiplier=${multiplier} threshold=${minimumAmount}`, + ), + ); + this.metrics.recordCreditUsageSpike(event.resource); + await this.sendUsageSpikeAlert({ + tenantId: event.userId, + resource: event.resource, + eventAmount, + windowTotal, + baselineAverage, + multiplier, + minimumAmount, + }); + } + + private async sendUsageSpikeAlert(input: { + tenantId: string; + resource: string; + eventAmount: number; + windowTotal: number; + baselineAverage: number; + multiplier: number; + minimumAmount: number; + }): Promise { + try { + const host = this.config.get('SMTP_HOST'); + const user = this.config.get('SMTP_USER'); + const pass = this.config.get('SMTP_PASS'); + const from = this.config.get('SMTP_FROM_EMAIL'); + const supportEmail = this.config.get('SUPPORT_EMAIL'); + if (!host || !user || !pass || !from || !supportEmail) { + return; + } + + const transport = createTransport({ + host, + port: this.config.get('SMTP_PORT', 587), + secure: this.config.get('SMTP_SECURE', 'false') === 'true', + auth: { user, pass }, + }); + const template = Handlebars.compile( + '{{company}} credit usage spike alert: tenant {{tenantId}} recorded {{windowTotal}} minor units for {{resource}} in the current {{windowMinutes}} minute window (event {{eventAmount}}, baseline {{baselineAverage}}, multiplier {{multiplier}}, minimum {{minimumAmount}}). Please investigate.', + ); + await transport.sendMail({ + from, + to: supportEmail, + subject: `${this.config.get('COMPANY_NAME', 'ManageHub')} credit usage spike alert`, + text: template({ + company: this.config.get('COMPANY_NAME', 'ManageHub'), + tenantId: input.tenantId, + resource: input.resource, + windowMinutes: this.getUsageSpikeWindowMinutes(), + eventAmount: input.eventAmount, + windowTotal: input.windowTotal, + baselineAverage: input.baselineAverage.toFixed(2), + multiplier: input.multiplier, + minimumAmount: input.minimumAmount, + }), + }); + } catch (error) { + this.logger.warn( + withRequestId( + `Unable to send credit usage spike alert email: ${ + error instanceof Error ? error.message : String(error) + }`, + ), + ); + } + } + + private getUsageSpikeWindowMinutes(): number { + return this.readPositiveNumberConfig( + 'CREDITS_USAGE_SPIKE_WINDOW_MINUTES', + 60, + ); + } + + private getUsageSpikeMultiplier(): number { + return this.readPositiveNumberConfig( + 'CREDITS_USAGE_SPIKE_MULTIPLIER', + 5.0, + ); + } + + private getUsageSpikeMinimumAmount(): number { + return this.readPositiveNumberConfig( + 'CREDITS_USAGE_SPIKE_MIN_AMOUNT', + 10_000, + ); + } + + private getUsageSpikeMinimumSamples(): number { + return this.readPositiveIntegerConfig( + 'CREDITS_USAGE_SPIKE_MIN_SAMPLES', + 3, + ); + } + + private readPositiveNumberConfig(key: string, fallback: number): number { + const configured = Number(this.config.get(key, fallback)); + return Number.isFinite(configured) && configured > 0 + ? configured + : fallback; + } + + private readPositiveIntegerConfig(key: string, fallback: number): number { + return Math.floor(this.readPositiveNumberConfig(key, fallback)); } async listForUser(userId: string, limit = 100): Promise { diff --git a/backend/src/database/migrations/1790003000000-CreateReconciliationRunsTable.ts b/backend/src/database/migrations/1790003000000-CreateReconciliationRunsTable.ts new file mode 100644 index 00000000..3200b1ea --- /dev/null +++ b/backend/src/database/migrations/1790003000000-CreateReconciliationRunsTable.ts @@ -0,0 +1,41 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class CreateReconciliationRunsTable1790003000000 implements MigrationInterface { + name = 'CreateReconciliationRunsTable1790003000000'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + CREATE TABLE "reconciliation_runs" ( + "id" uuid NOT NULL DEFAULT gen_random_uuid(), + "started_at" timestamptz NOT NULL DEFAULT now(), + "finished_at" timestamptz, + "duration_ms" integer, + "candidates" integer NOT NULL DEFAULT 0, + "resolved" integer NOT NULL DEFAULT 0, + "pending" integer NOT NULL DEFAULT 0, + "provider_errors" integer NOT NULL DEFAULT 0, + "escalated_to_manual_review" integer NOT NULL DEFAULT 0, + "expired_swept" integer NOT NULL DEFAULT 0, + "outcome" text, + "error" text, + "details" jsonb, + CONSTRAINT "pk_reconciliation_runs" PRIMARY KEY ("id") + ) + `); + + await queryRunner.query(` + CREATE INDEX "idx_reconciliation_runs_started_at" + ON "reconciliation_runs" ("started_at") + `); + await queryRunner.query(` + CREATE INDEX "idx_reconciliation_runs_outcome" + ON "reconciliation_runs" ("outcome") + `); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX "idx_reconciliation_runs_outcome"`); + await queryRunner.query(`DROP INDEX "idx_reconciliation_runs_started_at"`); + await queryRunner.query(`DROP TABLE "reconciliation_runs"`); + } +} diff --git a/backend/src/payments/dto/reconciliation-run-response.dto.spec.ts b/backend/src/payments/dto/reconciliation-run-response.dto.spec.ts new file mode 100644 index 00000000..c690a6b2 --- /dev/null +++ b/backend/src/payments/dto/reconciliation-run-response.dto.spec.ts @@ -0,0 +1,36 @@ +import { ReconciliationRunResponseDto } from './reconciliation-run-response.dto'; + +describe('ReconciliationRunResponseDto', () => { + it('maps a persisted reconciliation run without exposing entity-only state', () => { + const run = { + id: 'run-1', + startedAt: new Date('2025-01-01T00:00:00.000Z'), + finishedAt: new Date('2025-01-01T00:00:01.000Z'), + durationMs: 1000, + candidates: 3, + resolved: 1, + pending: 1, + providerErrors: 1, + escalatedToManualReview: 0, + expiredSwept: 0, + outcome: 'succeeded' as const, + error: null, + details: { attemptCapEscalations: 0 }, + }; + + const dto = ReconciliationRunResponseDto.fromEntity(run as any); + + expect(dto).toMatchObject({ + id: 'run-1', + candidates: 3, + resolved: 1, + pending: 1, + providerErrors: 1, + outcome: 'succeeded', + error: null, + details: { attemptCapEscalations: 0 }, + }); + expect(dto.durationMs).toBe(1000); + expect(dto.finishedAt).toBe(run.finishedAt); + }); +}); diff --git a/backend/src/payments/dto/reconciliation-run-response.dto.ts b/backend/src/payments/dto/reconciliation-run-response.dto.ts new file mode 100644 index 00000000..96252396 --- /dev/null +++ b/backend/src/payments/dto/reconciliation-run-response.dto.ts @@ -0,0 +1,47 @@ +import { ApiProperty } from '@nestjs/swagger'; +import { + ReconciliationRun, + ReconciliationRunOutcome, +} from '../entities/reconciliation-run.entity'; + +export class ReconciliationRunResponseDto { + @ApiProperty() id: string; + @ApiProperty() startedAt: Date; + @ApiProperty({ nullable: true }) finishedAt: Date | null; + @ApiProperty({ nullable: true }) durationMs: number | null; + @ApiProperty() candidates: number; + @ApiProperty() resolved: number; + @ApiProperty() pending: number; + @ApiProperty() providerErrors: number; + @ApiProperty() escalatedToManualReview: number; + @ApiProperty() expiredSwept: number; + @ApiProperty({ + enum: ['succeeded', 'failed'], + nullable: true, + description: 'Null only while a run is still in progress.', + }) + outcome: ReconciliationRunOutcome | null; + @ApiProperty({ nullable: true }) error: string | null; + @ApiProperty({ nullable: true, type: Object }) details: Record< + string, + unknown + > | null; + + static fromEntity(run: ReconciliationRun): ReconciliationRunResponseDto { + const dto = new ReconciliationRunResponseDto(); + dto.id = run.id; + dto.startedAt = run.startedAt; + dto.finishedAt = run.finishedAt; + dto.durationMs = run.durationMs; + dto.candidates = run.candidates; + dto.resolved = run.resolved; + dto.pending = run.pending; + dto.providerErrors = run.providerErrors; + dto.escalatedToManualReview = run.escalatedToManualReview; + dto.expiredSwept = run.expiredSwept; + dto.outcome = run.outcome; + dto.error = run.error; + dto.details = run.details; + return dto; + } +} diff --git a/backend/src/payments/entities/reconciliation-run.entity.ts b/backend/src/payments/entities/reconciliation-run.entity.ts new file mode 100644 index 00000000..4abad7d2 --- /dev/null +++ b/backend/src/payments/entities/reconciliation-run.entity.ts @@ -0,0 +1,64 @@ +import { + Column, + CreateDateColumn, + Entity, + Index, + PrimaryGeneratedColumn, +} from 'typeorm'; + +export type ReconciliationRunOutcome = 'succeeded' | 'failed'; + +/** + * Immutable-ish audit record for one scheduled reconciliation pass. The + * counters are copied from ReconciliationSummary when the pass completes; + * a row remains queryable with nullable completion fields if the process + * fails before it can finish the pass, and `error` captures that failure. + */ +@Entity('reconciliation_runs') +@Index('idx_reconciliation_runs_started_at', ['startedAt']) +@Index('idx_reconciliation_runs_outcome', ['outcome']) +export class ReconciliationRun { + @PrimaryGeneratedColumn('uuid') + id: string; + + @CreateDateColumn({ type: 'timestamptz', name: 'started_at' }) + startedAt: Date; + + @Column({ type: 'timestamptz', name: 'finished_at', nullable: true }) + finishedAt: Date | null; + + @Column({ type: 'int', name: 'duration_ms', nullable: true }) + durationMs: number | null; + + @Column({ type: 'int', name: 'candidates', default: 0 }) + candidates: number; + + @Column({ type: 'int', name: 'resolved', default: 0 }) + resolved: number; + + @Column({ type: 'int', name: 'pending', default: 0 }) + pending: number; + + @Column({ type: 'int', name: 'provider_errors', default: 0 }) + providerErrors: number; + + @Column({ + type: 'int', + name: 'escalated_to_manual_review', + default: 0, + }) + escalatedToManualReview: number; + + @Column({ type: 'int', name: 'expired_swept', default: 0 }) + expiredSwept: number; + + @Column({ type: 'text', name: 'outcome', nullable: true }) + outcome: ReconciliationRunOutcome | null; + + @Column({ type: 'text', name: 'error', nullable: true }) + error: string | null; + + /** Additional run context such as cap configuration and escalation counts. */ + @Column({ type: 'jsonb', name: 'details', nullable: true }) + details: Record | null; +} diff --git a/backend/src/payments/payment-confirmation.service.ts b/backend/src/payments/payment-confirmation.service.ts index eeec181a..903ef8b4 100644 --- a/backend/src/payments/payment-confirmation.service.ts +++ b/backend/src/payments/payment-confirmation.service.ts @@ -55,7 +55,10 @@ export class PaymentConfirmationService { /** * Applies a confirmation outcome for the Payment matching * `providerReference`. Returns the (possibly unchanged) Payment, or null - * if no Payment matches the reference at all. + * if no Payment matches the reference at all. Scheduled confirmation + * checks are bounded by PAYMENT_CONFIRMATION_MAX_ATTEMPTS and + * PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK; reaching either cap + * moves the payment to MANUAL_REVIEW before another poll is scheduled. */ async apply( providerReference: string, @@ -153,7 +156,11 @@ export class PaymentConfirmationService { * * Hard rule: `verified: false` (timeout, provider error, or a `pending` * response) NEVER implies a status change. Only an authoritative - * confirmed/failed response from the provider moves the Payment. + * confirmed/failed response from the provider moves the Payment. This + * method performs one bounded provider check per invocation; the total + * scheduled confirmation checks a payment may receive are capped by + * PAYMENT_CONFIRMATION_MAX_ATTEMPTS and + * PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK in ReconciliationService. */ async verifyOnReturn( payment: Payment, diff --git a/backend/src/payments/payments-admin.controller.ts b/backend/src/payments/payments-admin.controller.ts index 2d0a2c54..9e67e2ed 100644 --- a/backend/src/payments/payments-admin.controller.ts +++ b/backend/src/payments/payments-admin.controller.ts @@ -5,11 +5,13 @@ import { Param, ParseUUIDPipe, Post, + Query, UseGuards, } from '@nestjs/common'; import { ApiBearerAuth, ApiOperation, + ApiQuery, ApiResponse, ApiTags, } from '@nestjs/swagger'; @@ -24,6 +26,7 @@ import { RefundsService } from './refunds.service'; import { AdminActionLogService } from '../admin-audit/admin-action-log.service'; import { AdminActionType } from '../admin-audit/admin-action-type.enum'; import { PaymentResponseDto } from './dto/payment-response.dto'; +import { ReconciliationRunResponseDto } from './dto/reconciliation-run-response.dto'; import { ResolvePaymentManuallyDto } from './dto/resolve-payment-manually.dto'; import { VoidPaymentDto } from './dto/void-payment.dto'; import { CreateRefundDto } from './dto/create-refund.dto'; @@ -69,6 +72,26 @@ export class PaymentsAdminController { return this.reconciliationService.getMetrics(); } + @Get('reconciliation-runs') + @ApiOperation({ + summary: 'List persisted reconciliation runs, newest first', + description: + 'Returns the most recent reconciliation run records. The optional limit is clamped to 1–500 and defaults to 50.', + }) + @ApiQuery({ + name: 'limit', + required: false, + type: Number, + description: 'Number of runs to return (1–500, default 50)', + }) + @ApiResponse({ status: 200, type: [ReconciliationRunResponseDto] }) + async listReconciliationRuns( + @Query('limit') limit?: string, + ): Promise { + const runs = await this.reconciliationService.listRecentRuns(limit ?? 50); + return runs.map((run) => ReconciliationRunResponseDto.fromEntity(run)); + } + @Post(':id/force-reconcile') @ApiOperation({ summary: diff --git a/backend/src/payments/payments.module.ts b/backend/src/payments/payments.module.ts index e703149b..517e2acf 100644 --- a/backend/src/payments/payments.module.ts +++ b/backend/src/payments/payments.module.ts @@ -6,6 +6,7 @@ import { BullModule } from '@nestjs/bull'; import { Payment } from './entities/payment.entity'; import { ConfirmationEvent } from './entities/confirmation-event.entity'; import { Refund } from './entities/refund.entity'; +import { ReconciliationRun } from './entities/reconciliation-run.entity'; import { PaymentsService } from './payments.service'; import { PaymentConfirmationService } from './payment-confirmation.service'; import { ReconciliationService } from './reconciliation.service'; @@ -39,7 +40,12 @@ import { @Module({ imports: [ - TypeOrmModule.forFeature([Payment, ConfirmationEvent, Refund]), + TypeOrmModule.forFeature([ + Payment, + ConfirmationEvent, + Refund, + ReconciliationRun, + ]), WalletsModule, AdminAuditModule, BullModule.registerQueue({ name: SOROBAN_ESCROW_QUEUE }), diff --git a/backend/src/payments/reconciliation.service.spec.ts b/backend/src/payments/reconciliation.service.spec.ts index aed0ba4d..743820e2 100644 --- a/backend/src/payments/reconciliation.service.spec.ts +++ b/backend/src/payments/reconciliation.service.spec.ts @@ -112,12 +112,58 @@ function makePaymentRepository(seed: Payment[]) { }; } +function makeRunRepository(seed: any[] = []) { + const rows = [...seed]; + let sequence = rows.length; + + return { + create: jest.fn((entity: any) => ({ id: `run-${++sequence}`, ...entity })), + save: jest.fn(async (entity: any) => { + const row = { ...entity }; + if (!row.id) { + row.id = `run-${++sequence}`; + } + const index = rows.findIndex((candidate) => candidate.id === row.id); + if (index >= 0) { + rows[index] = { ...rows[index], ...row }; + } else { + rows.push(row); + } + return { ...row }; + }), + update: jest.fn(async (id: string, partial: any) => { + const index = rows.findIndex((candidate) => candidate.id === id); + if (index >= 0) { + rows[index] = { ...rows[index], ...partial }; + } + return { affected: index >= 0 ? 1 : 0 }; + }), + find: jest.fn(async (options?: { order?: any; take?: number }) => { + let result = [...rows]; + if (options?.order?.startedAt === 'DESC') { + result.sort( + (left, right) => + new Date(right.startedAt).getTime() - + new Date(left.startedAt).getTime(), + ); + } + if (options?.take) { + result = result.slice(0, options.take); + } + return result.map((row) => ({ ...row })); + }), + _rows: rows, + }; +} + function makeConfigService(overrides: Record = {}) { const values: Record = { PAYMENT_RECONCILE_DUE_AFTER_MINUTES: 5, PAYMENT_RECONCILE_BACKOFF_BASE_MINUTES: 5, PAYMENT_RECONCILE_BACKOFF_MAX_MINUTES: 60, PAYMENT_MANUAL_REVIEW_AFTER_HOURS: 24, + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 20, + PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK: 10, PAYMENT_VERIFY_TIMEOUT_MS: 3000, PAYMENT_RECONCILE_MAX_BATCH: 500, PAYMENT_MANUAL_REVIEW_ALERT_THRESHOLD: 20, @@ -152,18 +198,21 @@ describe('ReconciliationService', () => { function build( seed: Payment[], configOverrides: Record = {}, + runSeed: any[] = [], ) { const paymentRepository = makePaymentRepository(seed); + const runRepository = makeRunRepository(runSeed); const config = makeConfigService(configOverrides); - const service = new ReconciliationService( - paymentRepository as any, - confirmationService as any, - railRegistry as any, - gateway as any, - config as any, - metrics as any, - ); - return { service, paymentRepository }; + const service = new ReconciliationService( + paymentRepository as any, + runRepository as any, + confirmationService as any, + railRegistry as any, + gateway as any, + config as any, + metrics as any, + ); + return { service, paymentRepository, runRepository }; } const minutesAgo = (n: number, from: Date = new Date()) => @@ -238,6 +287,111 @@ describe('ReconciliationService', () => { }); }); + describe('reconcileDueBatch — confirmation polling caps', () => { + it('continues polling a payment that is below the attempt cap', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: minutesAgo(10, now), + reconciliationAttempts: 1, + }); + railAdapter.verifyByReference.mockResolvedValue({ outcome: 'pending' }); + + const { service } = build([payment], { + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 3, + }); + const summary = await service.reconcileDueBatch(now); + + expect(summary.candidates).toBe(1); + expect(summary.pending).toBe(1); + expect(railAdapter.verifyByReference).toHaveBeenCalledWith( + payment.providerReference, + ); + }); + + it('escalates at the attempt cap and never polls the capped payment again', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: minutesAgo(10, now), + reconciliationAttempts: 2, + }); + railAdapter.verifyByReference.mockResolvedValue({ outcome: 'pending' }); + + const { service, paymentRepository } = build([payment], { + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 3, + }); + const first = await service.reconcileDueBatch(now); + + expect(first.escalatedToManualReview).toBe(1); + expect(first.attemptCapEscalations).toBe(1); + const updated = await paymentRepository.findOne({ + where: { id: payment.id }, + }); + expect(updated!.status).toBe(PaymentStatus.MANUAL_REVIEW); + expect(updated!.manualReviewReason).toMatch(/attempt cap 3/); + expect(updated!.manualReviewReason).toMatch( + /3 total reconciliation attempts/, + ); + + const second = await service.reconcileDueBatch( + new Date(now.getTime() + 60_000), + ); + expect(second.candidates).toBe(0); + expect(railAdapter.verifyByReference).toHaveBeenCalledTimes(1); + }); + + it('escalates a pre-capped AWAITING_CONFIRMATION row without another poll', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: minutesAgo(10, now), + reconciliationAttempts: 3, + }); + const { service, paymentRepository } = build([payment], { + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 3, + }); + + const summary = await service.reconcileDueBatch(now); + + expect(summary.candidates).toBe(0); + expect(summary.escalatedToManualReview).toBe(1); + expect(railAdapter.verifyByReference).not.toHaveBeenCalled(); + const updated = await paymentRepository.findOne({ + where: { id: payment.id }, + }); + expect(updated!.status).toBe(PaymentStatus.MANUAL_REVIEW); + }); + + it('escalates at the provider-error-streak cap even on the erroring attempt', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: hoursAgo(48, now), + reconciliationAttempts: 2, + providerErrorStreak: 2, + }); + railAdapter.verifyByReference.mockRejectedValue( + new Error('provider unreachable'), + ); + + const { service, paymentRepository } = build([payment], { + PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK: 3, + }); + const summary = await service.reconcileDueBatch(now); + + expect(summary.providerErrors).toBe(1); + expect(summary.escalatedToManualReview).toBe(1); + expect(summary.providerErrorStreakCapEscalations).toBe(1); + const updated = await paymentRepository.findOne({ + where: { id: payment.id }, + }); + expect(updated!.status).toBe(PaymentStatus.MANUAL_REVIEW); + expect(updated!.manualReviewReason).toMatch( + /provider-error streak cap 3/, + ); + expect(updated!.manualReviewReason).toMatch( + /3 consecutive provider errors/, + ); + }); + }); + describe('reconcileDueBatch — provider outage does not cause false-positive escalation', () => { it('does not escalate an old payment to MANUAL_REVIEW when this attempt is itself a provider outage', async () => { const now = new Date(); @@ -405,6 +559,43 @@ describe('ReconciliationService', () => { expect(railAdapter.verifyByReference).toHaveBeenCalledTimes(500); }); + it('persists a started run and completes it with the pass summary', async () => { + railAdapter.verifyByReference.mockResolvedValue({ outcome: 'pending' }); + const payment = makePayment({ createdAt: minutesAgo(10) }); + const { service, runRepository } = build([payment]); + + await service.handleCron(); + + expect(runRepository.save).toHaveBeenCalledTimes(1); + expect(runRepository.update).toHaveBeenCalledWith( + 'run-1', + expect.objectContaining({ + outcome: 'succeeded', + candidates: 1, + pending: 1, + finishedAt: expect.any(Date), + durationMs: expect.any(Number), + }), + ); + }); + + it('marks a failed pass without masking the original error and resets the guard', async () => { + const { service, paymentRepository, runRepository } = build([]); + paymentRepository.find.mockRejectedValueOnce(new Error('database unavailable')); + + await expect(service.handleCron()).rejects.toThrow('database unavailable'); + expect(runRepository.update).toHaveBeenCalledWith( + 'run-1', + expect.objectContaining({ + outcome: 'failed', + error: 'database unavailable', + finishedAt: expect.any(Date), + }), + ); + + await expect(service.handleCron()).resolves.toBeUndefined(); + }); + it('skips a cron tick that fires while the previous pass is still running', async () => { const now = new Date(); const payment = makePayment({ createdAt: minutesAgo(10, now) }); @@ -560,6 +751,31 @@ describe('ReconciliationService', () => { expect(result.map((p) => p.id)).toEqual(['p1']); }); + it('lists recent reconciliation runs newest first with a bounded limit', async () => { + const { service, runRepository } = build( + [], + {}, + [ + { + id: 'run-old', + startedAt: new Date('2025-01-01T00:00:00.000Z'), + }, + { + id: 'run-new', + startedAt: new Date('2025-01-02T00:00:00.000Z'), + }, + ], + ); + + const result = await service.listRecentRuns(1); + + expect(result.map((run) => run.id)).toEqual(['run-new']); + expect(runRepository.find).toHaveBeenCalledWith({ + order: { startedAt: 'DESC' }, + take: 1, + }); + }); + it('getMetrics reports the manual-review queue depth and alerting flag', async () => { const payments = Array.from({ length: 25 }, (_, i) => makePayment({ id: `p${i}`, status: PaymentStatus.MANUAL_REVIEW }), @@ -570,6 +786,8 @@ describe('ReconciliationService', () => { expect(metrics.manualReviewQueueDepth).toBe(25); expect(metrics.alertThreshold).toBe(20); expect(metrics.alerting).toBe(true); + expect(metrics.confirmationMaxAttempts).toBe(20); + expect(metrics.confirmationMaxProviderErrorStreak).toBe(10); }); }); }); diff --git a/backend/src/payments/reconciliation.service.ts b/backend/src/payments/reconciliation.service.ts index 26be2615..56a6dc23 100644 --- a/backend/src/payments/reconciliation.service.ts +++ b/backend/src/payments/reconciliation.service.ts @@ -10,6 +10,7 @@ import { InjectRepository } from '@nestjs/typeorm'; import { Cron, CronExpression } from '@nestjs/schedule'; import { LessThan, Repository } from 'typeorm'; import { Payment } from './entities/payment.entity'; +import { ReconciliationRun } from './entities/reconciliation-run.entity'; import { PaymentStatus } from './enums/payment-status.enum'; import { PaymentFailureReason } from './enums/payment-failure-reason.enum'; import { ConfirmationSource } from './enums/confirmation-source.enum'; @@ -30,15 +31,28 @@ export interface ReconciliationSummary { providerErrors: number; escalatedToManualReview: number; expiredSwept: number; + /** Payments escalated specifically because the total-attempt cap tripped. */ + attemptCapEscalations: number; + /** Payments escalated specifically because the provider-error streak cap tripped. */ + providerErrorStreakCapEscalations: number; } export interface ReconciliationMetrics { manualReviewQueueDepth: number; alertThreshold: number; alerting: boolean; + confirmationMaxAttempts: number; + confirmationMaxProviderErrorStreak: number; + attemptCapEscalations: number; + providerErrorStreakCapEscalations: number; } type ReconcileOutcome = 'resolved' | 'pending' | 'provider_error' | 'escalated'; +type ReconciliationCap = 'attempt' | 'provider_error_streak'; +type ReconcileResult = { + outcome: ReconcileOutcome; + caps?: ReconciliationCap[]; +}; /** * Scheduled reconciliation (issue #1572): treats provider truth as @@ -65,9 +79,16 @@ export class ReconciliationService { */ private isRunning = false; + private lastRunCapState = { + attemptCapEscalations: 0, + providerErrorStreakCapEscalations: 0, + }; + constructor( @InjectRepository(Payment) private readonly paymentRepository: Repository, + @InjectRepository(ReconciliationRun) + private readonly runRepository: Repository, private readonly confirmationService: PaymentConfirmationService, private readonly railRegistry: PaymentRailRegistry, private readonly gateway: PaymentsGateway, @@ -87,11 +108,20 @@ export class ReconciliationService { } this.isRunning = true; + const startedAt = new Date(); + let run: ReconciliationRun | null = null; try { - const started = Date.now(); + run = await this.startRunRecord(startedAt); const summary = await this.reconcileDueBatch(); - this.metrics.recordReconciliationPass(Date.now() - started); - this.logger.log(withRequestId(`Reconciliation pass: ${JSON.stringify(summary)}`)); + this.lastRunCapState = { + attemptCapEscalations: summary.attemptCapEscalations, + providerErrorStreakCapEscalations: + summary.providerErrorStreakCapEscalations, + }; + this.metrics.recordReconciliationPass(Date.now() - startedAt.getTime()); + this.logger.log( + withRequestId(`Reconciliation pass: ${JSON.stringify(summary)}`), + ); const metrics = await this.getMetrics(); this.metrics.setManualReviewDepth(metrics.manualReviewQueueDepth); if (metrics.alerting) { @@ -101,8 +131,15 @@ export class ReconciliationService { `exceeds threshold ${metrics.alertThreshold}`, ), ); - await this.sendManualReviewAlert(metrics.manualReviewQueueDepth, metrics.alertThreshold); + await this.sendManualReviewAlert( + metrics.manualReviewQueueDepth, + metrics.alertThreshold, + ); } + await this.completeRunRecord(run, summary, startedAt); + } catch (error) { + await this.failRunRecord(run, error, startedAt); + throw error; } finally { this.isRunning = false; } @@ -122,6 +159,10 @@ export class ReconciliationService { 'PAYMENT_RECONCILE_MAX_BATCH', 500, ); + // MANUAL_REVIEW is deliberately excluded by the status predicate, and + // isDueForPoll repeats that exclusion as a defensive guard. A payment + // that has crossed a cap is therefore never selected on a later pass, + // even if a stale read briefly shows its previous AWAITING state. const awaiting = await this.paymentRepository.find({ where: { status: PaymentStatus.AWAITING_CONFIRMATION }, take: maxBatch, @@ -134,15 +175,35 @@ export class ReconciliationService { providerErrors: 0, escalatedToManualReview: 0, expiredSwept, + attemptCapEscalations: 0, + providerErrorStreakCapEscalations: 0, }; for (const payment of awaiting) { + // A row that was already at a cap when this process started (for + // example, data written by the pre-cap deployment) is escalated + // without making one more provider call. This preserves the hard cap + // instead of merely skipping it forever in AWAITING_CONFIRMATION. + const reachedCaps = this.getReachedCaps(payment); + if (reachedCaps.length > 0) { + if (await this.escalateCappedPayment(payment, reachedCaps)) { + summary.escalatedToManualReview++; + for (const cap of reachedCaps) { + if (cap === 'attempt') { + summary.attemptCapEscalations++; + } else { + summary.providerErrorStreakCapEscalations++; + } + } + } + continue; + } if (!this.isDueForPoll(payment, now)) { continue; } summary.candidates++; - const outcome = await this.reconcileOne(payment, now); - switch (outcome) { + const result = await this.reconcileOne(payment, now); + switch (result.outcome) { case 'resolved': summary.resolved++; break; @@ -156,8 +217,20 @@ export class ReconciliationService { summary.escalatedToManualReview++; break; } + for (const cap of result.caps ?? []) { + if (cap === 'attempt') { + summary.attemptCapEscalations++; + } else { + summary.providerErrorStreakCapEscalations++; + } + } } + this.lastRunCapState = { + attemptCapEscalations: summary.attemptCapEscalations, + providerErrorStreakCapEscalations: + summary.providerErrorStreakCapEscalations, + }; return summary; } @@ -169,7 +242,12 @@ export class ReconciliationService { `Payment in status ${payment.status} is not awaiting confirmation`, ); } - await this.reconcileOne(payment, new Date()); + const reachedCaps = this.getReachedCaps(payment); + if (reachedCaps.length > 0) { + await this.escalateCappedPayment(payment, reachedCaps); + } else { + await this.reconcileOne(payment, new Date()); + } return this.getPaymentOrThrow(paymentId); } @@ -217,6 +295,20 @@ export class ReconciliationService { }); } + /** Returns persisted reconciliation history, newest first, with a bounded take. */ + async listRecentRuns( + limit: number | string = 50, + ): Promise { + const numericLimit = Number(limit); + const boundedLimit = Number.isFinite(numericLimit) + ? Math.min(Math.max(Math.floor(numericLimit), 1), 500) + : 50; + return this.runRepository.find({ + order: { startedAt: 'DESC' }, + take: boundedLimit, + }); + } + async getMetrics(): Promise { const manualReviewQueueDepth = await this.paymentRepository.count({ where: { status: PaymentStatus.MANUAL_REVIEW }, @@ -229,9 +321,139 @@ export class ReconciliationService { manualReviewQueueDepth, alertThreshold, alerting: manualReviewQueueDepth > alertThreshold, + confirmationMaxAttempts: this.getConfirmationMaxAttempts(), + confirmationMaxProviderErrorStreak: + this.getConfirmationMaxProviderErrorStreak(), + attemptCapEscalations: this.lastRunCapState.attemptCapEscalations, + providerErrorStreakCapEscalations: + this.lastRunCapState.providerErrorStreakCapEscalations, }; } + private async startRunRecord( + startedAt: Date, + ): Promise { + try { + const run = this.runRepository.create({ + startedAt, + finishedAt: null, + durationMs: null, + candidates: 0, + resolved: 0, + pending: 0, + providerErrors: 0, + escalatedToManualReview: 0, + expiredSwept: 0, + outcome: null, + error: null, + details: null, + }); + return await this.runRepository.save(run); + } catch (error) { + this.logRunPersistenceFailure('create', error); + return null; + } + } + + private async completeRunRecord( + run: ReconciliationRun | null, + summary: ReconciliationSummary, + startedAt: Date, + ): Promise { + if (!run) { + return; + } + const finishedAt = new Date(); + try { + await this.runRepository.update(run.id, { + candidates: summary.candidates, + resolved: summary.resolved, + pending: summary.pending, + providerErrors: summary.providerErrors, + escalatedToManualReview: summary.escalatedToManualReview, + expiredSwept: summary.expiredSwept, + finishedAt, + durationMs: finishedAt.getTime() - startedAt.getTime(), + outcome: 'succeeded', + error: null, + details: { + attemptCap: this.getConfirmationMaxAttempts(), + providerErrorStreakCap: + this.getConfirmationMaxProviderErrorStreak(), + attemptCapEscalations: summary.attemptCapEscalations, + providerErrorStreakCapEscalations: + summary.providerErrorStreakCapEscalations, + }, + }); + } catch (error) { + this.logRunPersistenceFailure('complete', error); + } + } + + private async failRunRecord( + run: ReconciliationRun | null, + error: unknown, + startedAt: Date, + ): Promise { + if (!run) { + return; + } + const finishedAt = new Date(); + try { + await this.runRepository.update(run.id, { + finishedAt, + durationMs: finishedAt.getTime() - startedAt.getTime(), + outcome: 'failed', + error: + error instanceof Error + ? error.message + : String(error ?? 'Unknown error'), + details: { + phase: 'reconciliation', + attemptCap: this.getConfirmationMaxAttempts(), + providerErrorStreakCap: + this.getConfirmationMaxProviderErrorStreak(), + }, + }); + } catch (persistenceError) { + this.logRunPersistenceFailure('failure', persistenceError); + } + } + + private logRunPersistenceFailure( + operation: 'create' | 'complete' | 'failure', + error: unknown, + ): void { + this.logger.warn( + withRequestId( + `Unable to ${operation} reconciliation run record: ${ + error instanceof Error ? error.message : String(error) + }`, + ), + ); + } + + private getConfirmationMaxAttempts(): number { + return this.readPositiveIntegerConfig( + 'PAYMENT_CONFIRMATION_MAX_ATTEMPTS', + 20, + ); + } + + private getConfirmationMaxProviderErrorStreak(): number { + return this.readPositiveIntegerConfig( + 'PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK', + 10, + ); + } + + private readPositiveIntegerConfig(key: string, fallback: number): number { + const configured = Number(this.config.get(key, fallback)); + return Number.isFinite(configured) && configured > 0 + ? Math.floor(configured) + : fallback; + } + private async sendManualReviewAlert( depth: number, threshold: number, @@ -311,7 +533,7 @@ export class ReconciliationService { private async reconcileOne( payment: Payment, now: Date, - ): Promise { + ): Promise { if (!payment.providerReference) { this.logger.warn( `Payment ${payment.id} is AWAITING_CONFIRMATION with no providerReference`, @@ -371,7 +593,63 @@ export class ReconciliationService { : {}), }); - return 'resolved'; + return { outcome: 'resolved' }; + } + + private getReachedCaps(payment: Payment): ReconciliationCap[] { + const caps: ReconciliationCap[] = []; + if ( + payment.reconciliationAttempts >= this.getConfirmationMaxAttempts() + ) { + caps.push('attempt'); + } + if ( + payment.providerErrorStreak >= + this.getConfirmationMaxProviderErrorStreak() + ) { + caps.push('provider_error_streak'); + } + return caps; + } + + private async escalateCappedPayment( + payment: Payment, + caps: ReconciliationCap[], + ): Promise { + const fresh = await this.getPaymentOrThrow(payment.id); + if (fresh.status !== PaymentStatus.AWAITING_CONFIRMATION) { + return false; + } + assertValidTransition(fresh.status, PaymentStatus.MANUAL_REVIEW); + const maxAttempts = this.getConfirmationMaxAttempts(); + const maxProviderErrorStreak = + this.getConfirmationMaxProviderErrorStreak(); + const capDescriptions: string[] = []; + if (caps.includes('attempt')) { + capDescriptions.push( + `attempt cap ${maxAttempts} reached (${fresh.reconciliationAttempts} total reconciliation attempts)`, + ); + } + if (caps.includes('provider_error_streak')) { + capDescriptions.push( + `provider-error streak cap ${maxProviderErrorStreak} reached (${fresh.providerErrorStreak} consecutive provider errors)`, + ); + } + const description = capDescriptions.join('; '); + fresh.status = PaymentStatus.MANUAL_REVIEW; + fresh.manualReviewReason = + `Confirmation polling stopped: ${description}; ` + + `total reconciliation attempts: ${fresh.reconciliationAttempts}; ` + + `provider error streak: ${fresh.providerErrorStreak}.`; + this.logger.warn( + withRequestId( + `ALERT: payment ${fresh.id} reached a confirmation polling cap ` + + `(${description})`, + ), + ); + await this.paymentRepository.save(fresh); + this.gateway.emitPaymentUpdate(fresh.id, fresh.status); + return true; } /** @@ -383,17 +661,26 @@ export class ReconciliationService { * provider error — a provider-side outage must never mass-flag every * in-flight payment after a single bad run. Only an attempt that * successfully reached the provider (and got "still pending") can push - * an old-enough payment into MANUAL_REVIEW. + * an old-enough payment into MANUAL_REVIEW or trip the total-attempt cap. + * The provider-error streak cap is the deliberate exception: a long, + * bounded streak on one payment is evidence that this payment needs a + * human, whereas one failed run is evidence only that a provider may be + * experiencing a broad outage. If a provider-error attempt happens to + * reach the total-attempt cap first, the next scheduler pass escalates + * the already-capped row without making another provider call. */ private async recordAttempt( payment: Payment, now: Date, opts: { providerError: boolean }, - ): Promise { + ): Promise { const attempts = payment.reconciliationAttempts + 1; const providerErrorStreak = opts.providerError ? payment.providerErrorStreak + 1 : 0; + const maxAttempts = this.getConfirmationMaxAttempts(); + const maxProviderErrorStreak = + this.getConfirmationMaxProviderErrorStreak(); await this.paymentRepository.update(payment.id, { reconciliationAttempts: attempts, @@ -401,25 +688,68 @@ export class ReconciliationService { lastReconciledAt: now, }); + const caps: ReconciliationCap[] = []; + if (attempts >= maxAttempts) { + caps.push('attempt'); + } + if (providerErrorStreak >= maxProviderErrorStreak) { + caps.push('provider_error_streak'); + } + const manualReviewAfterHours = this.config.get( 'PAYMENT_MANUAL_REVIEW_AFTER_HOURS', 24, ); const ageMs = now.getTime() - payment.createdAt.getTime(); + const attemptCapReached = caps.includes('attempt'); + const providerErrorStreakCapReached = + caps.includes('provider_error_streak'); const shouldEscalate = - !opts.providerError && ageMs >= manualReviewAfterHours * 3_600_000; + providerErrorStreakCapReached || + (!opts.providerError && + (attemptCapReached || + ageMs >= manualReviewAfterHours * 3_600_000)); if (!shouldEscalate) { - return opts.providerError ? 'provider_error' : 'pending'; + return { + outcome: opts.providerError ? 'provider_error' : 'pending', + caps: [], + }; } const fresh = await this.getPaymentOrThrow(payment.id); assertValidTransition(fresh.status, PaymentStatus.MANUAL_REVIEW); fresh.status = PaymentStatus.MANUAL_REVIEW; - fresh.manualReviewReason = `Unresolved after ${attempts} reconciliation attempts (${(ageMs / 3_600_000).toFixed(1)}h old)`; + if (caps.length > 0) { + const capDescriptions: string[] = []; + if (caps.includes('attempt')) { + capDescriptions.push( + `attempt cap ${maxAttempts} reached (${attempts} total reconciliation attempts)`, + ); + } + if (caps.includes('provider_error_streak')) { + capDescriptions.push( + `provider-error streak cap ${maxProviderErrorStreak} reached (${providerErrorStreak} consecutive provider errors)`, + ); + } + fresh.manualReviewReason = + `Confirmation polling stopped: ${capDescriptions.join('; ')}; ` + + `total reconciliation attempts: ${attempts}; ` + + `provider error streak: ${providerErrorStreak}.`; + this.logger.warn( + withRequestId( + `ALERT: payment ${payment.id} reached a confirmation polling cap ` + + `(${capDescriptions.join('; ')})`, + ), + ); + } else { + fresh.manualReviewReason = + `Unresolved after ${attempts} reconciliation attempts ` + + `(${(ageMs / 3_600_000).toFixed(1)}h old)`; + } await this.paymentRepository.save(fresh); this.gateway.emitPaymentUpdate(fresh.id, fresh.status); - return 'escalated'; + return { outcome: 'escalated', caps }; } /** @@ -428,6 +758,16 @@ export class ReconciliationService { * often instead of every single cron tick. */ private isDueForPoll(payment: Payment, now: Date): boolean { + if (payment.status !== PaymentStatus.AWAITING_CONFIRMATION) { + return false; + } + if ( + payment.reconciliationAttempts >= this.getConfirmationMaxAttempts() || + payment.providerErrorStreak >= + this.getConfirmationMaxProviderErrorStreak() + ) { + return false; + } const dueAfterMinutes = this.config.get( 'PAYMENT_RECONCILE_DUE_AFTER_MINUTES', 5, diff --git a/backend/src/wallets/wallets.service.spec.ts b/backend/src/wallets/wallets.service.spec.ts index e353a9fe..b84c4972 100644 --- a/backend/src/wallets/wallets.service.spec.ts +++ b/backend/src/wallets/wallets.service.spec.ts @@ -311,27 +311,145 @@ describe('WalletsService', () => { }); describe('getWalletStatus', () => { + function makeStatusManager( + account: WalletAccount | null, + balance: number, + events: string[] = [], + ) { + const accountQueryBuilder: any = { + setLock: jest.fn(() => { + events.push('account-lock'); + return accountQueryBuilder; + }), + where: jest.fn(() => { + events.push('account-where'); + return accountQueryBuilder; + }), + getOne: jest.fn(async () => { + events.push('account-read'); + return account; + }), + }; + const ledgerQueryBuilder: any = { + select: jest.fn(() => ledgerQueryBuilder), + where: jest.fn(() => ledgerQueryBuilder), + getRawOne: jest.fn(async () => { + events.push('ledger-aggregate'); + return { balance: String(balance) }; + }), + }; + const accountRepository = { + createQueryBuilder: jest.fn(() => accountQueryBuilder), + }; + const ledgerRepository = { + createQueryBuilder: jest.fn(() => ledgerQueryBuilder), + }; + const manager = { + getRepository: jest.fn((entity: unknown) => + entity === WalletAccount ? accountRepository : ledgerRepository, + ), + }; + return { manager, accountQueryBuilder, ledgerQueryBuilder }; + } + it('reports an unprovisioned wallet as such, with a zero balance', async () => { - walletAccountRepository.findOne.mockResolvedValueOnce(null); + const { manager } = makeStatusManager(null, 0); + transaction.mockImplementationOnce((cb: any) => cb(manager)); const status = await service.getWalletStatus('user-1'); expect(status).toEqual({ account: null, balance: 0, currency: 'XLM' }); + expect(transaction).toHaveBeenCalledTimes(1); }); it('sums the ledger to compute the balance for a provisioned wallet', async () => { const account = makeAccount(); - walletAccountRepository.findOne.mockResolvedValueOnce(account); - const queryBuilder = { - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - getRawOne: jest.fn().mockResolvedValue({ balance: '5000' }), - }; - ledgerRepository.createQueryBuilder.mockReturnValue(queryBuilder); + const { manager, accountQueryBuilder, ledgerQueryBuilder } = + makeStatusManager(account, 5000); + transaction.mockImplementationOnce((cb: any) => cb(manager)); const status = await service.getWalletStatus('user-1'); expect(status).toEqual({ account, balance: 5000, currency: 'XLM' }); + expect(accountQueryBuilder.setLock).toHaveBeenCalledWith( + 'pessimistic_write', + ); + expect(ledgerQueryBuilder.getRawOne).toHaveBeenCalledTimes(1); + }); + + it('serializes concurrent balance reads around an interleaved writer', async () => { + const account = makeAccount(); + const events: string[] = []; + let balance = 100; + let releaseFirstAggregate!: () => void; + let firstAggregateStarted!: () => void; + const firstAggregateStartedPromise = new Promise((resolve) => { + firstAggregateStarted = resolve; + }); + const firstAggregateRelease = new Promise((resolve) => { + releaseFirstAggregate = resolve; + }); + const { manager, ledgerQueryBuilder } = makeStatusManager( + account, + balance, + events, + ); + ledgerQueryBuilder.getRawOne.mockImplementation(async () => { + const snapshot = balance; + events.push(`ledger-aggregate:${snapshot}`); + const aggregateCount = events.filter((event) => + event.startsWith('ledger-aggregate:'), + ).length; + if (aggregateCount === 1) { + firstAggregateStarted(); + await firstAggregateRelease; + } + return { balance: String(snapshot) }; + }); + + let queue = Promise.resolve(); + const enqueue = (work: () => Promise) => { + const run = queue.then(work); + queue = run.then( + () => undefined, + () => undefined, + ); + return run; + }; + transaction.mockImplementation((cb: any) => enqueue(() => cb(manager))); + const interleaveWrite = () => + enqueue(async () => { + events.push('writer-lock'); + balance = 200; + events.push('writer-commit'); + }); + + const first = service.getWalletStatus('user-1'); + await firstAggregateStartedPromise; + const write = interleaveWrite(); + const second = service.getWalletStatus('user-1'); + releaseFirstAggregate(); + const [firstStatus, secondStatus] = await Promise.all([ + first, + write, + second, + ]); + + expect(firstStatus.balance).toBe(100); + expect(secondStatus.balance).toBe(200); + expect(events).toEqual([ + 'account-lock', + 'account-where', + 'account-read', + 'ledger-aggregate:100', + 'writer-lock', + 'writer-commit', + 'account-lock', + 'account-where', + 'account-read', + 'ledger-aggregate:200', + ]); + expect(transaction).toHaveBeenCalledTimes(2); }); }); diff --git a/backend/src/wallets/wallets.service.ts b/backend/src/wallets/wallets.service.ts index 668b943b..6d6e1e1e 100644 --- a/backend/src/wallets/wallets.service.ts +++ b/backend/src/wallets/wallets.service.ts @@ -84,15 +84,43 @@ export class WalletsService { } } + /** + * PostgreSQL uses READ COMMITTED by default (unless a connection overrides + * it), so the account lookup and a bare ledger aggregate would otherwise + * be two independent statement snapshots. This transaction takes the same + * `pessimistic_write` row lock as fundCustodialWallet before reading the + * aggregate, and runs the aggregate through that transaction's EntityManager. + * `pessimistic_write` is chosen over a share/read lock for consistency over + * maximum read concurrency: it is the same lock fundCustodialWallet takes, + * so the two operations cannot pass one another. Against writers in this + * service, the lock serializes the read either + * before the writer (it sees the pre-write committed ledger) or after it + * (it waits for the writer's commit and sees the post-write ledger), never + * a partially applied application. + * + * This guarantee does not cover writers outside WalletsService that insert + * wallet_ledger_entries without first taking the wallet_accounts row lock + * (direct SQL, another service, or a future debit path). The current + * repository has no settlement/credit wallet-debit path; fundCustodialWallet + * is the only ledger writer and uses this lock, but any new writer must + * participate in the same locking protocol. + */ async getWalletStatus(userId: string): Promise { - const account = await this.walletAccountRepository.findOne({ - where: { userId }, - }); - if (!account) { - return { account: null, balance: 0, currency: LEDGER_ASSET }; - } - const balance = await this.getBalance(account.id); - return { account, balance, currency: LEDGER_ASSET }; + return this.walletAccountRepository.manager.transaction( + async (manager) => { + const account = await manager + .getRepository(WalletAccount) + .createQueryBuilder('wallet_account') + .setLock('pessimistic_write') + .where('wallet_account.user_id = :userId', { userId }) + .getOne(); + if (!account) { + return { account: null, balance: 0, currency: LEDGER_ASSET }; + } + const balance = await this.getBalance(manager, account.id); + return { account, balance, currency: LEDGER_ASSET }; + }, + ); } /** @@ -275,8 +303,13 @@ export class WalletsService { return repository.save(account); } - private async getBalance(walletAccountId: string): Promise { - const result = await this.ledgerRepository + /** Runs the ledger aggregate through the caller's locked transaction manager. */ + private async getBalance( + manager: EntityManager, + walletAccountId: string, + ): Promise { + const result = await manager + .getRepository(WalletLedgerEntry) .createQueryBuilder('entry') .select( `COALESCE(SUM(CASE WHEN entry.type = 'CREDIT' THEN entry.amount ELSE -entry.amount END), 0)`,