Merge pull request #1969 from Automattic/feat/1968-coordinator-state-adoption

Adopt exact Cloudflare coordinator state

Author: chubes4

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

@@ -42,6 +42,10 @@ export class D1RevisionCoordinator implements RevisionCoordinator {
     return readCommittedPointer(this.database, version)
   }
 
+  adopt(pointer: MarkdownPointer, version: number): Promise<{ pointer: MarkdownPointer; version: number }> {
+    return adoptWordPressState(this.database, pointer, version)
+  }
+
   async reset(): Promise<void> {
     await resetWordPressState(this.database)
   }
@@ -116,6 +120,36 @@ async function readCommittedPointer(database: D1Database, version: number): Prom
   return pointer
 }
 
+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.")
+  await ensureSchema(database)
+  const now = Date.now()
+  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 ((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,
+        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),
+  ])
+  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)
+  }
+  return { pointer, version }
+}
+
 async function resetWordPressState(database: D1Database): Promise<{ reset: true }> {
   await ensureSchema(database)
   await database.batch([
@@ -160,6 +194,10 @@ function validatePointer(pointer: unknown): asserts pointer is MarkdownPointer {
   }
 }
 
+function samePointer(left: MarkdownPointer | null, right: MarkdownPointer): boolean {
+  return !!left && left.revision === right.revision && left.manifestKey === right.manifestKey && left.persistedAt === right.persistedAt
+}
+
 function ensureSchema(database: D1Database): Promise<void> {
   const key = database as object
   const existing = schemaReady.get(key)

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

@@ -5,6 +5,7 @@ export type WorkerRequestRoute =
   | { kind: "r2-mutate" }
   | { kind: "operator-reset" }
   | { kind: "operator-restore" }
+  | { kind: "operator-adopt" }
   | { kind: "operator-publish" }
   | { kind: "probe"; phase: string }
 
@@ -16,6 +17,7 @@ export function routeWorkerRequest(request: Request): WorkerRequestRoute {
   if (phase === "r2-mutate") return { kind: "r2-mutate" }
   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-publish") return { kind: "operator-publish" }
   return { kind: "probe", phase }
 }

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

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

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

@@ -50,6 +50,10 @@ export class DurableObjectRevisionCoordinator implements RevisionCoordinator {
     return this.call("committed", { version })
   }
 
+  adopt(pointer: MarkdownPointer, version: number): Promise<{ pointer: MarkdownPointer; version: number }> {
+    return this.call("adopt", { pointer, version })
+  }
+
   async reset(): Promise<void> {
     await this.call("reset", {})
   }
@@ -103,6 +107,7 @@ export class WordPressStateCoordinator implements DurableObject {
     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 === "adopt") return Response.json(await this.adopt(body))
     if (action === "reset") return Response.json(await this.reset())
     return new Response("Unknown coordinator action.", { status: 404 })
   }
@@ -164,6 +169,26 @@ export class WordPressStateCoordinator implements DurableObject {
     return await this.state.storage.get<MarkdownPointer>(`wordpress-state-commit/${body.version}`) ?? null
   }
 
+  private async adopt(body: Record<string, unknown>): Promise<{ pointer: MarkdownPointer; version: number }> {
+    const record = await this.record()
+    const pointer = body.pointer as MarkdownPointer
+    const version = body.version
+    if (!pointer || typeof pointer.revision !== "string" || typeof pointer.manifestKey !== "string" || typeof pointer.persistedAt !== "string"
+      || !Number.isSafeInteger(version) || (version as number) < 1) throw new RevisionConflict("A complete canonical pointer and positive version are required for adoption.")
+    if (record.lease && record.lease.expiresAt > Date.now()) throw new RevisionConflict("Coordinator adoption requires no active lease.", record.lease.expiresAt)
+    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.")
+    record.pointer = pointer
+    record.version = version as number
+    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 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.")
@@ -190,3 +215,7 @@ export class WordPressStateCoordinator implements DurableObject {
     return this.state.storage.put(STORAGE_KEY, record)
   }
 }
+
+function samePointer(left: MarkdownPointer | null, right: MarkdownPointer): boolean {
+  return !!left && left.revision === right.revision && left.manifestKey === right.manifestKey && left.persistedAt === right.persistedAt
+}

packages/runtime-cloudflare/src/worker.ts

@@ -153,6 +153,7 @@ export function createCloudflareRuntime<Env extends RuntimeEnv>(resolveCoordinat
       if (uploadResponse) return uploadResponse
       if (route.kind === "operator-reset") return resetCanonicalWordPress(request, env, coordinator)
       if (route.kind === "operator-restore") return restoreCanonicalWordPress(request, env, coordinator)
+      if (route.kind === "operator-adopt") return adoptCanonicalWordPress(request, env, coordinator)
       if (route.kind === "operator-publish") return publishCanonicalWordPressPages(request, env, coordinator)
       if (route.kind === "probe") {
         return runBootProbe(route.phase, env.WORDPRESS_STATE_BUCKET)
@@ -322,6 +323,39 @@ async function restoreCanonicalWordPress(request: Request, env: RuntimeEnv, coor
   }
 }
 
+async function adoptCanonicalWordPress(request: Request, env: RuntimeEnv, coordinator: RevisionCoordinator): Promise<Response> {
+  if (request.method !== "POST") return new Response("Canonical adoption requires POST.", { status: 405 })
+  const authorization = request.headers.get("authorization")
+  if (!env.WORDPRESS_OPERATOR_TOKEN || !authorization || !await secretsMatch(authorization, `Bearer ${env.WORDPRESS_OPERATOR_TOKEN}`)) {
+    return new Response("Canonical adoption authorization failed.", { status: 401 })
+  }
+  let pointer: MarkdownPointer
+  let version: number
+  try {
+    const body = await request.json<{ pointer?: unknown; version?: unknown }>()
+    pointer = body.pointer as MarkdownPointer
+    version = body.version as number
+  } catch {
+    return new Response("Canonical adoption requires JSON state.", { status: 400 })
+  }
+  if (!isCanonicalRestorePointer(pointer) || !Number.isSafeInteger(version) || version < 1) return new Response("Canonical adoption state is invalid.", { status: 400 })
+  const manifest = await readMarkdownManifest(env.WORDPRESS_STATE_BUCKET, pointer)
+  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)
+  } catch (error) {
+    if (!(error instanceof RevisionConflict)) throw error
+    const headers = new Headers()
+    if (error.retryAt) headers.set("retry-after", String(Math.max(1, Math.ceil((error.retryAt - Date.now()) / 1000))))
+    return Response.json({ schema: "wp-codebox/cloudflare-coordinator-conflict/v1", message: error.message, retryAt: error.retryAt }, { status: 409, headers })
+  }
+  await discardCachedRuntime()
+  return Response.json({ adopted: true, ...adopted })
+}
+
 function isCanonicalRestorePointer(pointer: unknown): pointer is MarkdownPointer {
   if (!pointer || typeof pointer !== "object") return false
   const candidate = pointer as Partial<MarkdownPointer>

scripts/cloudflare-local-gate.mjs

@@ -26,6 +26,9 @@ try {
   await assertWordPressCronDisabled()
   await assertConcurrentMutations()
   await assertCoordinatorBackend()
+  console.log(`Coordinator adoption probe starting for ${coordinator}.`)
+  await assertCoordinatorAdoption()
+  console.log(`Coordinator adoption probe passed for ${coordinator}.`)
   const coldHome = await timedWordPressPage(origin, "cold explanatory homepage")
   const warmHome = await timedWordPressPage(origin, "warm explanatory homepage")
   await assertExplanatoryHomepage(warmHome.body)
@@ -102,8 +105,14 @@ async function startWorker() {
     env: { ...process.env, NO_PROXY: "wordpress.org,github.com,codeload.github.com", no_proxy: "wordpress.org,github.com,codeload.github.com" },
     stdio: ["ignore", "pipe", "pipe"],
   })
-  child.stdout.on("data", (chunk) => { output += chunk })
-  child.stderr.on("data", (chunk) => { output += chunk })
+  child.stdout.on("data", (chunk) => {
+    output += chunk
+    if (process.env.CLOUDFLARE_GATE_DEBUG) process.stdout.write(chunk)
+  })
+  child.stderr.on("data", (chunk) => {
+    output += chunk
+    if (process.env.CLOUDFLARE_GATE_DEBUG) process.stderr.write(chunk)
+  })
   await waitForServer()
 }
 
@@ -451,6 +460,30 @@ async function assertCoordinatorBackend() {
   }
 }
 
