refactor(platform-cloudflare): simplify workflow engine internals

- Fold loadExecutionName into loadExecution and share one detach helper for
  fire-and-forget promises
- Flatten run()'s discard branching and route the wake self-heal through
  resume()
- Deduplicate the per-workflow codec caches and drop the redundant conflict
  clause on the single-threaded deferred insert

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: claude

packages/platform/cloudflare/src/CloudflareDurableObjects.ts

@@ -47,7 +47,7 @@ import { armAlarm, earliestDeliverAt, ensureEntityStorage } from "./internal/ent
 import { decodeReplyFor, decodeRequest, encodeReplyFor } from "./internal/entityWire.ts"
 import type { WorkflowRunOptions, WorkflowStub } from "./internal/workflowRegistry.ts"
 import { makeWorkflowRuntime } from "./internal/workflowRuntime.ts"
-import { earliestClockWakeUp, ensureWorkflowStorage, loadExecutionName } from "./internal/workflowStorage.ts"
+import { earliestClockWakeUp, ensureWorkflowStorage, loadExecution } from "./internal/workflowStorage.ts"
 
 const notExposed = (className: string) => () => {
   throw new Error(
@@ -664,7 +664,7 @@ export class ClusterWorkflow extends DurableObject<unknown> {
     } else {
       // An alarm wake carries no `id.name`; recover it from the stored
       // execution so due clocks still fire after eviction.
-      const stored = loadExecutionName(ctx.storage.sql)
+      const stored = loadExecution(ctx.storage.sql)
       this.#name = stored === undefined ? undefined : encodeName(stored.workflowName, stored.executionId)
     }
     const wakeUp = earliestClockWakeUp(ctx.storage.sql)

packages/platform/cloudflare/src/internal/workflowRuntime.ts

@@ -84,6 +84,10 @@ export const makeWorkflowRuntime = (options: WorkflowRuntimeOptions): WorkflowRu
   let inflight: Inflight | undefined
   let resumeRequested = false
 
+  const detach = (promise: Promise<unknown>): void => {
+    options.waitUntil(promise.then(() => undefined, () => undefined))
+  }
+
   const isComplete = (result: string | undefined): boolean =>
     result !== undefined && (JSON.parse(result) as { readonly _tag?: unknown })._tag === "Complete"
 
@@ -139,7 +143,7 @@ export const makeWorkflowRuntime = (options: WorkflowRuntimeOptions): WorkflowRu
         if (parent !== undefined) options.waitUntil(resumeParent(parent))
       } else if (resumeRequested) {
         resumeRequested = false
-        options.waitUntil(startAttempt(row).then(() => undefined, () => undefined))
+        detach(startAttempt(row))
       } else {
         // This attempt observed every persisted deferred and still suspended,
         // so the pending resume (if any) has been serviced.
@@ -178,23 +182,20 @@ export const makeWorkflowRuntime = (options: WorkflowRuntimeOptions): WorkflowRu
     let row = WorkflowStorage.loadExecution(sql)
     if (row === undefined) {
       WorkflowStorage.createExecution(sql, workflowName, executionId, payload, opts.parent)
-      row = { workflowName, payload, parent: opts.parent, result: undefined, resumePending: false }
+      row = { workflowName, executionId, payload, parent: opts.parent, result: undefined, resumePending: false }
     } else if (opts.parent !== undefined && row.parent === undefined) {
       // An execution started standalone can gain a parent later; keep the
       // first parent so its completion still wakes that parent.
       WorkflowStorage.setParent(sql, opts.parent)
       row = { ...row, parent: opts.parent }
     }
-    if (inflight !== undefined) {
-      if (!opts.discard) return inflight.promise
-    } else if (row.result === undefined) {
-      const attempt = startAttempt(row)
-      if (!opts.discard) return attempt
-      options.waitUntil(attempt.then(() => undefined, () => undefined))
-    } else if (!opts.discard) {
-      return Promise.resolve(row.result)
+    if (opts.discard) {
+      if (inflight === undefined && row.result === undefined) detach(startAttempt(row))
+      return Promise.resolve("")
     }
-    return Promise.resolve("")
+    if (inflight !== undefined) return inflight.promise
+    if (row.result !== undefined) return Promise.resolve(row.result)
+    return startAttempt(row)
   }
 
   const resume = (): Promise<void> => {
@@ -204,7 +205,7 @@ export const makeWorkflowRuntime = (options: WorkflowRuntimeOptions): WorkflowRu
       resumeRequested = true
       return Promise.resolve()
     }
-    options.waitUntil(startAttempt(row).then(() => undefined, () => undefined))
+    detach(startAttempt(row))
     return Promise.resolve()
   }
 
@@ -266,14 +267,12 @@ export const makeWorkflowRuntime = (options: WorkflowRuntimeOptions): WorkflowRu
         return (pending ? resume() : Promise.resolve()).then(() => {
           // While a resume is pending a guard alarm stays armed, so a replay
           // lost with this isolate is retried instead of sleeping forever.
-          const earliest = WorkflowStorage.earliestClockWakeUp(sql)
-          const guard = pending ? options.now() + resumeGuardMillis : undefined
-          const target = earliest === undefined
-            ? guard
-            : guard === undefined
-            ? earliest
-            : Math.min(earliest, guard)
-          return target === undefined ? undefined : Effect.runPromise(armAlarm(options.alarm, target))
+          const targets = [
+            WorkflowStorage.earliestClockWakeUp(sql),
+            pending ? options.now() + resumeGuardMillis : undefined
+          ].filter((target) => target !== undefined)
+          if (targets.length === 0) return undefined
+          return Effect.runPromise(armAlarm(options.alarm, Math.min(...targets)))
         })
       })
     }).then(() => undefined)
