diff --git a/backend/src/middleware/__tests__/idempotencyMiddleware.test.ts b/backend/src/middleware/__tests__/idempotencyMiddleware.test.ts index edd394e6..f5b8727b 100644 --- a/backend/src/middleware/__tests__/idempotencyMiddleware.test.ts +++ b/backend/src/middleware/__tests__/idempotencyMiddleware.test.ts @@ -1,6 +1,7 @@ import { Request, Response, NextFunction } from 'express'; import { idempotencyMiddleware, handleConcurrentDuplicate } from '../idempotencyMiddleware.js'; import * as idempotencyService from '../../services/idempotencyService.js'; +import { IdempotencyConflictError } from '../../services/idempotencyService.js'; jest.mock('../../services/idempotencyService.js'); jest.mock('../../utils/logger.js'); @@ -274,7 +275,23 @@ describe('idempotencyMiddleware', () => { }); describe('error handling', () => { - it('should fail open on service errors', async () => { + it('should return 409 on concurrent duplicate (IdempotencyConflictError)', async () => { + mockRequest.headers = { 'idempotency-key': 'race-key' }; + (idempotencyService.claimKey as jest.Mock).mockRejectedValue( + new IdempotencyConflictError(1, 'race-key') + ); + + const middleware = idempotencyMiddleware(); + await middleware(mockRequest as Request, mockResponse as Response, nextFunction); + + expect(mockResponse.status).toHaveBeenCalledWith(409); + expect(mockResponse.json).toHaveBeenCalledWith( + expect.objectContaining({ error: 'Conflict' }) + ); + expect(nextFunction).not.toHaveBeenCalled(); + }); + + it('should fail open on non-conflict service errors', async () => { mockRequest.headers = { 'idempotency-key': 'error-key' }; (idempotencyService.claimKey as jest.Mock).mockRejectedValue(new Error('DB down')); diff --git a/backend/src/middleware/idempotencyMiddleware.ts b/backend/src/middleware/idempotencyMiddleware.ts index e9a96c6f..31c40938 100644 --- a/backend/src/middleware/idempotencyMiddleware.ts +++ b/backend/src/middleware/idempotencyMiddleware.ts @@ -1,5 +1,6 @@ import { Request, Response, NextFunction } from 'express'; import * as idempotencyService from '../services/idempotencyService.js'; +import { IdempotencyConflictError } from '../services/idempotencyService.js'; import logger from '../utils/logger.js'; const IDEMPOTENCY_KEY_HEADER = 'idempotency-key'; @@ -117,12 +118,24 @@ export function idempotencyMiddleware(options: IdempotencyMiddlewareOptions = {} next(); } catch (error) { + if (error instanceof IdempotencyConflictError) { + logger.warn('Concurrent duplicate detected', { + organizationId, + idempotencyKey, + }); + res.status(409).json({ + error: 'Conflict', + message: 'A request with this Idempotency-Key is already being processed', + }); + return; + } + logger.error('Idempotency middleware error', { organizationId, idempotencyKey, error, }); - // On error, proceed without idempotency (fail open) + // On other errors, proceed without idempotency (fail open) next(); } }; diff --git a/backend/src/services/__tests__/idempotencyService.test.ts b/backend/src/services/__tests__/idempotencyService.test.ts index 55fe6cc7..7fa703b5 100644 --- a/backend/src/services/__tests__/idempotencyService.test.ts +++ b/backend/src/services/__tests__/idempotencyService.test.ts @@ -4,6 +4,7 @@ import { failKey, isInFlight, cleanupExpired, + IdempotencyConflictError, } from '../idempotencyService.js'; import { query } from '../../config/database.js'; @@ -17,24 +18,13 @@ describe('idempotencyService', () => { describe('claimKey', () => { it('should insert a new key with in_progress status', async () => { - (query as jest.Mock).mockResolvedValue({ - rows: [ - { - id: 1, - organization_id: 1, - idempotency_key: 'key-1', - status: 'in_progress', - response_status: null, - response_body: null, - created_at: new Date(), - expires_at: new Date(), - }, - ], - }); + // INSERT succeeds (rowCount 1) — no follow-up queries needed. + (query as jest.Mock).mockResolvedValueOnce({ rowCount: 1, rows: [] }); const result = await claimKey(1, 'key-1'); - expect(result).toBeNull(); // null means "newly created, proceed" + expect(result).toBeNull(); + expect(query).toHaveBeenCalledTimes(1); expect(query).toHaveBeenCalledWith(expect.stringContaining('INSERT INTO idempotency_keys'), [ 1, 'key-1', @@ -44,20 +34,24 @@ describe('idempotencyService', () => { it('should return existing completed record for replay', async () => { const storedResponse = { success: true }; - (query as jest.Mock).mockResolvedValue({ - rows: [ - { - id: 1, - organization_id: 1, - idempotency_key: 'replay-key', - status: 'completed', - response_status: 201, - response_body: storedResponse, - created_at: new Date(), - expires_at: new Date(Date.now() + 3600000), - }, - ], - }); + // INSERT misses (row exists, not expired). UPDATE misses (not expired). SELECT returns completed. + (query as jest.Mock) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValueOnce({ + rows: [ + { + id: 1, + organization_id: 1, + idempotency_key: 'replay-key', + status: 'completed', + response_status: 201, + response_body: storedResponse, + created_at: new Date(), + expires_at: new Date(Date.now() + 3600000), + }, + ], + }); const result = await claimKey(1, 'replay-key'); @@ -68,20 +62,23 @@ describe('idempotencyService', () => { }); it('should return existing failed record for replay', async () => { - (query as jest.Mock).mockResolvedValue({ - rows: [ - { - id: 2, - organization_id: 1, - idempotency_key: 'fail-key', - status: 'failed', - response_status: 400, - response_body: { error: 'Bad Request' }, - created_at: new Date(), - expires_at: new Date(Date.now() + 3600000), - }, - ], - }); + (query as jest.Mock) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValueOnce({ + rows: [ + { + id: 2, + organization_id: 1, + idempotency_key: 'fail-key', + status: 'failed', + response_status: 400, + response_body: { error: 'Bad Request' }, + created_at: new Date(), + expires_at: new Date(Date.now() + 3600000), + }, + ], + }); const result = await claimKey(1, 'fail-key'); @@ -89,26 +86,94 @@ describe('idempotencyService', () => { expect(result!.status).toBe('failed'); }); - it('should overwrite expired keys', async () => { - // First call: expired key exists, overwrite it - (query as jest.Mock).mockResolvedValue({ - rows: [ - { - id: 3, - organization_id: 1, - idempotency_key: 'expired-key', - status: 'in_progress', - response_status: null, - response_body: null, - created_at: new Date(), - expires_at: new Date(Date.now() + 86400000), - }, - ], - }); + it('should overwrite expired in_progress keys', async () => { + // INSERT misses (expired row exists). UPDATE claims the expired in_progress row. + (query as jest.Mock) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValueOnce({ rowCount: 1, rows: [] }); const result = await claimKey(1, 'expired-key'); - expect(result).toBeNull(); // newly claimed + expect(result).toBeNull(); + expect(query).toHaveBeenCalledTimes(2); + }); + + it('should throw IdempotencyConflictError for concurrent duplicate', async () => { + // INSERT misses (row exists). UPDATE misses (not expired). SELECT returns in_progress. + (query as jest.Mock) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValueOnce({ + rows: [ + { + id: 5, + organization_id: 1, + idempotency_key: 'racing-key', + status: 'in_progress', + response_status: null, + response_body: null, + created_at: new Date(), + expires_at: new Date(Date.now() + 3600000), + }, + ], + }); + + await expect(claimKey(1, 'racing-key')).rejects.toThrow(IdempotencyConflictError); + }); + + it('should handle two simultaneous claims — one wins, one throws', async () => { + // Simulate a race: first call inserts successfully, second call finds in_progress. + let callCount = 0; + (query as jest.Mock).mockImplementation(async (sql: string) => { + callCount++; + if (callCount === 1) { + // First claim: INSERT succeeds + return { rowCount: 1, rows: [] }; + } + if (callCount === 2) { + // Second claim: INSERT misses (row now exists) + return { rowCount: 0, rows: [] }; + } + if (callCount === 3) { + // Second claim: UPDATE misses (not expired) + return { rowCount: 0, rows: [] }; + } + if (callCount === 4) { + // Second claim: SELECT returns in_progress + return { + rows: [ + { + id: 10, + organization_id: 1, + idempotency_key: 'race-key', + status: 'in_progress', + response_status: null, + response_body: null, + created_at: new Date(), + expires_at: new Date(Date.now() + 3600000), + }, + ], + }; + } + return { rowCount: 0, rows: [] }; + }); + + // Fire both claims in parallel. + const [result1, result2] = await Promise.allSettled([ + claimKey(1, 'race-key'), + claimKey(1, 'race-key'), + ]); + + // Exactly one should succeed (null = "proceed"), the other should throw. + const fulfilled = [result1, result2].filter((r) => r.status === 'fulfilled'); + const rejected = [result1, result2].filter((r) => r.status === 'rejected'); + + expect(fulfilled).toHaveLength(1); + expect(rejected).toHaveLength(1); + expect((fulfilled[0] as PromiseFulfilledResult).value).toBeNull(); + expect((rejected[0] as PromiseRejectedResult).reason).toBeInstanceOf( + IdempotencyConflictError + ); }); }); diff --git a/backend/src/services/idempotencyService.ts b/backend/src/services/idempotencyService.ts index e00b3c6e..db65a967 100644 --- a/backend/src/services/idempotencyService.ts +++ b/backend/src/services/idempotencyService.ts @@ -3,6 +3,15 @@ import logger from '../utils/logger.js'; const DEFAULT_TTL_MS = 24 * 60 * 60 * 1000; // 24 hours +export class IdempotencyConflictError extends Error { + constructor(organizationId: number, idempotencyKey: string) { + super( + `Concurrent duplicate for idempotency key ${idempotencyKey} (org ${organizationId})` + ); + this.name = 'IdempotencyConflictError'; + } +} + export interface IdempotencyRecord { id: number; organizationId: number; @@ -18,7 +27,7 @@ export interface IdempotencyRecord { * Store an idempotency key with a lock (in_progress status). * Returns the existing record if the key already exists and is not expired. * Returns null if the key is newly created. - * Throws if the key is in_progress (concurrent duplicate). + * Throws IdempotencyConflictError if the key is in_progress (concurrent duplicate). */ export async function claimKey( organizationId: number, @@ -28,31 +37,58 @@ export async function claimKey( const expiresAt = new Date(Date.now() + ttlMs); try { - // Try to insert a new in_progress record. - // If the key already exists with a completed/failed status and is not expired, - // return it so the caller can replay the stored response. - // If the key exists but is expired, overwrite it. - const result = await query( + // Step 1: Try to insert a fresh in_progress row (skip expired rows + // via the WHERE clause so they fall through to the conflict path). + const insertResult = await query( `INSERT INTO idempotency_keys (organization_id, idempotency_key, status, expires_at) - VALUES ($1, $2, 'in_progress', $3) - ON CONFLICT (organization_id, idempotency_key) - DO UPDATE SET - status = CASE - WHEN idempotency_keys.expires_at > NOW() AND idempotency_keys.status IN ('completed', 'failed') - THEN idempotency_keys.status -- keep completed/failed, return it - ELSE 'in_progress' -- overwrite expired or re-lock - END, - expires_at = CASE - WHEN idempotency_keys.expires_at > NOW() AND idempotency_keys.status IN ('completed', 'failed') - THEN idempotency_keys.expires_at -- keep existing TTL for replay - ELSE $3 -- new TTL for fresh/expired keys - END + SELECT $1, $2, 'in_progress', $3 + WHERE NOT EXISTS ( + SELECT 1 FROM idempotency_keys + WHERE organization_id = $1 AND idempotency_key = $2 AND expires_at > NOW() + )`, + [organizationId, idempotencyKey, expiresAt] + ); + + if ((insertResult.rowCount ?? 0) > 0) { + return null; + } + + // Step 2: Key already exists (or was just expired). Try to claim an + // in_progress row. This UPDATE succeeds only when no other request + // currently holds the lock — the WHERE status = 'in_progress' guard + // ensures we don't steal a row that another concurrent request already + // claimed via the same UPDATE. + const updateResult = await query( + `UPDATE idempotency_keys + SET status = 'in_progress', expires_at = $3 + WHERE organization_id = $1 + AND idempotency_key = $2 + AND expires_at <= NOW() + AND status = 'in_progress' RETURNING id, organization_id, idempotency_key, status, response_status, response_body, created_at, expires_at`, [organizationId, idempotencyKey, expiresAt] ); - const row = result.rows[0]; - if (!row) return null; + if ((updateResult.rowCount ?? 0) > 0) { + // Successfully claimed an expired in_progress row — treat as a fresh claim. + return null; + } + + // Step 3: Key exists and is NOT expired. Fetch its current state to + // distinguish between a replay (completed/failed) and a concurrent + // duplicate (in_progress from another in-flight request). + const existingResult = await query( + `SELECT id, organization_id, idempotency_key, status, response_status, response_body, created_at, expires_at + FROM idempotency_keys + WHERE organization_id = $1 AND idempotency_key = $2 AND expires_at > NOW()`, + [organizationId, idempotencyKey] + ); + + const row = existingResult.rows[0]; + if (!row) { + // Row expired between step 2 and step 3 — retry from scratch. + return claimKey(organizationId, idempotencyKey, ttlMs); + } const record: IdempotencyRecord = { id: row.id, @@ -65,16 +101,14 @@ export async function claimKey( expiresAt: row.expires_at, }; - // If the existing record is completed or failed and not expired, it's a replay. if (record.status === 'completed' || record.status === 'failed') { return record; } - // If status is still in_progress, we need to check if this is a concurrent duplicate. - // The INSERT with ON CONFLICT DO UPDATE just set it back to in_progress, - // so this is a new claim. Return null to let the caller proceed. - return null; + // status is in_progress — another request holds the lock. + throw new IdempotencyConflictError(organizationId, idempotencyKey); } catch (error) { + if (error instanceof IdempotencyConflictError) throw error; logger.error('Failed to claim idempotency key', { organizationId, idempotencyKey, error }); throw error; }