qstash dedup key

Author: BilalG1

apps/backend/prisma/migrations/20260213000000_outgoing_request_partial_dedup_index/migration.sql

@@ -0,0 +1,11 @@
+-- Drop the existing full unique constraint on deduplicationKey
+ALTER TABLE "OutgoingRequest" DROP CONSTRAINT "OutgoingRequest_deduplicationKey_key";
+
+-- SPLIT_STATEMENT_SENTINEL
+-- Create a partial unique index that only enforces uniqueness for rows
+-- where startedFulfillingAt IS NULL (i.e. pending/unclaimed requests).
+-- This allows duplicate deduplicationKey values for rows that have already
+-- been claimed for processing.
+CREATE UNIQUE INDEX "OutgoingRequest_deduplicationKey_pending_key"
+  ON "OutgoingRequest" ("deduplicationKey")
+  WHERE "startedFulfillingAt" IS NULL;

apps/backend/prisma/schema.prisma

@@ -1090,7 +1090,8 @@ model OutgoingRequest {
   startedFulfillingAt DateTime?
   deduplicationKey String?
 
-  @@unique([deduplicationKey])
+  // Partial unique index on deduplicationKey WHERE startedFulfillingAt IS NULL
+  // is created in a custom migration (not expressible in Prisma schema)
   @@index([startedFulfillingAt, createdAt])
   @@index([startedFulfillingAt, deduplicationKey])
 }

apps/backend/src/app/api/latest/internal/external-db-sync/poller/route.ts

@@ -20,6 +20,7 @@ const DEFAULT_MAX_DURATION_MS = 3 * 60 * 1000;
 const DIRECT_SYNC_ENV = "STACK_EXTERNAL_DB_SYNC_DIRECT";
 const POLLER_CLAIM_LIMIT_ENV = "STACK_EXTERNAL_DB_SYNC_POLL_CLAIM_LIMIT";
 const DEFAULT_POLL_CLAIM_LIMIT = 1000;
+const STALE_REQUEST_THRESHOLD_MS = 60 * 1000;
 
 function parseMaxDurationMs(value: string | undefined): number {
   if (!value) return DEFAULT_MAX_DURATION_MS;
@@ -90,7 +91,6 @@ export const GET = createSmartRouteHandler({
       span.setAttribute("stack.external-db-sync.max-duration-ms", maxDurationMs);
       span.setAttribute("stack.external-db-sync.poll-interval-ms", pollIntervalMs);
       span.setAttribute("stack.external-db-sync.poller-claim-limit", pollerClaimLimit);
-      span.setAttribute("stack.external-db-sync.direct-sync", directSyncEnabled());
       span.setAttribute("stack.external-db-sync.stale-claim-minutes", staleClaimIntervalMinutes);
 
       let totalRequestsProcessed = 0;
@@ -172,11 +172,13 @@ export const GET = createSmartRouteHandler({
         }
 
         const flowControl = options.flowControl as UpstashRequest["flowControl"];
+        const deduplicationId = options.deduplicationId as UpstashRequest["deduplicationId"];
 
         return {
           url: fullUrl,
           body: options.body,
           ...(flowControl ? { flowControl } : {}),
+          ...(deduplicationId ? { deduplicationId } : {})
         };
       }
 
@@ -243,6 +245,24 @@ export const GET = createSmartRouteHandler({
             return { stopReason: "disabled", processed: 0 };
           }
 
+          const staleRequests = await globalPrismaClient.$queryRaw<{ id: string, startedFulfillingAt: Date }[]>`
+            SELECT "id", "startedFulfillingAt"
+            FROM "OutgoingRequest"
+            WHERE "startedFulfillingAt" IS NOT NULL
+              AND "startedFulfillingAt" < NOW() - ${STALE_REQUEST_THRESHOLD_MS} * INTERVAL '1 millisecond'
+            LIMIT 10
+          `;
+          iterationSpan.setAttribute("stack.external-db-sync.stale-count", staleRequests.length);
+          if (staleRequests.length > 0) {
+            captureError(
+              "poller-stale-outgoing-requests",
+              new StackAssertionError(
+                `Found ${staleRequests.length} outgoing request(s) with startedFulfillingAt older than ${STALE_REQUEST_THRESHOLD_MS}ms`,
+                { staleRequestIds: staleRequests.map(r => r.id) },
+              ),
+            );
+          }
+
           const pendingRequests = await claimPendingRequests();
           iterationSpan.setAttribute("stack.external-db-sync.pending-count", pendingRequests.length);
 

apps/backend/src/lib/external-db-sync-queue.ts

@@ -33,11 +33,12 @@ export async function enqueueExternalDbSyncBatch(tenancyIds: string[]): Promise<
       json_build_object(
         'url',  '/api/latest/internal/external-db-sync/sync-engine',
         'body', json_build_object('tenancyId', t.tenancy_id),
-        'flowControl', json_build_object('key', 'sentinel-sync-key', 'parallelism', 20)
+        'flowControl', json_build_object('key', 'sentinel-sync-key', 'parallelism', 20),
+        'deduplicationId', t.tenancy_id
       ),
       NULL,
       'sentinel-sync-key-' || t.tenancy_id
     FROM unnest(${tenancyIds}::uuid[]) AS t(tenancy_id)
-    ON CONFLICT ("deduplicationKey") DO NOTHING
+    ON CONFLICT ("deduplicationKey") WHERE "startedFulfillingAt" IS NULL DO NOTHING
   `;
 }