Merge pull request #1977 from Automattic/feat/1976-cloudflare-cutover-fence

Fence Cloudflare coordinator cutovers

Author: chubes4

packages/runtime-cloudflare/README.md

@@ -26,6 +26,12 @@ WordPress server files, browser assets, and pinned runtime dependencies are sepa
 
 Canonical Cloudflare boots patch only the assembled PHP MEMFS copy of `/wordpress/wp-settings.php` after the R2 WordPress corpus is materialized and before WordPress executes. The disabled-cron scheduling policy requires exactly one canonical `do_action( 'init' );` needle and fails closed otherwise. When `DISABLE_WP_CRON` is true, it removes only core's `wp_cron`, `wp_schedule_delete_old_privacy_export_files`, and `wp_schedule_update_checks` callbacks through public `remove_action()` before executing the original `init` call once. This prevents browser requests from recreating cron work outside the explicit scheduled-Worker transaction.
 
+## Coordinator Cutover
+
+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.
+
 ## Static Artifact Import
 
 An authenticated `POST ?phase=operator-static-artifact-import` materializes a verified website artifact into the existing canonical site. The request is limited to 16 KiB and references an immutable JSON object at `sites/default/import-artifacts/<sha256>.json`; the Worker requires the declared R2 key, size, and SHA-256 to agree before acquiring a coordinator lease. The canonical `blocks-engine/php-transformer/site-artifact/v1` payload is limited to 4 MiB serialized, 500 safe unique files, 8 MiB per decoded file, and 32 MiB decoded in aggregate.

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

