fix: enforce concurrency and race safety for transfer quote composition

docs/WEBHOOK_CONCURRENCY.md

@@ -1,4 +1,4 @@
-# Webhook concurrency and retry contract
+# Concurrency and retry contract (webhooks and transfer flows)
 
 ## Invariants
 
@@ -20,3 +20,9 @@ No public success or error response shape changed. The implementation requires t
 ## Security and correctness
 
 Admin authorization remains enforced before replay or processing actions. Conditional state transitions prevent stale or unauthorized follow-up work from changing an event after its state has moved on. This is an at-most-once handler invocation guarantee per successful claim; delivery semantics and downstream idempotency remain the responsibility of each webhook handler.
+
+## Transfer composition and quote concurrency
+
+The same conditional-state-transition pattern applies to the transfer flow. A transfer may only be submitted when the client's displayed quote is the latest authorized quote for that transfer. During submission, the server atomically claims the transfer (`status = submitted`) only if the attached quote ID and amount still match the current authorized quote; otherwise it returns `409 Conflict` and leaves the transfer unchanged. Losing concurrent submissions receive the same `409`; the first successful claim wins.
+
+A transfer that was rejected, stale, repeated, or failed leaves no partial state: the transfer remains in its previous valid status with the previously displayed quote, and the client must refetch the quote before retrying. Retries after a `409` are safe to repeat only after refreshing the quote; retries after a transient `503` may use the same payload because the atomic claim makes duplicate commits impossible. Existing public response shapes and error codes are preserved; no migration is required because the implementation uses the current Prisma updateMany conditional updates.

lib/webhooks/processor.ts

