feat(backend): batch outgoing payments in worker jobs

packages/backend/src/app.ts

@@ -320,13 +320,22 @@ 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++) {
-        process.nextTick(() => this.processOutgoingPayment())
+        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())
@@ -867,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()
       })
   }
@@ -888,11 +901,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()
       })
   }
@@ -938,11 +955,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()
       })
   }

packages/backend/src/config/app.ts

@@ -127,16 +127,31 @@ 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'),
   authServiceApiUrl: envString('AUTH_SERVICE_API_URL'),
 
   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
+  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

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(

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(

packages/backend/src/open_payments/payment/incoming/service.ts

@@ -76,7 +76,7 @@ export interface IncomingPaymentService
     id: string,
     tenantId?: string
   ): Promise<IncomingPayment | IncomingPaymentError>
-  processNext(): Promise<string | undefined>
+  processNext(): Promise<string[] | undefined>
   update(
     options: UpdateOptions
   ): Promise<IncomingPayment | IncomingPaymentError>
@@ -313,43 +313,46 @@ async function getApprovedOrCanceledIncomingPayment(
 // Returns the id of the processed incoming payment (if any).
 async function processNextIncomingPayment(
   deps_: ServiceDependencies
-): Promise<string | undefined> {
+): Promise<string[] | undefined> {
   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)
   })
 }