Merge pull request #1996 from Automattic/ops/1976-cloudflare-d1-provision

Preserve lossless D1 rollback state

Author: chubes4

packages/runtime-cloudflare/README.md

@@ -36,7 +36,9 @@ Canonical Cloudflare boots patch only the assembled PHP MEMFS copy of `/wordpres
 
 The authenticated mutation-fence endpoints provide a bounded cutover window without interrupting anonymous published reads. `POST ?phase=operator-fence-acquire` accepts `{"ttlSeconds":30}` through `{"ttlSeconds":600}` and returns an opaque token plus expiry. While active, both coordinators reject new canonical leases, reset, adoption, manual publication, scheduled publication, and cron work with a conflict. `POST ?phase=operator-fence-renew` accepts the token and a fresh bounded TTL; `POST ?phase=operator-fence-release` accepts the token. Expiry automatically reopens mutations, and status responses never expose the token.
 
-Use authenticated `GET ?phase=operator-fence-status` after acquisition to export a coherent cutover envelope. It includes the coordinator store, pointer, version, matching commit receipt, validated R2 manifest identity, fence expiry, and a `coherent` verdict. Coherence also reads every referenced canonical object and verifies its declared size and SHA-256, so an incomplete target cannot be promoted. Adopt that exact pointer/version into the target coordinator, require the target status envelope to match, then promote the target Worker before the source fence expires. The first target mutation must commit version `N+1`. Rollback uses the same sequence in reverse after fencing the active target; relying on an unfenced status read is not a lossless cutover procedure.
+Use authenticated `GET ?phase=operator-fence-status` after acquisition to export a coherent cutover envelope. It includes the selected coordinator store, pointer, version, matching commit receipt, validated R2 manifest identity, fence expiry, and a `coherent` verdict. Coherence also reads every referenced canonical object and verifies its declared size and SHA-256, so an incomplete target cannot be promoted. Adopt that exact pointer/version into the target coordinator, require the target status envelope to match, then promote the target Worker before the source fence expires. The first target mutation must commit version `N+1`.
+
+For lossless D1-to-Durable-Object rollback, keep D1 deployed as the active Worker, acquire its fence, and capture its coherent status. Address the retained Durable Object only with `coordinator=durable-object` on authenticated `operator-fence-*` and `operator-adopt` requests. Fence that selected coordinator first, then adopt the D1 pointer and version with its matching `fenceToken`; selected-coordinator adoption fails unless that fence is active and matches. Require its selected status envelope to be coherent and identical, then promote the Durable Object Worker before the D1 fence expires and release its fence. Its first mutation must commit at `N+1`. The Durable Object binding and historical `WordPressStateCoordinator` export remain in the D1 profile for this procedure. Non-operator requests always retain the active coordinator, and selectors on unsupported or unknown operator routes fail closed.
 
 ## Static Artifact Import
 

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

@@ -66,8 +66,8 @@ export class D1RevisionCoordinator implements RevisionCoordinator {
     await releaseMutationFence(this.database, this.siteId, token)
   }
 
-  adopt(pointer: MarkdownPointer, version: number): Promise<{ pointer: MarkdownPointer; version: number }> {
-    return adoptWordPressState(this.database, this.siteId, pointer, version)
+  adopt(pointer: MarkdownPointer, version: number, fenceToken?: string, requireFence = false): Promise<{ pointer: MarkdownPointer; version: number }> {
+    return adoptWordPressState(this.database, this.siteId, pointer, version, fenceToken, requireFence)
   }
 
   async reset(): Promise<void> {
@@ -202,34 +202,41 @@ async function releaseMutationFence(database: D1Database, siteId: string, token:
   return { released: true }
 }
 
-async function adoptWordPressState(database: D1Database, siteId: string, pointer: MarkdownPointer, version: number): Promise<{ pointer: MarkdownPointer; version: number }> {
+async function adoptWordPressState(database: D1Database, siteId: string, pointer: MarkdownPointer, version: number, fenceToken?: string, requireFence = false): Promise<{ pointer: MarkdownPointer; version: number }> {
   validatePointer(pointer)
   if (!Number.isSafeInteger(version) || version < 1) throw new RevisionConflict("A positive canonical version is required for D1 adoption.")
   await ensureSchema(database, siteId)
   const now = Date.now()
+  const fence = await readMutationFence(database, siteId)
+  if (requireFence && !fence.active) throw new RevisionConflict("Selected coordinator adoption requires an active cutover fence.")
+  if (fence.active && !fenceToken) throw new RevisionConflict("D1 coordinator adoption is blocked by an unmatched fence.", fence.expiresAt)
   await database.batch([
     database.prepare(`UPDATE wp_codebox_state
       SET revision = ?, manifest_key = ?, persisted_at = ?, version = ?,
           lease_token = NULL, lease_base_revision = NULL, lease_version = NULL, lease_expires_at = NULL
       WHERE site_id = ? AND (lease_token IS NULL OR lease_expires_at <= ?)
