Catalog Ingestion & ILS Sync Pipelines: Architecture, Idempotency, and Production Patterns

Catalog ingestion is the discipline of moving bibliographic and holdings data from heterogeneous external sources into a canonical internal model, then synchronizing validated changes back into an Integrated Library System (ILS) without corrupting circulation state. This domain is written for library technical staff, ILS administrators, public sector developers, and Python automation engineers who own the machinery between vendor feeds and the public catalog. It sits alongside two companion domains on this site: the Core Architecture & Catalog Standards domain, which defines the schema and translation contracts every ingestion job depends on, and the Patron Validation & Privacy Data Routing domain, which governs how identity and circulation data are masked and routed. Ingestion reliability is the difference between a discovery layer that reflects the shelf and one that quietly drifts out of sync — duplicated holdings, stale availability, or lost authority linkages that take weeks to notice and days to unwind.

The pages below decompose ingestion into four implementation areas — record parsing, schema validation, asynchronous batch synchronization, and rate-limited API integration — each with its own deep-dive. This overview establishes the boundaries between them, the idempotency guarantees that hold the system together under retries and network partitions, and the compliance checkpoints that must be enforced before any record crosses into the catalog.

Domain Isolation & Data Flow Topology

The first architectural principle of catalog ingestion is strict domain isolation. Bibliographic records, authority files, and holdings operate under a different consistency model than circulation transactions, patron accounts, and financial ledgers. A production pipeline enforces a unidirectional flow for bibliographic normalization — external source to canonical model to ILS — while keeping any circulation-adjacent channel narrow, mediated, and privacy-scrubbed. Collapsing these two flows into a single bidirectional integration is the most common source of transactional anomalies, because a retry on the bibliographic side can then race a checkout on the circulation side and clobber patron-facing state.

Ingress arrives in several shapes. OAI-PMH harvests deliver incremental MARCXML over HTTP with resumption tokens. SFTP vendor drops deposit fixed-length MARC21 (.mrc) files on a schedule. Proprietary ILS REST endpoints expose JSON payloads that must be polled within vendor rate limits. Each transport produces heterogeneous payloads that share nothing except the requirement to become a single canonical internal model before crossing into the staging layer. That normalization begins with deterministic record parsing — Parsing MARC Records with pymarc covers leader-byte interpretation, field and subfield extraction, and encoding handling, and is the foundation every downstream stage assumes. When a record set is large enough that a full in-memory parse would exhaust the worker, defer to the streaming techniques in Optimizing pymarc Performance for Large Record Sets.

Egress routes toward the ILS core, the discovery layer, and analytics warehouses. The decoupling mechanism between ingress and egress is a message broker with durable queues. Producers (parsers) never call the ILS directly; they emit normalized deltas onto a topic, and consumers (sync workers) apply those deltas with ordering and idempotency guarantees. This asynchronous seam is what lets heavy transformation workloads run without blocking real-time OPAC queries or self-checkout terminals. The mechanics of that seam — aggregation windows, ordering keys, and conflict resolution — are detailed in Async Batch Processing for Catalog Updates, and when the broker fans work out to a distributed pool, Using Celery for Distributed Catalog Ingestion shows how to keep task routing and acknowledgement semantics correct.

Where the pipeline pulls from a live ILS rather than a batch drop, network behaviour becomes a first-class design concern. ILS REST API Polling & Rate Limiting governs how the pipeline paces requests against vendor quotas, and vendor-specific tuning — for example, Configuring Exponential Backoff for Sierra API Calls — keeps a burst of retries from tripping the vendor’s own throttling. The vendor payload quirks that survive polling are best neutralized at the boundary through the normalization layer described in ILS Schema Translation Patterns, so the rest of the pipeline only ever sees the canonical model.

Ingress transport contracts

Each transport imposes its own delivery guarantees, and the pipeline must map every one onto the same internal contract before records enter the broker.

Schema Interoperability & Transformation Standards

