From 88ed7012e3758c670e0e05638dac3e1a7b75c409 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Mon, 13 Jul 2026 12:12:43 +0200 Subject: [PATCH 01/15] feat(backend): batch outgoing payments in worker jobs --- packages/backend/src/config/app.ts | 21 ++--- .../payment/outgoing/lifecycle.test.ts | 80 +++++++++---------- .../payment/outgoing/service.test.ts | 6 +- .../open_payments/payment/outgoing/service.ts | 38 ++++----- .../open_payments/payment/outgoing/worker.ts | 27 ++++--- 5 files changed, 87 insertions(+), 85 deletions(-) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index 6a4b6e5218..9eda082a09 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -17,9 +17,9 @@ function envStringArray(name: string, value: string[]): string[] { return envValue == null ? value : envValue - .split(',') - .map((s) => s.trim()) - .filter((s) => s.length > 0) + .split(',') + .map((s) => s.trim()) + .filter((s) => s.length > 0) } function envInt(name: string, value: number): number { @@ -65,11 +65,11 @@ export const Config = { 'OPEN_TELEMETRY_COLLECTOR_URLS', envBool('LIVENET', false) ? [ - 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' + ] : [ - 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' + ] ), openTelemetryExportInterval: envInt('OPEN_TELEMETRY_EXPORT_INTERVAL', 15000), telemetryExchangeRatesUrl: envString( @@ -91,9 +91,9 @@ export const Config = { process.env.NODE_ENV === 'test' ? `${process.env.DATABASE_URL}_${process.env.JEST_WORKER_ID}` : envString( - 'DATABASE_URL', - 'postgresql://postgres:password@localhost:5432/development' - ), + 'DATABASE_URL', + 'postgresql://postgres:password@localhost:5432/development' + ), walletAddressUrl: envString( 'WALLET_ADDRESS_URL', 'http://127.0.0.1:3001/.well-known/pay' @@ -134,6 +134,7 @@ export const Config = { outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1), outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 10), // milliseconds + outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 1), incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1), incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds diff --git a/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts b/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts index 56029348f0..f4e1e2928e 100644 --- a/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts +++ b/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts @@ -270,8 +270,8 @@ describe('Lifecycle', (): void => { // advance time to ensure spents amounts created by processNext, if any, have // later createdAt so that fetching latest is accurate jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const finalPayment = await outgoingPaymentService.get({ id: payment.id @@ -336,8 +336,8 @@ describe('Lifecycle', (): void => { }) jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const finalPayment = await outgoingPaymentService.get({ id: payment.id @@ -418,8 +418,8 @@ describe('Lifecycle', (): void => { // advance time to ensure spents amounts created by processNext, if any, have // later createdAt so that fetching latest is accurate jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const finalPayment = await outgoingPaymentService.get({ id: payment.id @@ -494,8 +494,8 @@ describe('Lifecycle', (): void => { // advance time to ensure spents amounts created by processNext, if any, have // later createdAt so that fetching latest is accurate jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const finalPayment = await outgoingPaymentService.get({ id: payment.id @@ -582,8 +582,8 @@ describe('Lifecycle', (): void => { // advance time to ensure spents amounts created by processNext, if any, have // later createdAt so that fetching latest is accurate jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const latestGrantSpentAmounts = await OutgoingPaymentGrantSpentAmounts.query(knex) @@ -681,8 +681,8 @@ describe('Lifecycle', (): void => { // advance time to ensure spents amounts created by processNext, if any, have // later createdAt so that fetching latest is accurate jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(secondPayment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(secondPayment.id) const finalPayment = await outgoingPaymentService.get({ id: secondPayment.id @@ -771,8 +771,8 @@ describe('Lifecycle', (): void => { // advance time to ensure spents amounts created by processNext, if any, have // later createdAt so that fetching latest is accurate jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(secondPayment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(secondPayment.id) const finalPayment = await outgoingPaymentService.get({ id: secondPayment.id @@ -875,8 +875,8 @@ describe('Lifecycle', (): void => { // advance time to ensure spents amounts created by processNext, if any, have // later createdAt so that fetching latest is accurate jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(secondPayment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(secondPayment.id) const finalPayment = await outgoingPaymentService.get({ id: secondPayment.id @@ -949,9 +949,9 @@ describe('Lifecycle', (): void => { firstPayment ) ) - const id = await outgoingPaymentService.processNext() + const ids = await outgoingPaymentService.processNext() jest.advanceTimersByTime(500) - expect(id).toBe(firstPayment.id) + expect(ids).toContain(firstPayment.id) // Grant spent amounts should correspond to the first payment // Should not detect a difference and insert a new spent amount. @@ -1066,9 +1066,9 @@ describe('Lifecycle', (): void => { receive: firstPaymentSettledAmount })(accountingService, receiverWalletAddressId, firstPayment) ) - let id = await outgoingPaymentService.processNext() + let ids = await outgoingPaymentService.processNext() jest.advanceTimersByTime(500) - expect(id).toBe(firstPayment.id) + expect(ids).toContain(firstPayment.id) latestSpentAmounts[2] = await OutgoingPaymentGrantSpentAmounts.query(knex) @@ -1114,9 +1114,9 @@ describe('Lifecycle', (): void => { receive: secondPaymentSettledAmount })(accountingService, receiverWalletAddressId, secondPayment) ) - id = await outgoingPaymentService.processNext() + ids = await outgoingPaymentService.processNext() jest.advanceTimersByTime(500) - expect(id).toBe(secondPayment.id) + expect(ids).toContain(secondPayment.id) latestSpentAmounts[3] = await OutgoingPaymentGrantSpentAmounts.query(knex) @@ -1261,9 +1261,9 @@ describe('Lifecycle', (): void => { receive: firstPaymentSettledAmount })(accountingService, receiverWalletAddressId, firstPayment) ) - let id = await outgoingPaymentService.processNext() + let ids = await outgoingPaymentService.processNext() jest.advanceTimersByTime(500) - expect(id).toBe(firstPayment.id) + expect(ids).toContain(firstPayment.id) latestSpentAmounts[2] = await OutgoingPaymentGrantSpentAmounts.query(knex) @@ -1357,9 +1357,9 @@ describe('Lifecycle', (): void => { receive: secondPaymentSettledAmount })(accountingService, receiverWalletAddressId, secondPayment) ) - id = await outgoingPaymentService.processNext() + ids = await outgoingPaymentService.processNext() jest.advanceTimersByTime(500) - expect(id).toBe(secondPayment.id) + expect(ids).toContain(secondPayment.id) latestSpentAmounts[4] = await OutgoingPaymentGrantSpentAmounts.query(knex) @@ -1501,9 +1501,9 @@ describe('Lifecycle', (): void => { jest .spyOn(paymentMethodHandlerService, 'pay') .mockImplementationOnce(mockPayErrorFactory()()) - let id = await outgoingPaymentService.processNext() + let ids = await outgoingPaymentService.processNext() jest.advanceTimersByTime(500) - expect(id).toBe(firstPayment.id) + expect(ids).toContain(firstPayment.id) const failedPayment = await outgoingPaymentService.get({ id: firstPayment.id @@ -1552,9 +1552,9 @@ describe('Lifecycle', (): void => { secondPayment ) ) - id = await outgoingPaymentService.processNext() + ids = await outgoingPaymentService.processNext() jest.advanceTimersByTime(500) - expect(id).toBe(secondPayment.id) + expect(ids).toContain(secondPayment.id) const completedPayment = await outgoingPaymentService.get({ id: secondPayment.id @@ -1716,8 +1716,8 @@ describe('Lifecycle', (): void => { // Process payment after interval boundary (in February) jest.setSystemTime(new Date('2025-02-01T00:00:01Z')) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const finishSpentAmounts = await OutgoingPaymentGrantSpentAmounts.query( knex @@ -1799,8 +1799,8 @@ describe('Lifecycle', (): void => { // Process payment after interval boundary (in February) jest.setSystemTime(new Date('2025-02-01T00:00:01Z')) jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const endSpentAmounts = await OutgoingPaymentGrantSpentAmounts.query( knex @@ -1899,8 +1899,8 @@ describe('Lifecycle', (): void => { // Process payment after interval boundary (in February) jest.setSystemTime(new Date('2025-02-01T00:00:01Z')) jest.advanceTimersByTime(500) - const processedPaymentId = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(payment.id) + const processedPaymentIds = await outgoingPaymentService.processNext() + expect(processedPaymentIds).toContain(payment.id) const finalPayment = await outgoingPaymentService.get({ id: payment.id @@ -2090,9 +2090,9 @@ describe('Lifecycle', (): void => { })(accountingService, receiverWalletAddressId, firstPayment) ) - const processedPaymentId = + const processedPaymentIds = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(firstPayment.id) + expect(processedPaymentIds).toContain(firstPayment.id) const completedPayment = await outgoingPaymentService.get({ id: firstPayment.id @@ -2279,9 +2279,9 @@ describe('Lifecycle', (): void => { .spyOn(paymentMethodHandlerService, 'pay') .mockImplementationOnce(mockPayErrorFactory()()) - const processedPaymentId = + const processedPaymentIds = await outgoingPaymentService.processNext() - expect(processedPaymentId).toBe(firstPayment.id) + expect(processedPaymentIds).toContain(firstPayment.id) const failedPayment = await outgoingPaymentService.get({ id: firstPayment.id diff --git a/packages/backend/src/open_payments/payment/outgoing/service.test.ts b/packages/backend/src/open_payments/payment/outgoing/service.test.ts index 2ece3667cd..4b000f9d9c 100644 --- a/packages/backend/src/open_payments/payment/outgoing/service.test.ts +++ b/packages/backend/src/open_payments/payment/outgoing/service.test.ts @@ -126,7 +126,7 @@ describe('OutgoingPaymentService', (): void => { expectState: OutgoingPaymentState, expectedError?: string ): Promise { - await expect(outgoingPaymentService.processNext()).resolves.toBe(paymentId) + await expect(outgoingPaymentService.processNext()).resolves.toEqual([paymentId]) const payment = await outgoingPaymentService.get({ id: paymentId }) @@ -349,7 +349,7 @@ describe('OutgoingPaymentService', (): void => { if (receiver) { // "as any" to circumvent "readonly" check (compile time only) to allow overriding "isLocal" here // eslint-disable-next-line @typescript-eslint/no-explicit-any - ;(receiver.isLocal as any) = false + ; (receiver.isLocal as any) = false return receiver } return undefined @@ -3045,7 +3045,7 @@ describe('OutgoingPaymentService', (): void => { expect(result.spentDebitAmount?.assetScale).toBe(asset.scale) expect(result.spentReceiveAmount?.value).toBe( payment1.quote.receiveAmount.value + - payment2.quote.receiveAmount.value + payment2.quote.receiveAmount.value ) expect(result.spentReceiveAmount?.assetCode).toBe( payment2.quote.receiveAmount.assetCode diff --git a/packages/backend/src/open_payments/payment/outgoing/service.ts b/packages/backend/src/open_payments/payment/outgoing/service.ts index af3eb68714..65a0a95a0a 100644 --- a/packages/backend/src/open_payments/payment/outgoing/service.ts +++ b/packages/backend/src/open_payments/payment/outgoing/service.ts @@ -70,7 +70,7 @@ export interface OutgoingPaymentService fund( options: FundOutgoingPaymentOptions ): Promise - processNext(): Promise + processNext(): Promise getGrantSpentAmounts(options: { grantId: string limits?: Limits @@ -103,7 +103,7 @@ export async function createOutgoingPaymentService( create: (options) => createOutgoingPayment(deps, options), cancel: (options) => cancelOutgoingPayment(deps, options), fund: (options) => fundPayment(deps, options), - processNext: () => worker.processPendingPayment(deps), + processNext: () => worker.processPendingPayments(deps), getWalletAddressPage: (options) => getWalletAddressPage(deps, options), getGrantSpentAmounts: (options) => getGrantSpentAmounts(deps, options) } @@ -707,15 +707,15 @@ export async function calculateLegacyGrantSpentAmounts( intervalReceiveAmountValue, latestPayment: latestPayment ? { - id: latestPayment.id, - debitAmountValue: latestPayment.debitAmount.value, - debitAmountCode: latestPayment.debitAmount.assetCode, - debitAmountScale: latestPayment.debitAmount.assetScale, - receiveAmountValue: latestPayment.receiveAmount.value, - receiveAmountCode: latestPayment.receiveAmount.assetCode, - receiveAmountScale: latestPayment.receiveAmount.assetScale, - state: latestPayment.state - } + id: latestPayment.id, + debitAmountValue: latestPayment.debitAmount.value, + debitAmountCode: latestPayment.debitAmount.assetCode, + debitAmountScale: latestPayment.debitAmount.assetScale, + receiveAmountValue: latestPayment.receiveAmount.value, + receiveAmountCode: latestPayment.receiveAmount.assetCode, + receiveAmountScale: latestPayment.receiveAmount.assetScale, + state: latestPayment.state + } : null } } @@ -799,7 +799,7 @@ async function validateGrantAndAddSpentAmountsToPayment( latestSpentAmounts?.intervalEnd && paymentLimits.paymentInterval.end && paymentLimits.paymentInterval.end.toJSDate() <= - latestSpentAmounts.intervalStart + latestSpentAmounts.intervalStart ) { deps.logger.error( { @@ -859,10 +859,10 @@ async function validateGrantAndAddSpentAmountsToPayment( // detect if we need to restart interval sum at 0 or continue from last const isInIntervalAndFirstPayment = hasInterval ? !latestSpentAmounts || - (latestSpentAmounts.intervalEnd && - paymentLimits.paymentInterval?.start && - latestSpentAmounts.intervalEnd <= - paymentLimits.paymentInterval.start.toJSDate()) + (latestSpentAmounts.intervalEnd && + paymentLimits.paymentInterval?.start && + latestSpentAmounts.intervalEnd <= + paymentLimits.paymentInterval.start.toJSDate()) : false outgoingPaymentGrantSpentAmounts.grantTotalDebitAmountValue = @@ -952,7 +952,7 @@ function exceedsGrantLimits( limits.debitAmount.value - spentValue < payment.debitAmount.value) || (limits.receiveAmount && limits.receiveAmount.value - spentReceive < - payment.receiveAmount.value) || + payment.receiveAmount.value) || false ) } @@ -1165,9 +1165,9 @@ async function getRemainingGrantSpentAmounts( ) { if ( latestGrantSpentAmounts.intervalStart?.getTime() !== - latestPaymentSpentAmounts.intervalStart.getTime() || + latestPaymentSpentAmounts.intervalStart.getTime() || latestGrantSpentAmounts.intervalEnd?.getTime() !== - latestPaymentSpentAmounts.intervalEnd.getTime() + latestPaymentSpentAmounts.intervalEnd.getTime() ) { latestIntervalSpentAmounts = (await OutgoingPaymentGrantSpentAmounts.query(deps.knex) diff --git a/packages/backend/src/open_payments/payment/outgoing/worker.ts b/packages/backend/src/open_payments/payment/outgoing/worker.ts index 516e096890..581706a7a2 100644 --- a/packages/backend/src/open_payments/payment/outgoing/worker.ts +++ b/packages/backend/src/open_payments/payment/outgoing/worker.ts @@ -11,9 +11,9 @@ import { trace, Span } from '@opentelemetry/api' export const RETRY_BACKOFF_SECONDS = 10 // Returns the id of the processed payment (if any). -export async function processPendingPayment( +export async function processPendingPayments( deps_: ServiceDependencies -): Promise { +): Promise { const tracer = trace.getTracer('outgoing_payment_worker') return tracer.startActiveSpan( @@ -25,11 +25,11 @@ export async function processPendingPayment( callName: 'OutgoingPaymentWorker:processPendingPayment' } ) - const paymentId = await deps_.knex.transaction(async (trx) => { - const payment = await getPendingPayment(trx, deps_) - if (!payment) return + const paymentIds = await deps_.knex.transaction(async (trx) => { + const payments = await getPendingPayments(trx, deps_) + if (payments.length === 0) return - await handlePaymentLifecycle( + const promises = payments.map((payment) => handlePaymentLifecycle( { ...deps_, knex: trx, @@ -39,28 +39,29 @@ export async function processPendingPayment( }) }, payment - ) - return payment.id + )) + await Promise.all(promises) + return payments.map(payment => payment.id) }) stopTimer() span.end() - return paymentId + return paymentIds } ) } // Fetch (and lock) a payment for work. -async function getPendingPayment( +async function getPendingPayments( trx: Knex.Transaction, deps: ServiceDependencies -): Promise { +): Promise { const stopTimer = deps.telemetry.startTimer('get_pending_payment_ms', { callName: 'OutoingPaymentWorker:getPendingPayment' }) const now = new Date(Date.now()).toISOString() const payments = await OutgoingPayment.query(trx) - .limit(1) + .limit(deps.config.outgoingPaymentBatchSize) // Ensure the payment cannot be processed concurrently by multiple workers. .forUpdate() // Don't wait for a payment that is already being processed. @@ -77,7 +78,7 @@ async function getPendingPayment( }) .withGraphFetched('quote') stopTimer() - return payments[0] + return payments } async function handlePaymentLifecycle( From f42bc52b250754906b8a07b37c046edebf571c99 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Mon, 13 Jul 2026 12:18:54 +0200 Subject: [PATCH 02/15] feat: formatting, default batch size to 5 --- packages/backend/src/config/app.ts | 22 ++++++------ .../payment/outgoing/lifecycle.test.ts | 2 +- .../payment/outgoing/service.test.ts | 8 +++-- .../open_payments/payment/outgoing/service.ts | 34 +++++++++---------- .../open_payments/payment/outgoing/worker.ts | 26 +++++++------- 5 files changed, 48 insertions(+), 44 deletions(-) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index 9eda082a09..4ca99ea797 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -17,9 +17,9 @@ function envStringArray(name: string, value: string[]): string[] { return envValue == null ? value : envValue - .split(',') - .map((s) => s.trim()) - .filter((s) => s.length > 0) + .split(',') + .map((s) => s.trim()) + .filter((s) => s.length > 0) } function envInt(name: string, value: number): number { @@ -65,11 +65,11 @@ export const Config = { 'OPEN_TELEMETRY_COLLECTOR_URLS', envBool('LIVENET', false) ? [ - 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' + ] : [ - 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' + ] ), openTelemetryExportInterval: envInt('OPEN_TELEMETRY_EXPORT_INTERVAL', 15000), telemetryExchangeRatesUrl: envString( @@ -91,9 +91,9 @@ export const Config = { process.env.NODE_ENV === 'test' ? `${process.env.DATABASE_URL}_${process.env.JEST_WORKER_ID}` : envString( - 'DATABASE_URL', - 'postgresql://postgres:password@localhost:5432/development' - ), + 'DATABASE_URL', + 'postgresql://postgres:password@localhost:5432/development' + ), walletAddressUrl: envString( 'WALLET_ADDRESS_URL', 'http://127.0.0.1:3001/.well-known/pay' @@ -134,7 +134,7 @@ export const Config = { outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1), outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 10), // milliseconds - outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 1), + outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 5), incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1), incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds diff --git a/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts b/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts index f4e1e2928e..5f8829e5af 100644 --- a/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts +++ b/packages/backend/src/open_payments/payment/outgoing/lifecycle.test.ts @@ -171,7 +171,7 @@ describe('Lifecycle', (): void => { describe('Grant Spent Amounts', (): void => { beforeAll(async (): Promise => { - deps = initIocContainer(Config) + deps = initIocContainer({ ...Config, outgoingPaymentBatchSize: 1 }) appContainer = await createTestApp(deps) outgoingPaymentService = await deps.use('outgoingPaymentService') accountingService = await deps.use('accountingService') diff --git a/packages/backend/src/open_payments/payment/outgoing/service.test.ts b/packages/backend/src/open_payments/payment/outgoing/service.test.ts index 4b000f9d9c..d083b15225 100644 --- a/packages/backend/src/open_payments/payment/outgoing/service.test.ts +++ b/packages/backend/src/open_payments/payment/outgoing/service.test.ts @@ -126,7 +126,9 @@ describe('OutgoingPaymentService', (): void => { expectState: OutgoingPaymentState, expectedError?: string ): Promise { - await expect(outgoingPaymentService.processNext()).resolves.toEqual([paymentId]) + await expect(outgoingPaymentService.processNext()).resolves.toEqual([ + paymentId + ]) const payment = await outgoingPaymentService.get({ id: paymentId }) @@ -349,7 +351,7 @@ describe('OutgoingPaymentService', (): void => { if (receiver) { // "as any" to circumvent "readonly" check (compile time only) to allow overriding "isLocal" here // eslint-disable-next-line @typescript-eslint/no-explicit-any - ; (receiver.isLocal as any) = false + ;(receiver.isLocal as any) = false return receiver } return undefined @@ -3045,7 +3047,7 @@ describe('OutgoingPaymentService', (): void => { expect(result.spentDebitAmount?.assetScale).toBe(asset.scale) expect(result.spentReceiveAmount?.value).toBe( payment1.quote.receiveAmount.value + - payment2.quote.receiveAmount.value + payment2.quote.receiveAmount.value ) expect(result.spentReceiveAmount?.assetCode).toBe( payment2.quote.receiveAmount.assetCode diff --git a/packages/backend/src/open_payments/payment/outgoing/service.ts b/packages/backend/src/open_payments/payment/outgoing/service.ts index 65a0a95a0a..a8d4061d18 100644 --- a/packages/backend/src/open_payments/payment/outgoing/service.ts +++ b/packages/backend/src/open_payments/payment/outgoing/service.ts @@ -707,15 +707,15 @@ export async function calculateLegacyGrantSpentAmounts( intervalReceiveAmountValue, latestPayment: latestPayment ? { - id: latestPayment.id, - debitAmountValue: latestPayment.debitAmount.value, - debitAmountCode: latestPayment.debitAmount.assetCode, - debitAmountScale: latestPayment.debitAmount.assetScale, - receiveAmountValue: latestPayment.receiveAmount.value, - receiveAmountCode: latestPayment.receiveAmount.assetCode, - receiveAmountScale: latestPayment.receiveAmount.assetScale, - state: latestPayment.state - } + id: latestPayment.id, + debitAmountValue: latestPayment.debitAmount.value, + debitAmountCode: latestPayment.debitAmount.assetCode, + debitAmountScale: latestPayment.debitAmount.assetScale, + receiveAmountValue: latestPayment.receiveAmount.value, + receiveAmountCode: latestPayment.receiveAmount.assetCode, + receiveAmountScale: latestPayment.receiveAmount.assetScale, + state: latestPayment.state + } : null } } @@ -799,7 +799,7 @@ async function validateGrantAndAddSpentAmountsToPayment( latestSpentAmounts?.intervalEnd && paymentLimits.paymentInterval.end && paymentLimits.paymentInterval.end.toJSDate() <= - latestSpentAmounts.intervalStart + latestSpentAmounts.intervalStart ) { deps.logger.error( { @@ -859,10 +859,10 @@ async function validateGrantAndAddSpentAmountsToPayment( // detect if we need to restart interval sum at 0 or continue from last const isInIntervalAndFirstPayment = hasInterval ? !latestSpentAmounts || - (latestSpentAmounts.intervalEnd && - paymentLimits.paymentInterval?.start && - latestSpentAmounts.intervalEnd <= - paymentLimits.paymentInterval.start.toJSDate()) + (latestSpentAmounts.intervalEnd && + paymentLimits.paymentInterval?.start && + latestSpentAmounts.intervalEnd <= + paymentLimits.paymentInterval.start.toJSDate()) : false outgoingPaymentGrantSpentAmounts.grantTotalDebitAmountValue = @@ -952,7 +952,7 @@ function exceedsGrantLimits( limits.debitAmount.value - spentValue < payment.debitAmount.value) || (limits.receiveAmount && limits.receiveAmount.value - spentReceive < - payment.receiveAmount.value) || + payment.receiveAmount.value) || false ) } @@ -1165,9 +1165,9 @@ async function getRemainingGrantSpentAmounts( ) { if ( latestGrantSpentAmounts.intervalStart?.getTime() !== - latestPaymentSpentAmounts.intervalStart.getTime() || + latestPaymentSpentAmounts.intervalStart.getTime() || latestGrantSpentAmounts.intervalEnd?.getTime() !== - latestPaymentSpentAmounts.intervalEnd.getTime() + latestPaymentSpentAmounts.intervalEnd.getTime() ) { latestIntervalSpentAmounts = (await OutgoingPaymentGrantSpentAmounts.query(deps.knex) diff --git a/packages/backend/src/open_payments/payment/outgoing/worker.ts b/packages/backend/src/open_payments/payment/outgoing/worker.ts index 581706a7a2..a31edae60c 100644 --- a/packages/backend/src/open_payments/payment/outgoing/worker.ts +++ b/packages/backend/src/open_payments/payment/outgoing/worker.ts @@ -29,19 +29,21 @@ export async function processPendingPayments( const payments = await getPendingPayments(trx, deps_) if (payments.length === 0) return - const promises = payments.map((payment) => handlePaymentLifecycle( - { - ...deps_, - knex: trx, - logger: deps_.logger.child({ - payment: payment.id, - from_state: payment.state - }) - }, - payment - )) + const promises = payments.map((payment) => + handlePaymentLifecycle( + { + ...deps_, + knex: trx, + logger: deps_.logger.child({ + payment: payment.id, + from_state: payment.state + }) + }, + payment + ) + ) await Promise.all(promises) - return payments.map(payment => payment.id) + return payments.map((payment) => payment.id) }) stopTimer() From 0bc68b5cd05235eae214479a49e6b6fdec025af3 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Mon, 13 Jul 2026 14:12:52 +0200 Subject: [PATCH 03/15] chore: log for github action --- test/performance/scripts/create-outgoing-payments.js | 1 + 1 file changed, 1 insertion(+) diff --git a/test/performance/scripts/create-outgoing-payments.js b/test/performance/scripts/create-outgoing-payments.js index 5815dc565e..1ded499ac7 100644 --- a/test/performance/scripts/create-outgoing-payments.js +++ b/test/performance/scripts/create-outgoing-payments.js @@ -114,6 +114,7 @@ export default function (data) { } const createReceiverResponse = request(createReceiverPayload) + console.log(createReceiverResponse) const receiver = createReceiverResponse.createReceiver.receiver const createQuotePayload = { From e8e02f22cdbcb87c5d3e1cac8706949e32bf5c5c Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Mon, 13 Jul 2026 17:59:10 +0200 Subject: [PATCH 04/15] feat: fund outgoing payment in performance test --- .../scripts/create-outgoing-payments.js | 23 +++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/test/performance/scripts/create-outgoing-payments.js b/test/performance/scripts/create-outgoing-payments.js index 1ded499ac7..25fa5878b1 100644 --- a/test/performance/scripts/create-outgoing-payments.js +++ b/test/performance/scripts/create-outgoing-payments.js @@ -114,7 +114,6 @@ export default function (data) { } const createReceiverResponse = request(createReceiverPayload) - console.log(createReceiverResponse) const receiver = createReceiverResponse.createReceiver.receiver const createQuotePayload = { @@ -162,7 +161,27 @@ export default function (data) { } } - request(createOutgoingPaymentPayload) + const createOutgoingPaymentResponse = request(createOutgoingPaymentPayload) + const outgoingPayment = + createOutgoingPaymentResponse.createOutgoingPayment.payment + + const fundOutgoingPaymentPayload = { + query: ` + mutation DepositOutgoingPaymentLiquidity($input: DepositOutgoingPaymentLiquidityInput!) { + depositOutgoingPaymentLiquidity(input: $input) { + success + } + } + `, + variables: { + input: { + outgoingPaymentId: outgoingPayment.id, + idempotencyKey: uuidv4() + } + } + } + + request(fundOutgoingPaymentPayload) } export function handleSummary(data) { From 1baacb3e28db45b8835a4c73f6cf543d6229b0f6 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 09:24:55 +0200 Subject: [PATCH 05/15] feat: increase batch size to 100 --- packages/backend/src/config/app.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index 4ca99ea797..95b8906698 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -134,7 +134,7 @@ export const Config = { outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1), outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 10), // milliseconds - outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 5), + outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 100), incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1), incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds From 66478f67973915a916ac1fa3dc77c430fb51d924 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 09:38:27 +0200 Subject: [PATCH 06/15] feat: batch size to 500 --- packages/backend/src/config/app.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index 95b8906698..884f4e0e10 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -134,7 +134,7 @@ export const Config = { outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1), outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 10), // milliseconds - outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 100), + outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 500), incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1), incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds From 564edfeca192aedb313c16330bf2658599a85802 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 09:52:17 +0200 Subject: [PATCH 07/15] feat: batch size to 250 --- packages/backend/src/config/app.ts | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index 884f4e0e10..f2228cb559 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -17,9 +17,9 @@ function envStringArray(name: string, value: string[]): string[] { return envValue == null ? value : envValue - .split(',') - .map((s) => s.trim()) - .filter((s) => s.length > 0) + .split(',') + .map((s) => s.trim()) + .filter((s) => s.length > 0) } function envInt(name: string, value: number): number { @@ -65,11 +65,11 @@ export const Config = { 'OPEN_TELEMETRY_COLLECTOR_URLS', envBool('LIVENET', false) ? [ - 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' + ] : [ - 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' + ] ), openTelemetryExportInterval: envInt('OPEN_TELEMETRY_EXPORT_INTERVAL', 15000), telemetryExchangeRatesUrl: envString( @@ -91,9 +91,9 @@ export const Config = { process.env.NODE_ENV === 'test' ? `${process.env.DATABASE_URL}_${process.env.JEST_WORKER_ID}` : envString( - 'DATABASE_URL', - 'postgresql://postgres:password@localhost:5432/development' - ), + 'DATABASE_URL', + 'postgresql://postgres:password@localhost:5432/development' + ), walletAddressUrl: envString( 'WALLET_ADDRESS_URL', 'http://127.0.0.1:3001/.well-known/pay' @@ -134,7 +134,7 @@ export const Config = { outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1), outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 10), // milliseconds - outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 500), + outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 250), incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1), incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds From f22dcbb972a7f7ef6feeedbe155492ecab2cfa3c Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 09:56:25 +0200 Subject: [PATCH 08/15] chore: formatting --- packages/backend/src/config/app.ts | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index f2228cb559..8771a38644 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -17,9 +17,9 @@ function envStringArray(name: string, value: string[]): string[] { return envValue == null ? value : envValue - .split(',') - .map((s) => s.trim()) - .filter((s) => s.length > 0) + .split(',') + .map((s) => s.trim()) + .filter((s) => s.length > 0) } function envInt(name: string, value: number): number { @@ -65,11 +65,11 @@ export const Config = { 'OPEN_TELEMETRY_COLLECTOR_URLS', envBool('LIVENET', false) ? [ - 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://livenet-otel-collector-NLB-f7992547e797f23d.elb.eu-west-2.amazonaws.com:4317' + ] : [ - 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' - ] + 'http://otel-collector-NLB-e3172ff9d2f4bc8a.elb.eu-west-2.amazonaws.com:4317' + ] ), openTelemetryExportInterval: envInt('OPEN_TELEMETRY_EXPORT_INTERVAL', 15000), telemetryExchangeRatesUrl: envString( @@ -91,9 +91,9 @@ export const Config = { process.env.NODE_ENV === 'test' ? `${process.env.DATABASE_URL}_${process.env.JEST_WORKER_ID}` : envString( - 'DATABASE_URL', - 'postgresql://postgres:password@localhost:5432/development' - ), + 'DATABASE_URL', + 'postgresql://postgres:password@localhost:5432/development' + ), walletAddressUrl: envString( 'WALLET_ADDRESS_URL', 'http://127.0.0.1:3001/.well-known/pay' From e8d6ff99cdff13ad9508cad874c636b0d5803e7d Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 10:55:28 +0200 Subject: [PATCH 09/15] feat: added startup log --- packages/backend/src/index.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index 9f8654a747..52930be254 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -755,6 +755,11 @@ export const start = async ( logger.info(`Connector listening on ${config.connectorPort}`) logger.info('🐒 has 🚀. Get ready for 🍌🍌🍌🍌🍌') + logger.info( + `Running the Rafiki WW 2026 build with an outgoing payment batch size of ${config.outgoingPaymentBatchSize}.` + ) + logger.info(`Happy Testing - Nathan 🫡`) + if (config.enableAutoPeering) { await app.startAutoPeeringServer(config.autoPeeringServerPort) logger.info( From 6a7eec30a0681fee3670750b7e10d867238265b7 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 12:16:26 +0200 Subject: [PATCH 10/15] feat: change worker to use interval --- packages/backend/src/app.ts | 5 ++++- packages/backend/src/config/app.ts | 4 ++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts index fbaa03c0b8..8853406ca4 100644 --- a/packages/backend/src/app.ts +++ b/packages/backend/src/app.ts @@ -323,7 +323,10 @@ export class App { process.nextTick(() => this.processWalletAddress()) } for (let i = 0; i < this.config.outgoingPaymentWorkers; i++) { - process.nextTick(() => this.processOutgoingPayment()) + setInterval( + () => this.processOutgoingPayment(), + this.config.outgoingPaymentProcessingIntervalMs + ) } for (let i = 0; i < this.config.incomingPaymentWorkers; i++) { process.nextTick(() => this.processIncomingPayment()) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index 8771a38644..7d02e9b973 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -135,6 +135,10 @@ export const Config = { outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1), outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 10), // milliseconds outgoingPaymentBatchSize: envInt('OUTGOING_PAYMENT_WORKER_BATCH_SIZE', 250), + outgoingPaymentProcessingIntervalMs: envInt( + 'OUTGOING_PAYMENT_PROCESSING_INTERVAL_MS', + 200 + ), // milliseconds incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1), incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds From 253297fe306fe29bf20b51833eab87f83ce2e607 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 12:41:15 +0200 Subject: [PATCH 11/15] feat: setTimeout on processOutgoingPayment --- packages/backend/src/app.ts | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts index 8853406ca4..e420a1f897 100644 --- a/packages/backend/src/app.ts +++ b/packages/backend/src/app.ts @@ -891,11 +891,15 @@ export class App { return true }) .then((hasMoreWork) => { - if (hasMoreWork) process.nextTick(() => this.processOutgoingPayment()) + if (hasMoreWork) + setTimeout( + () => this.processOutgoingPayment(), + this.config.outgoingPaymentProcessingIntervalMs + ) else setTimeout( () => this.processOutgoingPayment(), - this.config.outgoingPaymentWorkerIdle + this.config.outgoingPaymentProcessingIntervalMs ).unref() }) } @@ -941,11 +945,15 @@ export class App { return true }) .then((hasMoreWork) => { - if (hasMoreWork) process.nextTick(() => this.processIncomingPayment()) + if (hasMoreWork) + setTimeout( + () => this.processIncomingPayment(), + this.config.incomingPaymentProcessingIntervalMs + ) else setTimeout( () => this.processIncomingPayment(), - this.config.incomingPaymentWorkerIdle + this.config.incomingPaymentProcessingIntervalMs ).unref() }) } From 4c3593ad51139f3cc62c6bbb69898bf1e506f385 Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 13:59:27 +0200 Subject: [PATCH 12/15] fix: config type error --- packages/backend/src/app.ts | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts index e420a1f897..fc17807ee3 100644 --- a/packages/backend/src/app.ts +++ b/packages/backend/src/app.ts @@ -945,15 +945,11 @@ export class App { return true }) .then((hasMoreWork) => { - if (hasMoreWork) - setTimeout( - () => this.processIncomingPayment(), - this.config.incomingPaymentProcessingIntervalMs - ) + if (hasMoreWork) process.nextTick(() => this.processIncomingPayment()) else setTimeout( () => this.processIncomingPayment(), - this.config.incomingPaymentProcessingIntervalMs + this.config.incomingPaymentWorkerIdle ).unref() }) } From bba93b21e1d9bd8e25cebf7787a53a89edbd1cfc Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 14:51:34 +0200 Subject: [PATCH 13/15] fix: proper recursive call --- packages/backend/src/app.ts | 7 ++- packages/backend/src/config/app.ts | 5 ++ .../payment/incoming/service.test.ts | 20 +++---- .../open_payments/payment/incoming/service.ts | 55 ++++++++++--------- 4 files changed, 49 insertions(+), 38 deletions(-) diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts index fc17807ee3..151dedcbab 100644 --- a/packages/backend/src/app.ts +++ b/packages/backend/src/app.ts @@ -323,13 +323,16 @@ export class App { process.nextTick(() => this.processWalletAddress()) } for (let i = 0; i < this.config.outgoingPaymentWorkers; i++) { - setInterval( + setTimeout( () => this.processOutgoingPayment(), this.config.outgoingPaymentProcessingIntervalMs ) } for (let i = 0; i < this.config.incomingPaymentWorkers; i++) { - process.nextTick(() => this.processIncomingPayment()) + setTimeout( + () => this.processIncomingPayment(), + this.config.incomingPaymentProcessingIntervalMs + ) } for (let i = 0; i < this.config.webhookWorkers; i++) { process.nextTick(() => this.processWebhook()) diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index 7d02e9b973..b61335863b 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -142,6 +142,11 @@ export const Config = { incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1), incomingPaymentWorkerIdle: envInt('INCOMING_PAYMENT_WORKER_IDLE', 200), // milliseconds + incomingPaymentBatchSize: envInt('INCOMING_PAYMENT_WORKER_BATCH_SIZE', 250), + incomingPaymentProcessingIntervalMs: envInt( + 'INCOMING_PAYMENT_PROCESSING_INTERVAL_MS', + 200 + ), // milliseconds pollIncomingPaymentCreatedWebhook: envBool( 'POLL_INCOMING_PAYMENT_CREATED_WEBHOOK', false diff --git a/packages/backend/src/open_payments/payment/incoming/service.test.ts b/packages/backend/src/open_payments/payment/incoming/service.test.ts index a9cc51e43a..b201e41246 100644 --- a/packages/backend/src/open_payments/payment/incoming/service.test.ts +++ b/packages/backend/src/open_payments/payment/incoming/service.test.ts @@ -772,7 +772,7 @@ describe('Incoming Payment Service', (): void => { jest.useFakeTimers() jest.setSystemTime(incomingPayment.expiresAt) - await expect(incomingPaymentService.processNext()).resolves.toBe( + await expect(incomingPaymentService.processNext()).resolves.toContain( incomingPayment.id ) await expect( @@ -803,7 +803,7 @@ describe('Incoming Payment Service', (): void => { }) jest.useFakeTimers() jest.setSystemTime(incomingPayment.expiresAt) - await expect(incomingPaymentService.processNext()).resolves.toBe( + await expect(incomingPaymentService.processNext()).resolves.toContain( incomingPayment.id ) @@ -886,9 +886,9 @@ describe('Incoming Payment Service', (): void => { if (eventType === IncomingPaymentEventType.IncomingPaymentExpired) { jest.useFakeTimers() jest.setSystemTime(incomingPayment.expiresAt) - await expect(incomingPaymentService.processNext()).resolves.toBe( - incomingPayment.id - ) + await expect( + incomingPaymentService.processNext() + ).resolves.toContain(incomingPayment.id) } else { await incomingPayment.onCredit({ // eslint-disable-next-line @typescript-eslint/no-non-null-assertion @@ -923,7 +923,7 @@ describe('Incoming Payment Service', (): void => { assert.ok(incomingPayment.processAt) jest.useFakeTimers() jest.setSystemTime(incomingPayment.processAt) - await expect(incomingPaymentService.processNext()).resolves.toBe( + await expect(incomingPaymentService.processNext()).resolves.toContain( incomingPayment.id ) const events = await IncomingPaymentEvent.query(knex) @@ -974,9 +974,9 @@ describe('Incoming Payment Service', (): void => { assert.ok(incomingPayment.processAt) jest.useFakeTimers() jest.setSystemTime(incomingPayment.processAt) - await expect(incomingPaymentService.processNext()).resolves.toBe( - incomingPayment.id - ) + await expect( + incomingPaymentService.processNext() + ).resolves.toContain(incomingPayment.id) const events = await IncomingPaymentEvent.query(knex) .where({ incomingPaymentId: incomingPayment.id, @@ -1127,7 +1127,7 @@ describe('Incoming Payment Service', (): void => { const future = new Date(Date.now() + 40_000) jest.useFakeTimers() jest.setSystemTime(future) - await expect(incomingPaymentService.processNext()).resolves.toBe( + await expect(incomingPaymentService.processNext()).resolves.toContain( incomingPayment.id ) await expect( diff --git a/packages/backend/src/open_payments/payment/incoming/service.ts b/packages/backend/src/open_payments/payment/incoming/service.ts index 3965a4c403..6498172387 100644 --- a/packages/backend/src/open_payments/payment/incoming/service.ts +++ b/packages/backend/src/open_payments/payment/incoming/service.ts @@ -76,7 +76,7 @@ export interface IncomingPaymentService id: string, tenantId?: string ): Promise - processNext(): Promise + processNext(): Promise update( options: UpdateOptions ): Promise @@ -313,43 +313,46 @@ async function getApprovedOrCanceledIncomingPayment( // Returns the id of the processed incoming payment (if any). async function processNextIncomingPayment( deps_: ServiceDependencies -): Promise { +): Promise { return deps_.knex.transaction(async (trx) => { const now = new Date(Date.now()).toISOString() const incomingPayments = await IncomingPayment.query(trx) - .limit(1) + .limit(deps_.config.incomingPaymentBatchSize) // Ensure the incoming payments cannot be processed concurrently by multiple workers. .forUpdate() // If an incoming payment is locked, don't wait — just come back for it later. .skipLocked() .where('processAt', '<=', now) - const incomingPayment = incomingPayments[0] - if (!incomingPayment) return + if (incomingPayments.length === 0) return - const asset = await deps_.assetService.get(incomingPayment.assetId) - if (asset) incomingPayment.asset = asset - - incomingPayment.walletAddress = await deps_.walletAddressService.get( - incomingPayment.walletAddressId - ) + const processIncoming = async (incomingPayment: IncomingPayment) => { + const asset = await deps_.assetService.get(incomingPayment.assetId) + if (asset) incomingPayment.asset = asset + incomingPayment.walletAddress = await deps_.walletAddressService.get( + incomingPayment.walletAddressId + ) - const deps = { - ...deps_, - knex: trx, - logger: deps_.logger.child({ - incomingPayment: incomingPayment.id - }) - } - if ( - incomingPayment.state === IncomingPaymentState.Expired || - incomingPayment.state === IncomingPaymentState.Completed - ) { - await handleDeactivated(deps, incomingPayment) - } else { - await handleExpired(deps, incomingPayment) + const deps = { + ...deps_, + knex: trx, + logger: deps_.logger.child({ + incomingPayment: incomingPayment.id + }) + } + if ( + incomingPayment.state === IncomingPaymentState.Expired || + incomingPayment.state === IncomingPaymentState.Completed + ) { + await handleDeactivated(deps, incomingPayment) + } else { + await handleExpired(deps, incomingPayment) + } + return incomingPayment.id } - return incomingPayment.id + + const promises = incomingPayments.map(processIncoming) + return Promise.all(promises) }) } From 305e6d12ff8248df7e9fd073afba29096e75eaaf Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Tue, 14 Jul 2026 14:53:32 +0200 Subject: [PATCH 14/15] feat: process incoming payments on interval --- packages/backend/src/app.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts index 151dedcbab..d2a7083171 100644 --- a/packages/backend/src/app.ts +++ b/packages/backend/src/app.ts @@ -948,11 +948,15 @@ export class App { return true }) .then((hasMoreWork) => { - if (hasMoreWork) process.nextTick(() => this.processIncomingPayment()) + if (hasMoreWork) + setTimeout( + () => this.processIncomingPayment(), + this.config.incomingPaymentProcessingIntervalMs + ) else setTimeout( () => this.processIncomingPayment(), - this.config.incomingPaymentWorkerIdle + this.config.incomingPaymentProcessingIntervalMs ).unref() }) } From 205ece4b875deb86744d366b3131b7dcbf0a7f4f Mon Sep 17 00:00:00 2001 From: Nathan Lie Date: Wed, 15 Jul 2026 12:17:51 +0200 Subject: [PATCH 15/15] feat: batch wallet addresses --- packages/backend/src/app.ts | 13 +++++++++--- packages/backend/src/config/app.ts | 5 +++++ .../wallet_address/service.test.ts | 8 ++++---- .../open_payments/wallet_address/service.ts | 20 ++++++------------- 4 files changed, 25 insertions(+), 21 deletions(-) diff --git a/packages/backend/src/app.ts b/packages/backend/src/app.ts index d2a7083171..d8641a1ec4 100644 --- a/packages/backend/src/app.ts +++ b/packages/backend/src/app.ts @@ -320,7 +320,10 @@ export class App { // Workers are in the way during tests. if (this.config.env !== 'test') { for (let i = 0; i < this.config.walletAddressWorkers; i++) { - process.nextTick(() => this.processWalletAddress()) + setTimeout( + () => this.processWalletAddress(), + this.config.walletAddressProcessingIntervalMs + ) } for (let i = 0; i < this.config.outgoingPaymentWorkers; i++) { setTimeout( @@ -873,11 +876,15 @@ export class App { return true }) .then((hasMoreWork) => { - if (hasMoreWork) process.nextTick(() => this.processWalletAddress()) + if (hasMoreWork) + setTimeout( + () => this.processWalletAddress(), + this.config.walletAddressProcessingIntervalMs + ) else setTimeout( () => this.processWalletAddress(), - this.config.walletAddressWorkerIdle + this.config.walletAddressProcessingIntervalMs ).unref() }) } diff --git a/packages/backend/src/config/app.ts b/packages/backend/src/config/app.ts index b61335863b..95f1f4dea3 100644 --- a/packages/backend/src/config/app.ts +++ b/packages/backend/src/config/app.ts @@ -127,6 +127,11 @@ export const Config = { walletAddressWorkers: envInt('WALLET_ADDRESS_WORKERS', 1), walletAddressWorkerIdle: envInt('WALLET_ADDRESS_WORKER_IDLE', 200), // milliseconds + walletAddressBatchSize: envInt('WALLET_ADDRESS_BATCH_SIZE', 250), + walletAddressProcessingIntervalMs: envInt( + 'WALLET_ADDRESS_PROCESSING_INTERVAL_MS', + 200 + ), // milliseconds authServerGrantUrl: envString('AUTH_SERVER_GRANT_URL'), authServerIntrospectionUrl: envString('AUTH_SERVER_INTROSPECTION_URL'), diff --git a/packages/backend/src/open_payments/wallet_address/service.test.ts b/packages/backend/src/open_payments/wallet_address/service.test.ts index 95c1860fb5..7d216cdb32 100644 --- a/packages/backend/src/open_payments/wallet_address/service.test.ts +++ b/packages/backend/src/open_payments/wallet_address/service.test.ts @@ -846,9 +846,9 @@ describe('Open Payments Wallet Address Service', (): void => { 'Does not process wallet address $description for withdrawal', async ({ processAt }): Promise => { await walletAddress.$query(knex).patch({ processAt }) - await expect( - walletAddressService.processNext() - ).resolves.toBeUndefined() + await expect(walletAddressService.processNext()).resolves.toHaveLength( + 0 + ) await expect( walletAddressService.get(walletAddress.id) ).resolves.toEqual(walletAddress) @@ -877,7 +877,7 @@ describe('Open Payments Wallet Address Service', (): void => { processAt: new Date(), totalEventsAmount }) - await expect(walletAddressService.processNext()).resolves.toBe( + await expect(walletAddressService.processNext()).resolves.toContain( walletAddress.id ) await expect( diff --git a/packages/backend/src/open_payments/wallet_address/service.ts b/packages/backend/src/open_payments/wallet_address/service.ts index b0a8aff516..5320b46d9b 100644 --- a/packages/backend/src/open_payments/wallet_address/service.ts +++ b/packages/backend/src/open_payments/wallet_address/service.ts @@ -79,7 +79,7 @@ export interface WalletAddressService { sortOrder?: SortOrder, tenantId?: string ): Promise - processNext(): Promise + processNext(): Promise triggerEvents(limit: number): Promise } @@ -130,7 +130,7 @@ export async function createWalletAddressService({ getOrPollByUrl: (url) => getOrPollByUrl(deps, url), getPage: (pagination?, sortOrder?, tenantId?) => getWalletAddressPage(deps, pagination, sortOrder, tenantId), - processNext: () => processNextWalletAddress(deps), + processNext: () => processNextWalletAddresses(deps), triggerEvents: (limit) => triggerWalletAddressEvents(deps, limit) } } @@ -453,14 +453,6 @@ async function getWalletAddressPage( return addresses } -// Returns the id of the processed wallet address (if any). -async function processNextWalletAddress( - deps: ServiceDependencies -): Promise { - const walletAddresses = await processNextWalletAddresses(deps, 1) - return walletAddresses[0]?.id -} - async function triggerWalletAddressEvents( deps: ServiceDependencies, limit: number @@ -473,12 +465,12 @@ async function triggerWalletAddressEvents( // Returns the processed accounts (if any). async function processNextWalletAddresses( deps_: ServiceDependencies, - limit: number -): Promise { + limit?: number +): Promise { return deps_.knex.transaction(async (trx) => { const now = new Date(Date.now()).toISOString() const walletAddresses = await WalletAddress.query(trx) - .limit(limit) + .limit(limit || deps_.config.walletAddressBatchSize) // Ensure the wallet addresses cannot be processed concurrently by multiple workers. .forUpdate() // If a wallet address is locked, don't wait — just come back for it later. @@ -504,7 +496,7 @@ async function processNextWalletAddresses( }) } - return walletAddresses + return walletAddresses.map((w) => w.id) }) }