Handling Partial Failures in Bulk Workloads

Unit 12, Lesson 5 covered what the Batch API itself does with failures: a submitted batch job produces both an output file for successful requests and a separate error file listing which requests failed and why, matched back by custom_id. That is failure handling at the API level — it tells you which HTTP-level requests didn't succeed. This lesson is about a different, harder problem: failure handling at the pipeline and business-logic level, where "the API call succeeded" and "this item was processed correctly" are not the same statement, and where a bulk job finishing with 9,850 out of 10,000 items succeeded raises real operational questions that a pass/fail count alone doesn't answer.

Partial Failure Is the Normal Case, Not the Exception

A synchronous, single-request feature can reasonably treat failure as exceptional — one user, one request, and if it fails you show an error and let them retry. A bulk job processing tens of thousands of independent items behaves completely differently: at that scale, some nonzero failure rate is close to guaranteed, whether from transient network issues, occasional malformed source data, content that trips a safety filter, or responses that don't parse the way you expected. Designing a bulk pipeline around the assumption that failures are rare and exceptional produces brittle code that either crashes the whole job on the first failure or silently drops failed items with no record of what happened to them. Designing around the assumption that failures are a normal, expected fraction of any large run produces a pipeline that finishes reliably and hands you a clear, actionable account of what needs attention.

A Failure Taxonomy: Not All Failures Are the Same

Treating every failure identically — log it and move on — throws away information that matters for what to do next. A useful taxonomy separates failures along two axes: where they occurred, and whether retrying is likely to help.

The last row deserves particular attention because it's the case most pipelines get wrong: if the model call succeeded but writing the result to your database failed, retrying "the item" naively means calling the model again, paying for a duplicate request, when the actual problem was entirely on the write side. This is exactly why the record-based pipeline design from Lesson 2 matters — a record's status should be granular enough to distinguish "model call failed" from "model call succeeded but downstream write failed," so the retry logic in Lesson 8 can act on the correct step rather than restarting the whole item from scratch.

from enum import Enum


class FailureStage(str, Enum):
    MODEL_CALL = "model_call"
    RESPONSE_PARSING = "response_parsing"
    DOWNSTREAM_WRITE = "downstream_write"
    SOURCE_DATA = "source_data"


class FailureCategory(str, Enum):
    TRANSIENT = "transient"      # safe and likely useful to retry
    PERMANENT = "permanent"      # retrying will not help

Designing a Per-Item Result Record for Failures

Extending the PipelineRecord from Lesson 2 with explicit failure metadata turns a vague error: Optional[str] field into something the pipeline's finalize stage — and a human reviewing the run afterward — can act on directly:

from dataclasses import dataclass, field
from typing import Optional


@dataclass
class FailureInfo:
    stage: FailureStage
    category: FailureCategory
    message: str
    raw_error: Optional[str] = None


@dataclass
class ItemResult:
    record_id: str
    success: bool
    output: Optional[str] = None
    failure: Optional[FailureInfo] = None


def classify_exception(exc: Exception, stage: FailureStage) -> FailureInfo:
    message = str(exc)
    lowered = message.lower()
    if "rate_limit" in lowered or "timeout" in lowered or "connection" in lowered:
        category = FailureCategory.TRANSIENT
    elif "invalid_request" in lowered or "content_policy" in lowered:
        category = FailureCategory.PERMANENT
    else:
        category = FailureCategory.TRANSIENT  # default to retryable unless proven otherwise
    return FailureInfo(stage=stage, category=category, message=message, raw_error=message)

classify_exception makes an explicit, reviewable decision about whether a given error is worth retrying, rather than leaving that judgment to be re-derived ad hoc every time someone looks at a failure log. Defaulting unknown errors to TRANSIENT is a deliberate, conservative choice: it's usually safer to retry an item that turns out not to need it (a wasted, cheap re-attempt) than to permanently give up on an item that would have succeeded on a second try.

Processing an Item Stage-by-Stage, Catching Failures at Each One

Wiring the taxonomy into actual item processing means wrapping each stage — model call, parsing, downstream write — in its own error handling, so a failure's stage field is always accurate:

