Actually, here's the code (MIT / Public Domain): Worker Side: ```ts import type { Id } from "../convex/_generated/dataModel"; import type { ConvexClient } from "convex/browser"; import type { FunctionReference } from "convex/server"; interface QueueApi { hasWork: FunctionReference<"query", "public", { workerKey: string }, boolean>; stealWork: FunctionReference< "mutation", "public", { workerKey: string }, { taskId: Id<"workerWip">; queueItem: Task; } | null >; markDone: FunctionReference< "mutation", "public", { workerKey: string; workId: Id<"workerWip">; error?: string | undefined }, boolean >; clearQueue: FunctionReference<"mutation", "public", { workerKey: string }, number>; } export async function runQueue( convex: ConvexClient, workerKey: string, concurrency: number, queue: QueueApi, process: (t: Task) => Promise, ): Promise { console.log("clearing queue, just in case"); const numDeleted = await convex.mutation(queue.clearQueue, { workerKey }); console.log("Removed", numDeleted, "docs with wip status"); const limiter = new ConcurrencyLimiter(concurrency); while (true) { await limiter.limit(); const job = await stealWork(convex, workerKey, queue); const promise = process(job.queueItem) .then(async () => { await convex.mutation(queue.markDone, { workerKey, workId: job.taskId, }); }) .catch(async (e) => { console.error(e); await convex.mutation(queue.markDone, { workerKey, workId: job.taskId, error: String(e), }); }); limiter.addPromise(promise); } } async function stealWork( convex: ConvexClient, workerKey: string, queue: QueueApi, ): Promise<{ taskId: Id<"workerWip">; queueItem: Task; }> { while (true) { const item = await convex.mutation(queue.stealWork, { workerKey, }); if (item != null) { return item; } const { promise, resolve } = Promise.withResolvers(); console.log("Waiting for job on queue"); const unsubscribe = convex.onUpdate(queue.hasWork, { workerKey }, (update) => { if (update) resolve(); }); await promise; console.log("got potential job :)"); unsubscribe(); } } class ConcurrencyLimiter { #inFlight = new Set>(); #maxParallel: number; #promise = Promise.withResolvers(); constructor(maxParallel: number) { this.#maxParallel = maxParallel; } addPromise(p: Promise) { this.#inFlight.add(p); void p.finally(() => { this.#inFlight.delete(p); if (this.#inFlight.size < this.#maxParallel) { this.#promise.resolve(); this.#promise = Promise.withResolvers(); } }); } limit() { if (this.#inFlight.size < this.#maxParallel) { return Promise.resolve(); } return this.#promise.promise; } } ```

Work Stealing API

Similar Threads

Similar Threads

Similar Threads