Skip to content

Commit 4b0a5f7

Browse files
committed
feat(runtime-cloudflare): expose allocation lifecycle controls
1 parent a811982 commit 4b0a5f7

6 files changed

Lines changed: 66 additions & 22 deletions

File tree

packages/runtime-cloudflare/src/allocation-lifecycle.ts

Lines changed: 16 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,8 @@ import type { SiteContext } from "./site-context.js"
33
export type AllocationLifecycleState = "active" | "deleting" | "tombstoned"
44
export interface AllocationIdentity { siteId: string; generation: number }
55
export interface AllocationRetentionPolicy { ttlMs: number; retainMs: number }
6-
export interface AllocationDeletionReceipt { schema: "wp-codebox/cloudflare-allocation-deletion-receipt/v1"; siteId: string; generation: number; deletedObjects: number; completedAt: string }
7-
export interface AllocationLifecycle { identity: AllocationIdentity; principal: string; state: AllocationLifecycleState; expiresAt: number; retainUntil: number; mutationFence: number; operationFence: number; cleanupCursor: string | null; deletedObjects: number; receipt: AllocationDeletionReceipt | null }
6+
export interface AllocationDeletionReceipt { schema: "wp-codebox/cloudflare-allocation-deletion-receipt/v1"; siteId: string; generation: number; deletedObjects: number; deletedBytes: number; retainedObjects: number; retainedBytes: number; unresolvedObjects: number; startedAt: string; completedAt: string; terminalState: "tombstoned" }
7+
export interface AllocationLifecycle { identity: AllocationIdentity; principal: string; state: AllocationLifecycleState; expiresAt: number; retainUntil: number; mutationFence: number; operationFence: number; cleanupCursor: string | null; deletedObjects: number; deletedBytes: number; receipt: AllocationDeletionReceipt | null }
88

