diff --git a/docs/WEBHOOK_CONCURRENCY.md b/docs/WEBHOOK_CONCURRENCY.md index 94f31276..cb0cc5ce 100644 --- a/docs/WEBHOOK_CONCURRENCY.md +++ b/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. diff --git a/lib/webhooks/processor.ts b/lib/webhooks/processor.ts index e3bf2b8d..8561772a 100644 --- a/lib/webhooks/processor.ts +++ b/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) => Promise -): Promise { +): Promise { + let event: Awaited>; + 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,20 +228,31 @@ export async function processWebhookEvent( export async function handleWebhookProcessingFailure( eventId: string, errorMessage: string -): Promise { + expectedUpdatedAt?: Date +): Promise { 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, @@ -169,6 +260,10 @@ export async function handleWebhookProcessingFailure( }, }); + 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' }], diff --git a/lib/webhooks/retry.ts b/lib/webhooks/retry.ts index 20cfca4c..f6da0576 100644 --- a/lib/webhooks/retry.ts +++ b/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(); + // Map of webhook source to handler functions const webhookHandlers: Record< string, @@ -33,7 +37,23 @@ async function processEvent(eventId: string, source: string): Promise { 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.`, }; } diff --git a/meridian-api/src/auth/providers/idempotency.provider.ts b/meridian-api/src/auth/providers/idempotency.provider.ts index 2ae122d8..d29a96de 100644 --- a/meridian-api/src/auth/providers/idempotency.provider.ts +++ b/meridian-api/src/auth/providers/idempotency.provider.ts @@ -1,27 +1,3 @@ -/** - * IdempotencyProvider - * - * Provides a deterministic idempotency boundary for authentication and recovery - * operations. Each operation must be supplied with a durable request key (nonce). - * Concurrent or retried requests with the same key share a single execution and - * return the same stored result. Reuse of a key with a different request is - * rejected with an `IdempotencyConflictError`. - * - * Design invariants: - * - One committed effect per key: the provider serializes executions per key and - * persists the final state in the store. - * - Deterministic success: after a completed operation, retries return the stored - * result without re-executing the business operation. - * - Retriable failures: a failed operation records the error and permits a retry - * with the same key, re-running the business operation once. - * - Conflict detection: the request body is hashed and bound to the key. If the - * hash differs, the caller receives a conflict and no state is modified. - * - Stale state: expired records are removed and treated as absent. - * - * The default `InMemoryIdempotencyStore` is suitable for tests and single-process - * deployments. For distributed deployments, supply an `IdempotencyStore` backed - * by a durable, atomic storage system. - */ import { createHash } from 'crypto'; export interface IdempotencyRecord { @@ -33,12 +9,15 @@ export interface IdempotencyRecord { createdAt: number; updatedAt: number; expiresAt: number; + version?: number; } export interface IdempotencyStore { get(key: string): Promise | undefined>; put(record: IdempotencyRecord): Promise; delete(key: string): Promise; + create?(record: IdempotencyRecord); Promise | undefined>; + update?(key: string, expectedVersion: number, record: IdempotencyRecord): Promise | null | undefined>; } export class InMemoryIdempotencyStore implements IdempotencyStore { @@ -55,6 +34,27 @@ export class InMemoryIdempotencyStore implements IdempotencyStore { async delete(key: string): Promise { this.records.delete(key); } + + async create(record: IdempotencyRecord): Promise | undefined> { + const key = record.key; + if (this.records.has(key)) { + return this.records.get(key) as IdempotencyRecord; + } + this.records.set(key, record as IdempotencyRecord); + return undefined; + } + + async update( + key: string, + expectedVersion: number, + record: IdempotencyRecord + ): Promise | null | undefined> { + const current = this.records.get(key) as IdempotencyRecord | undefined; + if (!current) return null; + if ((current.version ?? 0) !== expectedVersion) return current; + this.records.set(key, record as IdempotencyRecord); + return undefined; + } } export class IdempotencyConflictError extends Error { @@ -64,6 +64,15 @@ export class IdempotencyConflictError extends Error { } } +export class IdempotencyInProgressError extends Error { + readonly retryAfterMs: number; + constructor(key: string, retryAfterMs: number) { + super(`Idempotency key "${key}" is already being processed; retry after ${retryAfterMs}ms`); + this.name = 'IdempotencyInProgressError'; + this.retryAfterMs = retryAfterMs; + } +} + export interface IdempotencyProviderOptions { store?: IdempotencyStore; ttlMs?: number; @@ -94,16 +103,6 @@ export class IdempotencyProvider { this.ttlMs = options.ttlMs ?? 15 * 60 * 1000; } - /** - * Executes `operation` exactly once for a given idempotency `key`. - * - * @param key - Durable request key or nonce. - * @param request - The request payload. Its hash is bound to the key to reject - * conflicting reuse. - * @param operation - The business operation to perform. - * @returns The result of the operation (the same value for safe retries). - * @throws {IdempotencyConflictError} If the key is reused with a different payload. - */ async execute( key: string, request: unknown, @@ -116,73 +115,140 @@ export class IdempotencyProvider { const requestHash = this.hash(request); return this.getMutex(key).runExclusive(async () => { - const now = Date.now(); - const existing = await this.store.get(key); + let attempt = 0; + const maxAttempts = 10; - if (existing) { - if (existing.requestHash !== requestHash) { - throw new IdempotencyConflictError(key); + while (true) { + if (attempt++ >= maxAttempts) { + throw new IdempotencyInProgressError(key, this.ttlMs); } - if (existing.expiresAt < now) { + + const now = Date.now(); + let current = await this.store.get(key); + + if (current && current.expiresAt <= now) { await this.store.delete(key); - } else if (existing.status === 'completed') { - return existing.result as T; - } else if (existing.status === 'failed') { - // Retryable failure: reset the record and re-start the operation. - existing.status = 'in_progress'; - existing.error = undefined; - existing.updatedAt = now; - await this.store.put(existing); + current = undefined; } - } - const record: IdempotencyRecord = existing ?? { - key, - requestHash, - status: 'in_progress', - createdAt: now, - updatedAt: now, - expiresAt: now + this.ttlMs, - }; - if (!existing) { - await this.store.put(record); - } + if (!current) { + const newRecord: IdempotencyRecord = { + key, + requestHash, + status: 'in_progress', + createdAt: now, + updatedAt: now, + expiresAt: now + this.ttlMs, + version: 1, + }; - try { - const result = await operation(); - const completed: IdempotencyRecord = { - ...record, - status: 'completed', - result, - error: undefined, - updatedAt: Date.now(), - expiresAt: Date.now() + this.ttlMs, - }; - await this.store.put(completed); - return result; - } catch (error) { - const failed: IdempotencyRecord = { - ...record, - status: 'failed', - error, - updatedAt: Date.now(), - expiresAt: Date.now() + this.ttlMs, - }; - await this.store.put(failed); - throw error; + if (this.store.create) { + const existing = await this.store.create(newRecord); + if (!existing) { + return this.runOperation(newRecord, operation, newRecord.version!); + } + current = existing; + } else { + await this.store.put(newRecord); + return this.runOperation(newRecord, operation, newRecord.version!); + } + } + + if (current.requestHash !== requestHash) { + throw new IdempotencyConflictErrow(key); + } + + if (current.status === 'completed') { + return current.result as T; + } + + if (current.status === 'failed') { + const retryRecord: IdempotencyRecord = { + ...current, + status: 'in_progress', + error: undefined, + updatedAt: now, + expiresAt: now + this.ttlMs, + version: (current.version ?? 0) + 1, + }; + + if (this.store.update) { + const conflict = await this.store.update(key, current.version ?? 0, retryRecord); + if (conflict !== undefined) { + current = conflict ?? undefined; + continue; + } + return this.runOperation(retryRecord, operation, retryRecord.version!); + } else { + await this.store.put(retryRecord); + return this.runOperation(retryRecord, operation, retryRecord.version!); + } + } + + const retryAfterMs = Math.max(1, current.expiresAt - Date.now()); + throw new IdempotencyInProgressError(key, retryAfterMs); } }); } - /** - * Removes an idempotency record (e.g. post-logout or post-recovery). - */ async clear(key: string): Promise { await this.getMutex(key).runExclusive(async () => { await this.store.delete(key); }); } + private async runOperation( + record: IdempotencyRecord, + operation: () => Promise, + expectedVersion: number, + ): Promise { + try { + const result = await operation(); + const completed: IdempotencyRecord = { + ...record, + status: 'completed', + result, + error: undefined, + updatedAt: Date.now(), + expiresAt: Date.now() + this.ttlMs, + version: expectedVersion + 1, + }; + + if (this.store.update) { + const conflict = await this.store.update(record.key, expectedVersion, completed); + if (conflict !== undefined) { + throw new IdempotencyConflictError(record.key); + } + } else { + await this.store.put(completed); + } + return result; + } catch (error) { + if (error instanceof IdempotencyConflictError) { + throw error; + } + + const failed: IdempotencyRecord = { + ...record, + status: 'failed', + error, + updatedAt: Date.now(), + expiresAt: Date.now() + this.ttlMs, + version: expectedVersion + 1, + }; + + if (this.store.update) { + const conflict = await this.store.update(record.key, expectedVersion, failed); + if (conflict !== undefined) { + throw new IdempotencyConflictError(record.key); + } + } else { + await this.store.put(failed); + } + throw error; + } + } + private getMutex(key: string): Mutex { let mutex = this.mutexes.get(key); if (!mutex) { diff --git a/meridian-api/src/common/providers/database-transaction.provider.ts b/meridian-api/src/common/providers/database-transaction.provider.ts index 96f4befd..bd97dde1 100644 --- a/meridian-api/src/common/providers/database-transaction.provider.ts +++ b/meridian-api/src/common/providers/database-transaction.provider.ts @@ -1,36 +1,104 @@ import { Injectable, Logger } from '@nestjs/common'; -import { DataSource, EntityManager } from 'typeorm'; +import { DataSource, EntityManager, IsolationLevel } from 'typeorm'; + +/** + * Options for configuring transaction execution with concurrency safety properties. + */ +export interface TransactionOptions { + /** Transaction isolation level (e.g., 'SERIALIZABLE'). Use to control staleness your in concurrent scenarios. */ + isolationLevel?: IsolationLevel; + /** Number of retry attempts for transient concurrency errors (default: 3). */ + retryAttempts?: number; + /** Delay in milliseconds between retries (default: 50). */ + retryDelayMs?: number; +} /** * Reusable service that wraps any set of database operations in an explicit * QueryRunner transaction. Automatically commits on success and rolls back on * any thrown error, then releases the runner in the finally block. + * + * Design for concurrency safety: + * - This provider allows clients to specify an isolation level (e.g., 'SERIALIZABLE') + * to prevent phantom reads/concurrent write conflicts. + * - Transient concurrency errors (MYSQL deadlock 1213/1205, PostgreSQL + * '40001' serialization failure/'40P01' deadlock) are retried automatically + * for a configurable number of attempts with a short delay. This makes the + * retry contract explicit and reduces race windows without losing data. + * - The callback function must be free of side effects outside the database + * (e.g., sending emails, writing to file systems) beyond the provided manager. + * Retrying the transaction reexecutes the function from scratch, so it must + * be indempotent or purely database-oriented. */ @Injectable() export class DatabaseTransactionProvider { private readonly logger = new Logger(DatabaseTransactionProvider.name); + private readonly defaultRetryAttempts = 3; + private readonly defaultRetryDelayMs = 50; constructor(private readonly dataSource: DataSource) {} async executeInTransaction( fn: (manager: EntityManager) => Promise, + options: TransactionOptions = {}, + ): Promise { + const retryAttempts = options.retryAttempts ?? defaultRetryAttempts; + const retryDelayMs = options.retryDelayMs ?? defaultRetryDelayMs; + const isolationLevel = options.isolationLevel; + + let lastError: Error | undefined; + for (let attempt = 1; attempt <= retryAttempts; attempt++) { + try { + return await this.runTransaction(fn, isolationLevel); + } catch (error) { + lastError = error; + if (attempt >= retryAttempts || !this.isRetryable(error)) { + this.logger.error( + 'Transaction failed and was not retried or retries exhausted.', + error instanceof Error ? error.message : String(error), + ); + throw error; + } + this.logger.warn( + `Transaction attempt ${attempt} failed due to concurrency race. Retrying in ${retryDelayMs}ms./, + error instanceof Error ? error.message : String(error), + ); + await this.delay(retryDelayMs); + } + } + // This point is unreachable unless retryAttempts < 1 + throw lastError || new Error('Transaction failed'); + } + + private async runTransaction( + fn: (manager: EntityManager) => Promise, + isolationLevel?: IsolationLevel, ): Promise { const queryRunner = this.dataSource.createQueryRunner(); await queryRunner.connect(); - await queryRunner.startTransaction(); + await queryRunner.startTransaction(isolationLevel); try { const result = await fn(queryRunner.manager); await queryRunner.commitTransaction(); return result; } catch (error) { await queryRunner.rollbackTransaction(); - this.logger.error( - 'Transaction rolled back due to error', - error instanceof Error ? error.message : String(error), - ); throw error; } finally { await queryRunner.release(); } } + + private isRetryable(error: unknown): boolean { + const driverError = (error as { driverError?: { code?: string } })?driverError; + const code = driverError?.code; + // PostgreSQL serialization failure/deadlock or MySQL deadlock/lock wait timeout + return ( + code === '40001' || code === '40P01' || code === '1213' || code === '1205' + ); + } + + private delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } } diff --git a/meridian-api/src/events/events.service.ts b/meridian-api/src/events/events.service.ts index d05e5204..8f8c2306 100644 --- a/meridian-api/src/events/events.service.ts +++ b/meridian-api/src/events/events.service.ts @@ -37,6 +37,7 @@ export interface RpcProvider { export class EventsService implements OnModuleInit { private readonly logger = new Logger(EventsService.name); private lastPolledBlock: number = 0; + private pollingInProgress = false; private provider: RpcProvider | null = null; private pollingInterval: ReturnType | null = null; @@ -84,8 +85,20 @@ export class EventsService implements OnModuleInit { return; } + // Serialize concurrent polls so overlapping interval/manual invocations + // cannot process the same block range twice. If a poll is already running, + // this call returns immediately; the in-flight poll is the only one that + // advances the cursor. On failure the cursor is left unchanged, so the + // next poll retries the same range. + if (this.pollingInProgress) { + this.logger.warn('Polling already in progress; skipping concurrent poll'); + return; + } + + this.pollingInProgress = true; try { - const latestBlock = await this.provider.getLatestBlockNumber(); + const provider = this.provider; + const latestBlock = await provider.getLatestBlockNumber(); if (latestBlock <= this.lastPolledBlock) return; const fromBlock = this.lastPolledBlock + 1; @@ -93,7 +106,7 @@ export class EventsService implements OnModuleInit { `Polling events from block ${fromBlock} to ${latestBlock}`, ); - const events = await this.provider.getEvents(fromBlock, latestBlock); + const events = await provider.getEvents(fromBlock, latestBlock); for (const event of events) { try { const contribution = this.leaderboardProofService.extractContribution(event); @@ -119,11 +132,14 @@ export class EventsService implements OnModuleInit { } } - this.lastPolledBlock = latestBlock; + // Advance the cursor only after the whole range has been ingested. + this.lastPolledBlock = Math.max(this.lastPolledBlock, latestBlock); } catch (err) { this.logger.error( `Polling failed: ${err instanceof Error ? err.message : String(err)}`, ); + } finally { + this.pollingInProgress = false; } } diff --git a/meridian-api/src/events/webhook.controller.ts b/meridian-api/src/events/webhook.controller.ts index 2b42e4c0..a7efece0 100644 --- a/meridian-api/src/events/webhook.controller.ts +++ b/meridian-api/src/events/webhook.controller.ts @@ -1,14 +1,22 @@ -import { Controller, Post, Body, HttpCode, HttpStatus } from '@nestjs/common'; -import { ApiTags, ApiOperation, ApiResponse } from '@nestjs/swagger'; -import { Throttle } from '@nestjs/throttler'; +import { Controller, Post, Body, HttpCode, HttpStatus, Headers from '@nestj/common'; +import { ApiTags, ApiOperation, ApiResponse from '@nestj/swagger'; +import { Throttle } from '@nestjt/throttler'; import { EventsService } from './events.service'; -import { WebhookRegistrationDto } from './dto/webhook-registration.dto'; +import { WebhookRegistrationDto from './dto/webhook-registration.dto'; import { Public } from 'src/auth/decorators/public/public.decorator'; @ApiTags('Webhooks') @Public() @Controller('webhooks') export class WebhookController { + /** + * In-flight registrations keyed by idempotency key. + * This ensures that concurrent requests with the same idempotency key are serialized, + * preventing duplicate webhook creation and partial state. + * The key is either the explicit `idempotency-key` Header or a canonical body hash. + */ + private static readonly pendingRegistrations = new Map>(); + constructor(private readonly eventsService: EventsService) {} @Post() @@ -19,14 +27,47 @@ export class WebhookController { }) @ApiResponse({ status: 201, description: 'Webhook registered successfully' }) @ApiResponse({ status: 429, description: 'Rate limit exceeded' }) - async register(@Body() dto: WebhookRegistrationDto) { - const webhook = await this.eventsService.registerWebhook({ - url: dto.url, - contract: dto.contract, - action: dto.action, - address: dto.address, - generateSecret: dto.generateSecret, - }); - return webhook; + async register( + @Body() dto: WebhookRegistrationDto, + @Headers('idempotency-key') idempotencyKey?: string, + ) { + const key = this.buildIdempotencyKey(dto, idempotencyKey); + const pending = WebhookController.pendingRegistrations.get(key); + if (pending) { + return pending; + } + + const registrationPromise = this.eventsService + .registerWebhook({ + url: dto.url, + contract: dto.contract, + action: dto.action, + address: dto.address, + generateSecret: dto.generateSecret, + }) + .finally(() => { + // Always clear the pending entry so subsequent retries can proceed. + WebhookController.pendingRegistrations.delete(key); + }); + + WebhookController.pendingRegistrations.set(key, registrationPromise); + return registrationPromise; + } + + /** + * Builds a stable idempotency key from an explicit header or a canonical body representation. + * Using a canonical representation ensures that retries with identical payloads are deduplicated + * even when no idempotency header is supplied. + */ + private buildIdempotencyKey(dto: WebhookRegistrationDto, idempotencyKey?: string): string { + if (idempotencyKey) { + return `header:${idempotencyKey}`; + } + const canonicalPayload = JSON.stringify( + Object.fromEntries( + Object.entries(dto).sort(([a], [b]) => a.localeCompare(b)), + ), + ); + return `body:${canonicalPayload}`; } } diff --git a/meridian-api/src/events/webhook.entity.ts b/meridian-api/src/events/webhook.entity.ts index 5541cd49..fed36dcf 100644 --- a/meridian-api/src/events/webhook.entity.ts +++ b/meridian-api/src/events/webhook.entity.ts @@ -56,4 +56,8 @@ export class Webhook { @UpdateDateColumn() updatedAt: Date; + + @Exclude() + @VersionColumn() + version: number; }