Troubleshooting Ingestion Pipeline Failures

Every royalty ops team eventually gets the 3 a.m. page: a DSP batch that silently truncated, a cursor that stopped advancing, or a worker that died mid-parse with no stack trace worth reading. This reference collects the failure patterns that recur most often in production ingestion, as part of the Data Ingestion & Streaming Sync Pipelines guidance, and gives each one a diagnostic snippet and a concrete fix rather than a generic “check the logs” shrug. The goal is to get from symptom to root cause in minutes, because every hour a batch sits stuck is an hour of settlement data drifting further from what the DSP actually reported.

Most of these failures fall into six buckets: schema drift from a DSP that changed its ERN or CSV shape without notice, malformed payloads that pass a superficial check but fail deep validation, out-of-memory kills on large batch runs, 429 storms from overly aggressive polling, duplicate or partial batches from retried jobs, and cursors that get stuck pointing at a record that will never resolve. Each section below assumes you already have structural validation in place — see schema validation with Pydantic and automated CSV parsing for sales reports for how that layer is built — and focuses on what to do when it fails anyway.

Why is a DSP feed suddenly rejecting on fields that validated fine last week?

Schema drift is the single most common cause of an ingestion job that worked yesterday and fails today. DSPs revise their ERN profile or CSV column set without a formal changelog; a territory_code column gets renamed, a nullable field starts arriving as an empty string instead of being omitted, or a new repeating element appears in the ERN SalesTransaction block. The fastest way to confirm drift is to diff the incoming payload’s structural fingerprint against the last known-good batch:

import hashlib
import json
from collections import OrderedDict

def schema_fingerprint(record: dict) -> str:
    """Order-independent fingerprint of a record's key structure and types."""
    shape = OrderedDict(
        sorted((key, type(value).__name__) for key, value in record.items())
    )
    payload = json.dumps(shape, sort_keys=True).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()

def detect_drift(current_batch: list[dict], known_fingerprints: set[str]) -> list[dict]:
    drifted = [r for r in current_batch if schema_fingerprint(r) not in known_fingerprints]
    return drifted

Once drift is confirmed, do not patch the parser in place. Version the schema (ern_profile_v43 vs ern_profile_v44), route unrecognized fingerprints to a quarantine table, and only promote a new fingerprint to “known-good” after a human has reviewed a sample of the drifted records against the DSP’s published DDEX ERN profile.

Why does a batch fail deep validation even though it passed the initial ingest check?

Malformed CSV and ERN payloads often pass a shallow check — file opens, row count looks sane, XML is well-formed — but fail on semantic constraints: a right_share_pct column that sums to 97.5 instead of 100, an isrc that doesn’t match the 12-character DDEX pattern, or a CSV that silently switched delimiter from comma to semicolon partway through because it was concatenated from two source exports. Detect the second case by sampling delimiter consistency across the file rather than trusting the header row alone:

import csv
from io import StringIO

def detect_delimiter_drift(raw_text: str, expected_delimiter: str = ",") -> bool:
    sample_lines = raw_text.splitlines()[:200]
    counts = {line.count(expected_delimiter) for line in sample_lines if line.strip()}
    return len(counts) > 1  # inconsistent field counts across sampled rows

For ERN payloads, validate against the DDEX XSD before touching business rules — a file that fails XSD validation should never reach the split-sum or ISRC checks, because the parse itself is unreliable. lxml.etree.XMLSchema gives you a precise line/column error, which is worth logging verbatim into the quarantine record.

Why did last night’s ERN batch get OOM-killed?

Large ERN or CSV files loaded fully into memory — etree.parse() on a 4 GB sales report, or pandas.read_csv() without chunking — will exhaust container memory limits long before the file finishes parsing, because DOM parsing and dataframe construction both hold the entire structure plus per-node/per-cell overhead in RAM simultaneously. The kernel’s OOM killer terminates the worker with SIGKILL, leaving no Python traceback, which is why these incidents are so often misdiagnosed as “random” crashes. The fix is streaming parsing rather than full-document loads; the walkthrough on diagnosing OOM during large ERN batch runs covers iterparse, element clearing, and RSS-based leak detection in depth. As a quick triage step, check dmesg for an oom-kill entry matching the worker’s PID and timestamp before assuming the failure is a code bug rather than a memory ceiling.

Why is the DSP API returning waves of 429s that get worse the more we retry?

A naive retry loop that resends immediately on a 429 turns a brief rate-limit window into a sustained storm, because each retry adds to the same congested window the DSP is trying to shed load from. The retry itself becomes the outage. Confirm this pattern by checking whether your retry interval is shorter than the Retry-After header the DSP is actually sending:

import httpx

def diagnose_retry_storm(response: httpx.Response, configured_backoff_seconds: float) -> bool:
    retry_after = float(response.headers.get("Retry-After", "0"))
    return configured_backoff_seconds < retry_after

The durable fix is exponential backoff with jitter, tied to the Retry-After header rather than a fixed schedule, and a circuit breaker that stops issuing new requests once the rejection rate crosses a threshold within a rolling window. The full pattern — including per-DSP rate budgets — is covered in error handling and retry mechanisms.

How do duplicate or partial batches end up in the reconciliation tables?

Duplicates usually come from a retried job that succeeded on the DSP side before the client-side timeout fired, so the orchestrator reschedules a batch that was, in fact, already committed. Partial batches come from a worker that died after writing some rows but before marking the batch complete. Both are detectable by comparing the expected row count from the batch manifest against what actually landed:

def audit_batch_completeness(batch_id: str, manifest_row_count: int, committed_row_count: int) -> str:
    if committed_row_count == 0:
        return "not_started"
    if committed_row_count < manifest_row_count:
        return "partial"
    if committed_row_count > manifest_row_count:
        return "duplicate_rows"
    return "complete"

Prevent recurrence with an idempotent UPSERT keyed on (batch_id, isrc, reporting_period) rather than a plain INSERT, and mark batch completion only after a row-count assertion passes — not merely after the write call returns without raising.

Why has a sync cursor stopped advancing even though the job reports success?

A stuck cursor almost always means the job is catching an exception on one record, logging it, and then re-fetching from the same offset on the next run instead of past it — the batch “succeeds” in the sense that it doesn’t crash, but the cursor position never moves. This is common when a single poison record (a malformed iswc, a null payee_id that breaks a downstream join) sits at the head of the window on every attempt.

def diagnose_stuck_cursor(cursor_history: list[tuple[str, str]]) -> bool:
    """cursor_history: list of (run_timestamp, cursor_value) tuples, most recent last."""
    if len(cursor_history) < 3:
        return False
    recent_values = {value for _, value in cursor_history[-3:]}
    return len(recent_values) == 1

The fix is to advance the cursor past a record after it has been quarantined, not only after it has been processed successfully — otherwise a single bad row permanently blocks every record behind it.

Related

For the shape checks that catch drift and malformed payloads before they reach these failure modes, see schema validation with Pydantic and error handling and retry mechanisms; for the CSV-specific delimiter and encoding issues referenced above, see automated CSV parsing for sales reports. When the failure is specifically memory exhaustion on large batches, the OOM diagnosis walkthrough goes deeper into streaming parse strategies, and both pages sit under the broader Data Ingestion & Streaming Sync Pipelines reference.