+async function assertCoordinatorAdoption() {
+  const before = await (await fetch(`${origin}/?phase=r2-state`)).json()
+  const reset = await fetch(`${origin}/?phase=operator-reset`, { method: "POST", headers: { authorization: `Bearer ${operatorToken}` } })
+  if (!reset.ok) throw new Error(`Coordinator reset before adoption failed: ${reset.status} ${await reset.text()}.`)
+  const adoption = await fetch(`${origin}/?phase=operator-adopt`, {
+    method: "POST",
+    headers: { authorization: `Bearer ${operatorToken}`, "content-type": "application/json" },
+    body: JSON.stringify({ pointer: before.pointer, version: before.version }),
+  })
+  const adoptionBody = await adoption.text()
+  if (!adoption.ok) throw new Error(`Exact coordinator adoption failed: status=${adoption.status} body=${adoptionBody}.\nWorker output:\n${output}`)
+  const adopted = JSON.parse(adoptionBody)
+  if (!adoption.ok || !adopted.adopted || adopted.version !== before.version || adopted.pointer?.revision !== before.pointer?.revision) {
+    throw new Error(`Exact coordinator adoption failed: status=${adoption.status} payload=${JSON.stringify(adopted)}.`)
+  }
+  const divergent = await fetch(`${origin}/?phase=operator-adopt`, {
+    method: "POST",
+    headers: { authorization: `Bearer ${operatorToken}`, "content-type": "application/json" },
+    body: JSON.stringify({ pointer: before.pointer, version: before.version + 1 }),
+  })
+  if (divergent.status !== 409) throw new Error(`Divergent coordinator adoption was not rejected: ${divergent.status} ${await divergent.text()}.`)
+  await assertCoordinatorBackend()
+}
+
 async function assertWordPressPage(target, label) {
   const response = await request(target)
   const body = await response.text()

tests/cloudflare-runtime.test.ts

@@ -78,6 +78,7 @@ test("Cloudflare routing reserves phases while the phase-less route serves WordP
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=r2-mutate")), { kind: "r2-mutate" })
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-reset")), { kind: "operator-reset" })
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-restore")), { kind: "operator-restore" })
+  assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-adopt")), { kind: "operator-adopt" })
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-publish")), { kind: "operator-publish" })
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=seeded-wordpress")), { kind: "probe", phase: "seeded-wordpress" })
 })
