Durable Execution & the Temporal Model - Workflows vs Activities, Event History, Replay & Determinism

"""Event-sourced replay, the core of durable execution (Temporal, Cadence,
Azure Durable Functions, Restate and Inngest all use a variant of it).

- Workflow code is a generator that yields COMMANDS (run activity, start timer).
- The engine appends EVENTS to an append-only history.
- A worker that crashes loses its memory; the next worker rebuilds state by
  REPLAYING the code against the history: commands already in history get their
  recorded results back instead of running again.
- Replay only works if the code issues the same commands in the same order.
  Reading the wall clock inside workflow code breaks that. ctx.now() fixes it by
  returning the time recorded in history (like workflow.now() in Temporal SDKs).
"""

class NonDeterminismError(Exception):
    pass

class WorkerCrash(Exception):
    pass

class Ctx:
    def __init__(self, engine): self.e = engine
    def activity(self, name, arg): return ("activity", name, arg)
    def timer(self, seconds): return ("timer", f"{seconds}s", None)
    def now(self): return self.e.task_time          # deterministic: comes from history

ACTIVITIES = {
    "credit_score": lambda a: 640,
    "auto_approve": lambda a: f"approved:{a}",
    "manual_review": lambda a: f"queued-for-human:{a}",
}

class Engine:
    def __init__(self, history, wall_clock, crash_after_scheduling=None):
        self.h, self.wall, self.crash = history, wall_clock, crash_after_scheduling
        self.task_time = None

    def run(self, wf, arg):
        if not self.h:
            self.h.append({"type": "WorkflowExecutionStarted", "input": arg})
        cmd_events = [e for e in self.h if e["type"] in ("ActivityTaskScheduled", "TimerStarted")]
        done = {e["ref"]: e for e in self.h if e["type"] in ("ActivityTaskCompleted", "TimerFired")}
        task_times = [e["time"] for e in self.h if e["type"] == "WorkflowTaskStarted"]
        gen, send, i = wf(Ctx(self), arg), None, 0
        while True:
            # each decision point is one workflow task; its time is recorded once, then replayed
            if i < len(task_times):
                self.task_time = task_times[i]
            else:
                self.task_time = self.wall
                self.h.append({"type": "WorkflowTaskStarted", "time": self.wall})
            try:
                kind, name, carg = gen.send(send)
            except StopIteration as stop:
                self.h.append({"type": "WorkflowExecutionCompleted", "result": stop.value})
                return stop.value
            if i < len(cmd_events):                       # REPLAY: must match history
                ev = cmd_events[i]
                if ev["name"] != name:
                    raise NonDeterminismError(f"history has {ev['type']}({ev['name']}) at step {i+1}, code now asks for {kind}({name})")
            else:                                         # NEW command: record it first
                ev = {"type": "ActivityTaskScheduled" if kind == "activity" else "TimerStarted", "name": name, "id": i}
                self.h.append(ev)
                if self.crash == name:
                    raise WorkerCrash(name)
            if i not in done:                             # not finished yet: execute now
                result = ACTIVITIES[name](carg) if kind == "activity" else "fired"
                comp = {"type": "ActivityTaskCompleted" if kind == "activity" else "TimerFired", "ref": i, "result": result}
                self.h.append(comp); done[i] = comp
            send, i = done[i]["result"], i + 1

def hour(t): return int(t.split(":")[0])

def loan_bad(ctx, applicant):
    score = yield ctx.activity("credit_score", applicant)
    # BUG: wall clock inside workflow code; differs between first run and replay
    now = WALL_CLOCK_FOR_DEMO  # stands in for datetime.datetime.now()
    if score > 700 or hour(now) < 17:
        decision = yield ctx.activity("auto_approve", applicant)
    else:
        decision = yield ctx.activity("manual_review", applicant)
    yield ctx.timer(86400)                              # durable timer: survives restarts
    return decision

def loan_fixed(ctx, applicant):
    score = yield ctx.activity("credit_score", applicant)
    if score > 700 or hour(ctx.now()) < 17:              # time recorded in history
        decision = yield ctx.activity("auto_approve", applicant)
    else:
        decision = yield ctx.activity("manual_review", applicant)
    yield ctx.timer(86400)
    return decision

def show(h): return [e["type"] + (f"({e['name']})" if "name" in e else "") for e in h]

