Harden Cloudflare queue dispatch semantics

Author: chubes4

package.json

@@ -111,8 +111,8 @@
     "test:redaction": "tsx tests/redaction.test.ts",
     "test:browser-preview-routing": "tsx --test tests/browser-preview-routing.test.ts",
     "test:browser-routed-command-security": "tsx --test tests/browser-routed-command-security.test.ts",
-    "test:cloudflare-runtime": "node --test tests/cloudflare-d1-provisioner.test.mjs && tsx tests/cloudflare-site-context.test.ts && tsx tests/cloudflare-coordinator-site-partitioning.test.ts && tsx tests/cloudflare-d1-operation-repository.test.ts && tsx tests/cloudflare-allocation-lifecycle.test.ts && tsx tests/cloudflare-provisioning-api.test.ts && tsx tests/cloudflare-runtime.test.ts && tsx tests/cloudflare-public-reader.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit",
-    "test:cloudflare-queue": "tsx tests/cloudflare-d1-operation-repository.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit",
+    "test:cloudflare-runtime": "node --test tests/cloudflare-d1-provisioner.test.mjs && tsx tests/cloudflare-site-context.test.ts && tsx tests/cloudflare-coordinator-site-partitioning.test.ts && tsx tests/cloudflare-d1-operation-repository.test.ts && tsx tests/cloudflare-allocation-lifecycle.test.ts && tsx tests/cloudflare-provisioning-api.test.ts && tsx tests/cloudflare-queue-batch.test.ts && tsx tests/cloudflare-runtime.test.ts && tsx tests/cloudflare-public-reader.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit",
+    "test:cloudflare-queue": "tsx tests/cloudflare-queue-batch.test.ts && tsx tests/cloudflare-d1-operation-repository.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit",
     "test:cloudflare-administrator-claim": "tsx tests/cloudflare-provisioning-api.test.ts",
     "test:cloudflare-wordpress-auth": "tsx tests/cloudflare-wordpress-auth.test.ts",
     "test:cloudflare-wordpress-archive-corpus": "tsx tests/cloudflare-wordpress-archive-corpus.test.ts",

packages/runtime-cloudflare/src/allocation-lifecycle.ts

@@ -102,6 +102,7 @@ export class CloudflareAllocationLifecycle {
       ])
     }
     if (await tableExists(this.database, "wp_codebox_api_admin_claims")) await this.database.prepare("UPDATE wp_codebox_api_admin_claims SET state = 'expired', updated_at = ? WHERE site_id = ? AND state = 'pending'").bind(now, identity.siteId).run()