-        AND NOT EXISTS (SELECT 1 FROM wp_codebox_fences WHERE site_id = ? AND expires_at > ?)
+        AND (NOT EXISTS (SELECT 1 FROM wp_codebox_fences WHERE site_id = ? AND expires_at > ?)
+          OR EXISTS (SELECT 1 FROM wp_codebox_fences WHERE site_id = ? AND token = ? AND expires_at > ?))
         AND ((revision IS NULL AND manifest_key IS NULL AND persisted_at IS NULL)
-          OR (version = ? AND revision = ? AND manifest_key = ? AND persisted_at = ?))
+          OR (version = ? AND revision = ? AND manifest_key = ? AND persisted_at = ?)
+          OR version < ?)
         AND (NOT EXISTS (SELECT 1 FROM wp_codebox_commits WHERE site_id = ? AND version = ?)
           OR EXISTS (SELECT 1 FROM wp_codebox_commits WHERE site_id = ? AND version = ?
             AND revision = ? AND manifest_key = ? AND persisted_at = ?))`)
       .bind(pointer.revision, pointer.manifestKey, pointer.persistedAt, version, siteId, now, siteId, now,
-        version, pointer.revision, pointer.manifestKey, pointer.persistedAt,
+        siteId, fenceToken ?? null, now,
+        version, pointer.revision, pointer.manifestKey, pointer.persistedAt, version,
         siteId, version, siteId, version, pointer.revision, pointer.manifestKey, pointer.persistedAt),
     database.prepare(`INSERT OR IGNORE INTO wp_codebox_commits (site_id, version, revision, manifest_key, persisted_at)
       SELECT site_id, version, revision, manifest_key, persisted_at FROM wp_codebox_state
       WHERE site_id = ? AND version = ? AND revision = ? AND manifest_key = ? AND persisted_at = ? AND lease_token IS NULL
-        AND NOT EXISTS (SELECT 1 FROM wp_codebox_fences WHERE site_id = ? AND expires_at > ?)`)
-      .bind(siteId, version, pointer.revision, pointer.manifestKey, pointer.persistedAt, siteId, now),
+        AND (NOT EXISTS (SELECT 1 FROM wp_codebox_fences WHERE site_id = ? AND expires_at > ?)
+          OR EXISTS (SELECT 1 FROM wp_codebox_fences WHERE site_id = ? AND token = ? AND expires_at > ?))`)
+      .bind(siteId, version, pointer.revision, pointer.manifestKey, pointer.persistedAt, siteId, now, siteId, fenceToken ?? null, now),
   ])
   const [state, committed] = await Promise.all([readRow(database, siteId), readCommittedPointer(database, siteId, version)])
   if (state.version !== version || !samePointer(pointerFromRow(state), pointer) || !samePointer(committed, pointer)) {
-    throw new RevisionConflict("D1 coordinator adoption requires empty or exactly matching state without an active lease or fence.")
+    throw new RevisionConflict("D1 coordinator adoption requires empty, exactly matching, or monotonic forward state without an active lease or an unmatched fence.")
   }
   return { pointer, version }
 }

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

