Workflow Orchestration - Scheduled DAGs vs Durable Execution (Airflow, Temporal & Friends)

// The kernel every orchestrator shares, whichever family it belongs to:
//   durable state store + scheduler loop + task queue + workers + retry policy
//   + a heartbeat/lease so a silently dead worker is detected.
// Airflow calls the dead case a task whose heartbeat stopped (zombie/orphan task);
// Temporal calls it a Heartbeat or Start-To-Close timeout. Same idea.
// Simulated clock in seconds, so output is deterministic.

type State = "scheduled" | "running" | "retry_wait" | "success" | "failed";
interface Task { id: string; state: State; attempt: number; lastBeat: number; notBefore: number; worker?: string }
interface Policy { maxAttempts: number; initial: number; coeff: number; maxInterval: number; heartbeatTimeout: number }

const policy: Policy = { maxAttempts: 4, initial: 1, coeff: 2, maxInterval: 10, heartbeatTimeout: 5 };
const backoff = (attempt: number) => Math.min(policy.initial * policy.coeff ** (attempt - 1), policy.maxInterval);

const tasks: Task[] = [
  { id: "resize-images", state: "scheduled", attempt: 0, lastBeat: 0, notBefore: 0 },
  { id: "charge-card", state: "scheduled", attempt: 0, lastBeat: 0, notBefore: 0 },
];
// Script of what each worker does on a given attempt (the "real world").
const behaviour: Record<string, (attempt: number) => "ok" | "error" | "dies"> = {
  "resize-images": (a) => (a === 1 ? "dies" : "ok"),   // worker host vanishes on attempt 1
  "charge-card": (a) => (a < 3 ? "error" : "ok"),      // flaky dependency, succeeds on attempt 3
};
const workDuration = 3; // seconds each attempt takes
const inflight = new Map<string, { finishAt: number; outcome: "ok" | "error" | "dies" }>();
const log = (t: number, msg: string) => console.log(`t=${String(t).padStart(2)}s ${msg}`);

for (let now = 0; now <= 30; now++) {
  // 1. Scheduler loop: dispatch runnable tasks to the queue -> a worker picks it up.
  for (const task of tasks) {
    if ((task.state === "scheduled" || task.state === "retry_wait") && now >= task.notBefore) {
      task.attempt++; task.state = "running"; task.lastBeat = now; task.worker = `${task.id.split("-")[0]}-w${task.attempt}`;
      const outcome = behaviour[task.id](task.attempt);
      inflight.set(task.id, { finishAt: now + workDuration, outcome });
      log(now, `${task.id}: attempt ${task.attempt} leased to ${task.worker}`);
    }
  }
  // 2. Workers heartbeat while alive, and report results when done.
  for (const task of tasks.filter((t) => t.state === "running")) {
    const job = inflight.get(task.id)!;
    if (job.outcome !== "dies") task.lastBeat = now;
    if (now === job.finishAt && job.outcome !== "dies") {
      inflight.delete(task.id);
      if (job.outcome === "ok") { task.state = "success"; log(now, `${task.id}: success`); }
      else fail(task, now, "activity error");
    }
  }
  // 3. Lease check: no heartbeat within the timeout means the worker is presumed dead.
  for (const task of tasks.filter((t) => t.state === "running")) {
    if (now - task.lastBeat > policy.heartbeatTimeout) {
      inflight.delete(task.id);
      fail(task, now, `no heartbeat for ${now - task.lastBeat}s (worker ${task.worker} presumed dead)`);
    }
  }
}

function fail(task: Task, now: number, why: string) {
  if (task.attempt >= policy.maxAttempts) { task.state = "failed"; log(now, `${task.id}: FAILED for good (${why})`); return; }
  const wait = backoff(task.attempt);
  task.state = "retry_wait"; task.notBefore = now + wait;
  log(now, `${task.id}: ${why} -> retry in ${wait}s`);
}

console.log("final:", tasks.map((t) => `${t.id}=${t.state} after ${t.attempt} attempts`).join(", "));
console.log("note: the dead worker may still finish resize-images later -> tasks must be idempotent");

t= 0s resize-images: attempt 1 leased to resize-w1
t= 0s charge-card: attempt 1 leased to charge-w1
t= 3s charge-card: activity error -> retry in 1s
t= 4s charge-card: attempt 2 leased to charge-w2
t= 6s resize-images: no heartbeat for 6s (worker resize-w1 presumed dead) -> retry in 1s
t= 7s resize-images: attempt 2 leased to resize-w2
t= 7s charge-card: activity error -> retry in 2s
t= 9s charge-card: attempt 3 leased to charge-w3
t=10s resize-images: success
t=12s charge-card: success
final: resize-images=success after 2 attempts, charge-card=success after 3 attempts
note: the dead worker may still finish resize-images later -> tasks must be idempotent

