Building a Production Batch-Processing Pipeline

This lesson combines the decisions from every earlier lesson in this unit into one working system: a pipeline that processes a large set of records end to end, with bounded concurrency, adaptive backoff, progress tracking, classified failure handling, idempotent retries, and full crash resumability. This is not a repeat of the Batch API mechanics from Unit 12 or the simple worker pool from Lesson 4 — it is the architecture those pieces belong inside once a job needs to run unattended, at scale, and survive interruption.

The Scenario

Assume a system needs to generate a one-paragraph summary for each of a large set of customer feedback records stored in a source table, writing each summary into a destination table, and doing so reliably even if the process is restarted partway through a run of 50,000 records.

Architecture Overview

The pipeline follows the five-stage design from Lesson 2, with each stage's responsibility narrowed to exactly what was justified in the lessons that followed:

Ingest  → read source records, compute stable record_ids
Prepare → build prompts
Submit  → bounded-concurrency async workers with backoff (Lessons 4, 5)
Collect → checkpoint every outcome durably as it arrives (Lessons 6, 9)
Finalize → classify failures, route retryable ones, quarantine the rest (Lesson 7)

Resumability (Lesson 9) isn't a separate stage — it's a property of how Ingest and Collect are written: Ingest only enqueues records not already checkpointed as succeeded, and Collect writes every outcome durably the moment it's known.

The Checkpoint Store

This reuses the schema from Lesson 9, extended with a stage column so partial failures (Lesson 7) can be resumed from the correct step rather than from scratch.

import sqlite3
import time