@@ -0,0 +1,14 @@
+import type { RevisionCoordinator } from "./revision-coordinator.js"
+import type { WorkerRequestRoute } from "./request-routing.js"
+
+export function selectOperatorCoordinator(
+  active: RevisionCoordinator,
+  route: WorkerRequestRoute,
+  selector: string | null,
+  resolve?: (selector: string) => RevisionCoordinator | undefined,
+): { coordinator: RevisionCoordinator; selected: boolean } | null {
+  if (!selector) return { coordinator: active, selected: false }
+  if (route.kind !== "operator-adopt" && route.kind !== "operator-fence") return route.kind.startsWith("operator-") ? null : { coordinator: active, selected: false }
+  const coordinator = resolve?.(selector)
+  return coordinator ? { coordinator, selected: true } : null
+}

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

@@ -56,7 +56,7 @@ export interface RevisionCoordinator {
   acquireFence(ttlMs: number): Promise<MutationFence>
   renewFence(token: string, ttlMs: number): Promise<MutationFence>
   releaseFence(token: string): Promise<void>
-  adopt(pointer: MarkdownPointer, version: number): Promise<{ pointer: MarkdownPointer; version: number }>
+  adopt(pointer: MarkdownPointer, version: number, fenceToken?: string, requireFence?: boolean): Promise<{ pointer: MarkdownPointer; version: number }>
   reset(): Promise<void>
 }
 

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

@@ -71,8 +71,8 @@ export class DurableObjectRevisionCoordinator implements RevisionCoordinator {
     await this.call("fence-release", { token })
   }
 