for label, wf in [("BUGGY (reads wall clock)", loan_bad), ("FIXED (uses ctx.now())", loan_fixed)]:
    print(f"== {label}")
    history = []
    WALL_CLOCK_FOR_DEMO = "16:59"
    try:
        Engine(history, "16:59", crash_after_scheduling="auto_approve").run(wf, "ana")
    except WorkerCrash as c:
        print(f"  worker A (16:59) crashed right after scheduling {c}; history so far:")
        print("   ", show(history))
    WALL_CLOCK_FOR_DEMO = "17:01"                        # worker B picks up two minutes later
    try:
        out = Engine(history, "17:01").run(wf, "ana")
        print(f"  worker B (17:01) replayed and finished -> {out}")
        print("   ", show(history))
    except NonDeterminismError as e:
        print(f"  worker B (17:01) NonDeterminismError: {e}")
        print("    the workflow is stuck until the code is fixed (Temporal retries the workflow task)")

== BUGGY (reads wall clock)
  worker A (16:59) crashed right after scheduling auto_approve; history so far:
    ['WorkflowExecutionStarted', 'WorkflowTaskStarted', 'ActivityTaskScheduled(credit_score)', 'ActivityTaskCompleted', 'WorkflowTaskStarted', 'ActivityTaskScheduled(auto_approve)']
  worker B (17:01) NonDeterminismError: history has ActivityTaskScheduled(auto_approve) at step 2, code now asks for activity(manual_review)
    the workflow is stuck until the code is fixed (Temporal retries the workflow task)
== FIXED (uses ctx.now())
  worker A (16:59) crashed right after scheduling auto_approve; history so far:
    ['WorkflowExecutionStarted', 'WorkflowTaskStarted', 'ActivityTaskScheduled(credit_score)', 'ActivityTaskCompleted', 'WorkflowTaskStarted', 'ActivityTaskScheduled(auto_approve)']
  worker B (17:01) replayed and finished -> approved:ana
    ['WorkflowExecutionStarted', 'WorkflowTaskStarted', 'ActivityTaskScheduled(credit_score)', 'ActivityTaskCompleted', 'WorkflowTaskStarted', 'ActivityTaskScheduled(auto_approve)', 'ActivityTaskCompleted', 'WorkflowTaskStarted', 'TimerStarted(86400s)', 'TimerFired', 'WorkflowTaskStarted', 'WorkflowExecutionCompleted']

// Durable waits, messages and at-least-once activities in a replay engine.
// - A workflow waits for a Signal OR a durable Timer, whichever comes first.
// - A Query reads state by replaying history; it never appends events.
// - A Signal that arrives while no worker is alive is still recorded in history.
// - An activity whose worker dies after the side effect is retried after its
//   Start-To-Close timeout: activities are at-least-once, so the side effect
//   must be idempotent (here: an idempotency key the payment provider dedups on).

type Ev = { t: string; type: string; data?: string };
type Cmd = { kind: "race"; signal: string; timer: string } | { kind: "activity"; name: string; key: string };
interface Ctx { state: { status: string } }

function* expense(ctx: Ctx): Generator<Cmd, string, string> {
  ctx.state.status = "waiting-for-approval";
  const r = yield { kind: "race", signal: "approve", timer: "72h" };
  if (r === "timer") { ctx.state.status = "auto-rejected"; return "rejected"; }
  ctx.state.status = `paying (approved by ${r})`;
  const receipt = yield { kind: "activity", name: "pay", key: "expense-981" };
  ctx.state.status = "paid";
  return receipt;
}

// Replay the code against history; return state plus the first command that has no result yet.
function replay(h: Ev[]): { state: Ctx["state"]; pending?: Cmd; result?: string } {
  const ctx: Ctx = { state: { status: "new" } }; const g = expense(ctx); let send: string | undefined;
  for (;;) {
    const step = g.next(send as string);
    if (step.done) return { state: ctx.state, result: step.value };
    const c = step.value;
    if (c.kind === "race") {
      const sig = h.find((e) => e.type === "SignalReceived"); const fired = h.find((e) => e.type === "TimerFired");
      if (!sig && !fired) return { state: ctx.state, pending: c };
      send = sig ? sig.data! : "timer";
    } else {
      const done = h.find((e) => e.type === "ActivityTaskCompleted");
      if (!done) return { state: ctx.state, pending: c };
      send = done.data!;
    }
  }
}

