Designing Resumable AI Processing Jobs

Every previous lesson in this unit has assumed a job runs from start to finish without interruption. Real production jobs don't get that guarantee: a server restarts for a deployment, a process crashes on an unhandled exception, an operator needs to stop a job mid-run to fix a bug in the prompt template. A job that processes 100,000 records over six hours and has no way to resume after an interruption other than starting item one again is not production-ready, regardless of how well it handles concurrency, rate limits, and failures while it's running. This lesson covers how to design a job so that "restart" means "continue from where it left off," not "begin again."

What Makes a Job Resumable

A resumable job has three properties working together:

1. Durable state. Every record's current status is stored somewhere that survives the process ending — a database or a file on disk, never only in memory.
2. Idempotent restart logic. On startup, the job checks that durable state and skips anything already completed, using the idempotency techniques from Lesson 8 to make sure "checking status" and "acting on it" don't themselves race or duplicate work.
3. Frequent, atomic checkpoints. Status updates are written durably often enough that a crash loses at most a small, bounded amount of work — not the entire job's progress up to that point.

None of these are exotic techniques individually. The discipline is making sure all three are actually in place before a job runs at production scale, rather than discovering their absence during an outage.

Checkpointing with a Durable Store

The simplest reliable checkpoint store for a moderate-scale job (tens of thousands to low millions of records) is a SQLite database file — it requires no separate server process, supports atomic transactions out of the box, and survives a process crash because the data is written to disk, not held in memory.

import sqlite3
from contextlib import contextmanager


