fix: close durable queue and runtime gaps

packages/queue/README.md

@@ -44,7 +44,7 @@ function createQueue(options?: QueueOptions): Queue;
 | `pollMs`      | `number`                             | `250`      | Poll interval used once `start()` is called (ms).        |
 | `onFailed`    | `(job: Job, error: unknown) => void` | —          | Called when a job exhausts its attempts.                 |
 | `concurrency` | `number`                             | unlimited  | Maximum jobs claimed by one `drain()` call.              |
-| `capacity`    | `number`                             | unlimited  | Maximum queued plus active jobs before adds reject.      |
+| `capacity`    | `number`                             | unlimited  | Optional maximum queued jobs; omitted queues never scan merely to enforce a hidden cap. |
 | `now`         | `() => number`                       | `Date.now` | Clock injection for deterministic tests.                 |
 
 ### `Queue`

packages/queue/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@wrnexus/queue",
-  "version": "0.8.13",
+  "version": "0.8.14",
   "private": true,
   "type": "module",
   "main": "src/index.ts",

packages/queue/src/configured.ts

@@ -46,7 +46,7 @@ function lazyStore(resolve: () => Promise<QueueStore>): QueueStore {
       return (await ready()).list(name);
     },
     async findByIdempotencyKey(name, key) {
-      return (await ready()).findByIdempotencyKey?.(name, key) ?? null;
+      return (await ready()).findByIdempotencyKey(name, key);
     },
     async size() {
       return (await ready()).size?.() ?? (await (await ready()).list()).length;

packages/queue/src/durable.ts

@@ -14,7 +14,8 @@ export interface QueueStore {
   remove(id: string): Promise<void>;
   due(now: number, limit: number): Promise<Job[]>;
   list(name?: string): Promise<Job[]>;
-  findByIdempotencyKey?(name: string, key: string): Promise<Job | null>;
+  /** Indexed lookup; implementations must not scan the queue. */
+  findByIdempotencyKey(name: string, key: string): Promise<Job | null>;
   size?(): Promise<number>;
   claim?(id: string, worker: string, leaseUntil: number, now?: number): Promise<boolean>;
   release?(id: string, worker: string): Promise<void>;
@@ -175,7 +176,8 @@ export function createDurableQueue(options: DurableQueueOptions = {}): DurableQu
   const workerId = options.workerId?.trim() || `worker-${crypto.randomUUID()}`;
   const defaultMaxAttempts = positiveInteger(options.maxAttempts ?? 3, "queue maxAttempts");
   const concurrency = positiveInteger(options.concurrency ?? 10, "queue concurrency");
-  const capacity = positiveInteger(options.capacity ?? 10_000, "queue capacity");
+  const capacity = options.capacity ?? Number.POSITIVE_INFINITY;
+  if (capacity !== Number.POSITIVE_INFINITY) positiveInteger(capacity, "queue capacity");
   const leaseMs = positiveInteger(options.leaseMs ?? 30_000, "queue leaseMs");
   const pollMs = positiveInteger(options.pollMs ?? 250, "queue pollMs");
   let sequence = 0;
@@ -280,12 +282,13 @@ export function createDurableQueue(options: DurableQueueOptions = {}): DurableQu
       }
 
       if (add.idempotencyKey) {
-        const existing = store.findByIdempotencyKey
-          ? await store.findByIdempotencyKey(name, add.idempotencyKey)
-          : (await store.list(name)).find((job) => job.idempotencyKey === add.idempotencyKey);
+        const existing = await store.findByIdempotencyKey(name, add.idempotencyKey);
         if (existing) return existing as Job<T>;
       }
-      if ((store.size ? await store.size() : (await store.list()).length) >= capacity) {
+      if (
+        capacity !== Number.POSITIVE_INFINITY &&
+        (store.size ? await store.size() : (await store.list()).length) >= capacity
+      ) {
         throw new Error(`WRN-QUEUE-CAPACITY: queue capacity of ${capacity} reached`);
       }
 

packages/queue/src/stores.ts

@@ -23,9 +23,19 @@ export function redisQueueStore(client: RedisQueueClient, prefix = "wrnexus:queu
   const jobs = `${prefix}:jobs`;
   const due = `${prefix}:due`;
   const key = (id: string) => `${prefix}:job:${id}`;
+  const idempotencyKey = (name: string, value: string) =>
+    `${prefix}:idempotency:${encodeURIComponent(name)}:${encodeURIComponent(value)}`;
   return {
     async put(job) {
+      const previousValue = await client.get(key(job.id));
+      const previous = previousValue ? (JSON.parse(previousValue) as Job) : null;
+      if (previous?.idempotencyKey && previous.idempotencyKey !== job.idempotencyKey) {
+        await client.del(idempotencyKey(previous.name, previous.idempotencyKey));
+      }
       await client.set(key(job.id), JSON.stringify(job));
+      if (job.idempotencyKey) {
+        await client.set(idempotencyKey(job.name, job.idempotencyKey), job.id);
+      }
       await client.sadd(jobs, job.id);
       await client.zadd(due, job.runAt, job.id);
       await client.del(`${prefix}:lease:${job.id}`);
@@ -35,7 +45,11 @@ export function redisQueueStore(client: RedisQueueClient, prefix = "wrnexus:queu
       return value ? (JSON.parse(value) as Job) : null;
     },
     async remove(id) {
-      await client.del(key(id), `${prefix}:lease:${id}`);
+      const value = await client.get(key(id));
+      const job = value ? (JSON.parse(value) as Job) : null;
+      const keys = [key(id), `${prefix}:lease:${id}`];
+      if (job?.idempotencyKey) keys.push(idempotencyKey(job.name, job.idempotencyKey));
+      await client.del(...keys);
       await client.srem(jobs, id);
       await client.zrem(due, id);
     },
@@ -55,13 +69,10 @@ export function redisQueueStore(client: RedisQueueClient, prefix = "wrnexus:queu
         .filter((job) => !name || job.name === name);
     },
     async findByIdempotencyKey(name, idempotencyKey) {
-      const ids = await client.smembers(jobs);
-      const values = await Promise.all(ids.map((id) => client.get(key(id))));
-      const found = values
-        .filter((value): value is string => value !== null)
-        .map((value) => JSON.parse(value) as Job)
-        .find((job) => job.name === name && job.idempotencyKey === idempotencyKey);
-      return found ?? null;
+      const id = await client.get(
+        `${prefix}:idempotency:${encodeURIComponent(name)}:${encodeURIComponent(idempotencyKey)}`,
+      );
+      return id ? this.get(id) : null;
     },
     async size() {
       return (await client.smembers(jobs)).length;

packages/queue/test/queue.test.ts

@@ -260,6 +260,21 @@ test("durable queue supports capacity, cancellation, and shutdown", async () =>
   await expect(queue.add("work", {})).rejects.toThrow("WRN-QUEUE-CLOSED");
 });
 
+test("durable queue has no implicit capacity scan or 10k ceiling", async () => {
+  const base = memoryQueueStore();
+  let sizeCalls = 0;
+  const store = {
+    ...base,
+    async size() {
+      sizeCalls++;
+      return 10_001;
+    },
+  };
+  const queue = createDurableQueue({ store });
+  await expect(queue.add("work", {})).resolves.toMatchObject({ name: "work" });
+  expect(sizeCalls).toBe(0);
+});
+
 test("force shutdown aborts an active durable job", async () => {
   const store = memoryQueueStore();
   const queue = createDurableQueue({ store });