diff --git a/package.json b/package.json index 0010d201a..7677c998e 100644 --- a/package.json +++ b/package.json @@ -105,7 +105,7 @@ "generate:cloudflare-canonical-mdi-seed": "npm run generate:cloudflare-mdi-runtime-bundle && php scripts/build-cloudflare-canonical-mdi-seed.php", "smoke": "tsx scripts/run-smoke.ts", "test:redaction": "tsx tests/redaction.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-runtime.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-runtime.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit", "test:cloudflare-wordpress-auth": "tsx tests/cloudflare-wordpress-auth.test.ts", "test:cloudflare-wordpress-archive-corpus": "tsx tests/cloudflare-wordpress-archive-corpus.test.ts", "test:agent-task-contracts": "tsx tests/agent-task-contracts.test.ts && tsx tests/agent-task-canonical-evidence.test.ts && npm run test:agent-task-workflow-interface && npm run test:runtime-sources-materialization && tsx tests/agent-task-reusable-workflow.test.ts", diff --git a/packages/runtime-cloudflare/README.md b/packages/runtime-cloudflare/README.md index 68d70ce83..413fb0380 100644 --- a/packages/runtime-cloudflare/README.md +++ b/packages/runtime-cloudflare/README.md @@ -42,9 +42,13 @@ For lossless D1-to-Durable-Object rollback, keep D1 deployed as the active Worke ## 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/{siteId}/import-artifacts/.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. +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/{siteId}/import-artifacts/.json`; the Worker requires the declared R2 key, size, and SHA-256 to agree before creating work. 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. -The import transaction boots a dedicated runtime with the pinned Static Site Importer v1.3.4 archive (`8d27286021d7c6141609def40a97591322a14340b23a17d9405f7919ea145a29`) from R2, invokes its public `static-site-importer/import-website-artifact` ability as the operator-authorized administrator, and requires the canonical quality gate plus zero fallback, core HTML, freeform, and invalid blocks. SSI and its MU loader are runtime-owned and excluded from canonical mutable `wp-content`; generated themes, pages, options, and assets persist normally through MDI/R2. Failed or partial imports discard the PHP runtime without committing. Successful imports return compact provenance and canonical revision/version headers, enqueue affected publication routes, and store at most 20 idempotency receipts in canonical options. Exact replay returns the existing receipt without a new revision; reuse with different input returns a conflict. +The D1 profile registers the configured site identity and creates or converges on a durable operation before returning `202` with a `Location` for authenticated `GET ?phase=operator-static-artifact-operation&operationId=`. D1 enforces one active mutating operation per site and binds each site-scoped idempotency key to one immutable input fingerprint. Scheduled invocations claim one eligible operation, persist bounded attempt and progress records, and alternate operation/publication priority so neither queue starves. Claims are renewed around runtime boot, SSI execution, canonical persistence, and commit preparation. + +Before canonical promotion, the operation checkpoints the exact prepared R2 pointer and expected coordinator version. Recovery skips SSI only when the coordinator's immutable commit receipt matches that complete pointer; a version match alone cannot claim another mutation's commit. The operation remains `publication-pending` after canonical commit until the corresponding immutable R2 publication receipt records promotion, supersession, or orphaning. Its terminal receipt identifies the artifact digest, SSI report, canonical pointer/version, publication identity, site URL, and canonical and terminal completion times. Profiles without a D1 operation repository retain the direct import transaction. + +The import transaction boots a dedicated runtime with the pinned Static Site Importer v1.3.4 archive (`8d27286021d7c6141609def40a97591322a14340b23a17d9405f7919ea145a29`) from R2, invokes its public `static-site-importer/import-website-artifact` ability as the operator-authorized administrator, and requires the canonical quality gate plus zero fallback, core HTML, freeform, and invalid blocks. SSI and its MU loader are runtime-owned and excluded from canonical mutable `wp-content`; generated themes, pages, options, and assets persist normally through MDI/R2. Failed or partial imports discard the PHP runtime without committing. Successful imports persist compact provenance, enqueue affected publication routes, and store at most 20 idempotency receipts in canonical options as a second mutation fence behind the D1 operation record. Exact replay converges without a new revision; reuse with different input returns a conflict. SSI is extracted only for import requests, so normal browser, mutation, publication, and cron boots retain their existing memory and latency profile. The pinned normal plugin archive bundles Blocks Engine and supports website artifacts without the optional Figma zstd extension; compressed `.fig` import is outside this runtime contract. diff --git a/packages/runtime-cloudflare/src/d1-operation-repository.ts b/packages/runtime-cloudflare/src/d1-operation-repository.ts new file mode 100644 index 000000000..22482e456 --- /dev/null +++ b/packages/runtime-cloudflare/src/d1-operation-repository.ts @@ -0,0 +1,190 @@ +import type { MarkdownPointer } from "./revision-coordinator.js" +import type { SiteContext } from "./site-context.js" + +export const STATIC_ARTIFACT_OPERATION_SCHEMA = "wp-codebox/cloudflare-static-artifact-operation/v1" +export type StaticArtifactOperationState = "queued" | "running" | "retryable" | "publication-pending" | "succeeded" | "failed" +export interface StaticArtifactOperationInput { idempotencyKey: string; fingerprint: string; artifact: { r2Key: string; sha256: string; size: number }; options: { slug: string; name: string; siteTitle: string } } +export interface OperationAttempt { number: number; startedAt: string; completedAt: string | null; state: string; stage: string; error: { code: string; message: string } | null } +export interface OperationReceipt { input: StaticArtifactOperationInput; ssiResult: unknown; canonical: { pointer: MarkdownPointer; version: number }; publication: { status: "pending"; jobKey: string } | { status: "promoted"; jobKey: string; revision: string } | { status: "superseded" | "orphaned"; jobKey: string } | { status: "none" }; siteUrl: string; canonicalCompletedAt: string; terminalCompletedAt: string | null } +export interface StaticArtifactOperation { + schema: typeof STATIC_ARTIFACT_OPERATION_SCHEMA + site: Pick + operationId: string + state: StaticArtifactOperationState + input: StaticArtifactOperationInput + attempts: number + attemptHistory: OperationAttempt[] + stage: string + progress: number + retryAt: number | null + claimExpiresAt: number | null + prepared: { pointer: MarkdownPointer; version: number; ssiResult: unknown; publicationJobKey: string | null } | null + error: { code: string; message: string } | null + receipt: OperationReceipt | null +} + +export class OperationConflict extends Error {} +const schemaReady = new WeakMap>() + +export function shouldRecoverPreparedCommit(prepared: StaticArtifactOperation["prepared"], committed: MarkdownPointer | null): boolean { + return !!prepared && !!committed && prepared.pointer.revision === committed.revision && prepared.pointer.manifestKey === committed.manifestKey && prepared.pointer.persistedAt === committed.persistedAt +} + +export class D1OperationRepository { + constructor(private readonly database: D1Database, private readonly claimMs = 600_000) {} + + async createOrConverge(site: SiteContext, input: StaticArtifactOperationInput): Promise<{ operation: StaticArtifactOperation; created: boolean }> { + await ensureSchema(this.database) + const now = Date.now() + await this.database.prepare(`INSERT OR IGNORE INTO wp_codebox_sites (site_id, hostname, origin, state, created_at, activated_at, updated_at) VALUES (?, ?, ?, 'active', ?, ?, ?)`).bind(site.id, site.hostname, site.origin, now, now, now).run() + const registered = await this.database.prepare(`SELECT hostname, origin, state FROM wp_codebox_sites WHERE site_id = ?`).bind(site.id).first<{ hostname: string; origin: string; state: string }>() + if (!registered) throw new Error("D1 site registration was not observable.") + if (registered && (registered.hostname !== site.hostname || registered.origin !== site.origin || registered.state !== "active")) throw new OperationConflict("The configured site identity conflicts with its registered D1 site.") + const existing = await this.byKey(site.id, input.idempotencyKey) + if (existing) { + if (existing.input.fingerprint !== input.fingerprint) throw new OperationConflict("The idempotency key is already bound to a different immutable input.") + return { operation: existing, created: false } + } + try { + await this.database.prepare(`INSERT INTO wp_codebox_operations (site_id, operation_id, idempotency_key, fingerprint, artifact_key, artifact_sha256, artifact_size, slug, name, site_title, state, stage, progress, attempts, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', 'created', 0, 0, ?, ?)`) + .bind(site.id, crypto.randomUUID(), input.idempotencyKey, input.fingerprint, input.artifact.r2Key, input.artifact.sha256, input.artifact.size, input.options.slug, input.options.name, input.options.siteTitle, now, now).run() + } catch (error) { + const raced = await this.byKey(site.id, input.idempotencyKey) + if (raced?.input.fingerprint === input.fingerprint) return { operation: raced, created: false } + if (raced) throw new OperationConflict("The idempotency key is already bound to a different immutable input.") + const active = await this.database.prepare(`SELECT operation_id FROM wp_codebox_operations WHERE site_id = ? AND state IN ('queued','running','retryable')`).bind(site.id).first<{ operation_id: string }>() + if (active) throw new OperationConflict("A mutating operation is already active for this site.") + throw error + } + const operation = await this.byKey(site.id, input.idempotencyKey) + if (!operation) throw new Error("D1 operation insert was not observable.") + return { operation, created: true } + } + + async get(siteId: string, operationId: string): Promise { + await ensureSchema(this.database) + const row = await this.database.prepare(`${operationSelect} WHERE o.site_id = ? AND o.operation_id = ?`).bind(siteId, operationId).first() + return row ? this.hydrate(row) : null + } + + async claimNext(siteId: string, now = Date.now()): Promise<(StaticArtifactOperation & { claimToken: string }) | null> { + await ensureSchema(this.database) + const candidate = await this.database.prepare(`SELECT operation_id FROM wp_codebox_operations WHERE site_id = ? AND (state = 'queued' OR (state = 'retryable' AND retry_at <= ?) OR (state = 'running' AND claim_expires_at <= ?)) ORDER BY created_at LIMIT 1`).bind(siteId, now, now).first<{ operation_id: string }>() + if (!candidate) return null + const token = crypto.randomUUID() + const expiresAt = now + this.claimMs + const [update, attempt] = await this.database.batch([ + this.database.prepare(`UPDATE wp_codebox_operations SET state = 'running', stage = 'claimed', claim_token = ?, claim_expires_at = ?, attempts = attempts + 1, retry_at = NULL, error_code = NULL, error_message = NULL, updated_at = ? WHERE site_id = ? AND operation_id = ? AND (state = 'queued' OR (state = 'retryable' AND retry_at <= ?) OR (state = 'running' AND claim_expires_at <= ?))`).bind(token, expiresAt, now, siteId, candidate.operation_id, now, now), + this.database.prepare(`INSERT INTO wp_codebox_operation_attempts (site_id, operation_id, attempt_number, claim_token, started_at, state, stage) SELECT site_id, operation_id, attempts, ?, ?, 'running', 'claimed' FROM wp_codebox_operations WHERE site_id = ? AND operation_id = ? AND state = 'running' AND claim_token = ? AND claim_expires_at = ?`).bind(token, new Date(now).toISOString(), siteId, candidate.operation_id, token, expiresAt), + ]) + if (update.meta.changes !== 1) return null + if (attempt.meta.changes !== 1) throw new OperationConflict("Operation attempt creation did not match its claim.") + const operation = await this.get(siteId, candidate.operation_id) + if (!operation) throw new Error("Claimed operation is unavailable.") + return { ...operation, claimToken: token } + } + + async renew(siteId: string, operationId: string, token: string): Promise { + await this.ownedUpdate(siteId, operationId, token, `UPDATE wp_codebox_operations SET claim_expires_at = ?, updated_at = ?`, [Date.now() + this.claimMs, Date.now()]) + } + + async checkpoint(siteId: string, operationId: string, token: string, stage: string, progress: number): Promise { + await this.ownedUpdate(siteId, operationId, token, `UPDATE wp_codebox_operations SET stage = ?, progress = ?, updated_at = ?`, [stage, progress, Date.now()]) + } + + async prepareCommit(siteId: string, operationId: string, token: string, version: number, pointer: MarkdownPointer, ssiResult: unknown, publicationJobKey: string | null): Promise { + await this.ownedUpdate(siteId, operationId, token, `UPDATE wp_codebox_operations SET prepared_version = ?, prepared_revision = ?, prepared_manifest_key = ?, prepared_persisted_at = ?, prepared_result_json = ?, prepared_publication_job = ?, stage = 'prepared-commit', progress = 85, updated_at = ?`, [version, pointer.revision, pointer.manifestKey, pointer.persistedAt, JSON.stringify(ssiResult), publicationJobKey, Date.now()]) + } + + async recordCommit(siteId: string, operationId: string, token: string): Promise { + await this.ownedUpdate(siteId, operationId, token, `UPDATE wp_codebox_operations SET stage = 'committed', progress = 90, updated_at = ?`, [Date.now()]) + } + + async complete(siteId: string, operationId: string, token: string, ssiResult: unknown | undefined, publicationJobKey: string | undefined, siteUrl: string): Promise { + const operation = await this.get(siteId, operationId) + if (!operation?.prepared) throw new OperationConflict("A prepared canonical pointer is required before operation completion.") + const now = new Date().toISOString() + const result = ssiResult ?? operation.prepared.ssiResult + const jobKey = publicationJobKey ?? operation.prepared.publicationJobKey ?? undefined + const receipt: OperationReceipt = { input: operation.input, ssiResult: result, canonical: { pointer: operation.prepared.pointer, version: operation.prepared.version }, publication: jobKey ? { status: "pending", jobKey } : { status: "none" }, siteUrl, canonicalCompletedAt: now, terminalCompletedAt: jobKey ? null : now } + const state: StaticArtifactOperationState = jobKey ? "publication-pending" : "succeeded" + const stage = jobKey ? "canonical-completed-publication-pending" : "completed" + await this.terminalBatch(siteId, operationId, token, `UPDATE wp_codebox_operations SET state = ?, stage = ?, progress = ?, receipt_json = ?, error_code = NULL, error_message = NULL, claim_token = NULL, claim_expires_at = NULL, retry_at = NULL, completed_at = ?, updated_at = ?`, [state, stage, jobKey ? 95 : 100, JSON.stringify(receipt), jobKey ? null : now, Date.now()], state, stage, null) + } + + async retry(siteId: string, operationId: string, token: string, error: unknown, retryAt: number): Promise { await this.finish(siteId, operationId, token, "retryable", error, retryAt) } + async fail(siteId: string, operationId: string, token: string, error: unknown): Promise { await this.finish(siteId, operationId, token, "failed", error, null) } + + private async finish(siteId: string, operationId: string, token: string, state: "retryable" | "failed", error: unknown, retryAt: number | null): Promise { + const sanitized = sanitizeError(error) + const completedAt = state === "failed" ? new Date().toISOString() : null + await this.terminalBatch(siteId, operationId, token, `UPDATE wp_codebox_operations SET state = ?, stage = ?, error_code = ?, error_message = ?, retry_at = ?, claim_token = NULL, claim_expires_at = NULL, completed_at = ?, updated_at = ?`, [state, state, sanitized.code, sanitized.message, retryAt, completedAt, Date.now()], state, state, sanitized) + } + + private async ownedUpdate(siteId: string, operationId: string, token: string, update: string, values: unknown[]): Promise { + const result = await this.database.prepare(`${update} WHERE site_id = ? AND operation_id = ? AND state = 'running' AND claim_token = ? AND claim_expires_at > ?`).bind(...values, siteId, operationId, token, Date.now()).run() + if (result.meta.changes !== 1) throw new OperationConflict("Operation claim expired or changed.") + } + + private async terminalBatch(siteId: string, operationId: string, token: string, update: string, values: unknown[], state: string, stage: string, error: { code: string; message: string } | null): Promise { + const now = Date.now() + const [attempt, operation] = await this.database.batch([ + this.database.prepare(`UPDATE wp_codebox_operation_attempts SET completed_at = ?, state = ?, stage = ?, error_code = ?, error_message = ? WHERE site_id = ? AND operation_id = ? AND claim_token = ? AND completed_at IS NULL AND EXISTS (SELECT 1 FROM wp_codebox_operations WHERE site_id = ? AND operation_id = ? AND state = 'running' AND claim_token = ? AND claim_expires_at > ?)`).bind(new Date(now).toISOString(), state, stage, error?.code ?? null, error?.message ?? null, siteId, operationId, token, siteId, operationId, token, now), + this.database.prepare(`${update} WHERE site_id = ? AND operation_id = ? AND state = 'running' AND claim_token = ? AND claim_expires_at > ?`).bind(...values, siteId, operationId, token, now), + ]) + if (attempt.meta.changes !== 1 || operation.meta.changes !== 1) throw new OperationConflict("Operation claim expired or changed.") + } + + async reconcilePublication(siteId: string, jobKey: string, outcome: "promoted" | "superseded" | "orphaned", revision?: string): Promise { + if (outcome === "promoted" && !revision) return + const row = await this.database.prepare(`${operationSelect} WHERE o.site_id = ? AND o.state = 'publication-pending' AND o.prepared_publication_job = ?`).bind(siteId, jobKey).first() + if (!row) return + const operation = await this.hydrate(row) + if (!operation.receipt || operation.receipt.publication.status !== "pending") return + const completedAt = new Date().toISOString() + const receipt: OperationReceipt = { ...operation.receipt, publication: outcome === "promoted" ? { status: "promoted", jobKey, revision: revision! } : { status: outcome, jobKey }, terminalCompletedAt: completedAt } + const failed = outcome !== "promoted" + await this.database.prepare(`UPDATE wp_codebox_operations SET state = ?, stage = ?, progress = 100, receipt_json = ?, error_code = ?, error_message = ?, completed_at = ?, updated_at = ? WHERE site_id = ? AND operation_id = ? AND state = 'publication-pending' AND prepared_publication_job = ?`).bind(failed ? "failed" : "succeeded", failed ? `publication-${outcome}` : "published", JSON.stringify(receipt), failed ? `publication_${outcome}` : null, failed ? `Publication ${outcome}.` : null, completedAt, Date.now(), siteId, operation.operationId, jobKey).run() + } + + async pendingPublicationJobs(siteId: string, limit = 16): Promise { + await ensureSchema(this.database) + const rows = await this.database.prepare(`SELECT prepared_publication_job FROM wp_codebox_operations WHERE site_id = ? AND state = 'publication-pending' AND prepared_publication_job IS NOT NULL ORDER BY updated_at LIMIT ?`).bind(siteId, limit).all<{ prepared_publication_job: string }>() + return rows.results.map((row) => row.prepared_publication_job) + } + + private async byKey(siteId: string, key: string): Promise { + const row = await this.database.prepare(`${operationSelect} WHERE o.site_id = ? AND o.idempotency_key = ?`).bind(siteId, key).first() + return row ? this.hydrate(row) : null + } + + private async hydrate(row: Row): Promise { + const attempts = await this.database.prepare(`SELECT attempt_number, claim_token, started_at, completed_at, state, stage, error_code, error_message FROM wp_codebox_operation_attempts WHERE site_id = ? AND operation_id = ? ORDER BY attempt_number DESC LIMIT 10`).bind(row.site_id, row.operation_id).all() + return operationFromRow(row, attempts.results) + } +} + +const operationSelect = `SELECT o.*, s.hostname, s.origin FROM wp_codebox_operations o JOIN wp_codebox_sites s ON s.site_id = o.site_id` +interface Row { site_id: string; hostname: string; origin: string; operation_id: string; idempotency_key: string; fingerprint: string; artifact_key: string; artifact_sha256: string; artifact_size: number; slug: string; name: string; site_title: string; state: StaticArtifactOperationState; stage: string; progress: number; attempts: number; retry_at: number | null; claim_expires_at: number | null; prepared_version: number | null; prepared_revision: string | null; prepared_manifest_key: string | null; prepared_persisted_at: string | null; prepared_result_json: string | null; prepared_publication_job: string | null; error_code: string | null; error_message: string | null; receipt_json: string | null } +interface AttemptRow { attempt_number: number; claim_token: string; started_at: string; completed_at: string | null; state: string; stage: string; error_code: string | null; error_message: string | null } +function operationFromRow(row: Row, attempts: AttemptRow[]): StaticArtifactOperation { + const prepared = row.prepared_version !== null && row.prepared_revision && row.prepared_manifest_key && row.prepared_persisted_at && row.prepared_result_json ? { version: row.prepared_version, pointer: { revision: row.prepared_revision, manifestKey: row.prepared_manifest_key, persistedAt: row.prepared_persisted_at }, ssiResult: JSON.parse(row.prepared_result_json), publicationJobKey: row.prepared_publication_job } : null + const input = { idempotencyKey: row.idempotency_key, fingerprint: row.fingerprint, artifact: { r2Key: row.artifact_key, sha256: row.artifact_sha256, size: row.artifact_size }, options: { slug: row.slug, name: row.name, siteTitle: row.site_title } } + return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, site: { id: row.site_id, hostname: row.hostname, origin: row.origin }, operationId: row.operation_id, state: row.state, input, attempts: row.attempts, attemptHistory: attempts.map((attempt) => ({ number: attempt.attempt_number, startedAt: attempt.started_at, completedAt: attempt.completed_at, state: attempt.state, stage: attempt.stage, error: attempt.error_code ? { code: attempt.error_code, message: attempt.error_message ?? "Operation failed." } : null })), stage: row.stage, progress: row.progress, retryAt: row.retry_at, claimExpiresAt: row.claim_expires_at, prepared, error: row.error_code ? { code: row.error_code, message: row.error_message ?? "Operation failed." } : null, receipt: row.receipt_json ? JSON.parse(row.receipt_json) as OperationReceipt : null } +} +function sanitizeError(error: unknown): { code: string; message: string } { const message = error instanceof Error ? error.message : "Operation failed."; return { code: error instanceof OperationConflict ? "conflict" : "execution_failed", message: message.replace(/[\r\n\t]/g, " ").slice(0, 500) } } +async function ensureSchema(database: D1Database): Promise { + const existing = schemaReady.get(database as object) + if (!existing) { + const pending = (async () => { + await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_sites (site_id TEXT PRIMARY KEY, hostname TEXT NOT NULL, origin TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('active')), created_at INTEGER NOT NULL, activated_at INTEGER NOT NULL, updated_at INTEGER NOT NULL)`).run() + await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_operations (site_id TEXT NOT NULL, operation_id TEXT NOT NULL, idempotency_key TEXT NOT NULL, fingerprint TEXT NOT NULL, artifact_key TEXT NOT NULL, artifact_sha256 TEXT NOT NULL, artifact_size INTEGER NOT NULL, slug TEXT NOT NULL, name TEXT NOT NULL, site_title TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('queued','running','retryable','publication-pending','succeeded','failed')), stage TEXT NOT NULL, progress INTEGER NOT NULL CHECK (progress BETWEEN 0 AND 100), attempts INTEGER NOT NULL, retry_at INTEGER, claim_token TEXT, claim_expires_at INTEGER, prepared_version INTEGER, prepared_revision TEXT, prepared_manifest_key TEXT, prepared_persisted_at TEXT, prepared_result_json TEXT, prepared_publication_job TEXT, receipt_json TEXT, error_code TEXT, error_message TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, completed_at TEXT, PRIMARY KEY (site_id, operation_id), UNIQUE (site_id, idempotency_key), FOREIGN KEY (site_id) REFERENCES wp_codebox_sites(site_id))`).run() + await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_operation_attempts (site_id TEXT NOT NULL, operation_id TEXT NOT NULL, attempt_number INTEGER NOT NULL, claim_token TEXT NOT NULL, started_at TEXT NOT NULL, completed_at TEXT, state TEXT NOT NULL, stage TEXT NOT NULL, error_code TEXT, error_message TEXT, PRIMARY KEY (site_id, operation_id, attempt_number), FOREIGN KEY (site_id, operation_id) REFERENCES wp_codebox_operations(site_id, operation_id))`).run() + await database.prepare(`CREATE UNIQUE INDEX IF NOT EXISTS wp_codebox_one_active_operation ON wp_codebox_operations(site_id) WHERE state IN ('queued','running','retryable')`).run() + await database.prepare(`CREATE INDEX IF NOT EXISTS wp_codebox_operation_ready ON wp_codebox_operations(site_id, state, retry_at, claim_expires_at, created_at)`).run() + })() + schemaReady.set(database as object, pending) + pending.catch(() => schemaReady.delete(database as object)) + } + await schemaReady.get(database as object) +} diff --git a/packages/runtime-cloudflare/src/request-routing.ts b/packages/runtime-cloudflare/src/request-routing.ts index 431f3a2d6..7c9dea315 100644 --- a/packages/runtime-cloudflare/src/request-routing.ts +++ b/packages/runtime-cloudflare/src/request-routing.ts @@ -8,6 +8,7 @@ export type WorkerRequestRoute = | { kind: "operator-adopt" } | { kind: "operator-fence"; action: "status" | "acquire" | "renew" | "release" } | { kind: "operator-static-artifact-import" } + | { kind: "operator-static-artifact-operation"; operationId: string } | { kind: "operator-publish" } | { kind: "probe"; phase: string } @@ -25,6 +26,10 @@ export function routeWorkerRequest(request: Request): WorkerRequestRoute { 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-static-artifact-operation") { + const operationId = new URL(request.url).searchParams.get("operationId") + return operationId ? { kind: "operator-static-artifact-operation", operationId } : { kind: "probe", phase } + } if (phase === "operator-publish") return { kind: "operator-publish" } return { kind: "probe", phase } } diff --git a/packages/runtime-cloudflare/src/worker-d1.ts b/packages/runtime-cloudflare/src/worker-d1.ts index a3212a4f6..797103168 100644 --- a/packages/runtime-cloudflare/src/worker-d1.ts +++ b/packages/runtime-cloudflare/src/worker-d1.ts @@ -1,4 +1,5 @@ import { D1RevisionCoordinator } from "./d1-revision-coordinator.js" +import { D1OperationRepository } from "./d1-operation-repository.js" import { DurableObjectRevisionCoordinator, WordPressStateCoordinator } from "./state-coordinator.js" export { WordPressStateCoordinator } import type { SiteContext } from "./site-context.js" @@ -13,4 +14,4 @@ export default createCloudflareRuntime((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 -)) +), (env) => new D1OperationRepository(env.WORDPRESS_STATE_DATABASE)) diff --git a/packages/runtime-cloudflare/src/worker.ts b/packages/runtime-cloudflare/src/worker.ts index cb36627d4..a26c6e3fd 100644 --- a/packages/runtime-cloudflare/src/worker.ts +++ b/packages/runtime-cloudflare/src/worker.ts @@ -13,6 +13,7 @@ import { canonicalPublicRoute, MAX_PUBLISHED_PAGE_BYTES, MAX_PUBLISHED_REVISION_ import { RevisionConflict, type MarkdownPointer, type MutationFence, type RevisionCoordinator, type RevisionLease } from "./revision-coordinator.js" import { routeWorkerRequest } from "./request-routing.js" import { readStaticArtifactImport, STATIC_ARTIFACT_IMPORT_RESULT_SCHEMA, StaticArtifactImportError, type StaticArtifactImport } from "./static-artifact-import.js" +import { D1OperationRepository, OperationConflict, STATIC_ARTIFACT_OPERATION_SCHEMA, shouldRecoverPreparedCommit } from "./d1-operation-repository.js" import { toFetchResponse, toPHPRequest } from "./request-translation.js" import { DEFAULT_SITE_CONTEXT, parseSiteContexts, resolveSiteContextFromRequest, siteStorageKeys, type SiteContext } from "./site-context.js" import { validateUploadManifestFiles, validateUploadMetadata } from "./upload-persistence.js" @@ -214,6 +215,7 @@ export interface RuntimeEnv { export function createCloudflareRuntime( resolveCoordinator: (env: Env, site: SiteContext) => RevisionCoordinator, resolveOperatorCoordinator?: (env: Env, site: SiteContext, selector: string) => RevisionCoordinator | undefined, + resolveOperations?: (env: Env) => D1OperationRepository | undefined, ) { return { async fetch(request: Request, env: Env): Promise { @@ -242,7 +244,8 @@ export function createCloudflareRuntime( if (route.kind === "operator-restore") return restoreCanonicalWordPress(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-static-artifact-import") return importCanonicalStaticArtifact(request, env, coordinator, site, resolveOperations?.(env)) + if (route.kind === "operator-static-artifact-operation") return readStaticArtifactOperation(request, env, site, route.operationId, resolveOperations?.(env)) if (route.kind === "operator-publish") return publishCanonicalWordPressPages(request, env, coordinator, site) if (route.kind === "probe") { return runBootProbe(route.phase, env.WORDPRESS_STATE_BUCKET) @@ -262,10 +265,15 @@ export function createCloudflareRuntime( const sites = parseSiteContexts(env.WORDPRESS_SITE_CONTEXTS).sort((left, right) => left.id.localeCompare(right.id)) const site = sites[Math.floor(controller.scheduledTime / 60_000) % sites.length] const coordinator = resolveCoordinator(env, site) - const publication = await drainNextPublicationJob(env, coordinator, site) - // A publication boot is intentionally the only heavyweight runtime in this invocation. - const fence = publication ? undefined : await coordinator.fenceStatus() - const evidence = publication ?? (fence?.active + const operations = resolveOperations?.(env) + const fence = await coordinator.fenceStatus() + const siteCycle = Math.floor(controller.scheduledTime / (60_000 * sites.length)) + const operationFirst = siteCycle % 2 === 0 + // Alternate priority while allowing only one heavyweight runtime per invocation. + let operation = operationFirst && !fence.active ? await runNextStaticArtifactOperation(env, coordinator, site, operations) : null + const publication = operation ? null : await drainNextPublicationJob(env, coordinator, site, operations) + if (!operationFirst && !publication && !fence.active) operation = await runNextStaticArtifactOperation(env, coordinator, site, operations) + const evidence = operation ?? publication ?? (fence.active ? { schema: "wp-codebox/cloudflare-publication/v1" as const, status: "fenced" as const, jobKey: "coordinator" } : await runScheduledWordPressCron(env, coordinator, site, controller.scheduledTime)) console.log(JSON.stringify({ siteId: site.id, ...evidence })) @@ -516,7 +524,7 @@ function samePointer(left: MarkdownPointer, right: MarkdownPointer): boolean { return left.revision === right.revision && left.manifestKey === right.manifestKey && left.persistedAt === right.persistedAt } -async function importCanonicalStaticArtifact(request: Request, env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext): Promise { +async function importCanonicalStaticArtifact(request: Request, env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, operations?: D1OperationRepository): Promise { if (request.method !== "POST") return new Response("Static artifact import requires POST.", { status: 405 }) if (!await isAuthorizedOperator(request, env, site)) { return new Response("Static artifact import authorization failed.", { status: 401 }) @@ -528,6 +536,15 @@ async function importCanonicalStaticArtifact(request: Request, env: RuntimeEnv, if (error instanceof StaticArtifactImportError) return Response.json({ schema: STATIC_ARTIFACT_IMPORT_RESULT_SCHEMA, status: "rejected", error: error.message }, { status: error.status }) throw error } + if (operations) { + try { + const created = await operations.createOrConverge(site, { ...input, artifact: input.artifactReference }) + return Response.json(created.operation, { status: 202, headers: { location: `?phase=operator-static-artifact-operation&operationId=${created.operation.operationId}` } }) + } catch (error) { + if (error instanceof OperationConflict) return Response.json({ schema: STATIC_ARTIFACT_IMPORT_RESULT_SCHEMA, status: "conflict", error: error.message }, { status: 409 }) + throw error + } + } try { return await runCoordinatedWordPressRequest(request, env, coordinator, site, "static-artifact-import", input) } catch (error) { @@ -538,6 +555,53 @@ async function importCanonicalStaticArtifact(request: Request, env: RuntimeEnv, } } +async function readStaticArtifactOperation(request: Request, env: RuntimeEnv, site: SiteContext, operationId: string, operations?: D1OperationRepository): Promise { + if (request.method !== "GET") return new Response("Static artifact operation status requires GET.", { status: 405 }) + if (!await isAuthorizedOperator(request, env, site)) return new Response("Static artifact operation authorization failed.", { status: 401 }) + if (!operations) return new Response("Static artifact operation status is unavailable.", { status: 404 }) + const operation = await operations.get(site.id, operationId) + return operation ? Response.json(operation) : new Response("Static artifact operation is unavailable.", { status: 404 }) +} + +async function runNextStaticArtifactOperation(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, operations?: D1OperationRepository): Promise<{ schema: string; status: string; operationId: string } | null> { + if (!operations) return null + const claimed = await operations.claimNext(site.id) + if (!claimed) return null + try { + if (claimed.prepared) { + const receipt = await coordinator.committed(claimed.prepared.version) + if (shouldRecoverPreparedCommit(claimed.prepared, receipt)) { + await operations.complete(site.id, claimed.operationId, claimed.claimToken, undefined, undefined, site.origin) + return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, status: "recovered", operationId: claimed.operationId } + } + } + const heartbeat = async (stage: string, progress: number) => { + await operations.renew(site.id, claimed.operationId, claimed.claimToken) + await operations.checkpoint(site.id, claimed.operationId, claimed.claimToken, stage, progress) + } + await heartbeat("executing", 10) + const body = { schema: "wp-codebox/cloudflare-static-artifact-import-request/v1", idempotencyKey: claimed.input.idempotencyKey, artifact: claimed.input.artifact, import: claimed.input.options } + const input = await readStaticArtifactImport(new Request("https://scheduled.invalid/?phase=operator-static-artifact-import", { method: "POST", body: JSON.stringify(body) }), env.WORDPRESS_STATE_BUCKET, site) + const response = await runCoordinatedWordPressRequest(new Request("https://scheduled.invalid/?phase=operator-static-artifact-import", { method: "POST" }), env, coordinator, site, "static-artifact-import", input, async () => operations.recordCommit(site.id, claimed.operationId, claimed.claimToken), async (prepared) => operations.prepareCommit(site.id, claimed.operationId, claimed.claimToken, prepared.version, prepared.pointer, prepared.ssiResult, prepared.publicationJobKey), heartbeat) + const result = await response.json() + if (!result || typeof result !== "object" || (result as { status?: unknown }).status !== "imported") throw new Error("Static artifact import did not produce an import receipt.") + await operations.complete(site.id, claimed.operationId, claimed.claimToken, result, response.headers.get("x-wp-codebox-publication-job") ?? undefined, site.origin) + return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, status: "completed", operationId: claimed.operationId } + } catch (error) { + try { + if (error instanceof StaticArtifactImportError || claimed.attempts >= 3) { + await operations.fail(site.id, claimed.operationId, claimed.claimToken, error) + return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, status: "failed", operationId: claimed.operationId } + } + await operations.retry(site.id, claimed.operationId, claimed.claimToken, error, Date.now() + 30_000) + return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, status: "retryable", operationId: claimed.operationId } + } catch (transitionError) { + if (transitionError instanceof OperationConflict) return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, status: "claim-lost", operationId: claimed.operationId } + throw transitionError + } + } +} + function isCanonicalRestorePointer(pointer: unknown, site: SiteContext): pointer is MarkdownPointer { if (!pointer || typeof pointer !== "object") return false const candidate = pointer as Partial @@ -696,7 +760,11 @@ function publicationClaimObjectKey(job: PublicationJob, site: SiteContext): stri } function publicationReceiptObjectKey(job: PublicationJob, site: SiteContext): string { - return `${siteStorageKeys(site).publicationReceiptPrefix}/${job.key.split("/").at(-1)}` + return publicationReceiptObjectKeyForJob(job.key, site) +} + +function publicationReceiptObjectKeyForJob(jobKey: string, site: SiteContext): string { + return `${siteStorageKeys(site).publicationReceiptPrefix}/${jobKey.split("/").at(-1)}` } async function enqueuePublicationJob(bucket: R2Bucket, lease: Lease, canonical: MarkdownPointer, current: CurrentPublication | null, changes: PublicationChanges, site: SiteContext): Promise { @@ -781,7 +849,8 @@ async function releasePublicationClaim(bucket: R2Bucket, job: PublicationJob, si if (claim.etag) await bucket.put(publicationClaimObjectKey(job, site), JSON.stringify({ expiresAt: 0 }), { onlyIf: { etagMatches: claim.etag }, httpMetadata: { contentType: "application/json" } }) } -async function drainNextPublicationJob(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext): Promise { +async function drainNextPublicationJob(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, operations?: D1OperationRepository): Promise { + await reconcilePublicationReceipts(env.WORDPRESS_STATE_BUCKET, site, operations) let publicationLease: RevisionLease try { publicationLease = await coordinator.acquire() @@ -793,13 +862,30 @@ async function drainNextPublicationJob(env: RuntimeEnv, coordinator: RevisionCoo try { return await drainNextPublicationJobWhileLeased(env, coordinator, site, async () => { publicationLease = await coordinator.renew(publicationLease) - }) + }, operations) } finally { await abortLease(coordinator, site.origin, publicationLease) } } -async function drainNextPublicationJobWhileLeased(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, renewLease: () => Promise): Promise { +async function reconcilePublicationReceipts(bucket: R2Bucket, site: SiteContext, operations?: D1OperationRepository): Promise { + if (!operations) return + const pending = await operations.pendingPublicationJobs(site.id) + for (const jobKey of pending) { + const object = await bucket.get(publicationReceiptObjectKeyForJob(jobKey, site)) + if (!object || object.size > MAX_PUBLISHED_REVISION_BYTES) continue + let receipt: { status?: unknown; job?: unknown; publication?: unknown } + try { + receipt = JSON.parse(await object.text()) as { status?: unknown; job?: unknown; publication?: unknown } + } catch { + continue + } + if (receipt.job !== jobKey || !["promoted", "superseded", "orphaned"].includes(receipt.status as string)) continue + await operations.reconcilePublication(site.id, receipt.job, receipt.status as "promoted" | "superseded" | "orphaned", typeof receipt.publication === "string" ? receipt.publication : undefined) + } +} + +async function drainNextPublicationJobWhileLeased(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, renewLease: () => Promise, operations?: D1OperationRepository): Promise { const listed = await env.WORDPRESS_STATE_BUCKET.list({ prefix: `${siteStorageKeys(site).publicationJobPrefix}/`, limit: 16 }) if (!listed.objects.length) return null const state = await coordinator.state() @@ -814,6 +900,7 @@ async function drainNextPublicationJobWhileLeased(env: RuntimeEnv, coordinator: if (state.version > candidate.coordinatorVersion || Date.now() - Date.parse(candidate.createdAt) > PUBLICATION_CLAIM_MS) { await renewLease() await putImmutableJson(env.WORDPRESS_STATE_BUCKET, publicationReceiptObjectKey(candidate, site), JSON.stringify({ status: "orphaned", job: candidate.key, recordedAt: new Date().toISOString() })) + await operations?.reconcilePublication(site.id, candidate.key, "orphaned") await renewLease() await env.WORDPRESS_STATE_BUCKET.delete(candidate.key) } @@ -828,6 +915,7 @@ async function drainNextPublicationJobWhileLeased(env: RuntimeEnv, coordinator: if (!current || current.publication.canonicalVersion >= job.coordinatorVersion) { await renewLease() await putImmutableJson(env.WORDPRESS_STATE_BUCKET, publicationReceiptObjectKey(job, site), JSON.stringify({ status: "superseded", job: job.key, recordedAt: new Date().toISOString() })) + await operations?.reconcilePublication(site.id, job.key, "superseded") await renewLease() await env.WORDPRESS_STATE_BUCKET.delete([job.key, publicationProgressObjectKey(job, site)]) return { schema: "wp-codebox/cloudflare-publication/v1", status: "stale", jobKey: job.key } @@ -876,6 +964,7 @@ async function drainNextPublicationJobWhileLeased(env: RuntimeEnv, coordinator: await writePublicationProgress(env.WORDPRESS_STATE_BUCKET, job, { ...progress, completedAt: new Date().toISOString() }, site) await renewLease() await putImmutableJson(env.WORDPRESS_STATE_BUCKET, publicationReceiptObjectKey(job, site), JSON.stringify({ status: "promoted", job: job.key, publication: publication.revision, recordedAt: new Date().toISOString() })) + await operations?.reconcilePublication(site.id, job.key, "promoted", publication.revision) await renewLease() await env.WORDPRESS_STATE_BUCKET.delete([job.key, publicationProgressObjectKey(job, site)]) return { schema: "wp-codebox/cloudflare-publication/v1", status: "promoted", jobKey: job.key } @@ -936,7 +1025,7 @@ function validateWordPressPageSnapshot(snapshot: WordPressPageSnapshot, canonica || typeof snapshot.body !== "string") throw new Error("Published page artifact is invalid.") } -async function runCoordinatedWordPressRequest(request: Request, env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, route: "wordpress" | "health" | "r2-mutate" | "static-artifact-import", staticArtifactImport?: StaticArtifactImport): Promise { +async function runCoordinatedWordPressRequest(request: Request, env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, route: "wordpress" | "health" | "r2-mutate" | "static-artifact-import", staticArtifactImport?: StaticArtifactImport, onCommitted?: (committed: { pointer: MarkdownPointer; version: number }) => Promise, onPreparedCommit?: (prepared: { pointer: MarkdownPointer; version: number; ssiResult: unknown; publicationJobKey: string | null }) => Promise, heartbeat?: (stage: string, progress: number) => Promise): Promise { if (route === "r2-mutate" && request.method !== "POST") return new Response("WordPress state mutation requires POST.", { status: 405 }) if (route === "wordpress" && isCacheableWordPressPageRequest(request)) { const state = await coordinator.state() @@ -962,7 +1051,9 @@ async function runCoordinatedWordPressRequest(request: Request, env: RuntimeEnv, finalized = true return cachedPage } + if (route === "static-artifact-import") await heartbeat?.("booting-runtime", 20) runtime = await getRuntime(env, lease.pointer, site, route === "static-artifact-import") + if (route === "static-artifact-import") await heartbeat?.("runtime-ready", 35) const mutatesCanonicalState = isMutation(request, route) const diagnosticsStartedAt = Date.now() const retained = new MutationRetainedBytes() @@ -974,9 +1065,11 @@ async function runCoordinatedWordPressRequest(request: Request, env: RuntimeEnv, let publicationChanges: PublicationChanges | undefined if (route === "static-artifact-import") { if (!staticArtifactImport) throw new Error("Static artifact import input is unavailable.") + await heartbeat?.("executing-ssi", 45) initializePublicationChanges(runtime.php) runtime.php.writeFile(MARKDOWN_CHANGES_PATH, new TextEncoder().encode(JSON.stringify({ created: [], changed: [], deleted: [] }))) const imported = await runStaticArtifactImport(runtime, staticArtifactImport) + await heartbeat?.("ssi-completed", 60) if (imported.status !== "imported") { await discardRuntime(runtime, site) runtime = undefined @@ -1014,7 +1107,9 @@ async function runCoordinatedWordPressRequest(request: Request, env: RuntimeEnv, if (mutatesCanonicalState) { if (!canonicalChanges || !publicationChanges) throw new Error("Canonical mutation completed without its persistence evidence.") logMutationPhase(diagnosticsStartedAt, "php-request", retained, { canonicalCreated: canonicalChanges.created.length, canonicalChanged: canonicalChanges.changed.length, canonicalDeleted: canonicalChanges.deleted.length }) + if (route === "static-artifact-import") await heartbeat?.("persisting-canonical", 70) const next = await persistRuntime(env.WORDPRESS_STATE_BUCKET, runtime, canonicalChanges, site, diagnosticsStartedAt, retained) + if (route === "static-artifact-import") await heartbeat?.("canonical-prepared", 80) if (!response) { if (!phpResponse) throw new Error("WordPress mutation completed without a PHP response.") retained.retain(responseBodyBytes) @@ -1026,7 +1121,13 @@ async function runCoordinatedWordPressRequest(request: Request, env: RuntimeEnv, const publicationJob = await enqueuePublicationJob(env.WORDPRESS_STATE_BUCKET, lease, next, currentPublication, publicationChanges, site) response.headers.set("x-wp-codebox-publication", publicationJob ? "queued" : "unchanged") if (publicationJob) response.headers.set("x-wp-codebox-publication-job", publicationJob.key) + if (route === "static-artifact-import") { + await heartbeat?.("preparing-commit", 85) + await onPreparedCommit?.({ pointer: next, version: lease.version + 1, ssiResult: await response.clone().json(), publicationJobKey: publicationJob?.key ?? null }) + await heartbeat?.("committing", 90) + } const committed = await commitLease(coordinator, request.url, lease, next) + await onCommitted?.(committed) response.headers.set("x-wp-codebox-canonical-revision", committed.pointer.revision) response.headers.set("x-wp-codebox-canonical-version", String(committed.version)) logMutationPhase(diagnosticsStartedAt, "commit", retained, { publication: publicationJob ? "queued" : "unchanged" }) diff --git a/tests/cloudflare-d1-operation-repository.test.ts b/tests/cloudflare-d1-operation-repository.test.ts new file mode 100644 index 000000000..01d613bdf --- /dev/null +++ b/tests/cloudflare-d1-operation-repository.test.ts @@ -0,0 +1,93 @@ +import assert from "node:assert/strict" +import { DatabaseSync } from "node:sqlite" +import test from "node:test" +import { D1OperationRepository, OperationConflict, shouldRecoverPreparedCommit } from "../packages/runtime-cloudflare/src/d1-operation-repository.js" + +function database(): D1Database { + const sqlite = new DatabaseSync(":memory:") + let queued = Promise.resolve() + return { prepare(query: string) { const statement = sqlite.prepare(query); let values: unknown[] = []; return { bind(...next: unknown[]) { values = next; return this }, async run() { return { meta: { changes: statement.run(...values).changes } } }, async first() { return statement.get(...values) as T | null }, async all() { return { results: statement.all(...values) as T[] } } } }, async batch(statements: Array<{ run: () => Promise<{ meta: { changes: number } }> }>) { let release!: () => void; const previous = queued; queued = new Promise((resolve) => { release = resolve }); await previous; sqlite.exec("BEGIN"); try { const results = []; for (const statement of statements) results.push(await statement.run()); sqlite.exec("COMMIT"); return results } catch (error) { sqlite.exec("ROLLBACK"); throw error } finally { release() } } } as unknown as D1Database +} +const site = { id: "alpha", hostname: "alpha.example", origin: "https://alpha.example" } +const beta = { id: "beta", hostname: "beta.example", origin: "https://beta.example" } +function input(key = "key", fingerprint = "a".repeat(64)) { return { idempotencyKey: key, fingerprint, artifact: { r2Key: `sites/alpha/import-artifacts/${"b".repeat(64)}.json`, sha256: "b".repeat(64), size: 1 }, options: { slug: "site", name: "Site", siteTitle: "Site" } } } +const pointer = { revision: "revision", manifestKey: "sites/alpha/markdown/revisions/revision.json", persistedAt: "2026-07-23T00:00:00.000Z" } + +test("D1 operations converge by site/key/fingerprint and reject changed input or site identity", async () => { + const repository = new D1OperationRepository(database()) + const first = await repository.createOrConverge(site, input()) + const duplicate = await repository.createOrConverge(site, input()) + assert.equal(first.created, true) + assert.equal(duplicate.created, false) + assert.equal(duplicate.operation.operationId, first.operation.operationId) + await assert.rejects(() => repository.createOrConverge(site, input("key", "c".repeat(64))), OperationConflict) + await assert.rejects(() => repository.createOrConverge({ ...site, origin: "https://other.example" }, input("other")), /identity conflicts/) +}) + +test("D1 enforces one active mutation per site while independent sites claim concurrently", async () => { + const repository = new D1OperationRepository(database()) + await repository.createOrConverge(site, input("one")) + await assert.rejects(() => repository.createOrConverge(site, input("two")), /active/) + await repository.createOrConverge(beta, input("two")) + const [alphaClaim, racedAlphaClaim, betaClaim] = await Promise.all([repository.claimNext(site.id), repository.claimNext(site.id), repository.claimNext(beta.id)]) + assert.equal([alphaClaim, racedAlphaClaim].filter(Boolean).length, 1) + assert.ok(betaClaim) + assert.notEqual((alphaClaim ?? racedAlphaClaim)!.operationId, betaClaim.operationId) +}) + +test("expired claims recover, retries retain attempt history, and terminal receipts are immutable", async () => { + const repository = new D1OperationRepository(database()) + const created = await repository.createOrConverge(site, input()) + const claim = await repository.claimNext(site.id) + assert.ok(claim) + await repository.retry(site.id, claim.operationId, claim.claimToken, new Error("retry\nsecret"), Date.now()) + const recovered = await repository.claimNext(site.id) + assert.ok(recovered) + await repository.prepareCommit(site.id, recovered.operationId, recovered.claimToken, 7, pointer, { imported: true }, null) + await repository.complete(site.id, recovered.operationId, recovered.claimToken, { imported: true }, undefined, site.origin) + const terminal = await repository.get(site.id, created.operation.operationId) + assert.equal(terminal?.state, "succeeded") + assert.equal(terminal?.stage, "completed") + assert.equal(terminal?.attemptHistory.length, 2) + assert.equal(terminal?.attemptHistory[0].state, "succeeded") + assert.equal("token" in terminal!.attemptHistory[0], false) + assert.equal(terminal?.receipt?.publication.status, "none") + assert.ok(terminal?.receipt?.terminalCompletedAt) + assert.deepEqual(terminal?.receipt?.canonical, { pointer, version: 7 }) + assert.equal(await repository.claimNext(site.id), null) + await assert.rejects(() => repository.fail(site.id, recovered.operationId, recovered.claimToken, new Error("late")), OperationConflict) + + await repository.createOrConverge(beta, input("expiry")) + const expired = await repository.claimNext(beta.id) + assert.ok(expired) + const reclaimed = await repository.claimNext(beta.id, Date.now() + 600_001) + assert.ok(reclaimed) + assert.notEqual(reclaimed.claimToken, expired.claimToken) +}) + +test("claim and terminal batches keep attempt rows consistent and publication reconciliation is truthful", async () => { + const repository = new D1OperationRepository(database()) + const created = await repository.createOrConverge(site, input()) + const claim = await repository.claimNext(site.id) + assert.ok(claim) + let operation = await repository.get(site.id, created.operation.operationId) + assert.equal(operation?.attemptHistory.length, 1) + assert.equal(operation?.attemptHistory[0].completedAt, null) + await repository.prepareCommit(site.id, claim.operationId, claim.claimToken, 8, pointer, { imported: true }, "job-8") + await repository.complete(site.id, claim.operationId, claim.claimToken, { imported: true }, "job-8", site.origin) + operation = await repository.get(site.id, created.operation.operationId) + assert.equal(operation?.state, "publication-pending") + assert.equal(operation?.attemptHistory[0].completedAt !== null, true) + await repository.reconcilePublication(site.id, "job-8", "promoted", "publication-revision") + operation = await repository.get(site.id, created.operation.operationId) + assert.equal(operation?.state, "succeeded") + assert.deepEqual(operation?.receipt?.publication, { status: "promoted", jobKey: "job-8", revision: "publication-revision" }) + assert.ok(operation?.receipt?.terminalCompletedAt) +}) + +test("exact prepared-pointer recovery only skips SSI for the matching coordinator receipt", () => { + const prepared = { pointer, version: 7, ssiResult: { imported: true }, publicationJobKey: null } + assert.equal(shouldRecoverPreparedCommit(prepared, pointer), true) + assert.equal(shouldRecoverPreparedCommit(prepared, { ...pointer, revision: "unrelated", manifestKey: "sites/alpha/markdown/revisions/unrelated.json" }), false) + assert.equal(shouldRecoverPreparedCommit(null, pointer), false) +}) diff --git a/tests/cloudflare-runtime.test.ts b/tests/cloudflare-runtime.test.ts index 437642ad8..7c9c3f38c 100644 --- a/tests/cloudflare-runtime.test.ts +++ b/tests/cloudflare-runtime.test.ts @@ -106,6 +106,7 @@ test("Cloudflare routing reserves phases while the phase-less route serves WordP assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-fence-renew")), { kind: "operator-fence", action: "renew" }) assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-fence-release")), { kind: "operator-fence", action: "release" }) assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-static-artifact-import")), { kind: "operator-static-artifact-import" }) + assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-static-artifact-operation&operationId=operation-id")), { kind: "operator-static-artifact-operation", operationId: "operation-id" }) 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" }) }) @@ -471,8 +472,9 @@ test("Cloudflare runtime injects composable coordinators without moving PHP out assert.match(worker, /validateWpContentDeletedPaths\(manifest\.wpContentDeleted \?\? \[\]\)/) assert.match(worker, /materializeWpContentTombstones\(php, wpContentDeleted\)/) assert.match(worker, /async scheduled\(controller: ScheduledController, env: Env\)/) - assert.match(worker, /const publication = await drainNextPublicationJob\(env, coordinator, site\)/) - assert.match(worker, /const fence = publication \? undefined : await coordinator\.fenceStatus\(\)/) + assert.match(worker, /const publication = operation \? null : await drainNextPublicationJob\(env, coordinator, site, operations\)/) + assert.match(worker, /const fence = await coordinator\.fenceStatus\(\)/) + assert.match(worker, /const operationFirst = siteCycle % 2 === 0/) assert.match(worker, /: await runScheduledWordPressCron\(env, coordinator, site, controller\.scheduledTime\)/) assert.match(worker, /pathname === "\/wp-cron\.php"/) assert.match(worker, /MAX_CRON_EVENTS_PER_INVOCATION = 5/)