function runScenario(useIdempotencyKey: boolean) {
  const h: Ev[] = [{ t: "0h", type: "WorkflowExecutionStarted" }];
  const ledger: string[] = []; const seenKeys = new Set<string>();
  const provider = (key: string) => {           // the external payment API
    if (useIdempotencyKey && seenKeys.has(key)) return "rcpt-1 (deduplicated)";
    seenKeys.add(key); ledger.push("charge $120"); return `rcpt-${ledger.length}`;
  };
  // worker 1 starts the workflow: the race command creates a durable timer
  let r = replay(h);
  if (r.pending?.kind === "race") h.push({ t: "0h", type: "TimerStarted", data: r.pending.timer });
  console.log(`  0h  query status -> "${replay(h).state.status}" (history length still ${h.length})`);
  console.log("  5h  worker 1 crashes; no worker is running");
  h.push({ t: "6h", type: "SignalReceived", data: "manager-lee" });
  console.log("  6h  signal 'approve' accepted by the server and recorded while no worker is alive");
  r = replay(h);                                  // worker 2 replays from history
  if (r.pending?.kind === "activity") {
    h.push({ t: "7h", type: "TimerCanceled" }, { t: "7h", type: "ActivityTaskScheduled", data: r.pending.name });
    provider(r.pending.key);                      // attempt 1 charges the card...
    console.log("  7h  attempt 1 charged the card, then its worker died before reporting");
    console.log("  7h+ Start-To-Close timeout expires -> retry policy schedules attempt 2");
    const receipt = provider(r.pending.key);      // attempt 2
    h.push({ t: "7h", type: "ActivityTaskStarted", data: "attempt=2" }, { t: "7h", type: "ActivityTaskCompleted", data: receipt });
  }
  r = replay(h); h.push({ t: "7h", type: "WorkflowExecutionCompleted", data: r.result });
  console.log(`  done: result=${r.result}, status="${r.state.status}", charges=${ledger.length}`);
  console.log(`  history: ${h.map((e) => e.type).join(", ")}`);
}

console.log("Without an idempotency key:"); runScenario(false);
console.log("With an idempotency key:"); runScenario(true);

Without an idempotency key:
  0h  query status -> "waiting-for-approval" (history length still 2)
  5h  worker 1 crashes; no worker is running
  6h  signal 'approve' accepted by the server and recorded while no worker is alive
  7h  attempt 1 charged the card, then its worker died before reporting
  7h+ Start-To-Close timeout expires -> retry policy schedules attempt 2
  done: result=rcpt-2, status="paid", charges=2
  history: WorkflowExecutionStarted, TimerStarted, SignalReceived, TimerCanceled, ActivityTaskScheduled, ActivityTaskStarted, ActivityTaskCompleted, WorkflowExecutionCompleted
With an idempotency key:
  0h  query status -> "waiting-for-approval" (history length still 2)
  5h  worker 1 crashes; no worker is running
  6h  signal 'approve' accepted by the server and recorded while no worker is alive
  7h  attempt 1 charged the card, then its worker died before reporting
  7h+ Start-To-Close timeout expires -> retry policy schedules attempt 2
  done: result=rcpt-1 (deduplicated), status="paid", charges=1
  history: WorkflowExecutionStarted, TimerStarted, SignalReceived, TimerCanceled, ActivityTaskScheduled, ActivityTaskStarted, ActivityTaskCompleted, WorkflowExecutionCompleted

import asyncio
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from activities import credit_score, auto_approve, manual_review, pay

@workflow.defn
class LoanWorkflow:
    def __init__(self) -> None:
        self.approved_by: str | None = None

    @workflow.run
    async def run(self, applicant: str) -> str:
        opts = dict(start_to_close_timeout=timedelta(seconds=30),
                    retry_policy=RetryPolicy(maximum_attempts=5, non_retryable_error_types=["CardDeclined"]))
        score = await workflow.execute_activity(credit_score, applicant, **opts)
        if score > 700 or workflow.now().hour < 17:          # replay-safe clock
            return await workflow.execute_activity(auto_approve, applicant, **opts)
        try:                                                   # durable wait: signal or 3 days
            await workflow.wait_condition(lambda: self.approved_by is not None, timeout=timedelta(days=3))
        except asyncio.TimeoutError:
            return "rejected"
        return await workflow.execute_activity(pay, f"loan-{workflow.info().workflow_id}", **opts)

    @workflow.signal
    def approve(self, manager: str) -> None:
        self.approved_by = manager

    @workflow.query
    def status(self) -> str:
        return "approved" if self.approved_by else "waiting"