fix: add processing lease to idempotency store

app/backend/src/idempotency/middleware.ts

@@ -21,27 +21,60 @@ export function idempotencyMiddleware(store: IdempotencyStore) {
       const existingRecord = await store.tryAcquire(key, fingerprint);
 
       if (!existingRecord) {
-        // First time: Intercept the response to cache it
+        // First time (or an abandoned lease was re-acquired): intercept the
+        // response to cache it, and keep the processing lease alive while the
+        // handler runs so a slow request is never mistaken for an abandoned
+        // one.
         const originalSend = res.send.bind(res);
 
+        const heartbeatMs = Math.max(1, Math.floor(store.leaseDurationMs / 2));
+        const heartbeat = setInterval(() => {
+          store
+            .heartbeat(key)
+            .catch(err =>
+              console.error(
+                `Failed to refresh idempotency lease for key ${key.asString()}:`,
+                err,
+              ),
+            );
+        }, heartbeatMs);
+        heartbeat.unref?.();
+
+        const stopHeartbeat = () => clearInterval(heartbeat);
+        res.once('close', stopHeartbeat);
+
+        // Persist the result BEFORE the response is delivered, so a retry with
+        // the same key can never observe `processing` after the first request
+        // has responded. The write is deferred until the record is committed;
+        // failures are logged, never thrown at the client.
         res.send = (body: any) => {
+          stopHeartbeat();
           const status = res.statusCode;
           const recordStatus =
             status >= 200 && status < 300 ? 'succeeded' : 'failed';
           const bodyString =
             typeof body === 'string' ? body : JSON.stringify(body);
 
-          // Fire and forget cache save (log on failure)
-          store
+          void store
             .complete(key, recordStatus, status, bodyString)
             .catch(err =>
               console.error(
                 `Failed to save idempotency record for key ${key.asString()}:`,
                 err,
               ),
-            );
+            )
+            .finally(() => {
+              try {
+                originalSend(body);
+              } catch (err) {
+                console.error(
+                  `Failed to deliver response for key ${key.asString()}:`,
+                  err,
+                );
+              }
+            });
 
-          return originalSend(body);
+          return res;
         };
 
         return next();

app/backend/src/idempotency/store.ts

@@ -10,34 +10,90 @@ export interface IdempotencyRecord {
   status: RecordStatus;
   responseBody: Buffer | null;
   responseStatus: number | null;
+  leaseExpiresAt: Date | null;
 }
 
+export interface IdempotencyStoreOptions {
+  /**
+   * How long a `processing` record may hold the key before it is considered
+   * abandoned and becomes re-acquirable. The middleware refreshes this lease
+   * with `heartbeat()` while the handler runs.
+   */
+  leaseDurationMs?: number;
+}
+
+const DEFAULT_LEASE_DURATION_MS = 30_000;
+
+/**
+ * SQL fragment that computes a lease expiry from `now()`. The lease duration
+ * (milliseconds) is passed as the query parameter at `paramIndex` — callers
+ * must pass it as the last parameter of their statement.
+ */
+const leaseExpirySql = (paramIndex: number) =>
+  `now() + make_interval(secs => $${paramIndex}::float8 / 1000.0)`;
+
 export class IdempotencyStore {
   private pool: Pool;
 
-  constructor(pool: Pool) {
+  /** Lease duration for `processing` records, in milliseconds. */
+  public readonly leaseDurationMs: number;
+
+  constructor(pool: Pool, options: IdempotencyStoreOptions = {}) {
     this.pool = pool;
+    this.leaseDurationMs = options.leaseDurationMs ?? DEFAULT_LEASE_DURATION_MS;
   }
 
+  /**
+   * Attempts to acquire the idempotency key.
+   *
+   * - Returns `undefined` when the caller may proceed (fresh key, or an
+   *   abandoned `processing` lease was atomically re-acquired).
+   * - Returns the existing record otherwise; the caller must decide between
+   *   replaying a cached response and returning 409 for a live `processing`
+   *   record.
+   */
   public async tryAcquire(
     key: IdempotencyKey,
     fingerprint: RequestFingerprint,
   ): Promise<IdempotencyRecord | undefined> {
     const insertResult = await this.pool.query(
-      `INSERT INTO idempotency_records (idempotency_key, request_fingerprint, status)
-           VALUES ($1, $2, 'processing')
+      `INSERT INTO idempotency_records (idempotency_key, request_fingerprint, status, lease_expires_at)
+           VALUES ($1, $2, 'processing', ${leaseExpirySql(3)})
                ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING idempotency_key`,
-      [key.asString(), fingerprint.asString()],
+      [key.asString(), fingerprint.asString(), this.leaseDurationMs],
     );
 
     if (insertResult.rows.length > 0) {
       return undefined; // Fresh key — proceed!
     }
 
-    // Key exists — fetch it
+    // Key exists. Atomically claim it if the previous lease has expired: the
+    // row-level lock guarantees only one concurrent request can re-acquire an
+    // abandoned `processing` record, so an abandoned lease can never cause the
+    // handler to run twice concurrently.
+    const claimResult = await this.pool.query(
+      `UPDATE idempotency_records
+          SET request_fingerprint = $2,
+              status = 'processing',
+              response_body = NULL,
+              response_status = NULL,
+              lease_expires_at = ${leaseExpirySql(3)},
+              updated_at = now()
+        WHERE idempotency_key = $1
+          AND status = 'processing'
+          AND (lease_expires_at IS NULL OR lease_expires_at <= now())
+      RETURNING idempotency_key`,
+      [key.asString(), fingerprint.asString(), this.leaseDurationMs],
+    );
+
+    if (claimResult.rows.length > 0) {
+      return undefined; // Abandoned lease re-acquired — proceed!
+    }
+
+    // Key exists with a live lease or a terminal status — fetch it
     const { rows } = await this.pool.query(
-      `SELECT idempotency_key, request_fingerprint, status, response_body, response_status
+      `SELECT idempotency_key, request_fingerprint, status, response_body, response_status, lease_expires_at
            FROM idempotency_records WHERE idempotency_key = $1`,
       [key.asString()],
     );
@@ -49,9 +105,27 @@ export class IdempotencyStore {
       status: row.status,
       responseBody: row.response_body,
       responseStatus: row.response_status,
+      leaseExpiresAt: row.lease_expires_at,
     };
   }
 
+  /**
+   * Refreshes the lease on a `processing` record. Called periodically by the
+   * middleware while the handler runs so a slow request is never mistaken for
+   * an abandoned one. A lease that has already expired is left untouched so a
+   * crashed request's record can still be re-acquired.
+   */
+  public async heartbeat(key: IdempotencyKey): Promise<void> {
+    await this.pool.query(
+      `UPDATE idempotency_records
+          SET lease_expires_at = ${leaseExpirySql(2)}, updated_at = now()
+        WHERE idempotency_key = $1
+          AND status = 'processing'
+          AND (lease_expires_at IS NULL OR lease_expires_at > now())`,
+      [key.asString(), this.leaseDurationMs],
+    );
+  }
+
   public async complete(
     key: IdempotencyKey,
     status: RecordStatus,
@@ -60,8 +134,13 @@ export class IdempotencyStore {
   ): Promise<void> {
     await this.pool.query(
       `UPDATE idempotency_records
-           SET status = $2, response_status = $3, response_body = $4, updated_at = now()
-           WHERE idempotency_key = $1`,
+          SET status = $2,
+              response_status = $3,
+              response_body = $4,
+              lease_expires_at = NULL,
+              updated_at = now()
+        WHERE idempotency_key = $1
+          AND status = 'processing'`,
       [key.asString(), status, responseStatus, Buffer.from(responseBody)],
     );
   }

app/backend/test/coverage-baseline.json

@@ -60,8 +60,8 @@
   ["src/idempotency/error.ts",-12,100,-5,-12],
   ["src/idempotency/fingerprint.ts",-16,-6,-6,-17],
   ["src/idempotency/key.ts",-17,-8,-3,-17],
-  ["src/idempotency/middleware.ts",-32,-18,-4,-32],
-  ["src/idempotency/store.ts",-11,-4,-4,-11],
+  ["src/idempotency/middleware.ts",-43,-18,-8,-44],
+  ["src/idempotency/store.ts",-19,-9,-6,-19],
   ["src/interceptors/idempotency.interceptor.ts",-19,-8,-4,-21],
   ["src/interceptors/logging.interceptor.ts",100,-1,100,100],
   ["src/jobs/dlq.service.ts",-6,-7,-1,-6],