@@ -1,4 +1,4 @@
-import { RevisionConflict, type MarkdownPointer, type RevisionCoordinator, type RevisionLease, type RevisionState } from "./revision-coordinator.js"
+import { mutationFenceExpiresAt, revisionLeaseExpiresAt, RevisionConflict, type MarkdownPointer, type MutationFence, type MutationFenceStatus, type RevisionCoordinator, type RevisionLease, type RevisionState } from "./revision-coordinator.js"
 
 interface StateRow {
   revision: string | null
@@ -11,6 +11,11 @@ interface StateRow {
   lease_expires_at: number | null
 }
 
+interface FenceRow {
+  token: string
+  expires_at: number
+}
+
 const SITE_ID = "default"
 const LEASE_MS = 90_000
 const schemaReady = new WeakMap<object, Promise<void>>()
@@ -22,8 +27,12 @@ export class D1RevisionCoordinator implements RevisionCoordinator {
     return readWordPressState(this.database)
   }
 
-  acquire(): Promise<RevisionLease> {
-    return beginStateLease(this.database, this.leaseMs)
+  acquire(ttlMs = this.leaseMs): Promise<RevisionLease> {
+    return beginStateLease(this.database, ttlMs)
+  }
+
+  renew(lease: RevisionLease, ttlMs = this.leaseMs): Promise<RevisionLease> {
+    return renewStateLease(this.database, lease, ttlMs)
   }
 
   async release(lease: RevisionLease): Promise<void> {
@@ -42,6 +51,22 @@ export class D1RevisionCoordinator implements RevisionCoordinator {
     return readCommittedPointer(this.database, version)
   }
 
+  fenceStatus(): Promise<MutationFenceStatus> {
+    return readMutationFence(this.database)
+  }
+
+  acquireFence(ttlMs: number): Promise<MutationFence> {
+    return acquireMutationFence(this.database, ttlMs)
+  }
+
+  renewFence(token: string, ttlMs: number): Promise<MutationFence> {
+    return renewMutationFence(this.database, token, ttlMs)
+  }
+
+  async releaseFence(token: string): Promise<void> {
+    await releaseMutationFence(this.database, token)
+  }
+
   adopt(pointer: MarkdownPointer, version: number): Promise<{ pointer: MarkdownPointer; version: number }> {
     return adoptWordPressState(this.database, pointer, version)
   }
@@ -66,13 +91,16 @@ async function beginStateLease(database: D1Database, leaseMs = LEASE_MS): Promis
   await ensureSchema(database)
   const now = Date.now()
   const token = crypto.randomUUID()
-  const expiresAt = now + leaseMs
+  const expiresAt = revisionLeaseExpiresAt(leaseMs, now)
   const result = await database.prepare(`UPDATE wp_codebox_state
     SET lease_token = ?, lease_base_revision = revision, lease_version = version, lease_expires_at = ?
-    WHERE site_id = ? AND (lease_token IS NULL OR lease_expires_at <= ?)`)
-    .bind(token, expiresAt, SITE_ID, now).run()
+    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 > ?)`)
+    .bind(token, expiresAt, SITE_ID, now, SITE_ID, now).run()
   if (result.meta.changes !== 1) {
     const active = await readRow(database)
+    const fence = await readFenceRow(database)
+    if (fence && fence.expires_at > now) throw new RevisionConflict("Canonical WordPress mutations are fenced for coordinator cutover.", fence.expires_at)
     throw new RevisionConflict("A canonical WordPress lease is active.", active.lease_expires_at ?? undefined)
   }
   const row = await readRow(database)
@@ -85,6 +113,18 @@ async function releaseStateLease(database: D1Database, lease: RevisionLease): Pr
   return { released: true }
 }
 
+async function renewStateLease(database: D1Database, lease: RevisionLease, leaseMs: number): Promise<RevisionLease> {
+  await ensureSchema(database)
+  const now = Date.now()
+  const expiresAt = revisionLeaseExpiresAt(leaseMs, now)
+  const result = await database.prepare(`UPDATE wp_codebox_state SET lease_expires_at = ?
+    WHERE site_id = ? AND lease_token = ? AND lease_expires_at > ?
+      AND NOT EXISTS (SELECT 1 FROM wp_codebox_fences WHERE site_id = ? AND expires_at > ?)`)
+    .bind(expiresAt, SITE_ID, lease.token, now, SITE_ID, now).run()
+  if (result.meta.changes !== 1) throw new RevisionConflict("The canonical WordPress lease cannot renew because it expired or changed.")
+  return { ...lease, expiresAt }
+}
+
 async function abortStateLease(database: D1Database, lease: RevisionLease): Promise<{ aborted: true }> {
   await finishLease(database, lease, "abort")
   return { aborted: true }
@@ -120,6 +160,49 @@ async function readCommittedPointer(database: D1Database, version: number): Prom
   return pointer
 }
 
+async function readMutationFence(database: D1Database): Promise<MutationFenceStatus> {
+  await ensureSchema(database)
+  const row = await readFenceRow(database)
+  return row && row.expires_at > Date.now() ? { active: true, expiresAt: row.expires_at } : { active: false }
+}
+
+async function acquireMutationFence(database: D1Database, ttlMs: number): Promise<MutationFence> {
+  await ensureSchema(database)
+  const now = Date.now()
+  const fence = { token: crypto.randomUUID(), expiresAt: mutationFenceExpiresAt(ttlMs, now) }
+  const result = await database.prepare(`INSERT INTO wp_codebox_fences (site_id, token, expires_at)
+    SELECT ?, ?, ? WHERE EXISTS (
+      SELECT 1 FROM wp_codebox_state WHERE site_id = ? AND (lease_token IS NULL OR lease_expires_at <= ?)
+    )
+    ON CONFLICT(site_id) DO UPDATE SET token = excluded.token, expires_at = excluded.expires_at
+      WHERE wp_codebox_fences.expires_at <= ?`)
+    .bind(SITE_ID, fence.token, fence.expiresAt, SITE_ID, now, now).run()
+  if (result.meta.changes !== 1) {
+    const [state, active] = await Promise.all([readRow(database), readFenceRow(database)])
+    if (state.lease_token && state.lease_expires_at && state.lease_expires_at > now) throw new RevisionConflict("A canonical WordPress lease is active.", state.lease_expires_at)
+    throw new RevisionConflict("A coordinator cutover fence is already active.", active?.expires_at)
+  }
+  return fence
+}
+
+async function renewMutationFence(database: D1Database, token: string, ttlMs: number): Promise<MutationFence> {
+  await ensureSchema(database)
+  const now = Date.now()
+  const fence = { token, expiresAt: mutationFenceExpiresAt(ttlMs, now) }
+  const result = await database.prepare(`UPDATE wp_codebox_fences SET expires_at = ?
+    WHERE site_id = ? AND token = ? AND expires_at > ?`).bind(fence.expiresAt, SITE_ID, token, now).run()
+  if (result.meta.changes !== 1) throw new RevisionConflict("The coordinator cutover fence token is invalid or expired.")
+  return fence
+}
+
+async function releaseMutationFence(database: D1Database, token: string): Promise<{ released: true }> {
+  await ensureSchema(database)
+  const result = await database.prepare(`DELETE FROM wp_codebox_fences
+    WHERE site_id = ? AND token = ? AND expires_at > ?`).bind(SITE_ID, token, Date.now()).run()
+  if (result.meta.changes !== 1) throw new RevisionConflict("The coordinator cutover fence token is invalid or expired.")
+  return { released: true }
+}
+
 async function adoptWordPressState(database: D1Database, pointer: MarkdownPointer, version: number): 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.")
@@ -130,35 +213,47 @@ async function adoptWordPressState(database: D1Database, pointer: MarkdownPointe
       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 ((revision IS NULL AND manifest_key IS NULL AND persisted_at IS NULL)
           OR (version = ? AND revision = ? AND manifest_key = ? AND persisted_at = ?))
         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, SITE_ID, now,
+      .bind(pointer.revision, pointer.manifestKey, pointer.persistedAt, version, SITE_ID, now, SITE_ID, now,
         version, pointer.revision, pointer.manifestKey, pointer.persistedAt,
         SITE_ID, version, SITE_ID, 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`)
-      .bind(SITE_ID, version, pointer.revision, pointer.manifestKey, pointer.persistedAt),
+      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(SITE_ID, version, pointer.revision, pointer.manifestKey, pointer.persistedAt, SITE_ID, now),
   ])
   const [state, committed] = await Promise.all([readRow(database), readCommittedPointer(database, version)])
