From 30e7e44542b507fe211f39253e6aaffaf13014f4 Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:46:44 +0100 Subject: [PATCH 01/10] feat(common): add a bounded-concurrency map helper `mapWithConcurrency(items, n, worker)` runs `worker` over `items` with at most `n` in flight, returns `PromiseSettledResult`s in input order, and never rejects. Replaces the `for..of await` sweeps whose runtime was the sum of every item's latency (#236, #237, #238). --- backend/src/common/concurrency/index.ts | 1 + .../concurrency/map-with-concurrency.ts | 46 +++++++++++++++++++ 2 files changed, 47 insertions(+) create mode 100644 backend/src/common/concurrency/index.ts create mode 100644 backend/src/common/concurrency/map-with-concurrency.ts diff --git a/backend/src/common/concurrency/index.ts b/backend/src/common/concurrency/index.ts new file mode 100644 index 0000000..49f11b6 --- /dev/null +++ b/backend/src/common/concurrency/index.ts @@ -0,0 +1 @@ +export * from './map-with-concurrency'; diff --git a/backend/src/common/concurrency/map-with-concurrency.ts b/backend/src/common/concurrency/map-with-concurrency.ts new file mode 100644 index 0000000..fce663a --- /dev/null +++ b/backend/src/common/concurrency/map-with-concurrency.ts @@ -0,0 +1,46 @@ +/** + * Run an async `worker` over `items` with at most `concurrency` calls in flight + * at once, and never reject: results come back in input order as + * `PromiseSettledResult`s, the same shape `Promise.allSettled` returns. + * + * Replaces the `for (const x of items) await work(x)` sweeps in the background + * workers, whose total runtime was the *sum* of every item's latency — + * pathological when each item does its own network I/O with retries + * (#236, #237, #238). + * + * `concurrency` is clamped to at least 1; a value >= `items.length` runs every + * item at once. + */ +export async function mapWithConcurrency( + items: readonly T[], + concurrency: number, + worker: (item: T, index: number) => Promise, +): Promise[]> { + const limit = Math.max(1, Math.floor(concurrency) || 1); + const results: PromiseSettledResult[] = new Array(items.length); + let cursor = 0; + + const runner = async (): Promise => { + for (;;) { + const index = cursor; + cursor += 1; + if (index >= items.length) return; + try { + results[index] = { status: 'fulfilled', value: await worker(items[index], index) }; + } catch (reason) { + results[index] = { status: 'rejected', reason }; + } + } + }; + + await Promise.all( + Array.from({ length: Math.min(limit, items.length) }, () => runner()), + ); + + return results; +} + +/** How many `PromiseSettledResult`s in `results` rejected. */ +export function countRejected(results: PromiseSettledResult[]): number { + return results.reduce((n, r) => (r.status === 'rejected' ? n + 1 : n), 0); +} From 3ed2211c0e50f53c6e5a77d54685daf809bd196e Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:46:49 +0100 Subject: [PATCH 02/10] test(common): cover mapWithConcurrency --- .../concurrency/map-with-concurrency.spec.ts | 51 +++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 backend/src/common/concurrency/map-with-concurrency.spec.ts diff --git a/backend/src/common/concurrency/map-with-concurrency.spec.ts b/backend/src/common/concurrency/map-with-concurrency.spec.ts new file mode 100644 index 0000000..e7c1895 --- /dev/null +++ b/backend/src/common/concurrency/map-with-concurrency.spec.ts @@ -0,0 +1,51 @@ +import { mapWithConcurrency, countRejected } from './map-with-concurrency'; + +const sleep = (ms: number): Promise => new Promise(r => setTimeout(r, ms)); + +describe('mapWithConcurrency', () => { + it('returns results in input order regardless of completion order', async () => { + const results = await mapWithConcurrency([30, 10, 20, 0], 4, async ms => { + await sleep(ms); + return ms; + }); + expect(results.map(r => (r.status === 'fulfilled' ? r.value : null))).toEqual([30, 10, 20, 0]); + }); + + it('never runs more than `concurrency` workers at once', async () => { + let inFlight = 0; + let peak = 0; + await mapWithConcurrency(Array.from({ length: 20 }, (_, i) => i), 3, async () => { + inFlight += 1; + peak = Math.max(peak, inFlight); + await sleep(5); + inFlight -= 1; + }); + expect(peak).toBe(3); + }); + + it('is faster than sequential for many slow items', async () => { + const items = Array.from({ length: 12 }, (_, i) => i); + const start = Date.now(); + await mapWithConcurrency(items, 6, () => sleep(20)); + const elapsed = Date.now() - start; + // Sequential would be ~240ms; 6-wide is ~40ms. Allow generous slack. + expect(elapsed).toBeLessThan(150); + }); + + it('captures a rejection per item instead of aborting the batch', async () => { + const results = await mapWithConcurrency([1, 2, 3], 2, async n => { + if (n === 2) throw new Error('boom'); + return n; + }); + expect(results[0]).toEqual({ status: 'fulfilled', value: 1 }); + expect(results[1].status).toBe('rejected'); + expect(results[2]).toEqual({ status: 'fulfilled', value: 3 }); + expect(countRejected(results)).toBe(1); + }); + + it('clamps concurrency to at least 1 and handles an empty list', async () => { + expect(await mapWithConcurrency([], 4, async () => 1)).toEqual([]); + const results = await mapWithConcurrency([1, 2], 0, async n => n); + expect(results.map(r => (r.status === 'fulfilled' ? r.value : null))).toEqual([1, 2]); + }); +}); From bbc13244fa2d3b70c57529ce16c3e8b815182cb0 Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:46:54 +0100 Subject: [PATCH 03/10] perf(gig): expire gigs with bounded concurrency and guard against overlapping sweeps (#236) `runOnce()` fanned out `expire()` one gig at a time; each `expire` appends an outbox row the relay delivers with retries, so a big batch with a slow endpoint serialised all that latency and could outrun the sweep interval. Now bounded-concurrent (GIG_EXPIRY_SWEEP_CONCURRENCY, default 8), a failed expire is counted+logged not fatal, and a `sweeping` flag skips a tick while a previous sweep is still running. Closes #236 --- backend/src/gig/gig-expiry-worker.service.ts | 25 +++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/backend/src/gig/gig-expiry-worker.service.ts b/backend/src/gig/gig-expiry-worker.service.ts index 1de72b2..dfdfe97 100644 --- a/backend/src/gig/gig-expiry-worker.service.ts +++ b/backend/src/gig/gig-expiry-worker.service.ts @@ -2,8 +2,11 @@ import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/commo import { GigService } from './gig.service'; import { DEFAULT_GIG_EXPIRY_SWEEP_INTERVAL_MS } from './gig.entity'; import { DistributedLockService } from '../common/redis/distributed-lock.service'; +import { mapWithConcurrency, countRejected } from '../common/concurrency'; const LOCK_KEY = 'lock:gig-expiry-sweep'; +/** How many gigs to expire in parallel within one sweep (#236). */ +const SWEEP_CONCURRENCY = Number(process.env.GIG_EXPIRY_SWEEP_CONCURRENCY) || 8; /** * Periodically sweeps the DB for open gig solicitations whose response deadline has @@ -28,6 +31,8 @@ export class GigExpiryWorkerService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(GigExpiryWorkerService.name); private timer?: NodeJS.Timeout; private currentLockToken?: string; + /** Guards against a slow sweep still running when the next tick fires (#236). */ + private sweeping = false; constructor( private readonly gigService: GigService, @@ -58,14 +63,20 @@ export class GigExpiryWorkerService implements OnModuleInit, OnModuleDestroy { } private async tick(intervalMs: number): Promise { + if (this.sweeping) { + this.logger.warn('Previous gig expiry sweep still in flight — skipping this tick'); + return; + } const token = await this.lock.tryAcquire(LOCK_KEY, Math.ceil(intervalMs * 1.5)); if (!token) { return; // another instance is holding the lock for this tick } this.currentLockToken = token; + this.sweeping = true; try { await this.runOnce(); } finally { + this.sweeping = false; await this.lock.release(LOCK_KEY, token); this.currentLockToken = undefined; } @@ -75,9 +86,17 @@ export class GigExpiryWorkerService implements OnModuleInit, OnModuleDestroy { async runOnce(): Promise { const expirable = await this.gigService.findExpirable(); - for (const gig of expirable) { - const expired = await this.gigService.expire(gig.id); - if (!expired) continue; + // Expire gigs with bounded concurrency instead of one-at-a-time: each + // `expire()` appends an outbox row the relay then delivers with retries, + // so a fully sequential loop over a big batch serialised all of that + // latency and could outrun the sweep interval (#236). A failed `expire` + // no longer aborts the rest of the sweep — it is counted and logged. + const results = await mapWithConcurrency(expirable, SWEEP_CONCURRENCY, gig => + this.gigService.expire(gig.id), + ); + const failed = countRejected(results); + if (failed > 0) { + this.logger.warn(`Gig expiry sweep: ${failed}/${expirable.length} gigs failed to expire`); } } From 3b8e22197bc224bbf01c5923becc2680c93ff837 Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:47:01 +0100 Subject: [PATCH 04/10] test(gig): concurrent sweep speed and tick non-overlap Refs #236 --- .../src/gig/gig-expiry-worker.service.spec.ts | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/backend/src/gig/gig-expiry-worker.service.spec.ts b/backend/src/gig/gig-expiry-worker.service.spec.ts index ec1d2d4..bfaa1f0 100644 --- a/backend/src/gig/gig-expiry-worker.service.spec.ts +++ b/backend/src/gig/gig-expiry-worker.service.spec.ts @@ -144,3 +144,73 @@ describe('GigExpiryWorkerService', () => { }); }); }); + +describe('GigExpiryWorkerService — concurrency & non-overlap (#236)', () => { + const originalInterval = process.env.GIG_EXPIRY_SWEEP_INTERVAL_MS; + + afterEach(() => { + if (originalInterval === undefined) delete process.env.GIG_EXPIRY_SWEEP_INTERVAL_MS; + else process.env.GIG_EXPIRY_SWEEP_INTERVAL_MS = originalInterval; + jest.useRealTimers(); + }); + + function slowGigService( + perGigMs: number, + ): jest.Mocked> { + const gigs = Array.from({ length: 10 }, (_, i) => makeGig(`g${i}`, GigStatus.OPEN)); + return { + findExpirable: jest.fn().mockResolvedValue(gigs), + expire: jest.fn().mockImplementation( + () => new Promise(r => setTimeout(() => r(true), perGigMs)), + ), + }; + } + + it('expires gigs concurrently — a 10-gig sweep is far faster than 10x one gig', async () => { + const gigService = slowGigService(20); + const worker = new GigExpiryWorkerService( + gigService as unknown as GigService, + fakeLock() as unknown as DistributedLockService, + ); + + const start = Date.now(); + await worker.runOnce(); + const elapsed = Date.now() - start; + + expect(gigService.expire).toHaveBeenCalledTimes(10); + // Sequential would be ~200ms; bounded-concurrency finishes well under half. + expect(elapsed).toBeLessThan(120); + }); + + it('skips a tick while a previous sweep is still running', async () => { + jest.useFakeTimers(); + process.env.GIG_EXPIRY_SWEEP_INTERVAL_MS = '10'; + + let resolveSweep!: () => void; + const gigService: jest.Mocked> = { + findExpirable: jest.fn().mockReturnValue(new Promise(() => {})), + expire: jest.fn(), + }; + // First sweep hangs until we release it. + gigService.findExpirable.mockImplementationOnce( + () => new Promise(res => { resolveSweep = () => res([]); }), + ); + + const lock = fakeLock(); + const worker = new GigExpiryWorkerService( + gigService as unknown as GigService, + lock as unknown as DistributedLockService, + ); + worker.onModuleInit(); + + jest.advanceTimersByTime(35); // several ticks while the first sweep is stuck + await flushPromises(); + + // Only the first tick got past the in-flight guard. + expect(gigService.findExpirable).toHaveBeenCalledTimes(1); + + resolveSweep(); + await flushPromises(); + worker.onModuleDestroy(); + }); +}); From b87b5ee43d250367452224123d667e31e7951143 Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:47:09 +0100 Subject: [PATCH 05/10] perf(repin): reconcile degraded pins with bounded concurrency (#237) `runOnce()` awaited `pinningService.reconcile(cid)` serially; each call makes per-provider network requests, so sweep time scaled with pin count x latency. Now bounded-concurrent (IPFS_REPIN_SWEEP_CONCURRENCY, default 8) with the per-CID try/catch kept inside the worker so one bad CID stays isolated, plus the same overlapping-sweep guard. Closes #237 --- .../src/ipfs-pinning/repin-worker.service.ts | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/backend/src/ipfs-pinning/repin-worker.service.ts b/backend/src/ipfs-pinning/repin-worker.service.ts index 25316d2..1edc4f7 100644 --- a/backend/src/ipfs-pinning/repin-worker.service.ts +++ b/backend/src/ipfs-pinning/repin-worker.service.ts @@ -2,8 +2,11 @@ import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/commo import { IpfsPinningService } from './ipfs-pinning.service'; import { DEFAULT_REPIN_INTERVAL_MS, PinStatus } from './ipfs-pinning.types'; import { DistributedLockService } from '../common/redis/distributed-lock.service'; +import { mapWithConcurrency } from '../common/concurrency'; const LOCK_KEY = 'lock:repin-sweep'; +/** How many CIDs to reconcile in parallel within one sweep (#237). */ +const SWEEP_CONCURRENCY = Number(process.env.IPFS_REPIN_SWEEP_CONCURRENCY) || 8; /** * Periodically sweeps every pin record that isn't fully healthy and re-reconciles it, @@ -24,6 +27,8 @@ export class RepinWorkerService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(RepinWorkerService.name); private timer?: NodeJS.Timeout; private currentLockToken?: string; + /** Guards against a slow sweep still running when the next tick fires. */ + private sweeping = false; constructor( private readonly pinningService: IpfsPinningService, @@ -54,14 +59,20 @@ export class RepinWorkerService implements OnModuleInit, OnModuleDestroy { } private async tick(intervalMs: number): Promise { + if (this.sweeping) { + this.logger.warn('Previous re-pin sweep still in flight — skipping this tick'); + return; + } const token = await this.lock.tryAcquire(LOCK_KEY, Math.ceil(intervalMs * 1.5)); if (!token) { return; // another instance is holding the lock for this tick } this.currentLockToken = token; + this.sweeping = true; try { await this.runOnce(); } finally { + this.sweeping = false; await this.lock.release(LOCK_KEY, token); this.currentLockToken = undefined; } @@ -73,7 +84,12 @@ export class RepinWorkerService implements OnModuleInit, OnModuleDestroy { .findAll() .filter(record => record.status === PinStatus.DEGRADED || record.status === PinStatus.FAILED); - for (const record of targets) { + // Reconcile CIDs with bounded concurrency rather than serially — each + // `reconcile()` makes per-provider network calls, so a sweep over many + // degraded pins scaled linearly with pin count x provider latency (#237). + // The per-CID try/catch is kept inside the worker so one bad CID is + // isolated and logged, exactly as before. + await mapWithConcurrency(targets, SWEEP_CONCURRENCY, async record => { try { await this.pinningService.reconcile(record.cid); } catch (error) { @@ -83,7 +99,7 @@ export class RepinWorkerService implements OnModuleInit, OnModuleDestroy { }`, ); } - } + }); } private getIntervalMs(): number { From 495bf8a02c85d1b0b24e81a561b1b81bf9b4c16c Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:47:19 +0100 Subject: [PATCH 06/10] test(repin): concurrent reconcile with per-CID error isolation Refs #237 --- .../ipfs-pinning/repin-worker.service.spec.ts | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/backend/src/ipfs-pinning/repin-worker.service.spec.ts b/backend/src/ipfs-pinning/repin-worker.service.spec.ts index 3413255..ef1b0d4 100644 --- a/backend/src/ipfs-pinning/repin-worker.service.spec.ts +++ b/backend/src/ipfs-pinning/repin-worker.service.spec.ts @@ -135,3 +135,28 @@ describe('RepinWorkerService', () => { }); }); }); + +describe('RepinWorkerService — concurrency & isolation (#237)', () => { + it('reconciles degraded pins concurrently and keeps per-CID error isolation', async () => { + const records = Array.from({ length: 8 }, (_, i) => makeRecord(`cid-${i}`, PinStatus.DEGRADED)); + const pinningService: jest.Mocked> = { + findAll: jest.fn().mockReturnValue(records), + reconcile: jest.fn().mockImplementation((cid: string) => + cid === 'cid-3' + ? Promise.reject(new Error('provider down')) + : new Promise(r => setTimeout(r, 20)), + ), + }; + const worker = new RepinWorkerService( + pinningService as unknown as IpfsPinningService, + fakeLock() as unknown as DistributedLockService, + ); + + const start = Date.now(); + await expect(worker.runOnce()).resolves.toBeUndefined(); // one bad CID must not throw + const elapsed = Date.now() - start; + + expect(pinningService.reconcile).toHaveBeenCalledTimes(8); + expect(elapsed).toBeLessThan(100); // sequential would be ~140ms + }); +}); From f8f3de791bee3be92c2be17b4429c923793e1a3a Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:47:30 +0100 Subject: [PATCH 07/10] perf(event-ingestion): process independent escrows in parallel, preserve per-escrow order (#238) New `processEventBatch()` (used by `ingestEvents` and `ingestSingleLedger`): phase 1 runs every id-less event (escrow_created, unknown types) strictly in original order so no later keyed event can observe a missing escrow; phase 2 groups the `topic[1]`-keyed events per escrow and runs the groups in parallel (EVENT_PROCESSING_CONCURRENCY, default 8), sequential within a group. Full concurrency-safety analysis is in the method's doc comment. Closes #238 --- .../event-ingestion.service.ts | 87 ++++++++++++++++--- 1 file changed, 75 insertions(+), 12 deletions(-) diff --git a/backend/src/event-ingestion/event-ingestion.service.ts b/backend/src/event-ingestion/event-ingestion.service.ts index fb3a6e5..adc788f 100644 --- a/backend/src/event-ingestion/event-ingestion.service.ts +++ b/backend/src/event-ingestion/event-ingestion.service.ts @@ -3,6 +3,10 @@ import { rpc as SorobanRpc } from '@stellar/stellar-sdk'; import { LedgerCursorService, LedgerCheckpoint } from './ledger-cursor.service'; import { EventProcessorService, SorobanEvent, ProcessedEvent } from './event-processor.service'; import { STELLAR_CONFIG } from '../stellar/stellar.config'; +import { mapWithConcurrency } from '../common/concurrency'; + +/** How many independent escrows to process in parallel per ingestion batch (#238). */ +const EVENT_PROCESSING_CONCURRENCY = Number(process.env.EVENT_PROCESSING_CONCURRENCY) || 8; @Injectable() export class EventIngestionService implements OnModuleInit, OnModuleDestroy { @@ -73,12 +77,7 @@ export class EventIngestionService implements OnModuleInit, OnModuleDestroy { this.logger.log(`Ingesting events from ledger ${startLedger} to ${endLedger}`); const events = await this.fetchEvents(contractId, startLedger, endLedger); - const processedEvents: ProcessedEvent[] = []; - - for (const event of events) { - const result = await this.eventProcessorService.processEvent(event); - processedEvents.push(result); - } + const processedEvents = await this.processEventBatch(events); const latestProcessedLedger = events.length > 0 ? events[events.length - 1].ledger : endLedger; const networkHash = await this.getNetworkHash(); @@ -95,12 +94,7 @@ export class EventIngestionService implements OnModuleInit, OnModuleDestroy { async ingestSingleLedger(contractId: string, ledger: number): Promise { const events = await this.fetchEvents(contractId, ledger, ledger); - const processedEvents: ProcessedEvent[] = []; - - for (const event of events) { - const result = await this.eventProcessorService.processEvent(event); - processedEvents.push(result); - } + const processedEvents = await this.processEventBatch(events); const networkHash = await this.getNetworkHash(); await this.ledgerCursorService.updateCursor( @@ -113,6 +107,75 @@ export class EventIngestionService implements OnModuleInit, OnModuleDestroy { return processedEvents; } + /** + * Process a fetched batch of events with per-escrow ordering preserved and + * independent escrows processed in parallel (#238). + * + * Concurrency-safety analysis: + * - `escrow_funded` / `escrow_released` / `escrow_disputed` carry the + * escrow id in `topic[1]`. Events with the *same* `topic[1]` MUST keep + * their relative order (e.g. funded before released), so they are grouped + * and each group runs sequentially. + * - `escrow_created` (and any unknown type) has no `topic[1]`. It cannot be + * correlated to a specific later keyed event, and a keyed event may + * depend on it (created must precede funded for the same escrow). So all + * id-less events run first, strictly in their original order, before any + * keyed group starts — no keyed handler can observe a missing escrow. + * - Groups for different escrow ids are independent (they touch different + * rows), so they run concurrently, bounded by `EVENT_PROCESSING_CONCURRENCY`. + * - `processEvent` swallows its own errors (returns `{ success: false }`), + * so per-event isolation is unchanged. + * + * The returned array is ordered id-less-first then by group; callers use it + * for counts/status only (the ledger cursor is advanced from `events`, not + * from this array). + */ + async processEventBatch(events: SorobanEvent[]): Promise { + const unkeyed: SorobanEvent[] = []; + const keyed = new Map(); + + for (const event of events) { + const key = event.topic[1]; + if (key === undefined || key === '') { + unkeyed.push(event); + continue; + } + let bucket = keyed.get(key); + if (!bucket) { + bucket = []; + keyed.set(key, bucket); + } + bucket.push(event); + } + + const processed: ProcessedEvent[] = []; + + // Phase 1 — id-less events (escrow_created, unknown), strict original order. + for (const event of unkeyed) { + processed.push(await this.eventProcessorService.processEvent(event)); + } + + // Phase 2 — one group per escrow id, groups in parallel, sequential within. + const groupResults = await mapWithConcurrency( + [...keyed.values()], + EVENT_PROCESSING_CONCURRENCY, + async group => { + const groupOut: ProcessedEvent[] = []; + for (const event of group) { + groupOut.push(await this.eventProcessorService.processEvent(event)); + } + return groupOut; + }, + ); + for (const result of groupResults) { + if (result.status === 'fulfilled') { + processed.push(...result.value); + } + } + + return processed; + } + private async fetchEvents( contractId: string, startLedger: number, From 4fb4d02c922b02c212028685df9d4362b2bbd284 Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:47:41 +0100 Subject: [PATCH 08/10] test(event-ingestion): same-escrow ordering preserved under parallelism Refs #238 --- .../event-ingestion.service.spec.ts | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/backend/src/event-ingestion/event-ingestion.service.spec.ts b/backend/src/event-ingestion/event-ingestion.service.spec.ts index 5a9894e..a559ad8 100644 --- a/backend/src/event-ingestion/event-ingestion.service.spec.ts +++ b/backend/src/event-ingestion/event-ingestion.service.spec.ts @@ -204,3 +204,64 @@ describe('EventIngestionService', () => { }); }); }); + +describe('EventIngestionService.processEventBatch — per-escrow ordering (#238)', () => { + const mkEvent = ( + id: string, + eventType: string, + escrowId?: string, + ): import('./event-processor.service').SorobanEvent => ({ + id, + ledger: 1, + contractId: 'C1', + eventType, + topic: escrowId ? [eventType, escrowId] : [eventType], + value: {}, + xdr: '', + createdAt: new Date(), + }); + + it('processes same-escrow events in order and independent escrows in parallel', async () => { + const order: string[] = []; + jest + .spyOn(eventProcessorService, 'processEvent') + .mockImplementation(async event => { + order.push(event.id); + // Make the FIRST call slow so a naive parallel-all would reorder. + if (event.id === 'A-fund') await new Promise(r => setTimeout(r, 30)); + return { eventId: event.id, ledger: 1, success: true, processedAt: new Date() }; + }); + + const events = [ + mkEvent('A-create', 'escrow_created'), + mkEvent('B-create', 'escrow_created'), + mkEvent('A-fund', 'escrow_funded', 'A'), + mkEvent('A-release', 'escrow_released', 'A'), + mkEvent('B-fund', 'escrow_funded', 'B'), + ]; + + const results = await service.processEventBatch(events); + + expect(results).toHaveLength(5); + // Phase 1: both creates ran first, in order. + expect(order.slice(0, 2)).toEqual(['A-create', 'B-create']); + // Within escrow A, fund strictly precedes release despite fund being slow. + expect(order.indexOf('A-fund')).toBeLessThan(order.indexOf('A-release')); + // Escrow B was not blocked behind slow escrow A. + expect(order.indexOf('B-fund')).toBeLessThan(order.indexOf('A-release')); + }); + + it('does not throw when an individual event fails', async () => { + jest.spyOn(eventProcessorService, 'processEvent').mockImplementation(async event => ({ + eventId: event.id, + ledger: 1, + success: event.id !== 'bad', + processedAt: new Date(), + })); + const results = await service.processEventBatch([ + mkEvent('ok', 'escrow_funded', 'X'), + mkEvent('bad', 'escrow_funded', 'Y'), + ]); + expect(results.map(r => r.success).sort()).toEqual([false, true]); + }); +}); From ea91a92e8d9ae4c12be7765c56161d8d89168238 Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:47:55 +0100 Subject: [PATCH 09/10] fix(webhook): use the shared retry helper; fix isRetryable's 5xx match (#239) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `retry.helper.ts` was dead code — `WebhookService.sendWithRetry` re- implemented the loop inline. It now calls `withRetry(...)`, and `isRetryable` no longer classifies any message containing the digit 5 as retryable: it matches transient network errors and a real HTTP 5xx status (the bare 3-digit message `send()` throws, or `HTTP 5xx`). `withRetry` gained a `shouldRetry` predicate so a 4xx stops immediately. Closes #239 --- backend/src/webhook/retry.helper.ts | 47 ++++++++++++++++++++------ backend/src/webhook/webhook.service.ts | 30 ++++++++-------- 2 files changed, 51 insertions(+), 26 deletions(-) diff --git a/backend/src/webhook/retry.helper.ts b/backend/src/webhook/retry.helper.ts index 211f844..f828936 100644 --- a/backend/src/webhook/retry.helper.ts +++ b/backend/src/webhook/retry.helper.ts @@ -1,24 +1,51 @@ +/** + * Shared retry-with-backoff for webhook delivery. `WebhookService.sendWithRetry` + * uses this rather than re-implementing the loop inline (#239). + */ export async function withRetry( fn: () => Promise, maxAttempts: number, baseDelayMs: number, + shouldRetry: (error: unknown) => boolean = () => true, ): Promise { - let lastError: Error | undefined; + let lastError: unknown; for (let attempt = 1; attempt <= maxAttempts; attempt++) { try { return await fn(); } catch (err) { - lastError = err instanceof Error ? err : new Error(String(err)); - if (attempt < maxAttempts) await new Promise(r => setTimeout(r, baseDelayMs * attempt)); + lastError = err; + if (!shouldRetry(err)) break; + if (attempt < maxAttempts) { + await new Promise(r => setTimeout(r, baseDelayMs * attempt)); + } } } - throw lastError ?? new Error('All retries exhausted'); + throw lastError instanceof Error + ? lastError + : new Error(String(lastError ?? 'All retries exhausted')); } -export function isRetryable(error: Error): boolean { - return ( - error.message.includes('ECONNREFUSED') || - error.message.includes('ETIMEDOUT') || - error.message.includes('5') - ); +/** + * Whether an error from a webhook delivery attempt is worth retrying. + * + * Retries transient network failures and real HTTP 5xx responses only. The + * previous implementation used `error.message.includes('5')`, which matched + * the digit 5 anywhere in the message — a URL, a byte count, a port number — + * so a 4xx or an unrelated failure could be retried indefinitely (#239). + * + * `WebhookService.send()` rejects with `new Error(String(statusCode))`, so a + * bare 3-digit message *is* the HTTP status; `"HTTP 502 ..."` is matched too. + */ +export function isRetryable(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error ?? ''); + + if ( + /ECONNREFUSED|ECONNRESET|ETIMEDOUT|EAI_AGAIN|ENOTFOUND|EPIPE|socket hang up/i.test(message) + ) { + return true; + } + + const status = + message.trim().match(/^(\d{3})\b/)?.[1] ?? message.match(/\bHTTP\s*(\d{3})\b/i)?.[1]; + return status !== undefined && /^5\d{2}$/.test(status); } diff --git a/backend/src/webhook/webhook.service.ts b/backend/src/webhook/webhook.service.ts index 0c39b89..6ed54de 100644 --- a/backend/src/webhook/webhook.service.ts +++ b/backend/src/webhook/webhook.service.ts @@ -1,6 +1,10 @@ import { Injectable } from '@nestjs/common'; import * as https from 'https'; import * as http from 'http'; +import { withRetry, isRetryable } from './retry.helper'; + +/** Base backoff between webhook delivery attempts; grows linearly per attempt. */ +const WEBHOOK_RETRY_BASE_DELAY_MS = 1000; interface WebhookPayload { event: string; @@ -31,22 +35,16 @@ export class WebhookService { if (failed) throw failed.reason; } - private async sendWithRetry( - url: string, - payload: WebhookPayload, - retries: number, - ): Promise { - let lastError: unknown; - for (let i = 0; i < retries; i++) { - try { - await this.send(url, payload); - return; - } catch (error) { - lastError = error; - if (i < retries - 1) await new Promise(r => setTimeout(r, 1000 * (i + 1))); - } - } - throw lastError instanceof Error ? lastError : new Error(String(lastError)); + private sendWithRetry(url: string, payload: WebhookPayload, retries: number): Promise { + // Shared retry/backoff + retryability logic — no longer a second inline + // copy of what `retry.helper.ts` already provides, and non-retryable + // failures (4xx, unrelated errors) now stop immediately (#239). + return withRetry( + () => this.send(url, payload), + retries, + WEBHOOK_RETRY_BASE_DELAY_MS, + isRetryable, + ); } private send(url: string, payload: WebhookPayload): Promise { From ff90886d5c74ab4fe38f836a41b7b0277d209f68 Mon Sep 17 00:00:00 2001 From: stephanieoghenemega-eng <288002234+stephanieoghenemega-eng@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:48:09 +0100 Subject: [PATCH 10/10] test(webhook): retry.helper.spec.ts for the fixed retryability logic Refs #239 --- backend/src/webhook/retry.helper.spec.ts | 64 ++++++++++++++++++++++++ 1 file changed, 64 insertions(+) create mode 100644 backend/src/webhook/retry.helper.spec.ts diff --git a/backend/src/webhook/retry.helper.spec.ts b/backend/src/webhook/retry.helper.spec.ts new file mode 100644 index 0000000..71cc1fb --- /dev/null +++ b/backend/src/webhook/retry.helper.spec.ts @@ -0,0 +1,64 @@ +import { withRetry, isRetryable } from './retry.helper'; + +describe('isRetryable (#239)', () => { + it('retries transient network errors', () => { + for (const code of ['ECONNREFUSED', 'ECONNRESET', 'ETIMEDOUT', 'ENOTFOUND', 'EAI_AGAIN']) { + expect(isRetryable(new Error(`connect ${code} 127.0.0.1:443`))).toBe(true); + } + expect(isRetryable(new Error('socket hang up'))).toBe(true); + }); + + it('retries real HTTP 5xx responses', () => { + expect(isRetryable(new Error('500'))).toBe(true); + expect(isRetryable(new Error('503'))).toBe(true); + expect(isRetryable(new Error('HTTP 502 Bad Gateway'))).toBe(true); + }); + + it('does not retry 4xx responses', () => { + expect(isRetryable(new Error('400'))).toBe(false); + expect(isRetryable(new Error('404'))).toBe(false); + expect(isRetryable(new Error('429'))).toBe(false); + }); + + it('does not retry a message that merely contains the digit 5', () => { + expect(isRetryable(new Error('POST https://hooks.example.com/v5/deliver failed'))).toBe(false); + expect(isRetryable(new Error('payload was 512 bytes'))).toBe(false); + expect(isRetryable(new Error('connection to port 5000 was refused'))).toBe(false); + expect(isRetryable(new Error('unexpected token at position 5'))).toBe(false); + }); + + it('handles non-Error values without throwing', () => { + expect(isRetryable('503')).toBe(true); + expect(isRetryable(undefined)).toBe(false); + expect(isRetryable({ weird: true })).toBe(false); + }); +}); + +describe('withRetry (#239)', () => { + it('resolves on the first success without retrying', async () => { + const fn = jest.fn().mockResolvedValue('ok'); + await expect(withRetry(fn, 3, 1)).resolves.toBe('ok'); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('retries up to maxAttempts then throws the last error', async () => { + const fn = jest.fn().mockRejectedValue(new Error('503')); + await expect(withRetry(fn, 3, 1, isRetryable)).rejects.toThrow('503'); + expect(fn).toHaveBeenCalledTimes(3); + }); + + it('stops immediately when shouldRetry returns false', async () => { + const fn = jest.fn().mockRejectedValue(new Error('404')); + await expect(withRetry(fn, 3, 1, isRetryable)).rejects.toThrow('404'); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('recovers if a later attempt succeeds', async () => { + const fn = jest + .fn() + .mockRejectedValueOnce(new Error('ETIMEDOUT')) + .mockResolvedValueOnce('recovered'); + await expect(withRetry(fn, 3, 1, isRetryable)).resolves.toBe('recovered'); + expect(fn).toHaveBeenCalledTimes(2); + }); +});