@@ -295,9 +294,8 @@ export const makeWorkflowRuntime = (options: WorkflowRuntimeOptions): WorkflowRu
 
   // Self-heal on wake: a resume recorded by deferredDone but lost with the
   // previous isolate replays now instead of waiting for external contact.
-  const stored = WorkflowStorage.loadExecution(sql)
-  if (stored !== undefined && stored.resumePending && !isComplete(stored.result)) {
-    options.waitUntil(startAttempt(stored).then(() => undefined, () => undefined))
+  if (WorkflowStorage.loadExecution(sql)?.resumePending === true) {
+    void resume()
   }
 
   return runtime

packages/platform/cloudflare/src/internal/workflowStorage.ts

@@ -41,9 +41,15 @@ export const ensureWorkflowStorage = (sql: SqlStorage): void => {
   }
 }
 
-/** @internal */
+/**
+ * The stored `(workflowName, executionId)` also serve to recover the object
+ * name on an alarm wake, where `ctx.id.name` is undefined.
+ *
+ * @internal
+ */
 export interface ExecutionRow {
   readonly workflowName: string
+  readonly executionId: string
   readonly payload: string
   readonly parent: { readonly workflowName: string; readonly executionId: string } | undefined
   readonly result: string | undefined
@@ -53,12 +59,13 @@ export interface ExecutionRow {
 /** @internal */
 export const loadExecution = (sql: SqlStorage): ExecutionRow | undefined => {
   const row = sql.exec(
-    `SELECT workflow_name, payload, parent_name, parent_execution_id, result, resume_pending
+    `SELECT workflow_name, execution_id, payload, parent_name, parent_execution_id, result, resume_pending
      FROM workflow_execution WHERE id = 0`
   ).toArray()[0]
   if (row === undefined) return undefined
   return {
     workflowName: String(row.workflow_name),
+    executionId: String(row.execution_id),
     payload: String(row.payload),
     parent: typeof row.parent_name === "string" && typeof row.parent_execution_id === "string"
       ? { workflowName: row.parent_name, executionId: row.parent_execution_id }
@@ -68,20 +75,6 @@ export const loadExecution = (sql: SqlStorage): ExecutionRow | undefined => {
   }
 }
 
-/**
- * The stored `(workflowName, executionId)` of this object's execution, used to
- * recover the object name on an alarm wake where `ctx.id.name` is undefined.
- *
- * @internal
- */
-export const loadExecutionName = (
-  sql: SqlStorage
-): { readonly workflowName: string; readonly executionId: string } | undefined => {
-  const row = sql.exec("SELECT workflow_name, execution_id FROM workflow_execution WHERE id = 0").toArray()[0]
-  if (row === undefined) return undefined
-  return { workflowName: String(row.workflow_name), executionId: String(row.execution_id) }
-}
-
 /** @internal */
 export const createExecution = (
   sql: SqlStorage,
@@ -141,10 +134,15 @@ export const loadDeferred = (sql: SqlStorage, name: string): string | undefined
   return row === undefined ? undefined : String(row.exit)
 }
 
-/** @internal */
+/**
+ * First write wins; safe without a conflict clause because a Durable Object's
+ * SQLite access is single-threaded.
+ *
+ * @internal
+ */
 export const saveDeferred = (sql: SqlStorage, name: string, exit: string): boolean => {
   if (loadDeferred(sql, name) !== undefined) return false
-  sql.exec("INSERT OR IGNORE INTO workflow_deferreds (name, exit) VALUES (?, ?)", name, exit)
+  sql.exec("INSERT INTO workflow_deferreds (name, exit) VALUES (?, ?)", name, exit)
   return true
 }
 

packages/platform/cloudflare/src/internal/workflowWire.ts

@@ -30,30 +30,26 @@ export const decodeExit = (
     context
   )
 
-const resultCodecs = new WeakMap<Workflow.Any, Schema.Top>()
-
-const resultCodec = (workflow: Workflow.Any): Schema.Top => {
-  let codec = resultCodecs.get(workflow)
-  if (codec === undefined) {
-    codec = Schema.toCodecJson(Workflow.Result({
-      success: workflow.successSchema as any,
-      error: workflow.errorSchema as any
-    }))
-    resultCodecs.set(workflow, codec)
+const cachedCodec = (compute: (workflow: Workflow.Any) => Schema.Top): (workflow: Workflow.Any) => Schema.Top => {
+  const cache = new WeakMap<Workflow.Any, Schema.Top>()
+  return (workflow) => {
+    let codec = cache.get(workflow)
+    if (codec === undefined) {
+      codec = compute(workflow)
+      cache.set(workflow, codec)
+    }
+    return codec
   }
-  return codec
 }
 
-const payloadCodecs = new WeakMap<Workflow.Any, Schema.Top>()
+const resultCodec = cachedCodec((workflow) =>
+  Schema.toCodecJson(Workflow.Result({
+    success: workflow.successSchema as any,
+    error: workflow.errorSchema as any
+  }))
+)
 
-const payloadCodec = (workflow: Workflow.Any): Schema.Top => {
-  let codec = payloadCodecs.get(workflow)
-  if (codec === undefined) {
-    codec = Schema.toCodecJson(workflow.payloadSchema)
-    payloadCodecs.set(workflow, codec)
-  }
-  return codec
-}
+const payloadCodec = cachedCodec((workflow) => Schema.toCodecJson(workflow.payloadSchema))
 
 /** @internal */
 export const encodeResult = (

packages/platform/cloudflare/test/CloudflareWorkflowEngine.test.ts

@@ -3,7 +3,7 @@ import * as CloudflareWorkflowEngine from "@effect/platform-cloudflare/Cloudflar
 import { encodeName } from "@effect/platform-cloudflare/internal/clusterName"
 import type { EntityAlarm } from "@effect/platform-cloudflare/internal/entityStorage"
 import { makeWorkflowRuntime, type WorkflowRuntime } from "@effect/platform-cloudflare/internal/workflowRuntime"
-import { loadExecutionName } from "@effect/platform-cloudflare/internal/workflowStorage"
+import { loadExecution } from "@effect/platform-cloudflare/internal/workflowStorage"
 import { assert, describe, it } from "@effect/vitest"
 import { Effect, Exit, Layer, Option, Schema } from "effect"
 import { Activity, DurableClock, DurableDeferred, Workflow, WorkflowEngine } from "effect/unstable/workflow"
@@ -61,7 +61,7 @@ class FakeSql {
       const exit = this.activities.get(String(bindings[0]))
       return exit === undefined ? [] : [{ exit }]
     }
-    if (query.includes("INSERT OR IGNORE INTO workflow_deferreds")) {
+    if (query.includes("INSERT INTO workflow_deferreds")) {
       if (!this.deferreds.has(String(bindings[0]))) {
         this.deferreds.set(String(bindings[0]), String(bindings[1]))
       }
@@ -297,7 +297,9 @@ describe("CloudflareWorkflowEngine", () => {
       ])
       assert.strictEqual(store.alarm.current, 30_000)
       // An alarm wake has no `id.name`; the stored execution recovers it.
-      assert.deepStrictEqual(loadExecutionName(store.sql.sql), { workflowName: "Sleeper", executionId })
+      const stored = loadExecution(store.sql.sql)
+      assert.strictEqual(stored?.workflowName, "Sleeper")
+      assert.strictEqual(stored?.executionId, executionId)
 
       namespace.now = 30_000
       yield* Effect.promise(() => namespace.fireDueAlarms())