@@ -605,6 +606,16 @@ test("Cloudflare coordinator serializes leases, promotes with CAS, and recovers
   const coordinator = new WordPressStateCoordinator(state as never, { WORDPRESS_STATE_BUCKET: bucket as never, COORDINATOR_LEASE_MS: 50 })
   const call = async (action: string, body: Record<string, unknown> = {}) => coordinator.fetch(new Request(`https://worker.example/?__wp_codebox_coordinator=${action}`, { method: action === "state" ? "GET" : "POST", headers: { "content-type": "application/json" }, body: action === "state" ? undefined : JSON.stringify(body) }))
 
+  const adoptedPointer = { revision: "adopted", manifestKey: "sites/default/markdown/revisions/adopted.json", persistedAt: "2026-01-01T00:00:00.000Z" }
+  assert.equal((await call("adopt", { pointer: adoptedPointer, version: 31 })).status, 200)
+  assert.deepEqual(await (await call("committed", { version: 31 })).json(), adoptedPointer)
+  assert.equal((await call("adopt", { pointer: adoptedPointer, version: 31 })).status, 200)
+  assert.equal((await call("adopt", { pointer: { ...adoptedPointer, revision: "divergent" }, version: 31 })).status, 409)
+  const adoptedLease = await (await call("begin")).json() as { token: string }
+  assert.equal((await call("adopt", { pointer: adoptedPointer, version: 31 })).status, 409)
+  assert.equal((await call("abort", { token: adoptedLease.token })).status, 200)
+  assert.equal((await call("reset")).status, 200)
+
   const first = await (await call("begin")).json() as { token: string; pointer: null; version: number }
   assert.equal(first.pointer, null)
   assert.equal((await call("begin")).status, 409)