-  adopt(pointer: MarkdownPointer, version: number): Promise<{ pointer: MarkdownPointer; version: number }> {
-    return this.call("adopt", { pointer, version })
+  adopt(pointer: MarkdownPointer, version: number, fenceToken?: string, requireFence = false): Promise<{ pointer: MarkdownPointer; version: number }> {
+    return this.call("adopt", { pointer, version, fenceToken, requireFence })
   }
 
   async reset(): Promise<void> {
@@ -252,7 +252,8 @@ export class WordPressStateCoordinator implements DurableObject {
   private async adopt(siteId: string, body: Record<string, unknown>): Promise<{ pointer: MarkdownPointer; version: number }> {
     const record = await this.record(siteId)
     const fence = await this.activeFence(record)
-    if (fence) throw new RevisionConflict("Coordinator adoption is blocked by an active cutover fence.", fence.expiresAt)
+    if (body.requireFence === true && !fence) throw new RevisionConflict("Selected coordinator adoption requires an active cutover fence.")
+    if (fence && body.fenceToken !== fence.token) throw new RevisionConflict("Coordinator adoption is blocked by an active cutover fence.", fence.expiresAt)
     const pointer = body.pointer as MarkdownPointer
     const version = body.version
     if (!pointer || typeof pointer.revision !== "string" || typeof pointer.manifestKey !== "string" || typeof pointer.persistedAt !== "string"
@@ -261,7 +262,10 @@ export class WordPressStateCoordinator implements DurableObject {
     if (record.lease) delete record.lease
     const empty = record.version === 0 && record.pointer === null
     const exact = record.version === version && samePointer(record.pointer, pointer)
-    if (!empty && !exact) throw new RevisionConflict("Coordinator adoption requires empty or exactly matching state.")
+    const forward = (version as number) > record.version
+    if (!empty && !exact && !forward) throw new RevisionConflict("Coordinator adoption requires empty, exactly matching, or monotonic forward state.")
+    const committed = await this.state.storage.get<MarkdownPointer>(`wordpress-state-commit/${version}`)
+    if (committed && !samePointer(committed, pointer)) throw new RevisionConflict("Coordinator adoption cannot replace an existing commit receipt.")
     record.pointer = pointer
     record.version = version as number
     await this.state.storage.transaction(async (transaction) => {

packages/runtime-cloudflare/src/worker-d1.ts

@@ -1,12 +1,16 @@
 import { D1RevisionCoordinator } from "./d1-revision-coordinator.js"
-export { WordPressStateCoordinator } from "./state-coordinator.js"
+import { DurableObjectRevisionCoordinator, WordPressStateCoordinator } from "./state-coordinator.js"
+export { WordPressStateCoordinator }
 import type { SiteContext } from "./site-context.js"
 import { createCloudflareRuntime, type RuntimeEnv } from "./worker.js"
 
 interface D1RuntimeEnv extends RuntimeEnv {
   WORDPRESS_STATE_DATABASE: D1Database
+  WORDPRESS_STATE: DurableObjectNamespace
 }
 
 export default createCloudflareRuntime<D1RuntimeEnv>((env, site: SiteContext) => (
   new D1RevisionCoordinator(env.WORDPRESS_STATE_DATABASE, site.id)
+), (env, site: SiteContext, selector) => (
+  selector === "durable-object" ? new DurableObjectRevisionCoordinator(env.WORDPRESS_STATE.getByName(site.id), site.id) : undefined
 ))

packages/runtime-cloudflare/src/worker.ts

@@ -8,6 +8,7 @@ import phpWasmModule from "../../../node_modules/@php-wasm/web-8-5/asyncify/8_5_
 import { CLOUDFLARE_RUNTIME_HEALTH_MARKER, CLOUDFLARE_RUNTIME_HEALTH_SCHEMA, cloudflareRuntimeHealthResponse } from "./health-envelope.js"
 import { leaseRetryDelayMs } from "./lease-retry.js"
 import { logMutationPhase, MutationRetainedBytes } from "./mutation-memory.js"
+import { selectOperatorCoordinator } from "./operator-coordinator.js"
 import { canonicalPublicRoute, MAX_PUBLISHED_PAGE_BYTES, MAX_PUBLISHED_REVISION_BYTES, MAX_PUBLISHED_ROUTES, normalizePublishedRoutes, PUBLISHED_PAGE_SCHEMA, PUBLISHED_REVISION_SCHEMA, publishedPageObjectKey, publishedRevisionObjectKey, validatePublishedRevision, type PublishedRevision } from "./published-reader.js"
 import { RevisionConflict, type MarkdownPointer, type MutationFence, type RevisionCoordinator, type RevisionLease } from "./revision-coordinator.js"
 import { routeWorkerRequest } from "./request-routing.js"
@@ -210,7 +211,10 @@ export interface RuntimeEnv {
   WORDPRESS_OPERATOR_TOKEN?: string
 }
 
-export function createCloudflareRuntime<Env extends RuntimeEnv>(resolveCoordinator: (env: Env, site: SiteContext) => RevisionCoordinator) {
+export function createCloudflareRuntime<Env extends RuntimeEnv>(
+  resolveCoordinator: (env: Env, site: SiteContext) => RevisionCoordinator,
+  resolveOperatorCoordinator?: (env: Env, site: SiteContext, selector: string) => RevisionCoordinator | undefined,
+) {
   return {
     async fetch(request: Request, env: Env): Promise<Response> {
       let site: SiteContext
@@ -223,17 +227,20 @@ export function createCloudflareRuntime<Env extends RuntimeEnv>(resolveCoordinat
       if (new URL(request.url).pathname === "/wp-cron.php") return new Response("WordPress cron is managed by the Cloudflare scheduled handler.", { status: 404 })
       const publishedResponse = await servePublishedWordPressPage(request, env.WORDPRESS_STATE_BUCKET, site)
       if (publishedResponse) return publishedResponse
-      const coordinator = resolveCoordinator(env, site)
+      const route = routeWorkerRequest(request)
+      const selector = new URL(request.url).searchParams.get("coordinator")
+      const selection = selectOperatorCoordinator(resolveCoordinator(env, site), route, selector, (selected) => resolveOperatorCoordinator?.(env, site, selected))
+      if (!selection) return new Response("Unsupported operator coordinator selector.", { status: 400 })
+      const coordinator = selection.coordinator
       const wpContentResponse = await serveWordPressWpContent(request, env.WORDPRESS_STATE_BUCKET, coordinator, site)
       if (wpContentResponse) return wpContentResponse
       const staticResponse = await serveWordPressStaticAsset(request, env.WORDPRESS_STATE_BUCKET)
       if (staticResponse) return staticResponse
-      const route = routeWorkerRequest(request)
       const uploadResponse = await serveWordPressUpload(request, env.WORDPRESS_STATE_BUCKET, coordinator, site)
       if (uploadResponse) return uploadResponse
       if (route.kind === "operator-reset") return resetCanonicalWordPress(request, env, coordinator, site)
       if (route.kind === "operator-restore") return restoreCanonicalWordPress(request, env, coordinator, site)
-      if (route.kind === "operator-adopt") return adoptCanonicalWordPress(request, env, coordinator, site)
+      if (route.kind === "operator-adopt") return adoptCanonicalWordPress(request, env, coordinator, site, selection.selected)
       if (route.kind === "operator-fence") return operateCanonicalMutationFence(request, env, coordinator, site, route.action)
       if (route.kind === "operator-static-artifact-import") return importCanonicalStaticArtifact(request, env, coordinator, site)
       if (route.kind === "operator-publish") return publishCanonicalWordPressPages(request, env, coordinator, site)
@@ -418,28 +425,31 @@ async function restoreCanonicalWordPress(request: Request, env: RuntimeEnv, coor
   }
 }
 
-async function adoptCanonicalWordPress(request: Request, env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext): Promise<Response> {
+async function adoptCanonicalWordPress(request: Request, env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, requireFence = false): Promise<Response> {
   if (request.method !== "POST") return new Response("Canonical adoption requires POST.", { status: 405 })
   if (!await isAuthorizedOperator(request, env, site)) {
     return new Response("Canonical adoption authorization failed.", { status: 401 })
   }
   let pointer: MarkdownPointer
   let version: number
+  let fenceToken: string | undefined
   try {
-    const body = await request.json<{ pointer?: unknown; version?: unknown }>()
+    const body = await request.json<{ pointer?: unknown; version?: unknown; fenceToken?: unknown }>()
     pointer = body.pointer as MarkdownPointer
     version = body.version as number
+    fenceToken = typeof body.fenceToken === "string" ? body.fenceToken : undefined
   } catch {
     return new Response("Canonical adoption requires JSON state.", { status: 400 })
   }
+  if (requireFence && !fenceToken) return new Response("Selected coordinator adoption requires its active fence token.", { status: 400 })
   if (!isCanonicalRestorePointer(pointer, site) || !Number.isSafeInteger(version) || version < 1) return new Response("Canonical adoption state is invalid.", { status: 400 })
   const manifest = await readMarkdownManifest(env.WORDPRESS_STATE_BUCKET, pointer, site)
   if (!manifest || manifest.revision !== pointer.revision || manifest.manifestKey !== pointer.manifestKey || manifest.persistedAt !== pointer.persistedAt || !Array.isArray(manifest.files)) {
     return new Response("Canonical adoption manifest is unavailable or inconsistent.", { status: 409 })
   }
   let adopted: { pointer: MarkdownPointer; version: number }
   try {
-    adopted = await coordinator.adopt(pointer, version)
+    adopted = await coordinator.adopt(pointer, version, fenceToken, requireFence)
   } catch (error) {
     if (!(error instanceof RevisionConflict)) throw error
     return coordinatorConflictResponse(error)

packages/runtime-cloudflare/wrangler.d1.jsonc

@@ -16,6 +16,14 @@
 			"database_id": "00000000-0000-0000-0000-000000000000"
 		}
 	],
+	"durable_objects": {
+		"bindings": [
+			{ "name": "WORDPRESS_STATE", "class_name": "WordPressStateCoordinator" }
+		]
+	},
+	"migrations": [
+		{ "tag": "v1", "new_sqlite_classes": ["WordPressStateCoordinator"] }
+	],
 	"rules": [
 		{ "type": "CompiledWasm", "globs": ["**/*.wasm"], "fallthrough": false },
 		{ "type": "Data", "globs": ["**/*.sqlite", "**/*-runtime.zip", "**/*-canonical-seed.zip"], "fallthrough": false }