feat(runtime-cloudflare): fence lifecycle teardown

Author: chubes4

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

@@ -3,8 +3,8 @@ import type { SiteContext } from "./site-context.js"
 export type AllocationLifecycleState = "active" | "deleting" | "tombstoned"
 export interface AllocationIdentity { siteId: string; generation: number }
 export interface AllocationRetentionPolicy { ttlMs: number; retainMs: number }
-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" }
-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 }
+export interface AllocationDeletionReceipt { schema: "wp-codebox/cloudflare-allocation-deletion-receipt/v1"; siteId: string; generation: number; deletedObjects: number; deletedBytes: number; deletedRecords: number; retainedObjects: number; retainedBytes: number; retainedRecords: number; unresolvedObjects: number; unresolvedRecords: number; startedAt: string; completedAt: string; terminalState: "tombstoned" }
+export interface AllocationLifecycle { identity: AllocationIdentity; principal: string; state: AllocationLifecycleState; expiresAt: number; retainUntil: number; mutationFence: number; operationFence: number; cleanupCursor: string | null; deletedObjects: number; deletedBytes: number; deletedRecords: number; receipt: AllocationDeletionReceipt | null }
 
 const RECEIPT_SCHEMA = "wp-codebox/cloudflare-allocation-deletion-receipt/v1" as const
 const schemaReady = new WeakMap<object, Promise<void>>()
