From 0efcb8e7c5aa62060dceb48ee3742be80b091b56 Mon Sep 17 00:00:00 2001 From: "Thomas F. K. Jorna" Date: Mon, 19 Oct 2026 22:04:44 +0200 Subject: [PATCH] Compactions get their own queue on staging, 32 at once; units stay at 8 The bulk queue went from 100 jobs at once to 8 on 2026-10-16, for commit units, which hold whole leaves. Compactions share that queue, and the commit compaction of 30M records has taken 10.5-12 h since. A merge holds about 12 MB and mostly waits on R2, so push.compact and push.compactPiece now go to underlay-staging-compact, which the bulk Worker consumes 32 at a time. push.commit, refs.compact*, prod and Node are unchanged. --- packages/server/src/jobs.ts | 25 +++++++++++++++++--- packages/server/src/worker.ts | 10 ++++++-- packages/server/test/jobs.test.ts | 36 +++++++++++++++++++++++++++++ packages/server/wrangler.bulk.jsonc | 17 +++++++++++++- packages/server/wrangler.jsonc | 3 ++- 5 files changed, 84 insertions(+), 7 deletions(-) diff --git a/packages/server/src/jobs.ts b/packages/server/src/jobs.ts index 8f6cf6d7..61f510dd 100644 --- a/packages/server/src/jobs.ts +++ b/packages/server/src/jobs.ts @@ -11,6 +11,9 @@ * a whole collection, and go to their own queue, one per invocation; the rest * are small and interactive (a webhook, a file check), and run several at a * time so they never wait behind a bulk push or a mirror backfill. + * + * A push's run compactions are bulk jobs with a queue of their own where one is + * bound (COMPACT_JOBS), so that they can run more at once than commit units. */ import { and, asc, eq, lte, or, sql } from 'drizzle-orm' @@ -54,6 +57,16 @@ export const BULK_JOBS: ReadonlySet = new Set([ export const isBulk = (type: string) => BULK_JOBS.has(type) +/** + * Bulk jobs that go to the compaction queue when one is bound, else with the + * rest. A merge holds about 12 MB (runs.ts runConfig) and mostly waits on R2, + * where a unit holds whole leaves: at the bulk queue's 8 at once, the commit + * compaction of 30M records took about 10.5 h at about 30 CPU-minutes an hour + * (staging, 2026-10-19). push.commit, which claims each next round and then + * runs the commit, stays on the bulk queue. + */ +export const COMPACT_JOBS: ReadonlySet = new Set(['push.compact', 'push.compactPiece']) + const handlers = new Map() export function registerJob(type: string, handler: JobHandler): void { @@ -73,13 +86,19 @@ interface CfQueue { sendBatch(messages: { body: unknown; delaySeconds?: number }[]): Promise } -/** Jobs on Cloudflare Queues: interactive and bulk queues (one queue for both if no bulk is bound). */ +/** + * Jobs on Cloudflare Queues: interactive, bulk and compaction queues. Without a + * compaction queue compactions go to the bulk one; without a bulk queue, + * everything goes to the one queue. + */ export class QueueJobs implements Jobs { constructor( readonly queue: CfQueue, readonly bulk: CfQueue = queue, + readonly compact: CfQueue = bulk, ) {} #for(type: string) { + if (COMPACT_JOBS.has(type)) return this.compact return isBulk(type) ? this.bulk : this.queue } async enqueue(job: JobMessage, opts?: { delaySeconds?: number }) { @@ -89,13 +108,13 @@ export class QueueJobs implements Jobs { ) } async enqueueBatch(jobs: JobMessage[]) { - for (const q of [this.queue, this.bulk]) { + // Each queue once: an unbound one is the same object as the queue it falls back to. + for (const q of new Set([this.queue, this.bulk, this.compact])) { const mine = jobs.filter((j) => this.#for(j.type) === q) // sendBatch takes at most 100 messages. for (let i = 0; i < mine.length; i += 100) { await q.sendBatch(mine.slice(i, i + 100).map((body) => ({ body }))) } - if (this.bulk === this.queue) break } } } diff --git a/packages/server/src/worker.ts b/packages/server/src/worker.ts index bcfff27a..564033ae 100644 --- a/packages/server/src/worker.ts +++ b/packages/server/src/worker.ts @@ -34,6 +34,8 @@ export interface Env { JOBS: Queue /** Bulk jobs (jobs.ts BULK_JOBS); without it, everything goes to JOBS. */ JOBS_BULK?: Queue + /** Run compactions (jobs.ts COMPACT_JOBS); without it, they go with the other bulk jobs. */ + JOBS_COMPACT?: Queue /** Usage events (billing/usage.ts), a request's to a message; without it, nothing is metered. */ USAGE?: Queue APP_URL: string @@ -141,7 +143,11 @@ function makePorts(env: Env, ctx: ExecutionContext, req?: Request): Ports { }), cache, signer: () => (signer ??= ed25519Signer(env.SIGNING_KEY)), - jobs: new QueueJobs(env.JOBS as never, (env.JOBS_BULK ?? env.JOBS) as never), + jobs: new QueueJobs( + env.JOBS as never, + (env.JOBS_BULK ?? env.JOBS) as never, + (env.JOBS_COMPACT ?? env.JOBS_BULK ?? env.JOBS) as never, + ), waitUntil: (p) => ctx.waitUntil(p), outboundFetch: (url, init) => fetch(url, init), locationFetch: (req) => fetch(req), @@ -258,7 +264,7 @@ export default { msg.retry({ delaySeconds: Math.min(3600, 2 ** msg.attempts) }) } } - // The bulk queue delivers one job per invocation (wrangler.jsonc); an + // The bulk and compaction queues deliver one job per invocation (wrangler.bulk.jsonc); an // interactive batch's small jobs run a few at a time, sharing its CPU. const queue = [...batch.messages] const lanes = batch.messages.some((m) => isBulk(m.body.type)) ? 1 : INTERACTIVE_LANES diff --git a/packages/server/test/jobs.test.ts b/packages/server/test/jobs.test.ts index 1daaf187..2ff1c8f3 100644 --- a/packages/server/test/jobs.test.ts +++ b/packages/server/test/jobs.test.ts @@ -50,6 +50,42 @@ describe('job queues', () => { expect(one.sent).toEqual(['commit.unit', 'files.verify']) }) + it('sends compactions to the compaction queue, or with the bulk jobs when there is none', async () => { + const interactive = fakeQueue() + const bulk = fakeQueue() + const compact = fakeQueue() + const jobs = new QueueJobs(interactive, bulk, compact) + await jobs.enqueue({ type: 'push.compact', sessionId: 's', seq: 2 }) + await jobs.enqueueBatch([ + { type: 'push.compactPiece', sessionId: 's', seq: 2, piece: 0 }, + { type: 'push.commit', sessionId: 's' }, + { type: 'commit.unit', unitId: 'u' }, + { type: 'webhooks.deliver', deliveryId: 'd' }, + ...Array.from({ length: 150 }, (_, piece) => ({ + type: 'push.compactPiece', + sessionId: 's', + seq: 3, + piece, + })), + ]) + expect(compact.sent).toHaveLength(152) + expect(compact.sent.slice(0, 2)).toEqual(['push.compact', 'push.compactPiece']) + // The commit, which claims each next round of compactions, stays with the units. + expect(bulk.sent).toEqual(['push.commit', 'commit.unit']) + expect(interactive.sent).toEqual(['webhooks.deliver']) + + // No compaction queue bound: compactions go with the other bulk jobs, once each. + const one = fakeQueue() + const two = fakeQueue() + await new QueueJobs(one, two).enqueueBatch([ + { type: 'push.compactPiece', sessionId: 's', seq: 2, piece: 0 }, + { type: 'commit.unit', unitId: 'u' }, + { type: 'webhooks.deliver', deliveryId: 'd' }, + ]) + expect(two.sent).toEqual(['push.compactPiece', 'commit.unit']) + expect(one.sent).toEqual(['webhooks.deliver']) + }) + it('runs interactive jobs before bulk ones on Node', async () => { const h = await harness() const ran: string[] = [] diff --git a/packages/server/wrangler.bulk.jsonc b/packages/server/wrangler.bulk.jsonc index 8d64cbdd..2f368f0a 100644 --- a/packages/server/wrangler.bulk.jsonc +++ b/packages/server/wrangler.bulk.jsonc @@ -1,6 +1,7 @@ // The bulk job Worker: the same code as wrangler.jsonc's Worker, deployed as a // second Worker that consumes only the bulk queue (jobs.ts BULK_JOBS: commits, -// compactions, indexing, mirrors, cleanup). It serves no routes. +// compactions, indexing, mirrors, cleanup) and, on staging, the compaction +// queue (jobs.ts COMPACT_JOBS). It serves no routes. // // Why a second Worker: Workers memory is 128 MB per isolate, and an isolate // runs several invocations at once. Bulk jobs are sized to share an isolate @@ -18,6 +19,9 @@ // the same values as the main Worker's: // R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY, SIGNING_KEY, LOCATION_KEY // +// Queues are made by hand before a deploy names them; the compaction queue: +// wrangler queues create underlay-staging-compact +// // Keep bindings and vars in step with wrangler.jsonc. { "$schema": "../../node_modules/wrangler/config-schema.json", @@ -58,6 +62,7 @@ "producers": [ { "binding": "JOBS", "queue": "underlay-staging-jobs" }, { "binding": "JOBS_BULK", "queue": "underlay-staging-bulk" }, + { "binding": "JOBS_COMPACT", "queue": "underlay-staging-compact" }, { "binding": "USAGE", "queue": "underlay-staging-usage" }, ], "consumers": [ @@ -68,6 +73,16 @@ "max_retries": 10, "dead_letter_queue": "underlay-staging-dead", }, + // Run compactions, more at once than units: a merge holds about 12 MB and + // mostly waits on R2 (jobs.ts COMPACT_JOBS). Prod has no compaction queue; + // its bulk queue runs 100 at once. + { + "queue": "underlay-staging-compact", + "max_batch_size": 1, + "max_concurrency": 32, + "max_retries": 10, + "dead_letter_queue": "underlay-staging-dead", + }, ], }, }, diff --git a/packages/server/wrangler.jsonc b/packages/server/wrangler.jsonc index fed74fa5..a594714b 100644 --- a/packages/server/wrangler.jsonc +++ b/packages/server/wrangler.jsonc @@ -88,6 +88,7 @@ "producers": [ { "binding": "JOBS", "queue": "underlay-staging-jobs" }, { "binding": "JOBS_BULK", "queue": "underlay-staging-bulk" }, + { "binding": "JOBS_COMPACT", "queue": "underlay-staging-compact" }, { "binding": "USAGE", "queue": "underlay-staging-usage" }, ], "consumers": [ @@ -98,7 +99,7 @@ "max_retries": 10, "dead_letter_queue": "underlay-staging-dead", }, - // The bulk queue's consumer is the bulk Worker (wrangler.bulk.jsonc). + // The bulk and compaction queues' consumer is the bulk Worker (wrangler.bulk.jsonc). // Usage events (billing/usage.ts): up to 100 requests' events per log object. { "queue": "underlay-staging-usage",