Build a Workflow That Survives a Crash Mid-Run

13 min read · updated August 4, 2026

A multi-step AI workflow — fetch, extract, enrich, verify, deliver — will be interrupted. The process is redeployed, the model call times out, the machine is preempted. What decides whether that is a non-event or an incident is whether progress is written down between steps, and whether re-running a step is safe. Both are achievable in a SQLite table and about a hundred and fifty lines.

Why a script is not enough

A script holds its progress in local variables. When it dies, the progress dies with it, and the recovery options are both bad: start again and pay for the model calls twice, or work out by hand what it finished.

The costs are specific rather than theoretical. A run halfway through a thousand documents has spent real money on tokens that are now unrecoverable. Worse, if step four sent an email and step five crashed, restarting sends the email again — and there is no way to know from the outside whether it already went.

The state machine, written out

Write the states down before writing code. Not as a diagram in a document, as a constant in the file that the code checks against.

STATES = {
  "pending":    ["fetching", "cancelled"],
  "fetching":   ["fetched", "failed_fetch", "pending"],       # -> pending = retry
  "fetched":    ["extracting"],
  "extracting": ["extracted", "failed_extract", "fetched"],
  "extracted":  ["verifying"],
  "verifying":  ["verified", "needs_human", "extracted"],
  "verified":   ["delivering"],
  "delivering": ["delivered", "failed_deliver", "verified"],
  "delivered":  [],                       # terminal
  "needs_human":["extracted", "cancelled"],
  "cancelled":  [],                       # terminal
  "failed_fetch":   ["pending"],          # terminal until a human requeues
  "failed_extract": ["fetched"],
  "failed_deliver": ["verified"],
}

def transition(conn, job_id, frm, to):
    assert to in STATES[frm], "illegal transition " + frm + " -> " + to
    n = conn.execute(
        "UPDATE job SET state=?, updated=?, attempts=0 "
        "WHERE id=? AND state=?",       # <- the guard that makes it safe
        (to, time.time(), job_id, frm)).rowcount
    conn.commit()
    return n == 1                        # False: somebody else moved it first

Two properties make this worth the ceremony. Illegal transitions fail loudly at the point of the bug rather than producing a job in an impossible state. And the WHERE state = ? guard means the update is a compare-and-swap: if another worker has already advanced the job, your update affects zero rows and you know to stop.

Notice that every working state has a transition back to the state before it. That is the retry edge, and having it in the table means “retry” is not special-case code.

The schema

CREATE TABLE job (
  id          TEXT PRIMARY KEY,          -- caller-supplied idempotency key
  state       TEXT NOT NULL,
  payload     TEXT NOT NULL,             -- JSON input
  result      TEXT,                      -- JSON accumulated output
  attempts    INTEGER NOT NULL DEFAULT 0,
  next_run_at REAL NOT NULL DEFAULT 0,
  lease_until REAL NOT NULL DEFAULT 0,
  lease_owner TEXT,
  last_error  TEXT,
  created     REAL NOT NULL,
  updated     REAL NOT NULL
);
CREATE INDEX job_ready ON job(state, next_run_at);

CREATE TABLE step_result (                -- the memo table: never redo work
  job_id  TEXT NOT NULL,
  step    TEXT NOT NULL,
  output  TEXT NOT NULL,
  cost    REAL NOT NULL DEFAULT 0,
  at      REAL NOT NULL,
  PRIMARY KEY (job_id, step)
);

CREATE TABLE outbox (                     -- side effects, exactly once
  id       INTEGER PRIMARY KEY,
  job_id   TEXT NOT NULL,
  kind     TEXT NOT NULL,
  payload  TEXT NOT NULL,
  sent_at  REAL,
  attempts INTEGER NOT NULL DEFAULT 0,
  UNIQUE (job_id, kind)
);

step_result is the piece that turns a resume from “start the step again” into “skip the step”. Before running any step, look for its row; if it exists, use it. A model call costing three cents is never paid for twice, and a run that crashes at step five resumes at step five rather than step one.

The job id being caller-supplied is the outer idempotency guarantee: submitting the same work twice with the same id is a no-op, so an upstream retry cannot duplicate the job. Idempotency keys at every boundary is the pattern, and workflow engines exist largely to give you it.

Leases, so two workers cannot both run a step

With more than one worker, two of them will claim the same job. A lease — a timestamped claim that expires — solves it without a lock service, because the expiry means a worker that dies holding a job releases it automatically.

import os, socket, time, uuid