async def process_item_with_failure_tracking(client, record) -> ItemResult:
    # Stage 1: model call
    try:
        response = await client.responses.create(
            model="gpt-5.6-terra",
            input=record.prompt,
        )
    except Exception as exc:
        return ItemResult(
            record_id=record.record_id,
            success=False,
            failure=classify_exception(exc, FailureStage.MODEL_CALL),
        )

    # Stage 2: parse the response into the expected structure
    try:
        parsed_output = parse_model_output(response.output_text)
    except Exception as exc:
        return ItemResult(
            record_id=record.record_id,
            success=False,
            failure=classify_exception(exc, FailureStage.RESPONSE_PARSING),
        )

    # Stage 3: write the result downstream
    try:
        await write_result_downstream(record.record_id, parsed_output)
    except Exception as exc:
        return ItemResult(
            record_id=record.record_id,
            success=False,
            output=parsed_output,  # the model result exists even though the write failed
            failure=classify_exception(exc, FailureStage.DOWNSTREAM_WRITE),
        )

    return ItemResult(record_id=record.record_id, success=True, output=parsed_output)


def parse_model_output(text: str) -> str:
    if not text or not text.strip():
        raise ValueError("empty model output")
    return text.strip()


async def write_result_downstream(record_id: str, output: str) -> None:
    # Placeholder for a real database or storage write.
    pass

The key detail is in the DOWNSTREAM_WRITE failure branch: output=parsed_output is preserved on the ItemResult even though the overall item is marked as failed. This is exactly the information that prevents a wasteful, incorrect retry later — Lesson 8 shows how a retry step can check for this preserved output and skip straight to re-attempting the write, instead of calling the model again for an item whose model output was already obtained successfully.

Quarantining Non-Retryable Failures

Permanent failures — a content policy rejection, a source record missing a required field — should not sit in the same retry queue as transient ones. Routing them to a separate destination (a "needs human review" table or file) makes the distinction operationally real rather than just a label on a log line:

def route_failed_results(results: list[ItemResult]) -> dict[str, list[ItemResult]]:
    retry_queue = []
    quarantine = []

    for result in results:
        if result.success:
            continue
        if result.failure.category == FailureCategory.TRANSIENT:
            retry_queue.append(result)
        else:
            quarantine.append(result)

    return {"retry": retry_queue, "quarantine": quarantine}

Separating these two lists means the pipeline's next run only attempts the items that have a real chance of succeeding, and a human reviewing the quarantine list is looking exclusively at items that genuinely need a data fix or a manual decision — not wading through a mix of both to find the ones worth their attention.

Testing Failure Classification and Routing

Because classify_exception and route_failed_results are pure functions with no I/O, they test cleanly with fake exceptions and fake results:

def test_classify_exception_identifies_transient_errors():
    info = classify_exception(TimeoutError("request timeout"), FailureStage.MODEL_CALL)
    assert info.category == FailureCategory.TRANSIENT
    print("PASS: timeout classified as transient")


def test_route_failed_results_separates_retry_and_quarantine():
    results = [
        ItemResult(record_id="1", success=True, output="ok"),
        ItemResult(
            record_id="2", success=False,
            failure=FailureInfo(FailureStage.MODEL_CALL, FailureCategory.TRANSIENT, "timeout"),
        ),
        ItemResult(
            record_id="3", success=False,
            failure=FailureInfo(FailureStage.MODEL_CALL, FailureCategory.PERMANENT, "policy"),
        ),
    ]
    routed = route_failed_results(results)
    assert len(routed["retry"]) == 1
    assert len(routed["quarantine"]) == 1
    assert routed["retry"][0].record_id == "2"
    assert routed["quarantine"][0].record_id == "3"
    print("PASS: failures routed to retry vs quarantine correctly")


test_classify_exception_identifies_transient_errors()
test_route_failed_results_separates_retry_and_quarantine()

Common Mistakes

• Treating every failure as equally worth retrying (or equally worth giving up on). A content-policy rejection will fail identically on every retry; retrying it wastes time and cost. A network timeout is usually fine on a second attempt. Conflating the two means either endless useless retries or giving up on items that would have succeeded.
• Discarding partial progress on a multi-stage item when only the last stage failed. If the model call succeeded and only the database write failed, redoing the model call on retry is an avoidable cost and, worse, sets up the duplication risk covered in Lesson 8.
• Reporting failures only as a raw count with no breakdown. "150 items failed" tells an operator nothing actionable; "150 items failed: 12 permanent content-policy rejections needing manual review, 138 transient timeouts queued for retry" tells them exactly what to do next.

Best Practices

• Classify every failure by stage and by retry-worthiness at the point it occurs, while you still have the original exception, rather than trying to reconstruct that information later from a generic error string.
• Preserve any partial output already obtained before a later stage fails, so a subsequent retry can resume from the correct step instead of repeating already-successful work.
• Route permanent and transient failures to separate destinations — an automatic retry queue for the former, a human-reviewable quarantine list for the latter — so the systems and people handling each can each see exactly the failures relevant to them.