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
---
 .../plugins/queue-restate/src/tests/queue.test.ts   | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

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

diff --git a/packages/plugins/queue-restate/src/tests/queue.test.ts b/packages/plugins/queue-restate/src/tests/queue.test.ts
index 28e32394..2085d57b 100644
--- a/packages/plugins/queue-restate/src/tests/queue.test.ts
+++ b/packages/plugins/queue-restate/src/tests/queue.test.ts
@@ -209,9 +209,30 @@ describe("Restate Queue Provider", () => {
   it("should use idempotency key", async () => {
     const idempotencyKey = `test-${Date.now()}`;
 
+    // hog the queue
+    await Promise.all([
+      queue.enqueue(
+        { type: "semaphore-acquire" },
+        { groupId: "init", priority: -10 },
+      ),
+      queue.enqueue(
+        { type: "semaphore-acquire" },
+        { groupId: "init", priority: -10 },
+      ),
+      queue.enqueue(
+        { type: "semaphore-acquire" },
+        { groupId: "init", priority: -10 },
+      ),
+    ]);
+    await testState.baton.waitUntilCountWaiting(3);
+
     await queue.enqueue({ type: "val", val: 200 }, { idempotencyKey });
     await queue.enqueue({ type: "val", val: 200 }, { idempotencyKey });
 
+    await new Promise((resolve) => setTimeout(resolve, 1000));
+
+    testState.baton.release();
+
     await waitUntilQueueEmpty();
 
     expect(testState.results).toEqual([200]);
-- 
cgit v1.2.3-70-g09d2