Drain Cloudflare publication in bounded jobs

Author: chubes4

packages/runtime-cloudflare/README.md

@@ -18,7 +18,7 @@ The same revision transaction persists bounded user-managed files under `wp-cont
 
 The Worker forwards browser cookies directly to Playground and disables Playground's internal cookie store, preventing an empty per-isolate store from replacing a valid browser session after cold restart.
 
-A minute Cloudflare Cron Trigger drains at most five due WordPress events or 25 seconds of work per invocation. Each event receives its own coordinator lease and canonical R2 revision transaction. Scheduled publication callbacks use the same affected-route compiler and post-commit conditional promotion as browser mutations. Direct `/wp-cron.php` requests are disabled so callbacks cannot bypass that transaction boundary.
+A minute Cloudflare Cron Trigger first drains at most one publication route, or otherwise drains at most five due WordPress events or 25 seconds of work. Canonical mutations persist immutable state, enqueue a revision- and coordinator-version-bound publication job, and return `x-wp-codebox-publication: queued` when public routes need work; they never render routes in the mutation request. A publication job boots only its committed canonical revision, writes one immutable page snapshot per scheduled invocation, then conditionally promotes a complete descriptor. A stale job cannot overwrite a later canonical version, and failed work leaves the prior publication readable for retry. Direct `/wp-cron.php` requests are disabled so callbacks cannot bypass that transaction boundary.
 
 The bundled MDI source is pinned to immutable commit `bf6d434d1673fdd86d777501f7eaec292d32ad1f`, including [MDI PR #141](https://github.com/Automattic/markdown-database-integration/pull/141). The bundle generator, worker provenance, and source-contract test use the same revision.
 
@@ -31,6 +31,6 @@ Canonical Cloudflare boots patch only the assembled PHP MEMFS copy of `/wordpres
 2. Run `npm run provision:cloudflare-wordpress-runtime-corpus -- --local --persist-to <directory>` to verify and upload all exact content-addressed artifacts into isolated local R2 storage. For an authorized deployment, run the provisioner with `--remote` and require every upload to succeed before deploying the Worker that imports their manifests.
 3. Run `npm run test:cloudflare-runtime` for routing, coordinator composition, canonical-state, artifact validation, source contract, and TypeScript coverage.
 4. Run `npm run cloudflare:dry-run` and `npm run cloudflare:dry-run:d1` to compile the Durable Object and D1 profiles without creating Cloudflare resources. The placeholder D1 database ID is for local/dry-run verification; an implementation supplies its provisioned binding at deployment.
-5. Run `npm run cloudflare:local-gate` and `npm run cloudflare:local-gate:d1` for the same isolated workerd workflow through both coordinator implementations. Each gate generates and provisions all artifacts, verifies the selected backend through the state envelope, injects stable test-only admin-password, auth-secret, and operator-token values, uploads and activates a real plugin ZIP, establishes an anonymous homepage and canonical-permalink publication, updates a published post through authenticated REST, and proves that the affected permalink and homepage promote automatically. The cron gate also proves that a scheduled post becomes readable from the publication path without an operator call. Restart verification requires the immutable R2 publication, while the remaining coverage includes login, concurrent canonical writes, media, representative frontend/admin/editor assets, PHP diagnostics, session recovery, and a fresh login after restart.
+5. Run `npm run cloudflare:local-gate` and `npm run cloudflare:local-gate:d1` for the same isolated workerd workflow through both coordinator implementations. Each gate generates and provisions all artifacts, verifies the selected backend through the state envelope, injects stable test-only admin-password, auth-secret, and operator-token values, uploads and activates a real plugin ZIP, establishes an anonymous homepage and canonical-permalink publication, updates a published post through authenticated REST without immediate rendering, restarts with pending publication work, and proves bounded scheduled draining eventually exposes the updated permalink through coordinator-free R2. The cron gate also proves that a scheduled post becomes readable from the publication path without an operator call. Restart verification requires the immutable R2 publication, while the remaining coverage includes login, concurrent canonical writes, media, representative frontend/admin/editor assets, PHP diagnostics, session recovery, and a fresh login after restart.
 
 This document describes local candidate verification only. It does not claim remote deployment.

packages/runtime-cloudflare/src/d1-revision-coordinator.ts

@@ -38,6 +38,10 @@ export class D1RevisionCoordinator implements RevisionCoordinator {
     return commitStateLease(this.database, lease, pointer)
   }
 
+  committed(version: number): Promise<MarkdownPointer | null> {
+    return readCommittedPointer(this.database, version)
+  }
+
   async reset(): Promise<void> {
     await resetWordPressState(this.database)
   }
