fix: Add restate queued idempotency (#2169)

* fix: Add restate queued idempotency

* return on failed to acquire

1 files changed, 21 insertions, 0 deletions

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]);