@@ -26,7 +26,7 @@ export class CloudflareAllocationLifecycle {
   }
   async get(identity: AllocationIdentity): Promise<AllocationLifecycle | null> {
     await this.initialize()
-    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>()
+    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, deleted_records, receipt_json FROM wp_codebox_site_lifecycles WHERE site_id = ? AND generation = ?").bind(identity.siteId, identity.generation).first<LifecycleRow>()
     return row ? hydrate(row) : null
   }
   async renew(identity: AllocationIdentity, principal: string, now = Date.now()): Promise<AllocationLifecycle> {
@@ -45,14 +45,20 @@ export class CloudflareAllocationLifecycle {
     const lifecycle = await this.get(identity)
     if (!lifecycle || lifecycle.state !== "active" || lifecycle.expiresAt <= now || lifecycle.mutationFence !== fence) throw new AllocationLifecycleConflict("Allocation mutation fence changed.")
   }
+  async assertActive(identity: AllocationIdentity, now = Date.now()): Promise<void> {
+    const lifecycle = await this.get(identity)
+    if (lifecycle && (lifecycle.state !== "active" || lifecycle.expiresAt <= now)) throw new AllocationLifecycleConflict("Allocation is no longer active.")
+  }
   async beginDeletion(identity: AllocationIdentity, principal: string | null, now = Date.now()): Promise<number> {
     await this.initialize()
     const owner = principal === null ? "" : " AND principal = ?"
     const values: unknown[] = [now + this.policy.retainMs, now, identity.siteId, identity.generation]
     if (principal !== null) values.push(principal)
     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()
     if (result.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation is not deletable by this owner.")
-    return (await this.get(identity))!.operationFence
+    const lifecycle = (await this.get(identity))!
+    await this.cancelWork(identity, now)
+    return lifecycle.operationFence
   }
   async expire(now = Date.now(), limit = 8): Promise<AllocationIdentity[]> {
     await this.initialize()
@@ -65,7 +71,7 @@ export class CloudflareAllocationLifecycle {
   }
   async pendingDeletions(limit = 8, now = Date.now()): Promise<Array<AllocationLifecycle>> {
     await this.initialize()
-    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>()
+    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, deleted_records, receipt_json FROM wp_codebox_site_lifecycles WHERE state = 'deleting' AND retain_until <= ? ORDER BY updated_at LIMIT ?").bind(now, limit).all<LifecycleRow>()
     return rows.results.map(hydrate)
   }
   async reclaim(bucket: Pick<R2Bucket, "list" | "delete">, identity: AllocationIdentity, operationFence: number, limit = 100, now = Date.now()): Promise<AllocationLifecycle> {
@@ -77,16 +83,42 @@ export class CloudflareAllocationLifecycle {
     if (page.objects.length) await bucket.delete(page.objects.map((object) => object.key))
     const cursor = page.truncated ? page.cursor : null
     const deletedBytes = page.objects.reduce((total, object) => total + (object.size ?? 0), 0)
-    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()
+    const cleaned = cursor ? { deleted: 0, unresolved: 0 } : await this.cleanupWork(identity.siteId, limit)
+    const updated = await this.database.prepare("UPDATE wp_codebox_site_lifecycles SET cleanup_cursor = ?, deleted_objects = deleted_objects + ?, deleted_bytes = deleted_bytes + ?, deleted_records = deleted_records + ?, updated_at = ? WHERE site_id = ? AND generation = ? AND state = 'deleting' AND operation_fence = ?").bind(cursor, page.objects.length, deletedBytes, cleaned.deleted, now, identity.siteId, identity.generation, operationFence).run()
     if (updated.meta.changes !== 1) throw new AllocationLifecycleConflict("Allocation deletion fence changed.")
-    if (cursor) return (await this.get(identity))!
-    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" }
+    if (cursor || cleaned.unresolved) return (await this.get(identity))!
+    const receipt: AllocationDeletionReceipt = { schema: RECEIPT_SCHEMA, siteId: identity.siteId, generation: identity.generation, deletedObjects: lifecycle.deletedObjects + page.objects.length, deletedBytes: lifecycle.deletedBytes + deletedBytes, deletedRecords: lifecycle.deletedRecords + cleaned.deleted, retainedObjects: 0, retainedBytes: 0, retainedRecords: 0, unresolvedObjects: 0, unresolvedRecords: 0, startedAt: new Date(lifecycle.expiresAt).toISOString(), completedAt: new Date(now).toISOString(), terminalState: "tombstoned" }
     await this.database.batch([
       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),
       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),
     ])
     return (await this.get(identity))!
   }
+  private async cancelWork(identity: AllocationIdentity, now: number): Promise<void> {
+    if (await tableExists(this.database, "wp_codebox_operations") && await tableExists(this.database, "wp_codebox_operation_attempts")) {
+      await this.database.batch([
+        this.database.prepare("UPDATE wp_codebox_operation_attempts SET completed_at = ?, state = 'failed', stage = 'allocation-deleted', error_code = 'allocation_deleted', error_message = 'Allocation deletion fenced this attempt.' WHERE site_id = ? AND completed_at IS NULL").bind(new Date(now).toISOString(), identity.siteId),
+        this.database.prepare("UPDATE wp_codebox_operations SET state = 'failed', stage = 'allocation-deleted', progress = 100, claim_token = NULL, claim_expires_at = NULL, retry_at = NULL, error_code = 'allocation_deleted', error_message = 'Allocation deletion fenced this operation.', completed_at = ?, updated_at = ? WHERE site_id = ? AND state IN ('queued','running','retryable','publication-pending')").bind(new Date(now).toISOString(), now, identity.siteId),
+      ])
+    }
+    if (await tableExists(this.database, "wp_codebox_api_admin_claims")) await this.database.prepare("UPDATE wp_codebox_api_admin_claims SET state = 'expired', updated_at = ? WHERE site_id = ? AND state = 'pending'").bind(now, identity.siteId).run()
+  }
+  private async cleanupWork(siteId: string, limit: number): Promise<{ deleted: number; unresolved: number }> {
+    let deleted = 0
+    if (await tableExists(this.database, "wp_codebox_operation_attempts")) {
+      const result = await this.database.prepare("DELETE FROM wp_codebox_operation_attempts WHERE rowid IN (SELECT rowid FROM wp_codebox_operation_attempts WHERE site_id = ? LIMIT ?)").bind(siteId, limit).run(); deleted += result.meta.changes
+    }
+    if (await tableExists(this.database, "wp_codebox_operations")) {
+      const result = await this.database.prepare("DELETE FROM wp_codebox_operations WHERE rowid IN (SELECT rowid FROM wp_codebox_operations WHERE site_id = ? LIMIT ?)").bind(siteId, limit).run(); deleted += result.meta.changes
+    }
+    if (await tableExists(this.database, "wp_codebox_api_admin_claims")) {
+      const result = await this.database.prepare("DELETE FROM wp_codebox_api_admin_claims WHERE site_id = ?").bind(siteId).run(); deleted += result.meta.changes
+    }
+    const tables = ["wp_codebox_operation_attempts", "wp_codebox_operations", "wp_codebox_api_admin_claims"]
+    let unresolved = 0
+    for (const table of tables) if (await tableExists(this.database, table)) unresolved += (await this.database.prepare(`SELECT COUNT(*) AS count FROM ${table} WHERE site_id = ?`).bind(siteId).first<{ count: number }>())?.count ?? 0
+    return { deleted, unresolved }
+  }
 }
 
 export class AllocationLifecycleConflict extends Error {}
@@ -95,14 +127,16 @@ export function allocationIdentity(siteId: string): AllocationIdentity {
   return { siteId, generation: match ? Number(match[1]) : 1 }
 }
 
-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 }
-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 } }
+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; deleted_records: number; receipt_json: string | null }
+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, deletedRecords: row.deleted_records, receipt: row.receipt_json ? JSON.parse(row.receipt_json) as AllocationDeletionReceipt : null } }
+async function tableExists(database: D1Database, name: string): Promise<boolean> { return !!await database.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?").bind(name).first() }
 async function ensureSchema(database: D1Database): Promise<void> {
   let pending = schemaReady.get(database as object)
   if (!pending) {
     pending = (async () => {
-      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()
+      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, deleted_records 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()
       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 }
+      try { await database.prepare("ALTER TABLE wp_codebox_site_lifecycles ADD COLUMN deleted_records INTEGER NOT NULL DEFAULT 0").run() } catch (error) { if (!(error instanceof Error) || !/duplicate column/i.test(error.message)) throw error }
       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()
       await database.prepare("CREATE INDEX IF NOT EXISTS wp_codebox_site_lifecycle_expiry ON wp_codebox_site_lifecycles(state, expires_at)").run()
     })()