+    if (await tableExists(this.database, "wp_codebox_runtime_dispatches")) await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'done', last_error = 'Allocation deletion fenced this dispatch.', completed_at = ?, updated_at = ? WHERE site_id = ? AND state NOT IN ('done','dead-letter')").bind(now, now, identity.siteId).run()
   }
   private async cleanupWork(siteId: string, limit: number): Promise<{ deleted: number; unresolved: number }> {
     let deleted = 0
@@ -114,7 +115,10 @@ export class CloudflareAllocationLifecycle {
     if (await tableExists(this.database, "wp_codebox_api_admin_claims")) {
       const result = await this.database.prepare("DELETE FROM wp_codebox_api_admin_claims WHERE site_id = ?").bind(siteId).run(); deleted += result.meta.changes
     }
-    const tables = ["wp_codebox_operation_attempts", "wp_codebox_operations", "wp_codebox_api_admin_claims"]
+    for (const table of ["wp_codebox_runtime_dead_letters", "wp_codebox_runtime_dispatches", "wp_codebox_runtime_fairness"]) if (await tableExists(this.database, table)) {
+      const result = await this.database.prepare(`DELETE FROM ${table} WHERE site_id = ?`).bind(siteId).run(); deleted += result.meta.changes
+    }
+    const tables = ["wp_codebox_operation_attempts", "wp_codebox_operations", "wp_codebox_api_admin_claims", "wp_codebox_runtime_dead_letters", "wp_codebox_runtime_dispatches", "wp_codebox_runtime_fairness"]
     let unresolved = 0
     for (const table of tables) if (await tableExists(this.database, table)) unresolved += (await this.database.prepare(`SELECT COUNT(*) AS count FROM ${table} WHERE site_id = ?`).bind(siteId).first<{ count: number }>())?.count ?? 0
     return { deleted, unresolved }

packages/runtime-cloudflare/src/d1-operation-repository.ts

@@ -1,5 +1,5 @@
 import { D1OperationRepository, OperationConflict, type StaticArtifactOperation, type StaticArtifactOperationInput } from "./d1-operation-repository.js"
-import type { RuntimeQueue } from "./queue-dispatch.js"
+import { parseRuntimeQueuePolicy, runtimeQueueMessage, type RuntimeQueue } from "./queue-dispatch.js"
 import { MAX_STATIC_ARTIFACT_BYTES, readBoundedRequestBytes, readStaticArtifactImport, StaticArtifactImportError, validateStaticArtifact } from "./static-artifact-import.js"
 import { allocatePreviewSiteContext, parseSiteContexts, previewDomain, siteStorageKeys, type SiteContext } from "./site-context.js"
 import { deriveSiteCredential } from "./wordpress-auth.js"
@@ -18,7 +18,7 @@ export interface ProvisioningAllocation {
   artifactSha256: string; artifactSize: number; options: StaticArtifactOperationInput["options"]
 }
 interface CreateInput { key: string; fingerprint: string; artifactSha256: string; artifactSize: number; options: StaticArtifactOperationInput["options"] }
-export interface ProvisioningEnv { WORDPRESS_STATE_DATABASE: D1Database; WORDPRESS_STATE_BUCKET: R2Bucket; WORDPRESS_RUNTIME_QUEUE?: RuntimeQueue; WORDPRESS_SITE_CONTEXTS?: string; WORDPRESS_PREVIEW_DOMAIN?: string; WORDPRESS_PREVIEW_HOST_SECRET?: string; WORDPRESS_API_TOKENS?: string; WORDPRESS_ADMIN_CLAIM_SECRET?: string; WORDPRESS_ADMIN_PASSWORD?: string }
+export interface ProvisioningEnv { WORDPRESS_STATE_DATABASE: D1Database; WORDPRESS_STATE_BUCKET: R2Bucket; WORDPRESS_RUNTIME_QUEUE?: RuntimeQueue; WORDPRESS_QUEUE_POLICY?: string; WORDPRESS_SITE_CONTEXTS?: string; WORDPRESS_PREVIEW_DOMAIN?: string; WORDPRESS_PREVIEW_HOST_SECRET?: string; WORDPRESS_API_TOKENS?: string; WORDPRESS_ADMIN_CLAIM_SECRET?: string; WORDPRESS_ADMIN_PASSWORD?: string }
 
 export async function routeProvisioningApi(request: Request, env: ProvisioningEnv, operations: D1OperationRepository): Promise<Response> {
   const parts = new URL(request.url).pathname.split("/").filter(Boolean)
@@ -39,9 +39,10 @@ export async function routeProvisioningApi(request: Request, env: ProvisioningEn
   return notFound()
 }
 
-async function dispatchOperation(env: ProvisioningEnv, operations: D1OperationRepository, site: SiteContext, operation: StaticArtifactOperation): Promise<void> {
-  const message = await operations.stageDispatch(site, "operation", operation.operationId)
-  try { if (!env.WORDPRESS_RUNTIME_QUEUE) throw new Error("Runtime queue binding is unavailable."); await env.WORDPRESS_RUNTIME_QUEUE.send(message); await operations.deliveredDispatch(message) } catch (error) { await operations.failedDispatch(message, error); console.error("Runtime queue producer is backpressured; reconciliation will retry.", error) }
+async function dispatchOperation(env: ProvisioningEnv, operations: D1OperationRepository, site: SiteContext, operation: StaticArtifactOperation, principal: string): Promise<"queued" | "backpressured"> {
+  const message = await operations.stageDispatch(site, "operation", operation.operationId, principal)
+  if (!await operations.admitDispatch(message, parseRuntimeQueuePolicy(env.WORDPRESS_QUEUE_POLICY))) return "backpressured"
+  try { if (!env.WORDPRESS_RUNTIME_QUEUE) throw new Error("Runtime queue binding is unavailable."); await env.WORDPRESS_RUNTIME_QUEUE.send(message); await operations.deliveredDispatch(message); return "queued" } catch (error) { await operations.failedDispatch(message, error); console.error("Runtime queue producer is backpressured; reconciliation will retry.", error); return "backpressured" }
 }
 
 async function lifecycleStatus(request: Request, env: ProvisioningEnv, siteId: string): Promise<Response> {
@@ -116,7 +117,9 @@ async function create(request: Request, env: ProvisioningEnv, operations: D1Oper
   try {
     const operation = await resumeProvisioningAllocation(env, site, operations)
     const claim = await new AdministratorClaimStore(env.WORDPRESS_STATE_DATABASE).issue(allocation, env)
-    return siteResource(site, operation, 202, claim)
+    const response = siteResource(site, operation, 202, claim)
+    if (operation) response.headers.set("x-wp-codebox-dispatch", dispatchHeader(await operations.dispatchState(runtimeQueueMessage(site, allocationIdentity(site.id).generation, "operation", operation.operationId))))
+    return response
   } catch (error) { if (error instanceof OperationConflict) return apiError(409, "idempotency_conflict", error.message); if (error instanceof AdministratorClaimError) return apiError(409, "administrator_claim_unavailable", "The administrator claim cannot continue provisioning."); throw error }
 }
 
@@ -159,9 +162,11 @@ async function importSite(request: Request, env: ProvisioningEnv, operations: D1
     const input = await readStaticArtifactImport(request, env.WORDPRESS_STATE_BUCKET, site)
     if (input.idempotencyKey !== key) return apiError(409, "idempotency_conflict", "Idempotency-Key must match the import request.")
     const result = await operations.createOrConverge(site, { ...input, artifact: input.artifactReference })
-    await dispatchOperation(env, operations, site, result.operation)
+    const dispatch = await dispatchOperation(env, operations, site, result.operation, token.principal)
     await store.linkOperation(token.principal, siteId, result.operation.operationId, "import", key)
-    return operationResource(siteId, result.operation, 202)
+    const response = operationResource(siteId, await operations.get(siteId, result.operation.operationId) ?? result.operation, 202)
+    response.headers.set("x-wp-codebox-dispatch", dispatch)
+    return response
   } catch (error) { return error instanceof OperationConflict ? apiError(409, "operation_conflict", error.message) : importError(error) }
 }
 
@@ -194,10 +199,12 @@ export async function resumeProvisioningAllocation(env: ProvisioningEnv, site: S
   const result = await operations.createOrConverge(site, input)
   await store.bindOperation(allocation, result.operation.operationId)
   await store.linkOperation(allocation.principal, site.id, result.operation.operationId, "provision", allocation.key)
-  await dispatchOperation(env, operations, site, result.operation)
-  return result.operation
+  await dispatchOperation(env, operations, site, result.operation, allocation.principal)
+  return await operations.get(site.id, result.operation.operationId) ?? result.operation
 }
 
+function dispatchHeader(state: string | null): "queued" | "backpressured" { return state && !["backpressured", "dead-letter"].includes(state) ? "queued" : "backpressured" }
+
 async function readCreate(request: Request, bucket: R2Bucket, key: string): Promise<CreateInput> {
   const bytes = await readBoundedRequestBytes(request)
   let body: Record<string, unknown>; try { body = JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(bytes)) } catch { throw new StaticArtifactImportError("Provisioning request must be valid UTF-8 JSON.", 400) }

packages/runtime-cloudflare/src/provisioning-api.ts

@@ -0,0 +1,33 @@
+import type { RuntimeQueueMessage } from "./queue-dispatch.js"
+
+export interface QueueDelivery { body: unknown; attempts: number; ack(): void; retry(options?: { delaySeconds?: number }): void }
+export interface ParsedQueueDelivery { raw: QueueDelivery; value: RuntimeQueueMessage }
+export type QueueExecutionResult = "ack" | "retry" | { retryAfterSeconds: number }
+
+/** One heavyweight turn per site keeps a delivery batch fair without cross-site serialization. */
+export async function dispatchQueueBatch(
+  deliveries: QueueDelivery[],
+  parse: (body: unknown) => RuntimeQueueMessage | null,
+  select: (siteId: string, deliveries: ParsedQueueDelivery[]) => Promise<ParsedQueueDelivery | null>,
+  execute: (delivery: ParsedQueueDelivery) => Promise<QueueExecutionResult>,
+): Promise<void> {
+  const lanes = new Map<string, ParsedQueueDelivery[]>()
+  for (const raw of deliveries) {
+    const value = parse(raw.body)
+    if (!value) { raw.ack(); continue }
+    const lane = lanes.get(value.siteId) ?? []
+    lane.push({ raw, value })
+    lanes.set(value.siteId, lane)
+  }
+  await Promise.all([...lanes.entries()].map(async ([siteId, lane]) => {
+    const selected = await select(siteId, lane)
+    for (const delivery of lane) {
+      if (delivery !== selected) delivery.raw.retry()
+    }
+    if (!selected) return
+    const result = await execute(selected)
+    if (result === "ack") selected.raw.ack()
+    else if (result === "retry") selected.raw.retry()
+    else selected.raw.retry({ delaySeconds: result.retryAfterSeconds })
+  }))
+}

packages/runtime-cloudflare/src/queue-batch.ts

@@ -2,20 +2,41 @@ import type { SiteContext } from "./site-context.js"
 
 export const RUNTIME_QUEUE_MESSAGE_SCHEMA = "wp-codebox/runtime-dispatch/v1"
 export const RUNTIME_QUEUE_MAX_ATTEMPTS = 3
+export const DEFAULT_RUNTIME_QUEUE_POLICY: RuntimeQueuePolicy = { maxActive: 100, maxActivePerPrincipal: 10 }
 
 /** Queue messages wake durable work; they never own canonical state. */
 export type RuntimeQueueMessage = { schema: typeof RUNTIME_QUEUE_MESSAGE_SCHEMA; siteId: string; generation: number; kind: "operation" | "publication"; identity: string }
+export interface RuntimeQueuePolicy { maxActive: number; maxActivePerPrincipal: number }
+
+const operationIdentity = /^[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}$/
+const publicationIdentity = /^sites\/([a-z0-9][a-z0-9-]{0,127})\/publications\/jobs\/[0-9]{20}-[a-f0-9-]{36}\.json$/
+
+export function validRuntimeQueueIdentity(siteId: string, kind: RuntimeQueueMessage["kind"], identity: string): boolean {
+  if (kind === "operation") return operationIdentity.test(identity)
+  return publicationIdentity.test(identity) && publicationIdentity.exec(identity)?.[1] === siteId
+}
 
 export function runtimeQueueMessage(site: Pick<SiteContext, "id">, generation: number, kind: RuntimeQueueMessage["kind"], identity: string): RuntimeQueueMessage {
-  if (!/^[a-z0-9][a-z0-9-]{0,127}$/.test(site.id) || !Number.isSafeInteger(generation) || generation < 1 || !identity || identity.length > 1024) throw new Error("Runtime queue dispatch identity is invalid.")
+  if (!/^[a-z0-9][a-z0-9-]{0,127}$/.test(site.id) || !Number.isSafeInteger(generation) || generation < 1 || !validRuntimeQueueIdentity(site.id, kind, identity)) throw new Error("Runtime queue dispatch identity is invalid.")
   return { schema: RUNTIME_QUEUE_MESSAGE_SCHEMA, siteId: site.id, generation, kind, identity }
 }
 
 export function parseRuntimeQueueMessage(value: unknown): RuntimeQueueMessage | null {
   if (!value || typeof value !== "object") return null
   const message = value as Partial<RuntimeQueueMessage>
-  if (message.schema !== RUNTIME_QUEUE_MESSAGE_SCHEMA || (message.kind !== "operation" && message.kind !== "publication") || typeof message.siteId !== "string" || !/^[a-z0-9][a-z0-9-]{0,127}$/.test(message.siteId) || !Number.isSafeInteger(message.generation) || message.generation! < 1 || typeof message.identity !== "string" || !message.identity || message.identity.length > 1024) return null
+  if (message.schema !== RUNTIME_QUEUE_MESSAGE_SCHEMA || (message.kind !== "operation" && message.kind !== "publication") || typeof message.siteId !== "string" || !/^[a-z0-9][a-z0-9-]{0,127}$/.test(message.siteId) || !Number.isSafeInteger(message.generation) || message.generation! < 1 || typeof message.identity !== "string" || !validRuntimeQueueIdentity(message.siteId, message.kind, message.identity)) return null
   return message as RuntimeQueueMessage
 }
 
 export interface RuntimeQueue { send(message: RuntimeQueueMessage): Promise<void> }
+
+export function parseRuntimeQueuePolicy(value: string | undefined): RuntimeQueuePolicy {
+  if (!value) return DEFAULT_RUNTIME_QUEUE_POLICY
+  let policy: Partial<RuntimeQueuePolicy>
+  try { policy = JSON.parse(value) as Partial<RuntimeQueuePolicy> } catch { throw new Error("Runtime queue policy is invalid.") }
+  if (!Number.isSafeInteger(policy.maxActive) || !Number.isSafeInteger(policy.maxActivePerPrincipal)
+    || policy.maxActive! < 1 || policy.maxActive! > 10_000 || policy.maxActivePerPrincipal! < 1 || policy.maxActivePerPrincipal! > policy.maxActive!) {
+    throw new Error("Runtime queue policy is invalid.")
+  }
+  return policy as RuntimeQueuePolicy
+}