Library Catalog Data Sync Pipelines

Building Batch Reconciliation Reports for Circulation

This page answers one narrow operational question: why do circulation records silently disappear during a nightly sync batch — where the number received never quite matches the number committed — and how do you build a reconciliation report that makes every lost record provable instead of invisible? It sits under Compliance Reporting for Circulation Data, which covers the reporting contract for circulation pipelines, and within the broader Patron Validation & Privacy Data Routing architecture. If you run a nightly batch that masks, routes and commits circulation events, the drift below is one you will eventually discover the hard way — during an audit, months after the records went missing.

Per-batch reconciliation of circulation sync counters A batch of circulation records enters with a received count. It fans out into four disposition counters — committed, masked-through, quarantined, and intentionally dropped — which all flow into a reconciliation step. The reconciliation step asserts that received equals the sum of the four dispositions. When the sum balances, the batch is marked reconciled and a report is emitted to the audit sink. When the sum does not balance, the batch is flagged as a discrepancy and routed to a review queue, and the same report records the shortfall. records_in received this batch committed written to the ILS masked_through PII masked, then committed quarantined held for human review dropped intentional, with a reason reconcile assert per batch_id: in == c + m + q + d no record unaccounted balanced report to audit sink discrepancy flag + review queue

Problem Framing

The symptom is not a crash. The nightly circulation sync runs to completion, the exit code is zero, and the dashboard turns green. Only later — often at audit time — does someone notice that the number of circulation events the source system handed off does not match the number that landed in the ILS, and there is no record of where the difference went. A batch that received 48,210 events committed 48,187, and the missing 23 left no trace: not in the commit table, not in quarantine, nowhere.

The drift is visible the moment you sum the disposition counters a batch already emits and compare them to the intake count:

text
$ grep '"batch_id":"2026-07-15-night"' /var/log/circsync/batch.jsonl \
    | jq -r 'select(.event=="batch_summary")'
{"event":"batch_summary","batch_id":"2026-07-15-night",
 "records_in":48210,"committed":48187,"masked_through":11902,
 "quarantined":15,"dropped":0}

# received:  48210
# accounted: committed(48187) + quarantined(15) + dropped(0) = 48202
# missing:   8   <-- vanished with no disposition

Eight records were received, were never committed, were never quarantined, and were never deliberately dropped. They simply evaporated somewhere between intake and commit — a masking exception that was swallowed, a database constraint that rejected a row inside a batch the code treated as all-or-nothing, a worker that died mid-flush. Because nothing asserts that the counts must add up, the loss is silent. The batch reports success because success was never defined as conservation.

Root Cause

The failure is the absence of a conservation invariant per batch. Each stage of the pipeline knows its own local outcome — the masker knows it masked a record, the committer knows it wrote one — but no component owns the whole-batch identity that ties intake to final disposition. Every record that enters a batch must leave it through exactly one of a fixed set of exits: it is committed, it is masked and then committed, it is quarantined, or it is intentionally dropped for a stated reason. A record that takes none of those paths is a bug, but nothing in the pipeline is watching for it.

Partial failures are what make this invisible. A try/except around a per-record masking call that logs and continues will drop a record without incrementing any disposition counter. A bulk insert that silently skips conflicting rows commits fewer records than it was given without raising. A worker that crashes after reading a chunk but before flushing it loses that chunk entirely. In each case the batch still finishes, and every counter that does exist is individually correct — the error is that a counter is missing, and you cannot see a number that was never written. Summary logs like the one above look complete precisely because the eye fills in the gap.

The fix is to make conservation an explicit, asserted property of every batch: track a typed counter for each disposition, require that they sum to the intake count, and treat any shortfall as a first-class error that produces a report — not a log line that scrolls away.

Solution

Give each batch a single owner for its counters and an invariant it must satisfy before it is allowed to close. The counters are a typed dataclass so no disposition can be forgotten; incrementing is the only way to move a record out of “received”; and closing the batch runs the reconciliation assertion records_in == committed + masked_through + quarantined + dropped. A mismatch raises, and either outcome renders an auditable report. Records that are deliberately discarded must pass through drop() with a reason, so “intentionally dropped” and “silently lost” are never the same thing.

python
from __future__ import annotations

import json
import logging
from dataclasses import dataclass, asdict, field
from datetime import datetime, timezone
from typing import TextIO

logger = logging.getLogger("circsync.reconcile")


class ReconciliationError(Exception):
    """Raised when a batch's dispositions do not sum to its intake count."""

    def __init__(self, batch_id: str, records_in: int, accounted: int) -> None:
        self.batch_id = batch_id
        self.records_in = records_in
        self.accounted = accounted
        self.missing = records_in - accounted
        super().__init__(
            f"batch {batch_id}: received {records_in}, "
            f"accounted {accounted}, unaccounted {self.missing}"
        )


@dataclass
class BatchCounters:
    """One typed counter per exit path a record may take out of a batch."""

    batch_id: str
    records_in: int
    committed: int = 0
    masked_through: int = 0
    quarantined: int = 0
    dropped: int = 0
    drop_reasons: dict[str, int] = field(default_factory=dict)

    @property
    def accounted(self) -> int:
        return self.committed + self.quarantined + self.dropped

    def commit(self, *, masked: bool) -> None:
        self.committed += 1
        if masked:
            self.masked_through += 1

    def quarantine(self) -> None:
        self.quarantined += 1

    def drop(self, reason: str) -> None:
        """Record an *intentional* discard. A reason is mandatory."""
        if not reason:
            raise ValueError("a dropped record must carry a reason")
        self.dropped += 1
        self.drop_reasons[reason] = self.drop_reasons.get(reason, 0) + 1

    def is_balanced(self) -> bool:
        # masked_through is a subset of committed, so it is not summed here.
        return self.accounted == self.records_in

