Bounded concurrency for the gig/repin/event sweeps + fix the dead webhook retry helper (#236–#239)

backend/src/common/concurrency/index.ts

@@ -0,0 +1 @@
+export * from './map-with-concurrency';

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<void> => 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]);
+  });
+});

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<T, R>(
+  items: readonly T[],
+  concurrency: number,
+  worker: (item: T, index: number) => Promise<R>,
+): Promise<PromiseSettledResult<R>[]> {
+  const limit = Math.max(1, Math.floor(concurrency) || 1);
+  const results: PromiseSettledResult<R>[] = new Array(items.length);
+  let cursor = 0;
+
+  const runner = async (): Promise<void> => {
+    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<unknown>[]): number {
+  return results.reduce((n, r) => (r.status === 'rejected' ? n + 1 : n), 0);
+}

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]);
+  });
+});

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<ProcessedEvent[]> {
     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<ProcessedEvent[]> {
+    const unkeyed: SorobanEvent[] = [];
+    const keyed = new Map<string, SorobanEvent[]>();
+
+    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,

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<Pick<GigService, 'findExpirable' | 'expire'>> {
+    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<Pick<GigService, 'findExpirable' | 'expire'>> = {
+      findExpirable: jest.fn().mockReturnValue(new Promise<never>(() => {})),
+      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();
+  });
+});