Bibliographic data remains anchored in MARC21, while linked-data adoption pushes institutions toward BIBFRAME 2.0. An ingestion pipeline has to treat both as first-class citizens without letting either leak its representation into downstream services. The canonical internal model is a neutral intermediate: deterministic mapping rules convert fixed-length control fields, variable-length data fields, indicators, and repeatable subfields into typed structures, and the reverse mapping reconstitutes valid MARC or BIBFRAME on egress. The field-level rules for that translation are documented in MARC21 Field Mapping for Modern Pipelines, and the harder graph-shaped conversions — work, instance, and item relationships plus authority linkage — belong to the BIBFRAME to MARC21 Conversion Workflows layer.

Normalization is only trustworthy if it is validated. Before any normalized record is allowed onto the broker, it passes through a validation gate that enforces mandatory field presence, subfield delimiter integrity, controlled-vocabulary membership, and institutional policy constraints. Schema Validation for Ingested Records implements that gate with JSON Schema (or XSD for MARCXML), and the most failure-prone piece — the fixed-length leader — has its own procedure in Validating MARC Leader Fields Before Database Insert. A record that fails validation is never silently dropped and never partially applied; it is routed to a quarantine queue with the specific rule it violated, so a cataloger can remediate the source rather than the pipeline guessing at intent.

Two normalization contracts are worth making explicit. First, the mapping must be total — every source field either maps to a canonical field, maps to a documented catch-all (for example the local 9XX range), or is explicitly discarded with a logged reason. Silent field loss is the failure mode that erodes trust in a catalog over months. Second, the mapping must be reversible where it claims to be: if the pipeline advertises round-trip fidelity for a set of fields, an integration test must prove that MARC in equals MARC out for that set. These contracts are what let the ILS Schema Translation Patterns layer absorb a new vendor without ripple effects elsewhere.

Idempotent Sync Patterns & Production Implementation

Idempotency is non-negotiable in library automation. Network partitions, duplicate broker deliveries, operator-triggered replays, and out-of-order arrivals must never produce duplicate holdings, overwritten timestamps, or corrupted item counts. The pipeline achieves this by deriving a deterministic idempotency key for every delta from a stable triple — the record control number, the source modification timestamp, and a checksum of the mapped fields — and by using upsert semantics keyed on the control number rather than blind inserts. A retried request carrying an already-applied key is a safe no-op; a delta whose timestamp is older than the currently stored record is rejected rather than applied, which is the version-aware conflict resolution that protects newer state from a late-arriving stale message.

The pattern below is a production-grade async synchronization worker. It uses an async context manager for connection lifecycle, typed dataclasses for the delta contract, explicit retry classification, structured logging, and a privacy scrubber that runs before any field leaves the process. It routes exhausted retries to a quarantine callback rather than losing them.

from __future__ import annotations

import asyncio
import hashlib
import logging
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Awaitable, Callable

import httpx
from tenacity import (
    AsyncRetrying,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)

logger = logging.getLogger("ils_sync_pipeline")

# Deny-by-default: only bibliographic/holdings fields cross into the catalog.
PII_SENSITIVE_KEYS: frozenset[str] = frozenset(
    {"patron_id", "reading_history", "fine_balance", "hold_queue", "email"}
)


@dataclass(frozen=True, slots=True)
class CatalogDelta:
    """One validated, normalized change destined for the ILS."""

    control_number: str
    modified_timestamp: str  # ISO-8601, from the source of truth
    fields: dict[str, Any]
    pii_markers: list[str] = field(default_factory=list)


def generate_idempotency_key(delta: CatalogDelta) -> str:
    """Stable key from control number + source timestamp + field checksum.

    Deterministic across processes and retries, so an at-least-once broker
    delivering the same delta twice yields the same key and a safe no-op.
    """
    canonical = repr(sorted(delta.fields.items())).encode("utf-8")
    field_hash = hashlib.sha256(canonical).hexdigest()[:16]
    return f"{delta.control_number}:{delta.modified_timestamp}:{field_hash}"


def enforce_privacy_boundaries(record: dict[str, Any]) -> dict[str, Any]:
    """Tokenize any circulation-adjacent PII before the record leaves the worker."""
    sanitized: dict[str, Any] = {}
    for key, value in record.items():
        if key.lower() in PII_SENSITIVE_KEYS:
            digest = hashlib.sha256(str(value).encode("utf-8")).hexdigest()[:12]
            sanitized[key] = f"tok_{digest}"
        else:
            sanitized[key] = value
    return sanitized