def init_checkpoint_db(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS job_records (
            record_id TEXT PRIMARY KEY,
            status TEXT NOT NULL,
            output TEXT,
            error TEXT,
            attempts INTEGER DEFAULT 0,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    conn.commit()
    return conn


def upsert_checkpoint(conn: sqlite3.Connection, record_id: str, status: str,
                       output: str = None, error: str = None) -> None:
    conn.execute(
        """
        INSERT INTO job_records (record_id, status, output, error, attempts)
        VALUES (?, ?, ?, ?, 1)
        ON CONFLICT(record_id) DO UPDATE SET
            status = excluded.status,
            output = excluded.output,
            error = excluded.error,
            attempts = job_records.attempts + 1,
            updated_at = CURRENT_TIMESTAMP
        """,
        (record_id, status, output, error),
    )
    conn.commit()

This is the same upsert pattern from Lesson 8, applied here to the job's own progress tracking rather than to a business-level side effect — record_id is the UNIQUE/PRIMARY KEY column, so calling upsert_checkpoint any number of times for the same record always results in exactly one row reflecting its latest known state. Calling conn.commit() after every single update, rather than batching commits, is a deliberate tradeoff: it's slower than batching, but it guarantees that a crash immediately after a commit leaves the database reflecting a fully consistent, durable state — there is no window where several updates are sitting uncommitted in memory and would be lost together. For very high-throughput jobs where per-item commits become a bottleneck, batching commits every N items or every few seconds is a reasonable optimization, but it explicitly trades a small, bounded amount of possible rework (re-processing whatever was uncommitted at crash time) for higher throughput — a tradeoff worth making consciously, not by accident.

Determining What Still Needs to Run

The core of resumability is a query, run once at job startup, that partitions the full set of records into "already done" and "still needs work":

def load_incomplete_records(conn: sqlite3.Connection, all_record_ids: list[str]) -> list[str]:
    """Return the subset of record_ids that are not yet successfully completed."""
    placeholders = ",".join("?" for _ in all_record_ids)
    done_rows = conn.execute(
        f"""
        SELECT record_id FROM job_records
        WHERE record_id IN ({placeholders}) AND status = 'succeeded'
        """,
        all_record_ids,
    ).fetchall()
    done_ids = {row[0] for row in done_rows}
    return [rid for rid in all_record_ids if rid not in done_ids]

Notice this checks specifically for status = 'succeeded' — a record sitting at in_progress (from a run that crashed mid-item) or failed (from a previous attempt) is correctly treated as still needing work. This is an important subtlety: a naive resumability check that only asks "does a row exist for this record_id?" would incorrectly skip records that were started but never finished, silently leaving gaps in the output. Resumability depends on the status value, not merely on the presence of a row.

Handling the "In Progress" Ambiguity

There is one genuinely tricky case: a record whose last known status is in_progress when the job restarts. This means the previous run was in the middle of processing that record — calling the model, or writing its result — when the crash happened, and you cannot know from the checkpoint alone whether that in-flight work actually completed on the server side before the crash.

The safe default is to treat in_progress records as not-yet-done and retry them, relying on the idempotency work from Lesson 8 (upserts, idempotency keys) to make that retry harmless even in the rare case where the original attempt had actually succeeded moments before the crash:

def reset_stuck_in_progress_records(conn: sqlite3.Connection) -> int:
    """On startup, any record still marked in_progress belongs to a
    run that never finished cleanly. Reset it to pending so it gets retried."""
    cursor = conn.execute(
        "UPDATE job_records SET status = 'pending' WHERE status = 'in_progress'"
    )
    conn.commit()
    return cursor.rowcount

Running this once at the start of every job launch — before computing load_incomplete_records — ensures no record is permanently stuck in a state that neither the "done" filter nor a human notices. This is a direct, practical consequence of the idempotency design from Lesson 8: because retrying a record whose downstream write already succeeded is safe (an upsert just overwrites the same value), resetting all in_progress records to pending on every restart is the correct, low-risk default rather than something to be avoided.

A Job State Machine

Putting the statuses together, a record's lifecycle across a resumable job follows an explicit state machine, which is worth writing down rather than leaving implicit in scattered string literals:

pending → in_progress → succeeded
                      ↘ failed → pending (on retry) → in_progress → ...

from enum import Enum


class JobRecordStatus(str, Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    SUCCEEDED = "succeeded"
    FAILED = "failed"


VALID_TRANSITIONS = {
    JobRecordStatus.PENDING: {JobRecordStatus.IN_PROGRESS},
    JobRecordStatus.IN_PROGRESS: {JobRecordStatus.SUCCEEDED, JobRecordStatus.FAILED},
    JobRecordStatus.FAILED: {JobRecordStatus.PENDING},   # retry re-queues it
    JobRecordStatus.SUCCEEDED: set(),                     # terminal state
}


def transition(current: JobRecordStatus, target: JobRecordStatus) -> JobRecordStatus:
    if target not in VALID_TRANSITIONS[current]:
        raise ValueError(f"invalid transition: {current} -> {target}")
    return target

Encoding valid transitions explicitly, and raising on an invalid one, catches an entire class of bugs at the moment they'd occur — for example, code that accidentally tries to mark a record in_progress a second time without it ever having been reset to pending, which would indicate a bug in the pipeline's control flow rather than a legitimate state change.

Putting It Together: A Resumable Run Function

async def run_resumable_job(db_path: str, all_records: list, concurrency: int = 10):
    conn = init_checkpoint_db(db_path)
    reset_count = reset_stuck_in_progress_records(conn)
    if reset_count:
        print(f"reset {reset_count} stuck in_progress record(s) from a previous run")

    all_ids = [r.record_id for r in all_records]
    incomplete_ids = load_incomplete_records(conn, all_ids)
    incomplete_records = [r for r in all_records if r.record_id in set(incomplete_ids)]

    print(f"{len(all_records) - len(incomplete_records)} already done, "
          f"{len(incomplete_records)} remaining")

    client = AsyncOpenAI()
    queue: asyncio.Queue = asyncio.Queue()
    for r in incomplete_records:
        queue.put_nowait(r)

    async def worker(queue):
        while True:
            record = await queue.get()
            if record is None:
                queue.task_done()
                break
            upsert_checkpoint(conn, record.record_id, "in_progress")
            try:
                response = await client.responses.create(
                    model="gpt-5.6-terra", input=record.prompt
                )
                upsert_checkpoint(conn, record.record_id, "succeeded", output=response.output_text)
            except Exception as exc:
                upsert_checkpoint(conn, record.record_id, "failed", error=str(exc))
            queue.task_done()

    workers = [asyncio.create_task(worker(queue)) for _ in range(concurrency)]
    await queue.join()
    for _ in workers:
        queue.put_nowait(None)
    await asyncio.gather(*workers)
    conn.close()

If this function is interrupted at any point — the process is killed, the machine restarts — calling it again with the same db_path and the same all_records list picks up exactly where it left off: records already checkpointed as succeeded are excluded from incomplete_records before a single new request is made, and anything left in_progress from the interrupted run is reset and retried safely.

Testing Resumability Logic

The state-machine and record-selection logic can be tested entirely with an in-memory database and no network calls:

def test_resumable_job_skips_completed_and_retries_stuck_records():
    conn = init_checkpoint_db(":memory:")
    upsert_checkpoint(conn, "1", "succeeded", output="done")
    upsert_checkpoint(conn, "2", "in_progress")
    upsert_checkpoint(conn, "3", "failed", error="timeout")

    reset_count = reset_stuck_in_progress_records(conn)
    assert reset_count == 1

    incomplete = load_incomplete_records(conn, ["1", "2", "3", "4"])
    assert set(incomplete) == {"2", "3", "4"}  # "1" is done; others need work
    print("PASS: resumable job correctly identifies remaining work after a restart")


test_resumable_job_skips_completed_and_retries_stuck_records()

Common Mistakes

• Keeping job state only in memory (a Python list or dict) with no durable checkpoint. This makes every crash equivalent to losing all progress, regardless of how far the job had gotten, and defeats the entire purpose of building resumability.
• Treating "a row exists for this record" as equivalent to "this record is done." A row can exist with status pending, in_progress, or failed — only succeeded means the work doesn't need to happen again.
• Not handling the in_progress state at restart at all, leaving records permanently stuck in a status that the "still needs work" query doesn't select and that never gets human attention either.

Best Practices

• Persist status changes as they happen, not in a final batch at the end. A checkpoint written only when the whole job finishes provides no protection against the exact failure mode — a mid-job crash — that resumability exists to handle.
• Reset in_progress records to pending at every job startup, and rely on idempotent operations (Lesson 8) to make the resulting retry safe even in the rare case where the original attempt had actually completed.
• Define the record status state machine explicitly, including which transitions are valid, so that bugs in the pipeline's control flow surface as an immediate error rather than as silently inconsistent checkpoint data discovered much later.