WORKER = socket.gethostname() + ":" + str(os.getpid())
LEASE_SECONDS = 120

def claim(conn):
    now = time.time()
    row = conn.execute(
        "SELECT id FROM job WHERE state NOT IN "
        "  ('delivered','cancelled','needs_human',"
        "   'failed_fetch','failed_extract','failed_deliver') "
        "AND next_run_at <= ? AND lease_until <= ? "
        "ORDER BY next_run_at LIMIT 1", (now, now)).fetchone()
    if not row:
        return None
    ok = conn.execute(
        "UPDATE job SET lease_until=?, lease_owner=? "
        "WHERE id=? AND lease_until<=?",
        (now + LEASE_SECONDS, WORKER, row[0], now)).rowcount
    conn.commit()
    return row[0] if ok == 1 else None      # lost the race; try again

def heartbeat(conn, job_id):
    """Call inside long steps, or the lease expires mid-model-call."""
    conn.execute("UPDATE job SET lease_until=? WHERE id=? AND lease_owner=?",
                 (time.time() + LEASE_SECONDS, job_id, WORKER))
    conn.commit()

The lost lease is the failure people do not anticipate. A model call takes 180 seconds against a 120-second lease, another worker picks the job up, and now two workers are running the same step and both will write a result. The memo table’s primary key stops the duplicate write, but the duplicate spend already happened. Either heartbeat during long steps or set the lease longer than your slowest call plus its timeout — and make sure the call has a timeout, because an HTTP request with no timeout can hang far longer than any lease you would choose.

Retry, backoff and the poison message

RETRYABLE = (TimeoutError, ConnectionError)      # plus HTTP 429, 5xx
MAX_ATTEMPTS = 5

def backoff(attempt):
    """Exponential with full jitter: 1-2s, 1-4s, 1-8s, 1-16s, 1-32s."""
    import random
    return random.uniform(1.0, min(32.0, 2.0 ** attempt))

def fail_step(conn, job_id, state, exc, retryable):
    attempts = conn.execute(
        "SELECT attempts FROM job WHERE id=?", (job_id,)).fetchone()[0] + 1
    if retryable and attempts < MAX_ATTEMPTS:
        conn.execute(
            "UPDATE job SET attempts=?, next_run_at=?, last_error=?, "
            "lease_until=0 WHERE id=?",
            (attempts, time.time() + backoff(attempts), repr(exc)[:500], job_id))
    else:
        conn.execute(
            "UPDATE job SET state=?, last_error=?, lease_until=0 WHERE id=?",
            (TERMINAL_FOR[state], repr(exc)[:500], job_id))
    conn.commit()

Full jitter rather than plain exponential backoff. Without the random component, everything that failed during an outage retries at the same instant when the outage ends, and the recovery attempt becomes the second outage. Jittered backoff is the standard answer and it costs one line.

The poison message is the other half. A job that fails deterministically — malformed input, a document that crashes the parser — retried five times is five times the cost and the same outcome. Distinguish it by the exception type: retry timeouts, connection errors, 429 and 5xx; never retry a 400, a schema validation failure or a parse error, because nothing about them will be different next time. A 429 additionally carries retry guidance worth honouring rather than guessing at.

Side effects: the outbox

Here is the problem in one sentence: you cannot atomically send an email and commit a database row, so whatever order you choose, a crash between them leaves you either double-sending or losing the send.

The outbox pattern makes the choice explicit and recoverable. In the same transaction that completes the step, insert a row describing the side effect. A separate loop sends unsent rows and marks them.

Step completes: within one transaction, write step_result, advance the state, and insert into outbox. All or nothing.
The sender selects rows with sent_at IS NULL, performs the effect, and sets sent_at.
A crash between performing and marking means it sends again. That is at-least-once, and it is the best you can do without cooperation from the receiver.
So get that cooperation: pass the outbox row id as an idempotency key to whatever you are calling. Payment APIs and most transactional email services accept one. Where none is accepted, a UNIQUE (job_id, kind) constraint at least bounds duplicates to one per kind per job.

Exactly-once delivery does not exist across a network. What exists is at-least-once delivery plus an idempotent receiver, and saying so plainly in the design is better than a comment claiming otherwise.

What resuming actually looks like