"""The same 3-step job run by the two orchestration families.

Family A, scheduled batch DAG (Airflow / Dagster / Prefect / Argo style):
  - a run exists per data interval; state is tracked per TASK
  - a crash inside a task means the WHOLE task is retried from its start
Family B, durable execution (Temporal / Step Functions / Restate / Inngest style):
  - a run exists per business event; state is a journal of completed STEPS
  - after a crash the function is re-run, but completed steps return their
    recorded result instead of executing again

Everything is simulated in-process so the output is deterministic.
"""

calls = {}  # side-effect counter: how many times each real action executed

def side_effect(name):
    calls[name] = calls.get(name, 0) + 1

class Crash(Exception):
    pass

# ---------------- Family A: task-level state ----------------
def run_dag(interval, crash_in_task=None, crash_after_step=None, state=None):
    """Tasks: extract -> transform -> load. Each task is 2 internal steps."""
    state = state if state is not None else {}
    tasks = {
        "extract":   ["download_part_1", "download_part_2"],
        "transform": ["clean", "aggregate"],
        "load":      ["write_partition", "publish_metrics"],
    }
    for task, steps in tasks.items():          # topological order
        if state.get(task) == "success":
            print(f"  [{interval}] {task:<9} skipped (already success)")
            continue
        try:
            for step in steps:                     # the orchestrator cannot see these
                side_effect(step)
                if task == crash_in_task and step == crash_after_step:
                    raise Crash(step)
            state[task] = "success"
            print(f"  [{interval}] {task:<9} success")
        except Crash as c:
            state[task] = "up_for_retry"
            print(f"  [{interval}] {task:<9} CRASH after {c} -> up_for_retry (whole task reruns)")
            return state
    return state

# ---------------- Family B: step-level journal ----------------
def durable_order_workflow(ctx, order_id):
    """Ordinary-looking code; every ctx.step is journaled by the engine."""
    ctx.step("reserve_stock", order_id)
    ctx.step("charge_card", order_id)
    ctx.step("create_shipment", order_id)
    ctx.step("send_email", order_id)
    return "done"

class DurableCtx:
    def __init__(self, journal, crash_after=None):
        self.journal, self.crash_after, self.i = journal, crash_after, 0
    def step(self, name, arg):
        if self.i < len(self.journal):             # replay: return recorded result
            rec = self.journal[self.i]; self.i += 1
            assert rec[0] == name, "non-deterministic workflow code"
            return rec[1]
        side_effect(name)                          # first time: really execute
        result = f"{name}:{arg}:ok"
        self.journal.append((name, result)); self.i += 1
        if name == self.crash_after:
            raise Crash(name)
        return result

def run_durable(order_id, crash_after=None, journal=None):
    journal = journal if journal is not None else []
    try:
        out = durable_order_workflow(DurableCtx(journal, crash_after), order_id)
        print(f"  [{order_id}] completed, journal={[j[0] for j in journal]}")
        return journal, out
    except Crash as c:
        print(f"  [{order_id}] CRASH after step {c}; journal has {len(journal)} steps")
        return journal, None

print("Family A: scheduled DAG, task-level state")
st = run_dag("2026-10-06", crash_in_task="transform", crash_after_step="aggregate")
st = run_dag("2026-10-06", state=st)            # retry of the same data interval
print("  side effects:", {k: calls[k] for k in ["download_part_1", "clean", "aggregate", "write_partition"]})

calls.clear()
print("\nFamily B: durable execution, step-level journal")
j, _ = run_durable("order-42", crash_after="charge_card")
j, _ = run_durable("order-42", journal=j)       # worker restarts, replays journal
print("  side effects:", calls)

print("\nTakeaway: A retried 'transform' from its first step (clean ran twice);")
print("B resumed after charge_card without charging twice.")

Family A: scheduled DAG, task-level state
  [2026-10-06] extract   success
  [2026-10-06] transform CRASH after aggregate -> up_for_retry (whole task reruns)
  [2026-10-06] extract   skipped (already success)
  [2026-10-06] transform success
  [2026-10-06] load      success
  side effects: {'download_part_1': 1, 'clean': 2, 'aggregate': 2, 'write_partition': 1}

Family B: durable execution, step-level journal
  [order-42] CRASH after step charge_card; journal has 2 steps
  [order-42] completed, journal=['reserve_stock', 'charge_card', 'create_shipment', 'send_email']
  side effects: {'reserve_stock': 1, 'charge_card': 1, 'create_shipment': 1, 'send_email': 1}

Takeaway: A retried 'transform' from its first step (clean ran twice);
B resumed after charge_card without charging twice.