99
const RECEIPT_SCHEMA = "wp-codebox/cloudflare-allocation-deletion-receipt/v1" as const
1010
const schemaReady = new WeakMap<object, Promise<void>>()
@@ -19,14 +19,14 @@ export class CloudflareAllocationLifecycle {
1919
async create(site: SiteContext, principal: string, now = Date.now()): Promise<AllocationLifecycle> {
2020
await this.initialize()
2121
const identity = allocationIdentity(site.id)
22-
await this.database.prepare("INSERT OR IGNORE INTO wp_codebox_site_lifecycles (site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, created_at, updated_at) VALUES (?, ?, ?, 'active', ?, ?, 1, 1, NULL, 0, ?, ?)").bind(site.id, identity.generation, principal, now + this.policy.ttlMs, now + this.policy.ttlMs + this.policy.retainMs, now, now).run()
22+
await this.database.prepare("INSERT OR IGNORE INTO wp_codebox_site_lifecycles (site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, deleted_bytes, created_at, updated_at) VALUES (?, ?, ?, 'active', ?, ?, 1, 1, NULL, 0, 0, ?, ?)").bind(site.id, identity.generation, principal, now + this.policy.ttlMs, now + this.policy.ttlMs + this.policy.retainMs, now, now).run()
2323
const lifecycle = await this.get(identity)
2424
if (!lifecycle || lifecycle.principal !== principal) throw new AllocationLifecycleConflict("Allocation identity conflicts with an existing owner.")
2525
return lifecycle
2626
}
2727
async get(identity: AllocationIdentity): Promise<AllocationLifecycle | null> {
2828
await this.initialize()
29-
const row = await this.database.prepare("SELECT site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, receipt_json FROM wp_codebox_site_lifecycles WHERE site_id = ? AND generation = ?").bind(identity.siteId, identity.generation).first<LifecycleRow>()
29+
const row = await this.database.prepare("SELECT site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, deleted_bytes, receipt_json FROM wp_codebox_site_lifecycles WHERE site_id = ? AND generation = ?").bind(identity.siteId, identity.generation).first<LifecycleRow>()
3030
return row ? hydrate(row) : null
3131
}
3232
async renew(identity: AllocationIdentity, principal: string, now = Date.now()): Promise<AllocationLifecycle> {
@@ -48,9 +48,9 @@ export class CloudflareAllocationLifecycle {
4848
async beginDeletion(identity: AllocationIdentity, principal: string | null, now = Date.now()): Promise<number> {
4949
await this.initialize()
5050
const owner = principal === null ? "" : " AND principal = ?"
51-
const values: unknown[] = [now, identity.siteId, identity.generation]
51+
const values: unknown[] = [now + this.policy.retainMs, now, identity.siteId, identity.generation]
5252
if (principal !== null) values.push(principal)
53-
const result = await this.database.prepare(`UPDATE wp_codebox_site_lifecycles SET state = 'deleting', mutation_fence = mutation_fence + 1, operation_fence = operation_fence + 1, cleanup_cursor = NULL, updated_at = ? WHERE site_id = ? AND generation = ?${owner} AND state = 'active'`).bind(...values).run()
53+
const result = await this.database.prepare(`UPDATE wp_codebox_site_lifecycles SET state = 'deleting', mutation_fence = mutation_fence + 1, operation_fence = operation_fence + 1, retain_until = MIN(retain_until, ?), cleanup_cursor = NULL, updated_at = ? WHERE site_id = ? AND generation = ?${owner} AND state = 'active'`).bind(...values).run()
5454
if (result.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation is not deletable by this owner.")
5555
return (await this.get(identity))!.operationFence
5656
}
@@ -63,22 +63,24 @@ export class CloudflareAllocationLifecycle {
6363
}
6464
return expired
6565
}
66-
async pendingDeletions(limit = 8): Promise<Array<AllocationLifecycle>> {
66+
async pendingDeletions(limit = 8, now = Date.now()): Promise<Array<AllocationLifecycle>> {
6767
await this.initialize()
68-
const rows = await this.database.prepare("SELECT site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, receipt_json FROM wp_codebox_site_lifecycles WHERE state = 'deleting' ORDER BY updated_at LIMIT ?").bind(limit).all<LifecycleRow>()
68+
const rows = await this.database.prepare("SELECT site_id, generation, principal, state, expires_at, retain_until, mutation_fence, operation_fence, cleanup_cursor, deleted_objects, deleted_bytes, receipt_json FROM wp_codebox_site_lifecycles WHERE state = 'deleting' AND retain_until <= ? ORDER BY updated_at LIMIT ?").bind(now, limit).all<LifecycleRow>()
6969
return rows.results.map(hydrate)
7070
}
7171
async reclaim(bucket: Pick<R2Bucket, "list" | "delete">, identity: AllocationIdentity, operationFence: number, limit = 100, now = Date.now()): Promise<AllocationLifecycle> {
7272
if (!Number.isInteger(limit) || limit < 1 || limit > 1000) throw new Error("Allocation reclamation limit is invalid.")
7373
const lifecycle = await this.get(identity)
7474
if (!lifecycle || lifecycle.state !== "deleting" || lifecycle.operationFence !== operationFence) throw new AllocationLifecycleConflict("Allocation deletion fence changed.")
75+
if (lifecycle.retainUntil > now) throw new AllocationLifecycleConflict("Allocation retention period is still active.")
7576
const page = await bucket.list({ prefix: `sites/${identity.siteId}/`, cursor: lifecycle.cleanupCursor ?? undefined, limit })
7677
if (page.objects.length) await bucket.delete(page.objects.map((object) => object.key))
7778
const cursor = page.truncated ? page.cursor : null
78-
const updated = await this.database.prepare("UPDATE wp_codebox_site_lifecycles SET cleanup_cursor = ?, deleted_objects = deleted_objects + ?, updated_at = ? WHERE site_id = ? AND generation = ? AND state = 'deleting' AND operation_fence = ?").bind(cursor, page.objects.length, now, identity.siteId, identity.generation, operationFence).run()
79+
const deletedBytes = page.objects.reduce((total, object) => total + (object.size ?? 0), 0)
80+
const updated = await this.database.prepare("UPDATE wp_codebox_site_lifecycles SET cleanup_cursor = ?, deleted_objects = deleted_objects + ?, deleted_bytes = deleted_bytes + ?, updated_at = ? WHERE site_id = ? AND generation = ? AND state = 'deleting' AND operation_fence = ?").bind(cursor, page.objects.length, deletedBytes, now, identity.siteId, identity.generation, operationFence).run()
7981
if (updated.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation deletion fence changed.")
8082
if (cursor) return (await this.get(identity))!
81-
const receipt: AllocationDeletionReceipt = { schema: RECEIPT_SCHEMA, siteId: identity.siteId, generation: identity.generation, deletedObjects: lifecycle.deletedObjects + page.objects.length, completedAt: new Date(now).toISOString() }
83+
const receipt: AllocationDeletionReceipt = { schema: RECEIPT_SCHEMA, siteId: identity.siteId, generation: identity.generation, deletedObjects: lifecycle.deletedObjects + page.objects.length, deletedBytes: lifecycle.deletedBytes + deletedBytes, retainedObjects: 0, retainedBytes: 0, unresolvedObjects: 0, startedAt: new Date(lifecycle.expiresAt).toISOString(), completedAt: new Date(now).toISOString(), terminalState: "tombstoned" }
8284
await this.database.batch([
8385
this.database.prepare("INSERT OR IGNORE INTO wp_codebox_site_deletion_receipts (site_id, generation, receipt_json, created_at) VALUES (?, ?, ?, ?)").bind(identity.siteId, identity.generation, JSON.stringify(receipt), now),
8486
this.database.prepare("UPDATE wp_codebox_site_lifecycles SET state = 'tombstoned', cleanup_cursor = NULL, receipt_json = ?, terminal_at = ?, updated_at = ? WHERE site_id = ? AND generation = ? AND state = 'deleting' AND operation_fence = ?").bind(JSON.stringify(receipt), now, now, identity.siteId, identity.generation, operationFence),
@@ -93,13 +95,14 @@ export function allocationIdentity(siteId: string): AllocationIdentity {
9395
return { siteId, generation: match ? Number(match[1]) : 1 }
9496
}
9597

96-
interface LifecycleRow { site_id: string; generation: number; principal: string; state: AllocationLifecycleState; expires_at: number; retain_until: number; mutation_fence: number; operation_fence: number; cleanup_cursor: string | null; deleted_objects: number; receipt_json: string | null }
97-
function hydrate(row: LifecycleRow): AllocationLifecycle { return { identity: { siteId: row.site_id, generation: row.generation }, principal: row.principal, state: row.state, expiresAt: row.expires_at, retainUntil: row.retain_until, mutationFence: row.mutation_fence, operationFence: row.operation_fence, cleanupCursor: row.cleanup_cursor, deletedObjects: row.deleted_objects, receipt: row.receipt_json ? JSON.parse(row.receipt_json) as AllocationDeletionReceipt : null } }
98+
interface LifecycleRow { site_id: string; generation: number; principal: string; state: AllocationLifecycleState; expires_at: number; retain_until: number; mutation_fence: number; operation_fence: number; cleanup_cursor: string | null; deleted_objects: number; deleted_bytes: number; receipt_json: string | null }
99+
function hydrate(row: LifecycleRow): AllocationLifecycle { return { identity: { siteId: row.site_id, generation: row.generation }, principal: row.principal, state: row.state, expiresAt: row.expires_at, retainUntil: row.retain_until, mutationFence: row.mutation_fence, operationFence: row.operation_fence, cleanupCursor: row.cleanup_cursor, deletedObjects: row.deleted_objects, deletedBytes: row.deleted_bytes, receipt: row.receipt_json ? JSON.parse(row.receipt_json) as AllocationDeletionReceipt : null } }
98100
async function ensureSchema(database: D1Database): Promise<void> {
99101
let pending = schemaReady.get(database as object)
100102
if (!pending) {
101103
pending = (async () => {
102-
await database.prepare("CREATE TABLE IF NOT EXISTS wp_codebox_site_lifecycles (site_id TEXT NOT NULL, generation INTEGER NOT NULL, principal TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('active','deleting','tombstoned')), expires_at INTEGER NOT NULL, retain_until INTEGER NOT NULL, mutation_fence INTEGER NOT NULL, operation_fence INTEGER NOT NULL, cleanup_cursor TEXT, deleted_objects INTEGER NOT NULL, receipt_json TEXT, terminal_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation))").run()
104+
await database.prepare("CREATE TABLE IF NOT EXISTS wp_codebox_site_lifecycles (site_id TEXT NOT NULL, generation INTEGER NOT NULL, principal TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('active','deleting','tombstoned')), expires_at INTEGER NOT NULL, retain_until INTEGER NOT NULL, mutation_fence INTEGER NOT NULL, operation_fence INTEGER NOT NULL, cleanup_cursor TEXT, deleted_objects INTEGER NOT NULL, deleted_bytes INTEGER NOT NULL DEFAULT 0, receipt_json TEXT, terminal_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation))").run()
105+
try { await database.prepare("ALTER TABLE wp_codebox_site_lifecycles ADD COLUMN deleted_bytes INTEGER NOT NULL DEFAULT 0").run() } catch (error) { if (!(error instanceof Error) || !/duplicate column/i.test(error.message)) throw error }
103106
await database.prepare("CREATE TABLE IF NOT EXISTS wp_codebox_site_deletion_receipts (site_id TEXT NOT NULL, generation INTEGER NOT NULL, receipt_json TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation))").run()
104107
await database.prepare("CREATE INDEX IF NOT EXISTS wp_codebox_site_lifecycle_expiry ON wp_codebox_site_lifecycles(state, expires_at)").run()
105108
})()

packages/runtime-cloudflare/src/d1-operation-repository.ts

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -72,13 +72,15 @@ export class D1OperationRepository {
7272

7373
async activeSites(): Promise<SiteContext[]> {
7474
await ensureSchema(this.database)
75-
const rows = await this.database.prepare("SELECT site_id, hostname, origin FROM wp_codebox_sites WHERE state = 'active' ORDER BY site_id").all<{ site_id: string; hostname: string; origin: string }>()
75+
const rows = await this.database.prepare("SELECT s.site_id, s.hostname, s.origin FROM wp_codebox_sites s LEFT JOIN wp_codebox_site_lifecycles l ON l.site_id = s.site_id AND l.generation = s.generation WHERE s.state = 'active' AND (l.site_id IS NULL OR l.state = 'active') ORDER BY s.site_id").all<{ site_id: string; hostname: string; origin: string }>()
7676
return rows.results.map((row) => ({ id: row.site_id, hostname: row.hostname, origin: row.origin }))
7777
}
7878

7979
async claimNext(siteId: string, now = Date.now()): Promise<(StaticArtifactOperation & { claimToken: string }) | null> {
8080
await ensureSchema(this.database)
81-
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 }>()
81+
// A dynamic allocation may be fenced between queue deliveries. Configured
82+
// static sites have no lifecycle row and retain their existing behavior.
83+
const candidate = await this.database.prepare(`SELECT o.operation_id FROM wp_codebox_operations o JOIN wp_codebox_sites s ON s.site_id = o.site_id LEFT JOIN wp_codebox_site_lifecycles l ON l.site_id = o.site_id AND l.generation = s.generation WHERE o.site_id = ? AND (l.site_id IS NULL OR l.state = 'active') AND (o.state = 'queued' OR (o.state = 'retryable' AND o.retry_at <= ?) OR (o.state = 'running' AND o.claim_expires_at <= ?)) ORDER BY o.created_at LIMIT 1`).bind(siteId, now, now).first<{ operation_id: string }>()
8284
if (!candidate) return null
8385
const token = crypto.randomUUID()
8486
const expiresAt = now + this.claimMs
@@ -188,6 +190,9 @@ async function ensureSchema(database: D1Database): Promise<void> {
188190
const pending = (async () => {
189191
await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_sites (site_id TEXT PRIMARY KEY, hostname TEXT NOT NULL, origin TEXT NOT NULL, generation INTEGER NOT NULL DEFAULT 1, state TEXT NOT NULL CHECK (state IN ('active')), created_at INTEGER NOT NULL, activated_at INTEGER NOT NULL, updated_at INTEGER NOT NULL)`).run()
190192
try { await database.prepare("ALTER TABLE wp_codebox_sites ADD COLUMN generation INTEGER NOT NULL DEFAULT 1").run() } catch (error) { if (!(error instanceof Error) || !/duplicate column/i.test(error.message)) throw error }
193+
// Lifecycle owns this schema; the minimal compatible creation keeps queue
194+
// fencing available when the operation repository initializes first.
195+
await database.prepare("CREATE TABLE IF NOT EXISTS wp_codebox_site_lifecycles (site_id TEXT NOT NULL, generation INTEGER NOT NULL, principal TEXT NOT NULL, state TEXT NOT NULL, expires_at INTEGER NOT NULL, retain_until INTEGER NOT NULL, mutation_fence INTEGER NOT NULL, operation_fence INTEGER NOT NULL, cleanup_cursor TEXT, deleted_objects INTEGER NOT NULL, receipt_json TEXT, terminal_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation))").run()
191196
await database.prepare(`CREATE UNIQUE INDEX IF NOT EXISTS wp_codebox_site_hostname ON wp_codebox_sites(hostname)`).run()
192197
await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_site_aliases (hostname TEXT PRIMARY KEY, site_id TEXT NOT NULL, created_at INTEGER NOT NULL, FOREIGN KEY (site_id) REFERENCES wp_codebox_sites(site_id))`).run()
193198
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()

0 commit comments

Comments
 (0)