Compliance Reporting for Circulation Data
Operating within the broader Patron Validation & Privacy Data Routing architecture, this guide covers the layer most circulation pipelines bolt on last and regret first: the reporting apparatus that has to prove, not merely assert, that every patron record moving through the system was handled the way FERPA, GDPR, and state library-record confidentiality law require. A masking stage that works is not the same as a masking stage you can defend to an auditor, a records officer, or a court order. The difference is evidence — an append-only, tamper-evident trail of what happened to which record, in which batch, under which disposition, plus batch-level reconciliation that shows the numbers add up and retention exports scoped to the jurisdiction that demanded them. This page walks through the audit-event contract, a hash-chained log schema, a production-grade emitter and reconciliation report generator, the checkpoints that keep the reports themselves from leaking the data they describe, and the error-handling, performance, and verification patterns that make the evidence trustworthy at scale.
Specification & Contract
A compliance report is a contract between the pipeline and everyone who might later doubt it: an auditor confirming FERPA controls, a data-protection officer answering a GDPR access request, a state library commission checking that patron circulation records were purged on the statutory clock. The contract has two obligations. First, the pipeline must record what it did to every record, in a form that cannot be quietly edited after the fact. Second, it must be able to summarize and export those records without ever re-exposing the patron data the whole exercise was meant to protect.
The foundation is the audit event — one immutable row per meaningful action a record undergoes. Each row is bound to the row before it by a hash chain: the emitter computes a digest over the event’s own fields plus the previous row’s digest, so a single altered or deleted row breaks every link after it and the tamper is detectable by recomputation alone. This is what turns an ordinary log table into evidence.
| Column | Type | Description |
|---|---|---|
event_id |
UUIDv7 | Monotonic, time-ordered primary key; sortable by creation without a separate sequence |
occurred_at |
timestamptz (UTC) | Instant the action completed; never local time, never nullable |
actor |
text | The service account or job name that performed the action, not a patron |
action |
enum | RECEIVED, MASKED, QUARANTINED, EXPORTED, PURGED — a closed vocabulary |
entity_type |
enum | patron, checkout, hold, fine — the kind of record acted on |
entity_id_hashed |
text | HMAC-SHA-256 surrogate of the record id; never the raw identifier |
batch_id |
UUID | Groups every event from one pipeline run for reconciliation |
disposition |
enum | ok, masked, quarantined, suppressed — the outcome for this record |
prev_hash |
text (hex) | SHA-256 over this row’s canonical fields plus the previous row’s hash |
Three rules govern the schema. First, no column ever holds raw patron data — the record being described is referenced only by entity_id_hashed, a domain-separated HMAC surrogate produced the same way the PII Masking in Patron Data Exports stage produces its join keys, so the audit log cannot become a shadow copy of the very data it audits. Second, the log is append-only at every layer: no UPDATE, no DELETE, enforced with database grants and, ideally, write-once storage, because a mutable audit log is not evidence. Third, prev_hash chains the whole table in insertion order — the reconciliation and retention reports downstream are only as trustworthy as a fresh chain-recomputation says they are, so verification is not optional cleanup but a precondition of trusting any report built on top.
The event vocabulary is deliberately small. A record is RECEIVED when it enters the pipeline, MASKED or QUARANTINED depending on whether it cleared validation, EXPORTED when it leaves in a report, and PURGED when retention enforcement deletes it. The full schema design — indexing strategy, chain-segment checkpoints, and the migration that makes the table append-only — is the subject of Designing an Append-Only Audit Log Schema.
Prerequisites & Environment Setup
The examples target Python 3.11+ with the standard library’s hashlib, hmac, and dataclasses — no ORM is assumed, because the audit writer must be simple enough to reason about line by line. The chaining secret is resolved at runtime from a secrets manager, never a literal, and the audit store is a table or object bucket on which the pipeline’s role has been granted INSERT and SELECT only. If the same role can UPDATE or DELETE the audit table, the hash chain protects against nothing an attacker with that role could not simply rewrite.
Core Implementation
The reporting layer runs in two movements: an emitter that appends one tamper-evident audit row per action during the pipeline run, and a report generator that reads a finished batch back and reconciles the counts. Keeping the two apart is what keeps the evidence honest — the emitter only ever writes, the generator only ever reads, and neither can quietly change what the other recorded.
Step 1 — Model the audit event as a typed record
Nothing is emitted until it is well-typed. A frozen dataclass makes the event immutable in memory and gives the hash function a stable, ordered set of fields to digest. The entity_id_hashed is already a surrogate by the time it reaches this constructor; the raw id never enters the audit path.
from __future__ import annotations
import hashlib
import hmac
import json
import logging
import uuid
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from enum import Enum
audit_logger = logging.getLogger("circ_compliance_audit")
GENESIS_HASH = "0" * 64
class Action(str, Enum):
RECEIVED = "RECEIVED"
MASKED = "MASKED"
QUARANTINED = "QUARANTINED"
EXPORTED = "EXPORTED"
PURGED = "PURGED"
class Disposition(str, Enum):
OK = "ok"
MASKED = "masked"
QUARANTINED = "quarantined"
SUPPRESSED = "suppressed"
@dataclass(frozen=True, slots=True)
class AuditEvent:
"""One immutable row in the hash-chained circulation audit log."""
event_id: str
occurred_at: str
actor: str
action: Action
entity_type: str
entity_id_hashed: str
batch_id: str
disposition: Disposition
prev_hash: str
def chain_hash(self) -> str:
"""SHA-256 over this row's canonical fields plus the prior hash."""
canonical = json.dumps(
{
"event_id": self.event_id,
"occurred_at": self.occurred_at,
"actor": self.actor,
"action": self.action.value,
"entity_type": self.entity_type,
"entity_id_hashed": self.entity_id_hashed,
"batch_id": self.batch_id,
"disposition": self.disposition.value,
"prev_hash": self.prev_hash,
},
sort_keys=True,
separators=(",", ":"),
)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
Pitfall: the canonical serialization must be stable. Using asdict() and hashing its default repr, or letting json.dumps emit keys in insertion order, means the same logical event can hash two ways across Python versions — and a chain that cannot be recomputed identically is a chain that always looks tampered.
Step 2 — Hash the entity id, never carry it raw
The audit log references a patron or checkout only through a domain-separated HMAC surrogate, resolved from the same runtime key discipline the masking stage uses. Domain separation keeps an audit surrogate from colliding with an export token for the same underlying id.
class AuditKeys:
"""Runtime-resolved HMAC material for entity-id surrogates."""
def __init__(self, key: bytes, domain: str) -> None:
if len(key) < 32:
raise ValueError("AUDIT_HMAC_KEY must be at least 32 bytes")
self._key = key
self._domain = domain
def entity_surrogate(self, entity_type: str, raw_id: str) -> str:
"""Domain-separated HMAC-SHA-256 surrogate for a record id."""
msg = f"{self._domain}\x1f{entity_type}\x1f{raw_id}".encode("utf-8")
return hmac.new(self._key, msg, hashlib.sha256).hexdigest()
Pitfall: do not reuse the masking-export salt as the audit key. If the two share a secret, an analyst holding an export token could confirm whether a specific patron appears in the audit log — a cross-report correlation the domain separation exists to prevent.
Step 3 — Append a chained event under a context manager
The emitter reads the previous row’s hash, builds the event, computes its chain hash, and appends atomically. A context manager owns the store connection so a mid-batch failure cannot leave a half-written row that breaks the chain for every run that follows.
from contextlib import contextmanager
from typing import Iterator, Protocol
class AuditStore(Protocol):
def last_hash(self) -> str: ...
def append(self, event: AuditEvent, chain_hash: str) -> None: ...
@contextmanager
def audit_session(store: AuditStore, keys: AuditKeys, *, actor: str,
batch_id: str) -> Iterator["AuditEmitter"]:
emitter = AuditEmitter(store, keys, actor=actor, batch_id=batch_id)
try:
yield emitter
except Exception:
audit_logger.error(
"audit session aborted",
extra={"batch_id": batch_id, "written": emitter.count},
)
raise
class AuditEmitter:
def __init__(self, store: AuditStore, keys: AuditKeys, *,
actor: str, batch_id: str) -> None:
self._store = store
self._keys = keys
self._actor = actor
self._batch_id = batch_id
self.count = 0
def record(self, action: Action, entity_type: str, raw_id: str,
disposition: Disposition) -> AuditEvent:
event = AuditEvent(
event_id=str(uuid.uuid4()),
occurred_at=datetime.now(timezone.utc).isoformat(),
actor=self._actor,
action=action,
entity_type=entity_type,
entity_id_hashed=self._keys.entity_surrogate(entity_type, raw_id),
batch_id=self._batch_id,
disposition=disposition,
prev_hash=self._store.last_hash(),
)
chain_hash = event.chain_hash()
self._store.append(event, chain_hash)
self.count += 1
audit_logger.info(
"audit event appended",
extra={
"batch_id": self._batch_id,
"action": action.value,
"disposition": disposition.value,
"chain_hash": chain_hash[:12],
},
)
return event
Pitfall: prev_hash must be read inside the append, not cached at session start. If two workers append to the same log concurrently off a stale last_hash, they fork the chain. Serialize appends per log (a single-writer worker, or a row lock on the tail) — the audit log is one place where throughput yields to correctness.
Step 4 — Generate the batch reconciliation report
Once a batch closes, the generator reads every event for that batch_id and reconciles the flow: how many records came in, how many were masked, how many were quarantined. The three must account for every received record, or the batch is not provably complete.
@dataclass(frozen=True, slots=True)
class ReconciliationReport:
batch_id: str
records_in: int
records_masked: int
records_quarantined: int
balanced: bool
def reconcile_batch(events: list[AuditEvent], batch_id: str) -> ReconciliationReport:
"""Count in vs masked vs quarantined for one batch and check the balance."""
received = sum(1 for e in events if e.action is Action.RECEIVED)
masked = sum(1 for e in events if e.action is Action.MASKED)
quarantined = sum(1 for e in events if e.action is Action.QUARANTINED)
balanced = received == masked + quarantined
if not balanced:
audit_logger.error(
"batch reconciliation imbalance",
extra={
"batch_id": batch_id,
"records_in": received,
"records_masked": masked,
"records_quarantined": quarantined,
"unaccounted": received - (masked + quarantined),
},
)
return ReconciliationReport(
batch_id=batch_id,
records_in=received,
records_masked=masked,
records_quarantined=quarantined,
balanced=balanced,
)
Pitfall: reconcile from the audit log, not from the pipeline’s own in-memory counters. Counters can be wrong in exactly the way the report exists to catch; reading the counts back out of the immutable log is what makes the reconciliation independent evidence rather than a restatement of the code’s assumptions. The full report format, per-disposition breakdowns, and the CSV/JSON emitters are developed in Building Batch Reconciliation Reports for Circulation.
PII & Compliance Checkpoints
The reporting layer inverts the usual risk: its whole purpose is to describe patron data, so it is uniquely positioned to re-leak it. Three checkpoints keep the reports from becoming the breach they were meant to prevent.
No raw identifiers in any report. Every id that appears in an audit row, a reconciliation summary, or a retention export is a hashed surrogate or an aggregate count — never a patron name, barcode, or email. This is the same hard boundary the PII Masking in Patron Data Exports stage enforces on outbound feeds, applied to the evidence trail itself: a compliance report that quotes a raw identifier has failed the standard it documents.
Aggregate before you export. Reconciliation reports carry counts, not rows. Where a report must be broken down by branch or patron type, generalize any dimension that could isolate an individual, keeping cells above a small-count threshold rather than publishing a count = 1 that names a person by inference. The retention clock those exports answer to is set in Data Retention Policies for Public Libraries, and the erasure obligations that intersect it are handled in Patron Consent and Right-to-Erasure Management.
Audit completeness. Every record that enters the pipeline must produce a RECEIVED event and a terminal MASKED or QUARANTINED event carrying the same batch_id. A batch whose audit log has fewer terminal events than received events is not defensible, because you cannot demonstrate the missing records were handled rather than silently dropped — which is exactly the imbalance reconcile_batch is built to surface.
Error Handling & Quarantine Patterns
The reporting layer distinguishes two failure classes and never lets either corrupt the evidence. A single record that cannot be audited is a per-record fault to be quarantined; a broken hash chain or an unresolvable key is a batch-level fault that must stop everything, because every report built after it would be untrustworthy.
class AuditIntegrityError(Exception):
"""The hash chain is broken; no report on this log can be trusted."""
class RecordAuditError(Exception):
"""A single record could not be audited and must be quarantined."""
def verify_chain(events: list[AuditEvent]) -> None:
"""Recompute prev_hash links; raise on the first broken link."""
expected_prev = GENESIS_HASH
for event in events:
if event.prev_hash != expected_prev:
raise AuditIntegrityError(
f"chain break at event_id={event.event_id}"
)
expected_prev = event.chain_hash()
def audit_record(emitter: AuditEmitter, record: dict, quarantine_sink) -> None:
try:
emitter.record(
Action.RECEIVED, record["entity_type"], record["id"], Disposition.OK
)
except (KeyError, ValueError) as exc:
# A malformed record cannot be audited; divert it, keep the batch flowing.
quarantine_sink.put({"reason": str(exc), "raw_ref": record.get("id", "?")})
raise RecordAuditError(str(exc)) from exc
A RecordAuditError is caught by the batch runner and diverted, so one unauditable row cannot fail an otherwise complete run — the same isolation discipline libraries use when they route bad ingest records to a schema validation quarantine queue. An AuditIntegrityError is deliberately not caught at the record level: a broken chain means the entire log is suspect, so verify_chain runs before any report generation and aborts the whole reporting job loudly if a link fails. The quarantine record stores only a reference, never the failed payload, so the quarantine queue does not become an unaudited copy of the circulation data.
Performance Considerations
Hashing is cheap; the two real costs are the serialized tail-read on every append and materializing a whole batch to reconcile it. The append serialization is a correctness requirement, not a tunable — but you can amortize it by batching appends behind a single-writer worker that reads the tail hash once, then chains a run of events in memory before one atomic multi-row insert, rather than one round-trip per event.
Reconciliation should stream. A month-end report over millions of events must not pull the batch into a list. Aggregate with a windowed reducer that reads the log in batch_id order and retains only running counts:
from collections import Counter
def stream_reconcile(cursor, batch_id: str, page_size: int = 10_000) -> Counter:
counts: Counter = Counter()
while True:
page = cursor.fetchmany(page_size)
if not page:
break
counts.update(row["action"] for row in page)
return counts
For very large retention or reconciliation jobs, offload paging to the broker rather than a single worker: the durable-topic, at-least-once consumer pattern developed for Async Batch Processing for Catalog Updates applies unchanged, with report aggregation as the consumer’s side effect. And because both the source records and the append target are often behind a vendor API, respect the throttle discipline in ILS REST API Polling & Rate Limiting so the reporting job does not stall the pipeline it is meant to observe.
Verification & Testing
Two properties must be proven by test, not assumed: tamper-evidence (any altered or deleted row breaks the chain) and reconciliation balance (received equals masked plus quarantined for a complete batch).
import pytest
def _event(prev: str, action: Action, batch: str = "b1") -> AuditEvent:
return AuditEvent(
event_id=str(uuid.uuid4()),
occurred_at="2026-07-15T00:00:00+00:00",
actor="reporter",
action=action,
entity_type="patron",
entity_id_hashed="deadbeef",
batch_id=batch,
disposition=Disposition.OK,
prev_hash=prev,
)
def _chain(actions: list[Action]) -> list[AuditEvent]:
events, prev = [], GENESIS_HASH
for a in actions:
e = _event(prev, a)
events.append(e)
prev = e.chain_hash()
return events
def test_intact_chain_verifies():
verify_chain(_chain([Action.RECEIVED, Action.MASKED]))
def test_tampered_row_breaks_chain():
events = _chain([Action.RECEIVED, Action.MASKED, Action.EXPORTED])
from dataclasses import replace
events[1] = replace(events[1], actor="attacker")
with pytest.raises(AuditIntegrityError):
verify_chain(events)
def test_deleted_row_breaks_chain():
events = _chain([Action.RECEIVED, Action.MASKED, Action.EXPORTED])
with pytest.raises(AuditIntegrityError):
verify_chain([events[0], events[2]])
def test_reconciliation_balances():
events = [
_event(GENESIS_HASH, Action.RECEIVED),
_event(GENESIS_HASH, Action.RECEIVED),
_event(GENESIS_HASH, Action.MASKED),
_event(GENESIS_HASH, Action.QUARANTINED),
]
report = reconcile_batch(events, "b1")
assert report.balanced
assert report.records_in == report.records_masked + report.records_quarantined
def test_no_raw_id_in_report():
report = reconcile_batch(_chain([Action.RECEIVED, Action.MASKED]), "b1")
assert "deadbeef" not in json.dumps(asdict(report))
Run verify_chain as a scheduled job over the live audit table, not only in unit fixtures — an integrity break introduced by a bad migration or a rogue UPDATE grant will pass every test written against in-memory events but surface the moment the whole-table recomputation runs.
Troubleshooting
The reconciliation report says a batch is imbalanced — records_in exceeds masked plus quarantined. What happened?
A record was RECEIVED but never reached a terminal MASKED or QUARANTINED event, usually because the worker crashed mid-record or an exception skipped the audit call. The imbalance is the report doing its job. Find the RECEIVED events with no matching terminal event for that entity_id_hashed in the batch, replay or quarantine those records, and confirm every stage emits its terminal event even on the failure path.
verify_chain raises AuditIntegrityError but nobody edited the log. Why?
Most often the canonical serialization changed — a field was reordered, a Python or library upgrade altered json.dumps behaviour, or an enum’s .value shifted — so recomputed hashes no longer match the stored ones even though no row was tampered. Pin the canonical form (explicit key list, sort_keys=True, fixed separators) and add a schema-version marker to the digest input so a deliberate format change is distinguishable from tampering. If the serialization is unchanged, treat it as a real integrity event and freeze the log for investigation.
Two concurrent workers forked the audit chain — some rows share a prev_hash. How do I prevent it?
Both workers read the same last_hash before either appended. The audit log requires a single writer per chain: route all appends for one log through one worker, or take a row lock on the chain tail so the second append blocks until the first commits. Throughput per log is intentionally capped at one writer; if you need more, shard by batch_id into independent chains rather than interleaving writers on one.
An auditor wants the raw patron ids behind an entity_id_hashed. Can the report provide them?
No, and that is the design. entity_id_hashed is a one-way HMAC surrogate with no reverse table — the audit log deliberately holds no path back to the patron. If a lawful request genuinely needs identity, it is answered by querying the source ILS under proper authorization and cross-referencing the surrogate, not by un-hashing the report, which the report cannot do.
The month-end retention export is running out of memory. How do I fix it?
The batch is being materialized into a list before aggregation. Stream it in batch_id order with fetchmany, reduce into running counts, and retain nothing across pages, as shown under Performance Considerations. The per-state scoping and schedule logic that drives which records an export selects is covered in Generating State-Specific Retention Exports.
Related
- Designing an Append-Only Audit Log Schema — the indexing, checkpointing, and write-once migration behind the hash-chained table
- Building Batch Reconciliation Reports for Circulation — the full records-in vs masked vs quarantined report format and emitters
- Generating State-Specific Retention Exports — scoping exports to per-jurisdiction retention schedules with hashed identifiers
- PII Masking in Patron Data Exports — the surrogate-token discipline the audit log reuses so reports never carry raw PII
- Data Retention Policies for Public Libraries — the retention clock the reporting layer proves compliance against