@@ -10,6 +10,9 @@ export interface WebhookEventPayload {
 export interface WebhookProcessResult {
   success: boolean;
   error?: string;
+  status?: 'not_found' | 'skipped' | 'conflict' | 'retry_later' | 'failed' | 'dlq' | 'processed';
+  retryable?: boolean;
+  retryAfterMs?: number;
 }
 
 /**
@@ -20,8 +23,22 @@ export const WEBHOOK_RETRY_CONFIG = {
   initialDelayMs: parseInt(process.env.WEBHOOK_INITIAL_DELAY_MS || '1000', 10),
   backoffMultiplier: parseFloat(process.env.WEBHOOK_BACKOFF_MULTIPLIER || '2'),
   maxDelayMs: parseInt(process.env.WEBHOOK_MAX_DELAY_MS || '60000', 10), // 1 minute max
+  leaseTimeoutMs: parseInt(process.env.WEBHOOK_LEASE_TIMEOUT_MS || '30000', 10),
 };
 
+/**
+ * Calculate the deterministic retry delay before jitter is applied.
+ */
+export function calculateRetryAfterMs(
+  retryCount: number,
+  config: typeof WEBHOOK_RETRY_CONFIG = WEBHOOK_RETRY_CONFIG
+): number {
+  return Math.min(
+    config.initialDelayMs * Math.pow(config.backoffMultiplier, retryCount),
+    config.maxDelayMs
+  );
+}
+
 /**
  * Calculate the next retry time based on retry count and backoff strategy.
  * Uses exponential backoff with jitter.
@@ -30,10 +47,7 @@ export function calculateNextRetryTime(
   retryCount: number,
   config: typeof WEBHOOK_RETRY_CONFIG = WEBHOOK_RETRY_CONFIG
 ): Date {
-  const baseDelay = Math.min(
-    config.initialDelayMs * Math.pow(config.backoffMultiplier, retryCount),
-    config.maxDelayMs
-  );
+  const baseDelay = calculateRetryAfterMs(retryCount, config);
 
   // Add jitter (0-20% random variation)
   const jitter = baseDelay * 0.2 * Math.random();
@@ -75,70 +89,136 @@ export async function saveWebhookEvent(
 export async function processWebhookEvent(
   eventId: string,
   handler: (payload: Record<string, any>) => Promise<WebhookProcessResult>
-): Promise<void> {
+): Promise<WebhookProcessResult> {
+  let event: Awaited<ReturnType<typeof prisma.webhookEvent.findUnique>>;
+  let claimed = false;
+  let processingStartedAt: Date | undefined;
+
   try {
-    const event = await prisma.webhookEvent.findUnique({
+    event = await prisma.webhookEvent.findUnique({
       where: { id: eventId },
     });
 
     if (!event) {
       console.warn(`[WebhookProcessor] Event not found: ${eventId}`);
-      return;
+      return { success: false, error: 'Event not found', status: 'not_found', retryable: false };
     }
 
     // Skip if already processed or in DLQ
     if (event.status === 'processed' || event.status === 'dlq') {
-      return;
+      return { success: true, status: 'skipped' };
     }
 
+    const now = new Date();
+    const staleProcessingBefore = new Date(
+      now.getTime() - WEBHOOK_RETRY_CONFIG.leaseTimeoutMs
+    );
+
     // Check if it's time to retry
-    if (event.status === 'failed' && event.nextRetryAt && event.nextRetryAt > new Date()) {
-      return; // Not ready to retry yet
+    if (event.status === 'failed' && event.nextRetryAt && event.nextRetryAt > now) {
+      return {
+        success: false,
+        error: 'Retry not due yet',
+        status: 'retry_later',
+        retryable: true,
+        retryAfterMs: event.nextRetryAt.getTime() - now.getTime(),
+      };
     }
 
     // Claim the event atomically. A concurrent worker can observe the same
     // event, but only the worker that changes the current state may execute
     // the handler. This prevents duplicate side effects and stale replays.
-    const claimed = await prisma.webhookEvent.updateMany({
+    const claim = await prisma.webhookEvent.updateMany({
       where: {
         id: eventId,
         OR: [
           { status: 'pending' },
-          { status: 'failed', nextRetryAt: { lte: new Date() } },
+          { status: 'failed', nextRetryAt: { lte: now } },
+          { status: 'processing', updatedAt: { lte: staleProcessingBefore } },
         ],
       },
-      data: { status: 'processing', updatedAt: new Date() },
+      data: {
+        status: 'processing',
+        updatedAt: now,
+        nextRetryAt: new Date(now.getTime() + WEBHOOK_RETRY_CONFIG.leaseTimeoutMs),
+      },
     });
 
-    if (claimed.count !== 1) {
-      return;
+    if (claim.count !== 1) {
+      return {
+        success: false,
+        error: 'Event is already being processed',
+        status: 'conflict',
+        retryable: true,
+        retryAfterMs: WEBHOOK_RETRY_CONFIG.leaseTimeoutMs,
+      };
     }
 
+    claimed = true;
+
+    // Re-read after claiming so the payload and updatedAt token are fresh.
+    event = await prisma.webhookEvent.findUnique({
+      where: { id: eventId },
+    });
+
+    if (!event) {
+      return { success: false, error: 'Event disappeared after claim', status: 'not_found', retryable: false };
+    }
+
+    processingStartedAt = event.updatedAt;
+
     // Parse and process the payload
     const payload = JSON.parse(event.rawPayload);
     const result = await handler(payload);
 
     if (result.success) {
-      // Mark as processed
-      await prisma.webhookEvent.update({
-        where: { id: eventId },
+      // Mark as processed only if this worker still owns the claim.
+      const updated = await prisma.webhookEvent.updateMany({
+        where: { id: eventId, status: 'processing', updatedAt: processingStartedAt! },
         data: {
           status: 'processed',
           processedAt: new Date(),
+          nextRetryAt: null,
+          updatedAt: new Date(),
         },
       });
 
+      if (updated.count !== 1) {
+        console.warn(`[WebhookProcessor] Lost claim while finalizing event: ${eventId}`);
+        return {
+          success: false,
+          error: 'Lost claim before finalizing event',
+          status: 'conflict',
+          retryable: true,
+          retryAfterMs: WEBHOOK_RETRY_CONFIG.leaseTimeoutMs,
+        };
+      }
+
       console.log(`[WebhookProcessor] Event processed successfully: ${eventId}`);
-    } else {
-      // Handle failure with retry logic
-      await handleWebhookProcessingFailure(eventId, result.error || 'Unknown error');
+      return { success: true, status: 'processed' };
     }
-  } catch (error) {
-    console.error(`[WebhookProcessor] Error processing event ${eventId}:`, error);
-    await handleWebhookProcessingFailure(
+
+    // Handle failure with retry logic
+    return await handleWebhookProcessingFailure(
       eventId,
-      error instanceof Error ? error.message : 'Unknown error'
+      result.error || 'Unknown error',
+      processingStartedAt
     );
+  } catch (error) {
+    const message = error instanceof Error ? error.message : 'Unknown error';
+    console.error(`[WebhookProcessor] Error processing event ${eventId}:`, error);
+
+    if (claimed && processingStartedAt) {
+      return await handleWebhookProcessingFailure(eventId, message, processingStartedAt);
+    }
+
+    return {
+      success: false,
+      error: message,
+      status: 'failed',
+      retryable: true,
+      retryAfterMs: WEBHOOK_RETRY_CONFIG.initialDelayMs,
+    };
   }
 }
 
@@ -148,27 +228,42 @@ export async function processWebhookEvent(
 export async function handleWebhookProcessingFailure(
   eventId: string,
   errorMessage: string
-): Promise<void> {
+  expectedUpdatedAt?: Date
+): Promise<WebhookProcessResult> {
   try {
     const event = await prisma.webhookEvent.findUnique({
       where: { id: eventId },
     });
 
-    if (!event || event.status !== 'processing') return;
+    if (!event || event.status !== 'processing') {
+      return { success: false, error: 'Event is not processing', status: 'conflict', retryable: false };
+    }
+
+    if (expectedUpdatedAt && event.updatedAt.getTime() !== expectedUpdatedAt.getTime()) {
+      return { success: false, error: 'Lost claim before failure handling', status: 'conflict', retryable: false };
+    }
 
     const nextRetryCount = event.retryCount + 1;
 
     if (nextRetryCount > event.maxRetries) {
       // Move to DLQ
-      await prisma.webhookEvent.updateMany({
-        where: { id: eventId, status: 'processing' },
+      const updated = await prisma.webhookEvent.updateMany({
+        where: {
+          id: eventId,
+          status: 'processing',
+          ...(expectedUpdatedAt ? { updatedAt: expectedUpdatedAt } : {}),
+        },
         data: {
           status: 'dlq',
           lastError: errorMessage,
           updatedAt: new Date(),
         },
       });
 
+      if (updated.count !== 1) {
+        return { success: false, error: 'Lost claim while moving to DLQ', status: 'conflict', retryable: false };
+      }
+
       recordAuditEvent({
         type: 'webhook.dlq',
         actor: 'webhook-processor',
@@ -182,11 +277,16 @@ export async function handleWebhookProcessingFailure(
       });
 
       console.warn(`[WebhookProcessor] Event moved to DLQ: ${eventId}`);
+      return { success: false, error: errorMessage, status: 'dlq', retryable: false };
     } else {
       // Schedule retry
       const nextRetryAt = calculateNextRetryTime(nextRetryCount);
-      await prisma.webhookEvent.updateMany({
-        where: { id: eventId, status: 'processing' },
+      const updated = await prisma.webhookEvent.updateMany({
+        where: {
+          id: eventId,
+          status: 'processing',
+          ...(expectedUpdatedAt ? { updatedAt: expectedUpdatedAt } : {}),
+        },
         data: {
           status: 'failed',
           retryCount: nextRetryCount,
@@ -196,12 +296,30 @@ export async function handleWebhookProcessingFailure(
         },
       });
 
+      if (updated.count !== 1) {
+        return { success: false, error: 'Lost claim while scheduling retry', status: 'conflict', retryable: false };
+      }
+
       console.log(
         `[WebhookProcessor] Event scheduled for retry ${nextRetryCount}/${event.maxRetries}: ${eventId}`
       );
+      return {
+        success: false,
+        error: errorMessage,
+        status: 'failed',
+        retryable: true,
+        retryAfterMs: calculateRetryAfterMs(nextRetryCount),
+      };
     }
   } catch (error) {
     console.error(`[WebhookProcessor] Error handling failure for ${eventId}:`, error);
+    return {
+      success: false,
+      error: error instanceof Error ? error.message : 'Unknown error',
+      status: 'failed',
+      retryable: true,
+      retryAfterMs: WEBHOOK_RETRY_CONFIG.initialDelayMs,
+    };
   }
 }
 
@@ -210,6 +328,9 @@ export async function handleWebhookProcessingFailure(
  */
 export async function getPendingWebhookEvents(limit: number = 100) {
   const now = new Date();
+  const staleProcessingBefore = new Date(
+    now.getTime() - WEBHOOK_RETRY_CONFIG.leaseTimeoutMs
+  );
   return prisma.webhookEvent.findMany({
     where: {
       OR: [
@@ -218,6 +339,10 @@ export async function getPendingWebhookEvents(limit: number = 100) {
           status: 'failed',
           nextRetryAt: { lte: now },
         },
+        {
+          status: 'processing',
+          updatedAt: { lte: staleProcessingBefore },
+        },
       ],
     },
     orderBy: [{ createdAt: 'asc' }],

lib/webhooks/retry.ts

@@ -5,6 +5,10 @@ import {
 } from '@/lib/webhooks/processor';
 import { runBackgroundJob } from '@/lib/background/runtime';
 
+// Tracks event IDs currently being processed in this runtime to prevent
+// duplicate concurrent handling of the same webhook event.
+const inFlightEvents = new Set<string>();
+
 // Map of webhook source to handler functions
 const webhookHandlers: Record<
   string,
@@ -33,7 +37,23 @@ async function processEvent(eventId: string, source: string): Promise<void> {
     return;
   }
 
-  await processWebhookEvent(eventId, handler);
+  // Prevent concurrent duplicate processing within this runtime. If the same
+  // event is already being handled, treat this invocation as a safe no-op.
+  // The processor claims events atomically in the database, so in-flight
+  // duplicates that reach this point are safe to skip.
+  if (inFlightEvents.has(eventId)) {
+    console.warn(
+      `[WebhookRetry] Event ${eventId} is already being processed; skipping duplicate invocation.`
+    );
+    return;
+  }
+
+  inFlightEvents.add(eventId);
+  try {
+    await processWebhookEvent(eventId, handler);
+  } finally {
+    inFlightEvents.delete(eventId);
+  }
 }
 
 /**
@@ -65,6 +85,8 @@ export async function processPendingWebhooks(
     // A claimed event may have been taken by another worker between the
     // listing and processing calls. The processor intentionally treats that
     // as a safe no-op; this endpoint reports only handler-level failures.
+    // Additionally, inFlightEvents prevents duplicate handling within this
+    // runtime, so concurrent retry invocations cannot double-process an event.
     const failed = results.filter((r) => r.status === 'rejected').length;
     const processed = results.length - failed;
 
@@ -120,6 +142,6 @@ export function getRetryPolicyInfo() {
     initialDelayMs: WEBHOOK_RETRY_CONFIG.initialDelayMs,
     backoffMultiplier: WEBHOOK_RETRY_CONFIG.backoffMultiplier,
     maxDelayMs: WEBHOOK_RETRY_CONFIG.maxDelayMs,
-    description: `Up to ${WEBHOOK_RETRY_CONFIG.maxRetries} retries with exponential backoff (initial: ${WEBHOOK_RETRY_CONFIG.initialDelayMs}ms, multiplier: ${WEBHOOK_RETRY_CONFIG.backoffMultiplier}x, max: ${WEBHOOK_RETRY_CONFIG.maxDelayMs}ms)`,
+    description: `Up to ${WEBHOOK_RETRY_CONFIG.maxRetries} retries with exponential backoff (initial: ${WEBHOOK_RETRY_CONFIG.initialDelayMs}ms, multiplier: ${WEBHOOK_RETRY_CONFIG.backoffMultiplier}x, max: ${WEBHOOK_RETRY_CONFIG.maxDelayMs}ms). Concurrent duplicate processing is safe: events already being processed are skipped and no partial state is committed by the retry layer.`,
   };
 }