-  if (state.lease_token !== null || 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.", state.lease_expires_at ?? undefined)
+  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.")
   }
   return { pointer, version }
 }
 
 async function resetWordPressState(database: D1Database): Promise<{ reset: true }> {
   await ensureSchema(database)
-  await database.batch([
+  const now = Date.now()
+  const [reset] = 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),
+      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 > ?)`)
+      .bind(SITE_ID, now, SITE_ID, now),
+    database.prepare(`DELETE FROM wp_codebox_commits WHERE site_id = ?
+      AND EXISTS (SELECT 1 FROM wp_codebox_state 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 > ?)`)
+      .bind(SITE_ID, SITE_ID, now, SITE_ID, now),
   ])
+  if (reset.meta.changes !== 1) {
+    const [state, fence] = await Promise.all([readRow(database), readFenceRow(database)])
+    throw new RevisionConflict("Coordinator reset is blocked by an active canonical lease or cutover fence.", fence?.expires_at ?? state.lease_expires_at ?? undefined)
+  }
   return { reset: true }
 }
 
@@ -179,6 +274,10 @@ async function readRow(database: D1Database): Promise<StateRow> {
   return row
 }
 
+async function readFenceRow(database: D1Database): Promise<FenceRow | null> {
+  return database.prepare(`SELECT token, expires_at FROM wp_codebox_fences WHERE site_id = ?`).bind(SITE_ID).first<FenceRow>()
+}
+
 function pointerFromRow(row: StateRow): MarkdownPointer | null {
   if (row.revision === null && row.manifest_key === null && row.persisted_at === null) return null
   const pointer = { revision: row.revision, manifestKey: row.manifest_key, persistedAt: row.persisted_at }
@@ -216,6 +315,7 @@ function ensureSchema(database: D1Database): Promise<void> {
     )`).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()
+    await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_fences (site_id TEXT PRIMARY KEY, token TEXT NOT NULL, expires_at INTEGER NOT NULL)`).run()
   })()
   schemaReady.set(key, pending)
   pending.catch(() => schemaReady.delete(key))

packages/runtime-cloudflare/src/request-routing.ts

@@ -6,6 +6,7 @@ export type WorkerRequestRoute =
   | { kind: "operator-reset" }
   | { kind: "operator-restore" }
   | { kind: "operator-adopt" }
+  | { kind: "operator-fence"; action: "status" | "acquire" | "renew" | "release" }
   | { kind: "operator-static-artifact-import" }
   | { kind: "operator-publish" }
   | { kind: "probe"; phase: string }
@@ -19,6 +20,10 @@ export function routeWorkerRequest(request: Request): WorkerRequestRoute {
   if (phase === "operator-reset") return { kind: "operator-reset" }
   if (phase === "operator-restore") return { kind: "operator-restore" }
   if (phase === "operator-adopt") return { kind: "operator-adopt" }
+  if (phase === "operator-fence-status") return { kind: "operator-fence", action: "status" }
+  if (phase === "operator-fence-acquire") return { kind: "operator-fence", action: "acquire" }
+  if (phase === "operator-fence-renew") return { kind: "operator-fence", action: "renew" }
+  if (phase === "operator-fence-release") return { kind: "operator-fence", action: "release" }
   if (phase === "operator-static-artifact-import") return { kind: "operator-static-artifact-import" }
   if (phase === "operator-publish") return { kind: "operator-publish" }
   return { kind: "probe", phase }

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

@@ -18,13 +18,44 @@ export interface RevisionState {
   version: number
 }
 
+export interface MutationFence {
+  token: string
+  expiresAt: number
+}
+
+export interface MutationFenceStatus {
+  active: boolean
+  expiresAt?: number
+}
+
+export const MIN_MUTATION_FENCE_MS = 30_000
+export const MAX_MUTATION_FENCE_MS = 600_000
+export const MAX_REVISION_LEASE_MS = 600_000
+
+export function mutationFenceExpiresAt(ttlMs: number, now = Date.now()): number {
+  if (!Number.isSafeInteger(ttlMs) || ttlMs < MIN_MUTATION_FENCE_MS || ttlMs > MAX_MUTATION_FENCE_MS) {
+    throw new RevisionConflict(`Mutation fence TTL must be between ${MIN_MUTATION_FENCE_MS / 1_000} and ${MAX_MUTATION_FENCE_MS / 1_000} seconds.`)
+  }
+  return now + ttlMs
+}
+
+export function revisionLeaseExpiresAt(ttlMs: number, now = Date.now()): number {
+  if (!Number.isSafeInteger(ttlMs) || ttlMs < 1 || ttlMs > MAX_REVISION_LEASE_MS) throw new RevisionConflict(`Revision lease TTL must be between 1 and ${MAX_REVISION_LEASE_MS} milliseconds.`)
+  return now + ttlMs
+}
+
 export interface RevisionCoordinator {
   state(): Promise<RevisionState>
-  acquire(): Promise<RevisionLease>
+  acquire(ttlMs?: number): Promise<RevisionLease>
+  renew(lease: RevisionLease, ttlMs?: number): Promise<RevisionLease>
   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>
+  fenceStatus(): Promise<MutationFenceStatus>
+  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 }>
   reset(): Promise<void>
 }