@@ -86,24 +90,41 @@ async function commitStateLease(database: D1Database, lease: RevisionLease, poin
   validatePointer(pointer)
   await ensureSchema(database)
   const baseRevision = lease.pointer?.revision ?? null
-  const result = await database.prepare(`UPDATE wp_codebox_state
+  const update = database.prepare(`UPDATE wp_codebox_state
     SET revision = ?, manifest_key = ?, persisted_at = ?, version = version + 1,
         lease_token = NULL, lease_base_revision = NULL, lease_version = NULL, lease_expires_at = NULL
     WHERE site_id = ? AND lease_token = ? AND lease_expires_at > ? AND version = ? AND lease_version = ?
       AND ((revision IS NULL AND ? IS NULL) OR revision = ?)
       AND ((lease_base_revision IS NULL AND ? IS NULL) OR lease_base_revision = ?)`)
     .bind(pointer.revision, pointer.manifestKey, pointer.persistedAt, SITE_ID, lease.token, Date.now(), lease.version, lease.version,
-      baseRevision, baseRevision, baseRevision, baseRevision).run()
-  if (result.meta.changes !== 1) throw new RevisionConflict("The canonical pointer changed before D1 promotion.")
+      baseRevision, baseRevision, baseRevision, baseRevision)
+  const receipt = database.prepare(`INSERT OR IGNORE INTO wp_codebox_commits (site_id, version, revision, manifest_key, persisted_at)
+    SELECT ?, ?, ?, ?, ? FROM wp_codebox_state WHERE site_id = ? AND version = ? AND revision = ?`)
+    .bind(SITE_ID, lease.version + 1, pointer.revision, pointer.manifestKey, pointer.persistedAt, SITE_ID, lease.version + 1, pointer.revision)
+  const [result, receiptResult] = await database.batch([update, receipt])
+  if (result.meta.changes !== 1 || receiptResult.meta.changes !== 1) throw new RevisionConflict("The canonical pointer changed before D1 promotion.")
   return { pointer, version: lease.version + 1 }
 }
 
+async function readCommittedPointer(database: D1Database, version: number): Promise<MarkdownPointer | null> {
+  if (!Number.isSafeInteger(version) || version < 1) throw new RevisionConflict("A canonical commit version is required.")
+  await ensureSchema(database)
+  const row = await database.prepare(`SELECT revision, manifest_key, persisted_at FROM wp_codebox_commits WHERE site_id = ? AND version = ?`).bind(SITE_ID, version).first<{ revision: string; manifest_key: string; persisted_at: string }>()
+  if (!row) return null
+  const pointer = { revision: row.revision, manifestKey: row.manifest_key, persistedAt: row.persisted_at }
+  validatePointer(pointer)
+  return pointer
+}
+
 async function resetWordPressState(database: D1Database): Promise<{ reset: true }> {
   await ensureSchema(database)
-  await database.prepare(`UPDATE wp_codebox_state
-    SET revision = NULL, manifest_key = NULL, persisted_at = NULL, version = version + 1,
-        lease_token = NULL, lease_base_revision = NULL, lease_version = NULL, lease_expires_at = NULL
-    WHERE site_id = ?`).bind(SITE_ID).run()
+  await database.batch([
+    database.prepare(`UPDATE wp_codebox_state
+      SET revision = NULL, manifest_key = NULL, persisted_at = NULL, version = version + 1,
+          lease_token = NULL, lease_base_revision = NULL, lease_version = NULL, lease_expires_at = NULL
+      WHERE site_id = ?`).bind(SITE_ID),
+    database.prepare(`DELETE FROM wp_codebox_commits WHERE site_id = ?`).bind(SITE_ID),
+  ])
   return { reset: true }
 }
 
@@ -156,6 +177,7 @@ function ensureSchema(database: D1Database): Promise<void> {
       lease_expires_at INTEGER
     )`).run()
     await database.prepare(`INSERT OR IGNORE INTO wp_codebox_state (site_id, version) VALUES (?, 0)`).bind(SITE_ID).run()
+    await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_commits (site_id TEXT NOT NULL, version INTEGER NOT NULL, revision TEXT NOT NULL, manifest_key TEXT NOT NULL, persisted_at TEXT NOT NULL, PRIMARY KEY (site_id, version))`).run()
   })()
   schemaReady.set(key, pending)
   pending.catch(() => schemaReady.delete(key))

packages/runtime-cloudflare/src/published-reader.ts

@@ -1,4 +1,5 @@
-export const PUBLISHED_REVISION_SCHEMA = "wp-codebox/published-revision/v2" as const
+export const PUBLISHED_REVISION_SCHEMA = "wp-codebox/published-revision/v3" as const
+const PREVIOUS_PUBLISHED_REVISION_SCHEMA = "wp-codebox/published-revision/v2"
 const LEGACY_PUBLISHED_REVISION_SCHEMA = "wp-codebox/published-revision/v1"
 export const PUBLISHED_PAGE_SCHEMA = "wp-codebox/wordpress-page/v2" as const
 export const R2_PUBLISHED_CURRENT_KEY = "sites/default/publications/current.json"
@@ -17,6 +18,7 @@ export interface PublishedRevision {
   schema: typeof PUBLISHED_REVISION_SCHEMA
   revision: string
   canonicalRevision: string
+  canonicalVersion: number
   publishedAt: string
   routes: PublishedRoute[]
 }
@@ -43,8 +45,10 @@ export function validatePublishedRevision(value: unknown): PublishedRevision {
   if (!value || typeof value !== "object") throw new Error("Published revision is invalid.")
   const revision = value as Partial<PublishedRevision>
   const legacy = (value as { schema?: unknown }).schema === LEGACY_PUBLISHED_REVISION_SCHEMA
-  if (![PUBLISHED_REVISION_SCHEMA, LEGACY_PUBLISHED_REVISION_SCHEMA].includes(revision.schema as string) || !isRevision(revision.revision) || !isRevision(revision.canonicalRevision)
+  const current = revision.schema === PUBLISHED_REVISION_SCHEMA
+  if (![PUBLISHED_REVISION_SCHEMA, PREVIOUS_PUBLISHED_REVISION_SCHEMA, LEGACY_PUBLISHED_REVISION_SCHEMA].includes(revision.schema as string) || !isRevision(revision.revision) || !isRevision(revision.canonicalRevision)
     || typeof revision.publishedAt !== "string" || !Number.isFinite(Date.parse(revision.publishedAt)) || !Array.isArray(revision.routes)
+    || (current && (typeof revision.canonicalVersion !== "number" || !Number.isSafeInteger(revision.canonicalVersion) || revision.canonicalVersion < 0))
     || revision.routes.length === 0 || revision.routes.length > MAX_PUBLISHED_ROUTES) throw new Error("Published revision is invalid.")
   let previous = ""
   const normalizedRoutes: PublishedRoute[] = []
@@ -57,7 +61,7 @@ export function validatePublishedRevision(value: unknown): PublishedRevision {
     previous = route.route
     normalizedRoutes.push({ route: route.route, objectKey: route.objectKey, canonicalRevision: routeRevision })
   }
-  return { ...revision, schema: PUBLISHED_REVISION_SCHEMA, routes: normalizedRoutes } as PublishedRevision
+  return { ...revision, schema: PUBLISHED_REVISION_SCHEMA, canonicalVersion: current ? revision.canonicalVersion! : 0, routes: normalizedRoutes } as PublishedRevision
 }
 
 export function normalizePublishedRoutes(value: unknown): string[] {

packages/runtime-cloudflare/src/revision-coordinator.ts

@@ -24,6 +24,7 @@ export interface RevisionCoordinator {
   release(lease: RevisionLease): Promise<void>
   abort(lease: RevisionLease): Promise<void>
   commit(lease: RevisionLease, pointer: MarkdownPointer): Promise<{ pointer: MarkdownPointer; version: number }>
+  committed(version: number): Promise<MarkdownPointer | null>
   reset(): Promise<void>
 }
 

packages/runtime-cloudflare/src/state-coordinator.ts

@@ -46,6 +46,10 @@ export class DurableObjectRevisionCoordinator implements RevisionCoordinator {
     return this.call("commit", { token: lease.token, baseRevision: lease.pointer?.revision ?? null, version: lease.version, pointer })
   }
 
+  committed(version: number): Promise<MarkdownPointer | null> {
+    return this.call("committed", { version })
+  }
+
   async reset(): Promise<void> {
     await this.call("reset", {})
   }
@@ -98,6 +102,7 @@ export class WordPressStateCoordinator implements DurableObject {
     if (action === "release") return Response.json(await this.release(body))
     if (action === "abort") return Response.json(await this.abort(body))
     if (action === "commit") return Response.json(await this.commit(body))
+    if (action === "committed") return Response.json(await this.committed(body))
     if (action === "reset") return Response.json(await this.reset())
     return new Response("Unknown coordinator action.", { status: 404 })
   }
@@ -147,10 +152,18 @@ export class WordPressStateCoordinator implements DurableObject {
     record.pointer = pointer
     record.version++
     delete record.lease
-    await this.save(record)
+    await this.state.storage.transaction(async (transaction) => {
+      await transaction.put(`wordpress-state-commit/${record.version}`, pointer)
+      await transaction.put(STORAGE_KEY, record)
+    })
     return { pointer, version: record.version }
   }
 
+  private async committed(body: Record<string, unknown>): Promise<MarkdownPointer | null> {
+    if (!Number.isSafeInteger(body.version) || (body.version as number) < 1) throw new RevisionConflict("A canonical commit version is required.")
+    return await this.state.storage.get<MarkdownPointer>(`wordpress-state-commit/${body.version}`) ?? null
+  }
+
   private requireLease(record: CoordinatorRecord, body: Record<string, unknown>): StoredLease {
     const lease = record.lease
     if (!lease || lease.expiresAt <= Date.now()) throw new RevisionConflict("The canonical WordPress lease has expired.")