def init_pipeline_db(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS records (
            record_id TEXT PRIMARY KEY,
            status TEXT NOT NULL DEFAULT 'pending',
            failure_stage TEXT,
            failure_category TEXT,
            output TEXT,
            error TEXT,
            attempts INTEGER DEFAULT 0,
            updated_at REAL
        )
        """
    )
    conn.commit()
    return conn


def upsert_record(conn, record_id, status, output=None, error=None,
                   failure_stage=None, failure_category=None):
    conn.execute(
        """
        INSERT INTO records (record_id, status, output, error, failure_stage,
                              failure_category, attempts, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, 1, ?)
        ON CONFLICT(record_id) DO UPDATE SET
            status = excluded.status,
            output = COALESCE(excluded.output, records.output),
            error = excluded.error,
            failure_stage = excluded.failure_stage,
            failure_category = excluded.failure_category,
            attempts = records.attempts + 1,
            updated_at = excluded.updated_at
        """,
        (record_id, status, output, error, failure_stage, failure_category, time.time()),
    )
    conn.commit()

COALESCE(excluded.output, records.output) is the detail that makes partial-failure recovery actually work end to end: if a record already has a saved output from a successful model call, and a later update comes in with output=None (because that update is reporting a downstream write failure, not a new model result), the existing output is preserved rather than being overwritten with nothing. This is the same "don't discard partial progress" principle from Lesson 7, now expressed at the storage layer.

The Adaptive, Checkpointed Worker

Each worker combines backoff (Lesson 5), stage-aware failure classification (Lesson 7), and immediate checkpointing (Lesson 9):

import asyncio
import random
from openai import AsyncOpenAI, RateLimitError


async def process_and_checkpoint(client, conn, record, max_retries=4):
    upsert_record(conn, record.record_id, "in_progress")

    # Stage 1: model call, with backoff on rate limits
    output_text = None
    for attempt in range(max_retries):
        try:
            response = await client.responses.create(
                model="gpt-5.6-terra",
                input=record.prompt,
            )
            output_text = response.output_text
            break
        except RateLimitError:
            if attempt == max_retries - 1:
                upsert_record(conn, record.record_id, "failed",
                              error="rate limit exhausted",
                              failure_stage="model_call", failure_category="transient")
                return
            await asyncio.sleep(2 ** attempt + random.uniform(0, 1))
        except Exception as exc:
            upsert_record(conn, record.record_id, "failed", error=str(exc),
                          failure_stage="model_call", failure_category="permanent")
            return

    # Stage 2: write result downstream
    try:
        await write_summary_to_destination(record.record_id, output_text)
        upsert_record(conn, record.record_id, "succeeded", output=output_text)
    except Exception as exc:
        upsert_record(conn, record.record_id, "failed", output=output_text,
                      error=str(exc), failure_stage="downstream_write",
                      failure_category="transient")


async def write_summary_to_destination(record_id: str, summary: str) -> None:
    # Placeholder for the real destination write (database, file, API call).
    pass

Note that output is passed to upsert_record in the downstream_write failure branch — this is what allows a later retry to skip the model call for that record entirely, exactly as designed in Lesson 8.

Resuming: Loading Only What's Left to Do

def load_remaining_records(conn, all_records: list) -> list:
    ids = [r.record_id for r in all_records]
    placeholders = ",".join("?" for _ in ids)
    rows = conn.execute(
        f"SELECT record_id, output FROM records "
        f"WHERE record_id IN ({placeholders}) AND status = 'succeeded'",
        ids,
    ).fetchall()
    done_ids = {row[0] for row in rows}
    conn.execute(
        "UPDATE records SET status = 'pending' WHERE status = 'in_progress'"
    )
    conn.commit()
    return [r for r in all_records if r.record_id not in done_ids]

Assembling the Full Run

async def run_production_pipeline(db_path: str, source_rows: list[dict],
                                    prompt_template: str, concurrency: int = 10):
    conn = init_pipeline_db(db_path)

    records = [
        PipelineRecord(record_id=str(row["id"]), source_data=row,
                        prompt=prompt_template.format(**row))
        for row in source_rows
    ]

    remaining = load_remaining_records(conn, records)
    print(f"{len(records) - len(remaining)} already done, {len(remaining)} to process")

    tracker = ProgressTracker(total=len(records))
    tracker.succeeded = len(records) - len(remaining)

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

    async def worker():
        while True:
            record = await queue.get()
            if record is None:
                queue.task_done()
                break
            tracker.report_started()
            try:
                await process_and_checkpoint(client, conn, record)
                row = conn.execute(
                    "SELECT status FROM records WHERE record_id = ?", (record.record_id,)
                ).fetchone()
                if row and row[0] == "succeeded":
                    tracker.report_succeeded()
                else:
                    tracker.report_failed()
            except Exception:
                tracker.report_failed()
            queue.task_done()

    async def report_loop():
        while tracker.pending > 0 or tracker.in_progress > 0:
            print(tracker.summary())
            await asyncio.sleep(5)
        print(tracker.summary())

    workers = [asyncio.create_task(worker()) for _ in range(concurrency)]
    reporter = asyncio.create_task(report_loop())

    await queue.join()
    for _ in workers:
        queue.put_nowait(None)
    await asyncio.gather(*workers)
    await reporter

    quarantined = conn.execute(
        "SELECT record_id, error FROM records WHERE failure_category = 'permanent'"
    ).fetchall()
    if quarantined:
        print(f"{len(quarantined)} record(s) need manual review:")
        for record_id, error in quarantined:
            print(f"  {record_id}: {error}")

    conn.close()

This function is what an operator or a scheduler actually invokes. Run it once, and it processes every remaining record. If it's killed at any point — deliberately or by a crash — running it again with the same db_path and the same source_rows resumes exactly where it stopped: already-succeeded records are excluded before any new request is made, stuck in_progress records are reset and safely retried, and partial output from downstream_write failures is preserved rather than recomputed.

Testing the Full Pipeline with a Fake Client

Because every dependency (client, conn) is passed in rather than constructed globally, the entire pipeline is testable end to end without touching the real API:

class FakePipelineClient:
    def __init__(self, fail_ids: set = frozenset()):
        self.fail_ids = fail_ids
        self.responses = self

    async def create(self, model, input):
        if any(fid in input for fid in self.fail_ids):
            raise RuntimeError("simulated permanent failure")

        class FakeResponse:
            output_text = f"summary of: {input}"
        return FakeResponse()


async def test_pipeline_resumes_after_simulated_crash():
    conn = init_pipeline_db(":memory:")
    records = [
        PipelineRecord(record_id=str(i), source_data={}, prompt=f"feedback {i}")
        for i in range(5)
    ]

    # First "run": process records 0-2, simulate a crash before 3 and 4 run.
    client = FakePipelineClient()
    for r in records[:3]:
        await process_and_checkpoint(client, conn, r)

    # "Restart": load_remaining_records should only return records 3 and 4.
    remaining = load_remaining_records(conn, records)
    remaining_ids = {r.record_id for r in remaining}
    assert remaining_ids == {"3", "4"}

    for r in remaining:
        await process_and_checkpoint(client, conn, r)

    succeeded = conn.execute(
        "SELECT COUNT(*) FROM records WHERE status = 'succeeded'"
    ).fetchone()[0]
    assert succeeded == 5
    print("PASS: pipeline resumes after simulated crash and completes all records")


asyncio.run(test_pipeline_resumes_after_simulated_crash())

This test simulates a crash by simply stopping short of processing every record in the first loop, then verifies that a fresh call to load_remaining_records — exactly what a real restart would do — correctly identifies only the unfinished records, and that the second pass completes the job. No real network call, no real delay, and no dependency on wall-clock timing is involved, which is what makes it suitable to run on every commit.

What This Pipeline Deliberately Leaves Out

A genuinely complete production deployment would add a few things beyond this lesson's scope: structured logging to a centralized log system rather than print, metrics exported to a monitoring system rather than only a terminal summary, and a supervisor process (a process manager, a container orchestrator, or a scheduled job runner) that actually restarts the pipeline process after a crash rather than requiring a human to notice and re-invoke it. None of that changes the pipeline's internal design — those are operational concerns that wrap around exactly the resumable, checkpointed, failure-aware core built in this lesson.

Common Mistakes

• Building the concurrency, retry, and checkpoint logic as three separate, uncoordinated systems that were each tested individually but never verified together — the interaction between backoff delays, checkpoint timing, and resumability is exactly where subtle bugs hide, which is why this lesson's test exercises the combination directly.
• Forgetting to reset in_progress records on restart in the full pipeline, which silently reproduces the exact "stuck record" bug from Lesson 9 even though the individual piece was solved there — integration is where previously-fixed problems reappear if they aren't wired in everywhere they're needed.
• Skipping an end-to-end resumability test and only testing each lesson's concept in isolation. Unit tests for upsert_record, process_and_checkpoint, and load_remaining_records individually can all pass while the full restart sequence still has a gap between them.

Best Practices

• Test the crash-and-resume path explicitly, not just the happy path. A test that processes some records, "restarts," and verifies the rest complete correctly is the single most valuable test for any pipeline built around checkpointing.
• Keep the checkpoint schema rich enough to support partial-failure recovery, not just a plain succeeded/failed flag — the failure_stage, failure_category, and preserved output columns are what make retries cheap and correct rather than wasteful.
• Treat quarantined (permanent-failure) records as a required output of every run, not an afterthought. A production pipeline is not finished when it stops running — it's finished when every record has either succeeded or been placed somewhere a human can act on it.