From 4c1dea77e1246de21c644a5141ab6b0e65841f61 Mon Sep 17 00:00:00 2001 From: Samraaj Bath Date: Mon, 6 Jul 2026 22:44:42 -0700 Subject: [PATCH] Parallelize the clone queue: in-flight dedup + WORKER_CONCURRENCY MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- packages/api/src/backends/db.ts | 7 +++++++ packages/db/src/queue.ts | 28 +++++++++++++++++++--------- packages/db/src/repo.ts | 15 ++++++++++++++- packages/worker/src/env.ts | 5 +++++ packages/worker/src/worker.ts | 26 +++++++++++++++++++++++--- 5 files changed, 68 insertions(+), 13 deletions(-) diff --git a/packages/api/src/backends/db.ts b/packages/api/src/backends/db.ts index cee0def..8b82b0a 100644 --- a/packages/api/src/backends/db.ts +++ b/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 }); diff --git a/packages/db/src/queue.ts b/packages/db/src/queue.ts index 057a5e4..44126f1 100644 --- a/packages/db/src/queue.ts +++ b/packages/db/src/queue.ts @@ -31,13 +31,23 @@ export async function enqueueClone(boss: PgBoss, jobId: string): Promise Promise): Promise { - 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, concurrency = 1): Promise { + 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; } diff --git a/packages/db/src/repo.ts b/packages/db/src/repo.ts index 0f7f089..d452f01 100644 --- a/packages/db/src/repo.ts +++ b/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 { 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 { + 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 { const rows = await db.delete(jobs).where(eq(jobs.id, id)).returning({ id: jobs.id }); return rows.length > 0; diff --git a/packages/worker/src/env.ts b/packages/worker/src/env.ts index ee57026..9b5f092 100644 --- a/packages/worker/src/env.ts +++ b/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, }; } diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts index b02760f..8014998 100644 --- a/packages/worker/src/worker.ts +++ b/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 { 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) => {