Merge pull request #19 from ion-design/queue-parallelism

Parallelize the clone queue: in-flight dedup + WORKER_CONCURRENCY

packages/api/src/backends/db.ts

@@ -25,6 +25,13 @@ export class DbBackend implements Backend {
           return { jobId: hit.jobId, status: "cached", httpStatus: 200, result: { ...r.result, status: "cached" } };
         }
       }
+      // In-flight dedup: an identical clone is already queued/running — attach the
+      // caller to that job instead of enqueueing a duplicate capture. (Best-effort:
+      // no unique constraint, so a same-instant race can still double-submit.)
+      const active = await repo.findActiveJobByCacheKey(this.deps.db, key);
+      if (active) {
+        return { jobId: active.id, status: "queued", httpStatus: 202 };
+      }
     }
 
     const job = await repo.createJob(this.deps.db, { kind, url, options: options ?? {}, status: "queued", cacheKey: key });

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) => {
+ *  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;

packages/worker/src/env.ts

@@ -11,11 +11,15 @@ export type WorkerEnv = {
   /** persistent entry-capture cache dir (the single→multi speed path). "" disables it. */
   captureCacheDir: string;
   tier: string;
+  /** concurrent clone jobs in this process (WORKER_CONCURRENCY, default 1). Each
+   *  slot runs a full Chromium capture — size the container accordingly. */
+  concurrency: number;
 };
 
 export function loadWorkerEnv(): WorkerEnv {
   const databaseUrl = process.env.DATABASE_URL;
   if (!databaseUrl) throw new Error("DATABASE_URL is required for the worker");
+  const concurrency = Math.max(1, parseInt(process.env.WORKER_CONCURRENCY ?? "1", 10) || 1);
   return {
     databaseUrl,
     artifactsDir: process.env.ARTIFACTS_DIR ?? join(process.cwd(), "local-data", "artifacts"),
@@ -23,5 +27,6 @@ export function loadWorkerEnv(): WorkerEnv {
     harnessDir: process.env.HARNESS_DIR ?? join(process.cwd(), "local-data", "harness"),
     captureCacheDir: process.env.CAPTURE_CACHE_DIR ?? join(process.cwd(), "local-data", "capture-cache"),
     tier: process.env.VERIFY_TIER ?? "stage2",
+    concurrency,
   };
 }

packages/worker/src/worker.ts

@@ -1,3 +1,4 @@
+import { join } from "node:path";
 import { runCloneJob } from "@cloner/core";
 import { createDb, createBoss, workClone } from "@cloner/db";
 import { artifactStoreFromEnv } from "@cloner/storage";
@@ -16,13 +17,32 @@ async function main(): Promise<void> {
     store,
     runJob: runCloneJob,
     cacheTtlMs: env.cacheStaleAfterMs,
-    harnessProvider: makeHarnessProvider(env.harnessDir),
     captureCacheDir: env.captureCacheDir,
     tier: env.tier,
   };
 
-  await workClone(boss, (jobId) => processCloneJob(deps, jobId));
-  console.log(JSON.stringify({ event: "worker_started", artifactsDir: env.artifactsDir, harnessDir: env.harnessDir, cacheStaleAfterMs: env.cacheStaleAfterMs }));
+  // One harness per concurrency slot: verify builds must not share a harness dir
+  // (concurrent framework builds collide). With concurrency 1 this is the plain
+  // harnessDir, preserving existing layouts. At most `concurrency` jobs run at
+  // once (one per poller), so a free slot always exists at checkout.
+  const harnesses = Array.from({ length: env.concurrency }, (_, i) =>
+    makeHarnessProvider(env.concurrency > 1 ? join(env.harnessDir, `slot-${i}`) : env.harnessDir),
+  );
+  const freeSlots = harnesses.map((_, i) => i);
+
+  await workClone(
+    boss,
+    async (jobId) => {
+      const slot = freeSlots.pop() ?? 0;
+      try {
+        await processCloneJob({ ...deps, harnessProvider: harnesses[slot]! }, jobId);
+      } finally {
+        freeSlots.push(slot);
+      }
+    },
+    env.concurrency,
+  );
+  console.log(JSON.stringify({ event: "worker_started", concurrency: env.concurrency, artifactsDir: env.artifactsDir, harnessDir: env.harnessDir, cacheStaleAfterMs: env.cacheStaleAfterMs }));
 }
 
 main().catch((e) => {