From d3da3be3ab227850523bc9cb20da8c63d0e80400 Mon Sep 17 00:00:00 2001 From: weare Date: Mon, 19 Oct 2026 12:29:29 -0800 Subject: [PATCH] feat: add horizontal scaling for payment workers (#945) --- backend/docs/PAYMENT_WORKER_POOL.md | 116 +++++++ .../__tests__/payment-worker-pool.test.ts | 163 ++++++++++ .../services/payments/payment-worker-pool.ts | 307 ++++++++++++++++++ 3 files changed, 586 insertions(+) create mode 100644 backend/docs/PAYMENT_WORKER_POOL.md create mode 100644 backend/src/services/payments/__tests__/payment-worker-pool.test.ts create mode 100644 backend/src/services/payments/payment-worker-pool.ts diff --git a/backend/docs/PAYMENT_WORKER_POOL.md b/backend/docs/PAYMENT_WORKER_POOL.md new file mode 100644 index 00000000..9cee6dae --- /dev/null +++ b/backend/docs/PAYMENT_WORKER_POOL.md @@ -0,0 +1,116 @@ +# Horizontal Scaling for Payment Workers (Issue #945) + +A distributed, horizontally scalable payment worker pool backed by BullMQ + +Redis. Multiple worker instances connect to the same Redis queue and compete +for payment jobs, giving automatic load distribution across processes and hosts +with at-least-once delivery, exponential back-off retries, and dead-lettering. + +## Problem + +Payment processing was handled synchronously in the request path +(`ProviderPaymentRouter.route()`), which blocks the caller and limits +throughput to a single instance. Under load, slow downstream providers (Stripe, +Stellar Horizon, EVM RPCs) exhaust the event loop and there is no way to +scale payment processing independently of the API. + +## Solution + +`PaymentWorkerPool` (`src/services/payments/payment-worker-pool.ts`) enqueues +payments as BullMQ jobs and a pool of workers processes them concurrently: + +``` +client ──► enqueue() ──► BullMQ queue (Redis) ◀──► Worker instance A + ▲ │ + └──► Worker instance B ◀──┘ + (competes for same jobs) +``` + +Each worker pulls jobs from the shared queue, so adding worker instances +automatically increases throughput without rebalancing. Jobs survive process +restarts and are retried with exponential back-off before being dead-lettered. + +## Architecture + +| Component | Responsibility | +|-----------|----------------| +| `PaymentWorkerPool` | Creates/manages BullMQ queue + workers, enqueues payments | +| `Queue` (BullMQ) | Persisted job store; shared across all instances | +| `Worker` (BullMQ) | Pulls jobs, processes via `providerRegistry`, reports completion/failure | +| `providerRegistry` | Routes payments to soroban/evm/fiat/credit providers | +| In-process fallback | When Redis is unavailable, jobs run locally on a 1s interval | + +## Usage + +### Enqueue a payment (fire-and-forget) + +```ts +import { getPaymentWorkerPool } from '../services/payments/payment-worker-pool.js'; + +const pool = getPaymentWorkerPool(); +await pool.start(); + +const paymentId = await pool.enqueue({ + amount: 100, + currency: 'USDC', + toAddress: 'GABC...', + network: 'stellar', + tenantId: 'tenant-1', +}, 'soroban'); // optional preferred provider +``` + +### Get metrics (for autoscaling / observability) + +```ts +const metrics = await pool.getMetrics(); +// { queued, active, completed, failed, delayed, concurrency, workerCount } +``` + +Use `queued` and `active` to drive horizontal autoscaling: add worker +instances when queue depth grows, remove when idle. + +### Graceful shutdown + +```ts +await pool.shutdown(); // closes workers and queue connection +``` + +## Configuration + +| Option | Env | Default | Meaning | +|--------|-----|---------|---------| +| `redisUrl` | `REDIS_URL` | undefined | Redis connection string | +| `concurrency` | — | 4 | Parallel workers per instance | +| `maxAttempts` | — | 5 | Max attempts per payment job | +| `initialDelayMs` | — | 1000 | First retry delay (ms) | +| `maxDelayMs` | — | 30000 | Max retry delay (ms) | +| `queueName` | — | `agenticpay:payments` | BullMQ queue name | +| `prefix` | — | `agenticpay` | BullMQ key prefix | + +Redis is required for true horizontal scaling. When `REDIS_ENABLED=true` and +`REDIS_URL` are set, all instances share the same queue. Without Redis the pool +falls back to an in-process queue (single instance only). + +## Testing + +```bash +# Backend tests +cd backend +npx vitest run src/services/payments/__tests__/payment-worker-pool.test.ts +``` + +## Horizontal scaling guide + +1. Deploy `N` worker instances (containers/pods) all running the same code. +2. Set `REDIS_ENABLED=true` and `REDIS_URL` to a shared Redis. +3. Each instance calls `pool.start()` — BullMQ workers automatically compete + for jobs from the shared queue. +4. Monitor `pool.getMetrics().queued` — scale `N` up/down based on depth. +5. On deploy, call `pool.shutdown()` to drain in-flight jobs before exit. + +## Security + +- Payment payloads are job data in Redis — ensure Redis is network-isolated + and requires authentication (`REDIS_URL` with password). +- No secrets are stored in job metadata. +- Each provider is an isolated circuit; a failing provider does not block + others (see circuit breaker integration, issue #948). diff --git a/backend/src/services/payments/__tests__/payment-worker-pool.test.ts b/backend/src/services/payments/__tests__/payment-worker-pool.test.ts new file mode 100644 index 00000000..6ad9a7dd --- /dev/null +++ b/backend/src/services/payments/__tests__/payment-worker-pool.test.ts @@ -0,0 +1,163 @@ +import { describe, it, expect, beforeEach, vi } from 'vitest'; +import { PaymentWorkerPool, resetPaymentWorkerPoolForTesting } from '../payment-worker-pool.js'; +import { PaymentProviderRegistry } from '../provider-registry.js'; +import { MockPaymentProvider } from '../providers/mock.js'; +import type { PaymentInput } from '../providers/types.js'; + +const baseInput: PaymentInput = { + amount: 10, + currency: 'XLM', + toAddress: 'GABC', + network: 'stellar', + tenantId: 'tenant-1', +}; + +describe('PaymentWorkerPool', () => { + beforeEach(() => { + resetPaymentWorkerPoolForTesting(); + }); + + it('constructs with defaults', () => { + const pool = new PaymentWorkerPool(); + expect(pool).toBeInstanceOf(PaymentWorkerPool); + }); + + it('selects provider by network', () => { + const pool = new PaymentWorkerPool(); + // @ts-ignore - accessing private method for unit verification + expect(pool['selectProviderId']({ ...baseInput, network: 'stellar' })).toBe('soroban'); + expect(pool['selectProviderId']({ ...baseInput, network: 'polygon' })).toBe('evm'); + expect(pool['selectProviderId']({ ...baseInput, network: 'other', currency: 'USD' })).toBe('fiat'); + expect(pool['selectProviderId']({ ...baseInput, network: 'other', currency: 'POINTS' })).toBe('credit'); + }); + + it('rejects invalid amounts before enqueue', async () => { + const pool = new PaymentWorkerPool(); + await expect(pool.processPayment({ ...baseInput, amount: 0 })).resolves.toMatchObject({ ok: false }); + await expect(pool.processPayment({ ...baseInput, amount: -5 })).resolves.toMatchObject({ ok: false }); + await expect(pool.processPayment({ ...baseInput, amount: Number.NaN })).resolves.toMatchObject({ ok: false }); + }); + + it('processes a successful payment through the registry', async () => { + const registry = new PaymentProviderRegistry(); + const provider = new MockPaymentProvider('test-provider'); + registry.register(provider); + const pool = new PaymentWorkerPool({}, registry); + + const result = await pool.processPayment(baseInput); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.providerId).toBe('test-provider'); + expect(registry.getMetrics('test-provider').successCount).toBe(1); + }); + + it('falls back to another provider when primary fails', async () => { + const registry = new PaymentProviderRegistry(); + const failing = new MockPaymentProvider('failing', { shouldFail: true }); + const healthy = new MockPaymentProvider('healthy'); + registry.register(failing); + registry.register(healthy); + const pool = new PaymentWorkerPool({}, registry); + + const result = await pool.processPayment({ ...baseInput, network: 'stellar' }); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.providerId).toBe('healthy'); + expect(registry.getMetrics('failing').errorCount).toBe(1); + }); + + it('returns PROVIDER_UNAVAILABLE when all providers fail', async () => { + const registry = new PaymentProviderRegistry(); + registry.register(new MockPaymentProvider('failing', { shouldFail: true })); + const pool = new PaymentWorkerPool({}, registry); + + const result = await pool.processPayment(baseInput); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe('PROVIDER_UNAVAILABLE'); + }); + + it('respects preferred provider override', async () => { + const registry = new PaymentProviderRegistry(); + registry.register(new MockPaymentProvider('soroban')); + registry.register(new MockPaymentProvider('evm')); + const pool = new PaymentWorkerPool({}, registry); + + const result = await pool.processPayment({ ...baseInput, network: 'stellar' }, 'evm'); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.providerId).toBe('evm'); + }); + + it('getMetrics returns serialisable shape without Redis', async () => { + const pool = new PaymentWorkerPool(); + await pool.start(); + const metrics = await pool.getMetrics(); + expect(typeof metrics.queued).toBe('number'); + expect(typeof metrics.active).toBe('number'); + expect(typeof metrics.completed).toBe('number'); + expect(typeof metrics.failed).toBe('number'); + expect(typeof metrics.concurrency).toBe('number'); + await pool.shutdown(); + }); + + it('enqueue + drainLocal processes a job in-process', async () => { + const registry = new PaymentProviderRegistry(); + const provider = new MockPaymentProvider('local'); + registry.register(provider); + const pool = new PaymentWorkerPool({}, registry); + await pool.start(); + + const paymentId = await pool.enqueue(baseInput); + expect(typeof paymentId).toBe('string'); + + // Allow the local drain interval to pick up the job + await new Promise((r) => setTimeout(r, 150)); + + expect(registry.getMetrics('local').successCount).toBe(1); + await pool.shutdown(); + }); + + it('shutdown stops workers and clears state', async () => { + const pool = new PaymentWorkerPool(); + await pool.start(); + await pool.shutdown(); + // Should not throw on double shutdown + await pool.shutdown(); + }); + + it('pause/resume without Redis does not throw', async () => { + const pool = new PaymentWorkerPool(); + await pool.start(); + await pool.pause(); + await pool.resume(); + await pool.shutdown(); + }); + + it('getPaymentWorkerPool returns a singleton', () => { + resetPaymentWorkerPoolForTesting(); + const a = getPaymentWorkerPool(); + const b = getPaymentWorkerPool(); + expect(a).toBe(b); + resetPaymentWorkerPoolForTesting(); + }); + + it('registry integration: records success and error metrics', async () => { + const registry = new PaymentProviderRegistry(); + const failing = new MockPaymentProvider('fail', { shouldFail: true }); + const ok = new MockPaymentProvider('ok'); + registry.register(failing); + registry.register(ok); + const pool = new PaymentWorkerPool({}, registry); + + await pool.processPayment({ ...baseInput, network: 'stellar' }); + expect(registry.getMetrics('fail').errorCount).toBe(1); + expect(registry.getMetrics('ok').successCount).toBe(1); + }); + + it('processPayment returns structured error for invalid amount', async () => { + const pool = new PaymentWorkerPool(); + const result = await pool.processPayment({ ...baseInput, amount: -100 }); + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.error.statusCode).toBe(400); + expect(result.error.code).toBe('INVALID_PAYMENT_AMOUNT'); + } + }); +}); diff --git a/backend/src/services/payments/payment-worker-pool.ts b/backend/src/services/payments/payment-worker-pool.ts new file mode 100644 index 00000000..a60fd057 --- /dev/null +++ b/backend/src/services/payments/payment-worker-pool.ts @@ -0,0 +1,307 @@ +/** + * payment-worker-pool.ts — Issue #945 + * + * Horizontally scalable payment worker pool backed by BullMQ + Redis. + * + * Multiple worker instances connect to the same Redis queue and compete for + * payment jobs, giving automatic load distribution across processes/hosts. + * Each instance runs `concurrency` parallel workers. Jobs are persisted in + * Redis so they survive process restarts, and failed jobs are retried with + * exponential back-off before being dead-lettered. + * + * When Redis is unavailable the pool falls back to an in-process queue so the + * service remains operational in single-instance mode. + */ + +import { Queue, Worker, type Job, type ConnectionOptions } from 'bullmq'; +import { ok, err, type Result } from '../../lib/result.js'; +import { providerRegistry, type PaymentProviderRegistry } from './provider-registry.js'; +import type { PaymentInput, PaymentOutput } from './providers/types.js'; + +export interface PaymentWorkerJob { + paymentId: string; + input: PaymentInput; + preferredProviderId?: string; + retries?: number; +} + +export interface PaymentWorkerResult { + paymentId: string; + providerId: string; + output: PaymentOutput; +} + +export interface PaymentWorkerPoolConfig { + redisUrl?: string; + concurrency?: number; + maxAttempts?: number; + initialDelayMs?: number; + maxDelayMs?: number; + queueName?: string; + prefix?: string; +} + +const DEFAULT_CONFIG: Required = { + redisUrl: undefined, + concurrency: 4, + maxAttempts: 5, + initialDelayMs: 1_000, + maxDelayMs: 30_000, + queueName: 'agenticpay:payments', + prefix: 'agenticpay', +}; + +export interface PoolMetrics { + queued: number; + active: number; + completed: number; + failed: number; + delayed: number; + concurrency: number; + workerCount: number; +} + +export class PaymentWorkerPool { + private readonly config: Required; + private readonly registry: PaymentProviderRegistry; + private queue: Queue | null = null; + private workers: Worker[] = []; + private running = false; + private localQueue: PaymentWorkerJob[] = []; + private localProcessing = new Set(); + private localMetrics = { queued: 0, active: 0, completed: 0, failed: 0 }; + + constructor(config: PaymentWorkerPoolConfig = {}, registry: PaymentProviderRegistry = providerRegistry) { + this.config = { ...DEFAULT_CONFIG, ...config }; + this.registry = registry; + } + + /** Start the pool: connect to Redis (or fall back to in-process queue) and spawn workers. */ + async start(): Promise { + if (this.running) return; + this.running = true; + + const redisUrl = this.config.redisUrl ?? process.env.REDIS_URL; + const redisEnabled = process.env.REDIS_ENABLED === 'true' && redisUrl; + + if (redisEnabled && redisUrl) { + try { + const connection = this.parseRedisUrl(redisUrl); + this.queue = new Queue(this.config.queueName, { + connection, + prefix: this.config.prefix, + defaultJobOptions: { + attempts: this.config.maxAttempts, + backoff: { type: 'exponential', delay: this.config.initialDelayMs, maxDelay: this.config.maxDelayMs }, + removeOnComplete: { count: 100 }, + removeOnFail: { count: 500 }, + }, + }); + + for (let i = 0; i < this.config.concurrency; i++) { + const worker = new Worker( + this.config.queueName, + async (job: Job) => this.processJob(job), + { connection, concurrency: 1 }, + ); + worker.on('completed', () => { + this.localMetrics.completed++; + this.localMetrics.active = Math.max(0, this.localMetrics.active - 1); + }); + worker.on('failed', () => { + this.localMetrics.failed++; + this.localMetrics.active = Math.max(0, this.localMetrics.active - 1); + }); + this.workers.push(worker); + } + + console.log(`[payment-worker-pool] Started ${this.config.concurrency} workers on queue "${this.config.queueName}"`); + } catch (error) { + console.error('[payment-worker-pool] Redis connection failed, falling back to in-process queue', error); + this.queue = null; + } + } + + if (!this.queue) { + // In-process fallback: process jobs locally on a short interval + const timer = setInterval(() => void this.drainLocal(), 1_000); + timer.unref(); + (this as unknown as { _localTimer: NodeJS.Timeout })._localTimer = timer; + console.log('[payment-worker-pool] Running in-process (Redis not configured)'); + } + } + + /** Enqueue a payment for asynchronous processing by a worker. */ + async enqueue(input: PaymentInput, preferredProviderId?: string): Promise { + const paymentId = `pay_${crypto.randomUUID().replace(/-/g, '')}`; + const job: PaymentWorkerJob = { paymentId, input, preferredProviderId }; + + if (this.queue) { + await this.queue.add('process-payment', job, { + jobId: `payment:${paymentId}`, + }); + this.localMetrics.queued++; + } else { + this.localQueue.push(job); + this.localMetrics.queued++; + } + + return paymentId; + } + + /** Process a single job (BullMQ worker handler or local drain). */ + private async processJob(job: Job): Promise { + const { paymentId, input, preferredProviderId } = job.data as PaymentWorkerJob; + + const result = await this.processPayment(input, preferredProviderId); + + if (!result.ok) { + throw new Error(result.error.message); + } + + return { + paymentId, + providerId: result.value.providerId, + output: result.value.output, + }; + } + + /** Route a payment through the provider registry (same logic as payment-router). */ + async processPayment(input: PaymentInput, preferredProviderId?: string): Promise> { + if (!Number.isFinite(input.amount) || input.amount <= 0) { + return err({ code: 'INVALID_PAYMENT_AMOUNT', message: 'Payment amount must be a finite number greater than zero', statusCode: 400 }); + } + + const primaryId = preferredProviderId && this.registry.get(preferredProviderId) + ? preferredProviderId + : this.selectProviderId(input); + + const candidateIds = [primaryId, ...this.registry.list().filter((id) => id !== primaryId)]; + + for (const id of candidateIds) { + const provider = this.registry.get(id); + if (!provider) continue; + + const start = Date.now(); + const providerResult = await provider.processPayment(input); + const latencyMs = Date.now() - start; + + if (providerResult.ok) { + this.registry.recordSuccess(id, latencyMs); + return ok({ ...providerResult.value, providerId: id }); + } + + this.registry.recordError(id); + } + + return err({ code: 'PROVIDER_UNAVAILABLE', message: 'All payment providers failed', statusCode: 502 }); + } + + private selectProviderId(input: PaymentInput): string { + const EVM_NETWORKS = new Set(['ethereum', 'polygon', 'arbitrum', 'optimism', 'base']); + const FIAT_CURRENCIES = new Set(['USD', 'EUR', 'GBP']); + if (input.network === 'stellar') return 'soroban'; + if (EVM_NETWORKS.has(input.network)) return 'evm'; + if (input.token ? FIAT_CURRENCIES.has(input.token) : FIAT_CURRENCIES.has(input.currency)) return 'fiat'; + return 'credit'; + } + + /** Drain the in-process fallback queue (one tick). */ + private async drainLocal(): Promise { + if (this.localProcessing.size >= this.config.concurrency || this.localQueue.length === 0) return; + + const batch = this.localQueue.splice(0, this.config.concurrency - this.localProcessing.size); + await Promise.all(batch.map((job) => this.processLocalJob(job))); + } + + private async processLocalJob(job: PaymentWorkerJob): Promise { + this.localProcessing.add(job.paymentId); + this.localMetrics.active++; + try { + await this.processPayment(job.input, job.preferredProviderId); + this.localMetrics.completed++; + } catch { + this.localMetrics.failed++; + } finally { + this.localProcessing.delete(job.paymentId); + this.localMetrics.active = Math.max(0, this.localMetrics.active - 1); + } + } + + /** Get current pool metrics. */ + async getMetrics(): Promise { + if (this.queue) { + const [waiting, active, completed, failed, delayed] = await Promise.all([ + this.queue.getWaitingCount(), + this.queue.getActiveCount(), + this.queue.getCompletedCount(), + this.queue.getFailedCount(), + this.queue.getDelayedCount(), + ]); + return { + queued: waiting, + active, + completed, + failed, + delayed, + concurrency: this.config.concurrency, + workerCount: this.workers.length, + }; + } + return { + ...this.localMetrics, + concurrency: this.config.concurrency, + workerCount: 1, + }; + } + + /** Pause the queue (no new jobs will be picked up). */ + async pause(): Promise { + await this.queue?.pause(); + } + + /** Resume the queue. */ + async resume(): Promise { + await this.queue?.resume(); + } + + /** Gracefully shut down all workers and close the queue connection. */ + async shutdown(): Promise { + this.running = false; + await Promise.all(this.workers.map((w) => w.close())); + this.workers = []; + await this.queue?.close(); + this.queue = null; + if (this.localQueue.length > 0) { + this.localQueue = []; + } + } + + private parseRedisUrl(url: string): ConnectionOptions { + try { + const parsed = new URL(url); + return { + host: parsed.hostname || 'localhost', + port: parseInt(parsed.port || '6379', 10), + password: parsed.password || undefined, + username: parsed.username || undefined, + tls: parsed.protocol === 'rediss:' ? {} : undefined, + }; + } catch { + const [host, port] = url.split(':'); + return { host: host || 'localhost', port: parseInt(port || '6379', 10) }; + } + } +} + +/** Singleton accessor so routes and services share one pool instance. */ +let _pool: PaymentWorkerPool | null = null; + +export function getPaymentWorkerPool(config?: PaymentWorkerPoolConfig): PaymentWorkerPool { + if (!_pool) _pool = new PaymentWorkerPool(config); + return _pool; +} + +export function resetPaymentWorkerPoolForTesting(): void { + _pool = null; +}