The reconciliation step is deliberately separate from the counting: closing a batch validates the invariant, emits a structured report to an append-only sink regardless of outcome, and raises on a discrepancy so the failure cannot be ignored by a caller that forgot to check a return value.

python
def reconcile_batch(counters: BatchCounters, report_sink: TextIO) -> dict[str, object]:
    """Assert conservation for one batch and emit an auditable report.

    Writes a report row whether the batch balances or not, then raises
    ReconciliationError on a shortfall so a partial failure halts the caller.
    """
    balanced = counters.is_balanced()
    report: dict[str, object] = {
        "event": "reconciliation_report",
        "batch_id": counters.batch_id,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "balanced": balanced,
        "records_in": counters.records_in,
        "committed": counters.committed,
        "masked_through": counters.masked_through,
        "quarantined": counters.quarantined,
        "dropped": counters.dropped,
        "drop_reasons": dict(counters.drop_reasons),
        "unaccounted": counters.records_in - counters.accounted,
    }

    # The report is itself an auditable artifact: append-only, one JSON line.
    report_sink.write(json.dumps(report, sort_keys=True) + "\n")
    report_sink.flush()

    if not balanced:
        logger.error(
            "batch_reconciliation_failed",
            extra={
                "batch_id": counters.batch_id,
                "records_in": counters.records_in,
                "unaccounted": report["unaccounted"],
            },
        )
        raise ReconciliationError(
            counters.batch_id, counters.records_in, counters.accounted
        )

    logger.info(
        "batch_reconciled",
        extra={"batch_id": counters.batch_id, "records_in": counters.records_in},
    )
    return report

The behavioural change is that “the batch finished” is no longer the same claim as “every record is accounted for.” Before, the eight lost records produced no signal because no code compared intake to disposition. After, reconcile_batch runs that comparison unconditionally, writes a report row that an auditor can read months later, and raises ReconciliationError the instant the sum falls short — turning a silent shortfall into a loud, dated, per-batch_id failure that blocks the run from being marked successful.

A small human-readable renderer on top of the same report data keeps the on-call view honest without ever reopening the raw records:

python
def render_report(report: dict[str, object]) -> str:
    status = "BALANCED" if report["balanced"] else "DISCREPANCY"
    lines = [
        f"batch {report['batch_id']}  [{status}]",
        f"  received       {report['records_in']:>8}",
        f"  committed      {report['committed']:>8}"
        f"  (masked {report['masked_through']})",
        f"  quarantined    {report['quarantined']:>8}",
        f"  dropped        {report['dropped']:>8}  {report['drop_reasons']}",
        f"  unaccounted    {report['unaccounted']:>8}",
    ]
    return "\n".join(lines)

Compliance or Privacy Impact

Reconciliation reports are an aggregate artifact, and they must stay that way. Every field in the report is a count or a reason code — records_in, committed, masked_through, a drop_reasons histogram — and none of them carries a patron identifier, a barcode, or a title. This is deliberate: the report proves that records were conserved without ever re-materializing the records themselves, so it can be retained and read at audit time without widening the PII surface the PII Masking in Patron Data Exports contract works to keep narrow.

Verification

Confirm the invariant with a test that proves a balanced batch passes and an under-counted batch raises with the exact shortfall — and that the report is written in both cases.

python
import io
import pytest


def test_balanced_batch_reconciles() -> None:
    counters = BatchCounters(batch_id="b-ok", records_in=3)
    counters.commit(masked=True)
    counters.commit(masked=False)
    counters.quarantine()
    sink = io.StringIO()

    report = reconcile_batch(counters, sink)

    assert report["balanced"] is True
    assert report["unaccounted"] == 0
    assert report["masked_through"] == 1
    # The report was persisted even on success.
    assert '"reconciliation_report"' in sink.getvalue()


def test_missing_records_raise_with_exact_shortfall() -> None:
    counters = BatchCounters(batch_id="b-lossy", records_in=10)
    for _ in range(7):
        counters.commit(masked=False)
    # Two records were quarantined; one vanished with no disposition.
    counters.quarantine()
    counters.quarantine()
    sink = io.StringIO()

    with pytest.raises(ReconciliationError) as exc:
        reconcile_batch(counters, sink)

    assert exc.value.missing == 1
    # A discrepancy report is still written for the auditor.
    assert '"balanced": false' in sink.getvalue().lower()


def test_dropped_records_require_a_reason() -> None:
    counters = BatchCounters(batch_id="b-drop", records_in=1)
    with pytest.raises(ValueError):
        counters.drop("")  # a silent drop is not allowed

For a running pipeline, wire the assertion into the batch’s exit path so no run can be marked successful without a balanced report: call reconcile_batch in a finally-guarded close step and let ReconciliationError fail the job. Then track the count of discrepancy reports over time — a nonzero rate is your early signal that a stage is swallowing records, exactly the way a rising unaccounted total surfaced the eight lost events above.