Compactions get their own queue on staging, 32 at once; units stay at 8

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'
 
@@ -59,6 +62,16 @@ export const BULK_JOBS: ReadonlySet<string> = 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<string> = new Set(['push.compact', 'push.compactPiece'])
+
 const handlers = new Map<string, JobHandler>()
 
 export function registerJob(type: string, handler: JobHandler): void {
@@ -78,13 +91,19 @@ interface CfQueue {
   sendBatch(messages: { body: unknown; delaySeconds?: number }[]): Promise<void>
 }
 
-/** 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 }) {
@@ -94,13 +113,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
     }
   }
 }

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

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[] = []

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",
+          },
         ],
       },
     },

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",