QuarantineHandler = Callable[[CatalogDelta, Exception], Awaitable[None]]


class IdempotentILSSync:
    """Applies catalog deltas to an ILS with idempotent, retried upserts."""

    def __init__(
        self,
        client: httpx.AsyncClient,
        base_url: str,
        quarantine: QuarantineHandler,
        max_attempts: int = 5,
    ) -> None:
        self._client = client
        self._base_url = base_url.rstrip("/")
        self._quarantine = quarantine
        self._max_attempts = max_attempts

    async def _put_once(self, delta: CatalogDelta) -> httpx.Response:
        payload = {
            "control_number": delta.control_number,
            "data": enforce_privacy_boundaries(delta.fields),
        }
        headers = {"Idempotency-Key": generate_idempotency_key(delta)}
        response = await self._client.put(
            f"{self._base_url}/catalog/records/{delta.control_number}",
            json=payload,
            headers=headers,
        )
        response.raise_for_status()
        return response

    async def upsert(self, delta: CatalogDelta) -> None:
        """Retry transient failures with jittered backoff; quarantine on exhaustion."""
        retryer = AsyncRetrying(
            stop=stop_after_attempt(self._max_attempts),
            wait=wait_exponential_jitter(initial=2, max=30),
            retry=retry_if_exception_type(
                (httpx.TransportError, httpx.HTTPStatusError)
            ),
            reraise=True,
        )
        try:
            async for attempt in retryer:
                with attempt:
                    await self._put_once(delta)
            logger.info(
                "synced control_number=%s idem_key=%s",
                delta.control_number,
                generate_idempotency_key(delta),
            )
        except (httpx.TransportError, httpx.HTTPStatusError) as exc:
            logger.error(
                "quarantining control_number=%s after %d attempts: %s",
                delta.control_number,
                self._max_attempts,
                exc,
            )
            await self._quarantine(delta, exc)

    async def process_batch(self, deltas: list[CatalogDelta]) -> None:
        await asyncio.gather(*(self.upsert(d) for d in deltas))


@asynccontextmanager
async def open_sync(
    base_url: str, api_key: str, quarantine: QuarantineHandler
) -> AsyncIterator[IdempotentILSSync]:
    """Own the HTTP client lifecycle so sockets are always released."""
    headers = {"Authorization": f"Bearer {api_key}"}
    async with httpx.AsyncClient(headers=headers, timeout=15.0) as client:
        yield IdempotentILSSync(client, base_url, quarantine)

Three properties make this safe in production. The Idempotency-Key header lets the ILS collapse duplicate deliveries server-side, so at-least-once broker semantics become effectively exactly-once at the business layer. wait_exponential_jitter spreads retry timing so a broker replay of a thousand deltas does not synchronize into a thundering herd against the vendor endpoint — the same reasoning that drives Configuring Exponential Backoff for Sierra API Calls. And the quarantine callback guarantees that a delta which cannot be applied is preserved with its failure context rather than dropped, which is the invariant the recovery section below depends on.

Compliance & Privacy Architecture

Privacy is not a downstream cleanup step; it is a gate positioned at the exact boundary where a record could first carry patron-identifying data into the catalog domain. Any circulation-adjacent field — patron identifiers, reading history, hold queues, fine balances, contact details — must be tokenized or irreversibly stripped before it crosses into catalog synchronization. The scrubber operates deny-by-default: only explicitly enumerated bibliographic and holdings fields are allowed through, and everything else is either dropped or replaced with a non-reversible token. This is the pipeline-side enforcement of the rules laid out in Data Privacy Boundaries in Library Systems, and it is the same masking contract that PII Masking in Patron Data Exports applies on the export side.

Regulatory obligations shape the design directly. FERPA constrains how student-patron records may be linked and disclosed; GDPR grants a right to erasure that a pipeline must be able to honour without breaking referential integrity in the catalog; and state-level library confidentiality statutes frequently forbid retaining borrower history beyond the return of an item. The pipeline satisfies these by keeping identity data out of the bibliographic flow entirely and by routing anything that must retain a patron linkage through the dedicated channels described in Circulation History Routing & Anonymization. Retention itself — how long a masked record may live and when it must be purged — follows the deterministic lifecycle rules in Data Retention Policies for Public Libraries.

