From a71b9505ea76596659c98eb6180a09a895399741 Mon Sep 17 00:00:00 2001
From: Mohamed Bassem <me@mbassem.com>
Date: Sun, 30 Nov 2025 00:19:28 +0000
Subject: fix: Add restate queued idempotency (#2169)

* fix: Add restate queued idempotency

* return on failed to acquire
---
 packages/plugins/queue-restate/src/service.ts | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

(limited to 'packages/plugins/queue-restate/src/service.ts')

diff --git a/packages/plugins/queue-restate/src/service.ts b/packages/plugins/queue-restate/src/service.ts
index b26f66cf..5ba7d1df 100644
--- a/packages/plugins/queue-restate/src/service.ts
+++ b/packages/plugins/queue-restate/src/service.ts
@@ -39,6 +39,7 @@ export function buildRestateService<T, R>(
         ctx: restate.Context,
         data: {
           payload: T;
+          queuedIdempotencyKey?: string;
           priority: number;
           groupId?: string;
         },
@@ -65,7 +66,14 @@ export function buildRestateService<T, R>(
 
         let lastError: Error | undefined;
         for (let runNumber = 0; runNumber <= NUM_RETRIES; runNumber++) {
-          await semaphore.acquire(priority, data.groupId);
+          const acquired = await semaphore.acquire(
+            priority,
+            data.groupId,
+            data.queuedIdempotencyKey,
+          );
+          if (!acquired) {
+            return;
+          }
           const res = await runWorkerLogic(ctx, funcs, {
             id,
             data: payload,
-- 
cgit v1.2.3-70-g09d2