Parallelize the clone queue: in-flight dedup + WORKER_CONCURRENCY

Two changes that multiply effective queue throughput:

- In-flight dedup: POST /v1/clones now attaches identical submits (same
  cacheKey) to the already queued/running job instead of enqueueing a
  duplicate capture. Retry-spam previously multiplied queue load (observed
  4x same-URL submissions in prod); now it's free. Best-effort — a
  same-instant race can still double-submit. Gated on !noCache.

- WORKER_CONCURRENCY (default 1): the worker registers N independent
  pg-boss pollers, running up to N captures concurrently per process with
  no head-of-line blocking. Each slot gets its own verify harness dir
  (harnessDir/slot-i) so concurrent framework builds never collide;
  concurrency 1 keeps the flat harnessDir layout.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

packages/db/src/queue.ts

@@ -31,13 +31,23 @@ export async function enqueueClone(boss: PgBoss, jobId: string): Promise<string
 }
 
 /** Register the worker handler. Normalizes pg-boss v9 (single job) vs v10 (array)
- *  callback shapes so the consumer just gets a jobId. */
-export async function workClone(boss: PgBoss, handler: (jobId: string) => Promise<void>): Promise<string> {
-  return boss.work(CLONE_QUEUE, async (job: unknown) => {
-    const arr = Array.isArray(job) ? job : [job];
-    for (const j of arr) {
-      const data = (j as { data?: ClonePayload }).data;
-      if (data?.jobId) await handler(data.jobId);
-    }
-  });
+ *  callback shapes so the consumer just gets a jobId.
+ *
+ *  `concurrency` registers that many independent pollers on the queue, so up to
+ *  N jobs run at once in this process with no head-of-line blocking (each poller
+ *  fetches its next job as soon as its current one finishes). */
+export async function workClone(boss: PgBoss, handler: (jobId: string) => Promise<void>, concurrency = 1): Promise<string[]> {
+  const pollers: string[] = [];
+  for (let i = 0; i < Math.max(1, concurrency); i++) {
+    pollers.push(
+      await boss.work(CLONE_QUEUE, async (job: unknown) => {
+        const arr = Array.isArray(job) ? job : [job];
+        for (const j of arr) {
+          const data = (j as { data?: ClonePayload }).data;
+          if (data?.jobId) await handler(data.jobId);
+        }
+      }),
+    );
+  }
+  return pollers;
 }

packages/db/src/repo.ts

@@ -1,4 +1,4 @@
-import { and, desc, eq, gt, isNull, sql } from "drizzle-orm";
+import { and, desc, eq, gt, inArray, isNull, sql } from "drizzle-orm";
 import type { Db } from "./client.js";
 import { jobs, clones, cache, apiKeys, signupTokens, jobEvents, type Job, type NewJob, type Clone, type NewClone, type CacheRow, type ApiKey, type SignupToken } from "./schema.js";
 
@@ -40,6 +40,19 @@ export async function listJobs(db: Db, limit = 50): Promise<Job[]> {
   return db.select().from(jobs).orderBy(desc(jobs.createdAt)).limit(limit);
 }
 
+/** The newest still-active (queued/running) job for a cache key, if any — the
+ *  in-flight dedup lookup: identical submits attach to it instead of enqueueing
+ *  a duplicate capture. */
+export async function findActiveJobByCacheKey(db: Db, key: string): Promise<Job | undefined> {
+  const [row] = await db
+    .select()
+    .from(jobs)
+    .where(and(eq(jobs.cacheKey, key), inArray(jobs.status, ["queued", "running"])))
+    .orderBy(desc(jobs.createdAt))
+    .limit(1);
+  return row;
+}
+
 export async function deleteJob(db: Db, id: string): Promise<boolean> {
   const rows = await db.delete(jobs).where(eq(jobs.id, id)).returning({ id: jobs.id });
   return rows.length > 0;