def run_once(conn):
    job_id = claim(conn)
    if not job_id:
        return False
    state, payload = conn.execute(
        "SELECT state, payload FROM job WHERE id=?", (job_id,)).fetchone()
    step = STEP_FOR_STATE[state]              # e.g. 'fetching' -> fetch_step

    memo = conn.execute(
        "SELECT output FROM step_result WHERE job_id=? AND step=?",
        (job_id, step.name)).fetchone()
    if memo:                                   # already done before the crash
        transition(conn, job_id, state, step.success_state)
        return True

    try:
        out, cost = step.run(json.loads(payload), conn, job_id)
    except Exception as e:
        fail_step(conn, job_id, state, e, isinstance(e, RETRYABLE))
        return True

    with conn:                                 # one transaction
        conn.execute("INSERT OR IGNORE INTO step_result "
                     "(job_id, step, output, cost, at) VALUES (?,?,?,?,?)",
                     (job_id, step.name, json.dumps(out), cost, time.time()))
        for effect in step.effects(out):
            conn.execute("INSERT OR IGNORE INTO outbox "
                         "(job_id, kind, payload) VALUES (?,?,?)",
                         (job_id, effect["kind"], json.dumps(effect)))
        conn.execute("UPDATE job SET state=?, attempts=0, lease_until=0, "
                     "updated=? WHERE id=? AND state=?",
                     (step.success_state, time.time(), job_id, state))
    return True

Kill the process at any line of that function and the next worker to claim the job does the right thing. If the crash was before the transaction, the step runs again — costing one model call, which the memo check would have saved had it completed. If it was after, the memo exists and the step is skipped. There is no third case, which is the entire point of putting the writes in one transaction.

Record cost per step while you are there. Cost per job by step is the report that tells you which stage of a workflow to optimise, and it is nearly free to collect at the moment the call returns.

That cost column needs a per-request number, which is awkward to assemble when a workflow spans several models and a fallback path. Multigrid returns the cost of each request with the response, so step_result.cost is a field you copy rather than a figure you reconstruct from token counts and a price table.

The four queries you will actually run

The schema above was chosen partly so that operating the thing is SQL rather than a log search. These four answer nearly every question anyone asks about a running workflow.

-- 1. Where is everything? The first thing you look at, every time.
SELECT state, COUNT(*), ROUND(AVG(strftime('%s','now') - updated)) AS avg_age_s
FROM job GROUP BY state ORDER BY 2 DESC;

-- 2. What is stuck? Jobs whose lease expired while they were being worked.
SELECT id, state, attempts, lease_owner, last_error
FROM job
WHERE lease_until > 0 AND lease_until < strftime('%s','now')
ORDER BY updated LIMIT 50;

-- 3. What is failing, and is it one cause or many?
SELECT substr(last_error, 1, 60) AS err, COUNT(*)
FROM job WHERE state LIKE 'failed%' GROUP BY err ORDER BY 2 DESC;

-- 4. Which step costs the money?
SELECT step, COUNT(*), ROUND(SUM(cost), 2) AS total, ROUND(AVG(cost), 4) AS mean
FROM step_result WHERE at > strftime('%s','now') - 86400
GROUP BY step ORDER BY total DESC;

Query one is the alert. A state whose count is rising and whose average age is rising with it is a stalled stage, and that pair of numbers identifies it faster than any error rate — because the characteristic failure of a workflow is not throwing exceptions, it is jobs quietly accumulating in one state while everything else looks healthy.

Alert on the oldest job in a non-terminal state, not on throughput. Throughput looks fine while one job is wedged.
Alert on the failed states having any rows at all if volume is low, or on their rate if it is high. Those states are terminal until a human acts, so nothing else will surface them.
Keep step_result after the job completes. It is small, it is the cost report, and it is what lets you rerun a single step against new code with the exact input it had.

When to reach for a real engine instead

The design above is right up to roughly the point where any of these becomes true:

Steps run on different machines and need to pass large payloads. SQLite as a shared bus stops being reasonable at that point.
You need timers measured in days. “Wait 30 days, then check again” is a first-class feature of workflow engines and an awkward one to build.
Human approval steps are common. A workflow that suspends for a week and resumes with a decision is exactly what those systems are for.
You need history and replay. Re-running a completed workflow against new code, with the old inputs, is a capability worth adopting a dependency for.

Below that line, a hundred and fifty lines of SQLite is easier to reason about than an orchestration framework, and — more to the point — you now know precisely what one is doing for you. Orchestration choices for AI pipelines and handling partial failure take that comparison further.

Build a Workflow That Survives a Crash Mid-Run

Why a script is not enough

The state machine, written out

The schema

Leases, so two workers cannot both run a step

Retry, backoff and the poison message

Side effects: the outbox

What resuming actually looks like

The four queries you will actually run

When to reach for a real engine instead

Related