Choosing & Operating Workflow Orchestrators - Airflow vs Dagster vs Prefect vs Temporal vs Step Functions vs Argo, Testing & Migration

"""DAG integrity tests: cheap CI checks that catch most Airflow-style outages
before a file ever reaches the scheduler. Real setups load every file with
DagBag (or `airflow dags list-import-errors`) inside pytest; here we check
three toy Dag files statically with Python's ast module so it runs anywhere.

Checks:
  1. the file parses (no import errors)
  2. no expensive top-level code (network, DB, Variable reads at parse time)
  3. no dynamic start_date such as datetime.now()
  4. every task sets retries, and the Dag sets catchup explicitly
  5. the dependency graph is acyclic
"""
import ast

FILES = {
"orders_daily.py": '''
from airflow.sdk import dag, task
import pendulum
@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def orders_daily():
    @task(retries=3)
    def extract(): ...
    @task(retries=3)
    def load(rows): ...
    load(extract())
''',
"crm_sync.py": '''
from airflow.sdk import dag, task, Variable
import datetime, requests
TOKEN = Variable.get("crm_token")                 # runs on EVERY parse
ACCOUNTS = requests.get("https://crm.example/accounts").json()
@dag(schedule="@hourly", start_date=datetime.datetime.now())
def crm_sync():
    @task
    def push(): ...
    push()
''',
"broken.py": '''
from airflow.sdk import dag, task
@dag(schedule=None
def broken(): ...
''',
}
EXPENSIVE = {("requests", "get"), ("Variable", "get"), ("psycopg2", "connect"), ("boto3", "client")}
DEPS = {"orders_daily.py": {"extract": [], "load": ["extract"]},
        "crm_sync.py": {"push": []}}

def dotted(call):
    f = call.func
    return (f.value.id, f.attr) if isinstance(f, ast.Attribute) and isinstance(f.value, ast.Name) else None

def acyclic(deps):
    seen, stack = set(), set()
    def dfs(n):
        if n in stack: return False
        if n in seen: return True
        stack.add(n); ok = all(dfs(u) for u in deps[n]); stack.discard(n); seen.add(n); return ok
    return all(dfs(n) for n in deps)

def check(name, src):
    problems = []
    try:
        tree = ast.parse(src)
    except SyntaxError as e:
        return [f"import error: {e.msg} (line {e.lineno})"]
    for node in tree.body:                                   # module-level statements only
        if isinstance(node, ast.FunctionDef): continue         # function bodies run at task time
        for call in [n for n in ast.walk(node) if isinstance(n, ast.Call)]:
            if dotted(call) in EXPENSIVE:
                problems.append(f"top-level call {'.'.join(dotted(call))}() runs on every parse")
    for fn in [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)]:
        for dec in fn.decorator_list:
            if not isinstance(dec, ast.Call):
                if isinstance(dec, ast.Name) and dec.id == "task": problems.append(f"task {fn.name}: retries not set")
                continue
            kw = {k.arg: k.value for k in dec.keywords}
            if getattr(dec.func, "id", "") == "dag":
                sd = kw.get("start_date")
                if sd is not None and "now" in ast.unparse(sd): problems.append("dynamic start_date (datetime.now())")
                if "catchup" not in kw: problems.append("catchup not set explicitly")
            if getattr(dec.func, "id", "") == "task" and "retries" not in kw:
                problems.append(f"task {fn.name}: retries not set")
    if name in DEPS and not acyclic(DEPS[name]): problems.append("cycle in task graph")
    return problems

failed = 0
for name, src in FILES.items():
    probs = check(name, src)
    failed += bool(probs)
    print(f"{'PASS' if not probs else 'FAIL'}  {name}")
    for p in probs: print(f"      - {p}")
print(f"\n{len(FILES) - failed} passed, {failed} failed -> {'block the deploy' if failed else 'ship it'}")

PASS  orders_daily.py
FAIL  crm_sync.py
      - top-level call Variable.get() runs on every parse
      - top-level call requests.get() runs on every parse
      - dynamic start_date (datetime.now())
      - catchup not set explicitly
      - task push: retries not set
FAIL  broken.py
      - import error: '(' was never closed (line 3)

1 passed, 2 failed -> block the deploy

// Replay tests: before deploying new workflow code, replay a sample of real
// event histories (exported from production) against it. Any history that no
// longer replays would get stuck in production with a non-determinism error.
// Temporal SDKs ship a Replayer for exactly this; this is a simplified model.

type RecordedHistory = { id: string; startedOn: string; cmds: string[]; markers: string[] };
type Candidate = { name: string; run: (h: RecordedHistory) => string[] };

// Recorded histories: three in-flight orders from v1 code, two from v2.
const histories: RecordedHistory[] = [
  { id: "order-101", startedOn: "v1", cmds: ["charge"], markers: [] },
  { id: "order-102", startedOn: "v1", cmds: ["charge", "ship"], markers: [] },
  { id: "order-103", startedOn: "v1", cmds: [], markers: [] },
  { id: "order-201", startedOn: "v2", cmds: ["fraud_check", "charge"], markers: ["add-fraud-check"] },
  { id: "order-202", startedOn: "v2", cmds: ["fraud_check"], markers: ["add-fraud-check"] },
];

// Each candidate returns the commands its code would issue while replaying a history.
const patchedPath = (h: RecordedHistory) => (h.markers.includes("add-fraud-check") || h.cmds.length === 0 ? ["fraud_check", "charge", "ship"] : ["charge", "ship"]);
const candidates: Candidate[] = [
  { name: "v3-unsafe (adds fraud_check, no patch)", run: () => ["fraud_check", "charge", "ship"] },
  { name: "v3-patched (keeps add-fraud-check patch)", run: patchedPath },
  { name: "v3-patch-removed-too-early", run: () => ["fraud_check", "charge", "ship"] },
  { name: "v3-reorders ship before charge", run: (h) => (h.markers.length || !h.cmds.length ? ["fraud_check", "ship", "charge"] : ["ship", "charge"]) },
];

function replays(c: Candidate, h: RecordedHistory): boolean {
  const issued = c.run(h);
  return h.cmds.every((cmd, i) => issued[i] === cmd);   // recorded prefix must match exactly
}

for (const c of candidates) {
  const failures = histories.filter((h) => !replays(c, h)).map((h) => h.id);
  const verdict = failures.length ? `FAIL on ${failures.join(", ")}` : "PASS on all 5 histories";
  console.log(`${failures.length ? "BLOCK" : "SHIP "}  ${c.name.padEnd(42)} ${verdict}`);
}
console.log("\nNote: v3-patch-removed-too-early fails only on v1 histories, so it becomes safe");
console.log("once no v1 execution is still open or needs to be replayed (for example, for queries or resets).");

BLOCK  v3-unsafe (adds fraud_check, no patch)     FAIL on order-101, order-102
SHIP   v3-patched (keeps add-fraud-check patch)   PASS on all 5 histories
BLOCK  v3-patch-removed-too-early                 FAIL on order-101, order-102
BLOCK  v3-reorders ship before charge             FAIL on order-101, order-102, order-201

Note: v3-patch-removed-too-early fails only on v1 histories, so it becomes safe
once no v1 execution is still open or needs to be replayed (for example, for queries or resets).