Every stage of the pipeline emits an append-only audit record so that a compliance review can reconstruct exactly what was transformed, masked, or rejected, and when. A minimal audit event carries enough to prove the privacy gate ran without itself storing any PII.

Because the audit log is the evidence trail for a regulator, it must be written on the same transaction boundary as the action it describes — an event that records a sync but whose sync later fails is worse than no event at all. Keep the log free of PII by design: reference records by control number, reference jobs by service-account id, and record the action taken on patron data (tokenized, stripped) rather than the data itself.

Operational Failure Modes & Recovery

Production pipelines are defined by how they behave when a dependency degrades, not by their happy path. The recurring failure modes in catalog ingestion are vendor outages, malformed source records, broker backpressure, and poisoned messages that fail deterministically on every attempt. Each has a distinct containment strategy, and none is allowed to lose data.

Malformed and policy-violating records are contained by the validation quarantine described in Schema Validation for Ingested Records: the record is set aside with the exact rule it broke, and the batch proceeds. Deltas that pass validation but fail to apply after exhausting retries — the case the sync worker’s quarantine callback handles — go to a dead-letter queue that preserves the original message, the idempotency key, and the terminal exception, so a poisoned message never blocks the ordered stream behind it. Recovery from a dead-letter queue is a deliberate, replayable operation rather than an automatic loop, and the backpressure and recovery mechanics belong to Async Batch Processing for Catalog Updates.

Vendor outages and maintenance windows are contained by a circuit breaker in front of every ILS endpoint. When failures cross a threshold the breaker opens, the pipeline stops hammering a dead endpoint, and buffered deltas accumulate in durable queues to be replayed in chronological order once the breaker half-opens and probes succeed. The breaker implementation and its tuning are covered in Implementing Circuit Breakers for ILS API Timeouts; the rate-limiting that keeps the pipeline from causing the outage in the first place lives in ILS REST API Polling & Rate Limiting.

Rollback is the last line of defence. Every catalog deployment should retain an immutable snapshot of pre-sync state for the records a batch touches, so that a faulty normalization rule — one that, say, mislinks an authority heading across a consortium — can be reverted to a known-good bibliographic baseline without re-harvesting from source. Alerting closes the loop: quarantine depth, dead-letter growth rate, breaker state transitions, and end-to-end delta latency are the four signals that tell an operator the pipeline is degrading before patrons notice a stale catalog. Operator controls must be able to halt ingestion at the broker level without losing in-flight state, which is what makes safe schema migrations and vendor API cutovers possible.

Reference Implementation Checklist

Use this as a per-deployment gate. Each item maps to one architectural layer above; none is optional in a system that feeds a live discovery layer.

• Ingress transports normalized to a single canonical model before the broker
• Deterministic record parsing with streaming for large .mrc files
• Schema validation gate enforcing leader, mandatory fields, and controlled vocabularies
• Failing records routed to a quarantine queue with the violated rule attached
• Privacy scrubber running deny-by-default before any field leaves the worker
• Idempotency key derived from control number + source timestamp + field checksum
• Upsert semantics with version-aware rejection of stale deltas
• Retries using exponential backoff with jitter, classified by exception type
• Dead-letter queue preserving message, idempotency key, and terminal exception
• Circuit breaker in front of every ILS endpoint, with buffered chronological replay
• Append-only, PII-free audit log written on the same boundary as each action
• Immutable pre-sync snapshots enabling rollback to a known-good baseline
• Alerting on quarantine depth, dead-letter growth, breaker state, and delta latency
• Broker-level halt control that pauses ingestion without losing in-flight state

Related

• Parsing MARC Records with pymarc — leader and field extraction that seeds the canonical model
• Schema Validation for Ingested Records — the validation and quarantine gate before the broker
• Async Batch Processing for Catalog Updates — broker seam, ordering, and backpressure
• ILS REST API Polling & Rate Limiting — pacing pulls against vendor quotas
• Core Architecture & Catalog Standards — the schema and translation contracts ingestion depends on
• Patron Validation & Privacy Data Routing — masking and routing rules for circulation-adjacent data