Circulation History Routing & Anonymization

Operating within the broader Patron Validation & Privacy Data Routing architecture, this guide covers the one stage where a data-integrity mistake becomes a reportable privacy breach: decoupling a patron’s checkout, renewal, and return history from their identity before that history is archived. Library-tech staff hit this problem the moment circulation telemetry has to outlive the loan that produced it — collection-development analysts want to know that a title circulated, statute says you may not keep who borrowed it, and the pipeline is the only place those two requirements are reconciled. This page walks through the event contract, the retention state machine, an irreversible HMAC anonymization step, idempotent archival, and the quarantine and audit patterns that keep the whole flow defensible, with production-grade Python throughout.

Specification & Contract

A circulation event enters this stage after it has already been validated and routed by the identity pipeline. What arrives here is a normalized transaction — one row per checkout, renewal, or return — carrying a patron reference, an item reference, timestamps, and the branch that recorded it. The routing engine is a stateless consumer: it reads a normalized payload from a SIP2/NCIP feed or a vendor REST export, evaluates it against a retention policy, and advances it through a finite lifecycle. That lifecycle is the contract every downstream table depends on.

The anonymize transition is the irreversible boundary. Everything before it is recoverable; nothing after it can be re-linked to a person. Because retention thresholds change by legislation rather than by code, the window that governs the retention_pending → anonymize transition must be externalized to a configuration service or rules engine — never compiled into the consumer. The windows themselves are mapped in Data Retention Policies for Public Libraries; this stage only enforces them.

The second half of the contract is field-level disposition. Not every field survives the anonymize transition, and the policy engine must classify each one before it can route the record.

Timestamp truncation matters as much as dropping the identifier: a full-precision return time plus a small branch can single out one patron even after patron_id is gone, so temporal generalization is part of the anonymization, not an afterthought.

Prerequisites & Environment Setup

The examples target Python 3.11+, Pydantic v2 (the model_validator API differs from v1 and code written against v1 fails silently after upgrade, so pin the major version), and psycopg for the archive. Cryptographic salts and broker credentials are injected through environment variables and, in production, sourced from an HSM or cloud KMS — never committed.

• Python >= 3.11 in the worker image (python --version)
• pydantic >= 2.5 pinned in requirements.txt (the v2 model_validator API is assumed throughout)
• psycopg[binary] >= 3.1 for idempotent archival upserts
• CIRC_HMAC_SALT exported as a secret (32+ random bytes, rotated on the jurisdictional schedule, resolved from KMS/HSM at runtime)
• RETENTION_POLICY_URL pointing at the externalized rules service (never a hard-coded window)
• CIRC_DLQ_TOPIC set to a durable dead-letter destination for malformed or conflicting records
• AUDIT_CORRELATION_ID propagated from the batch orchestrator for end-to-end tracing
• Broker acknowledgements configured for manual (post-persist) commit, so at-least-once delivery cannot skip an anonymization

python -m venv .venv && source .venv/bin/activate
pip install "pydantic>=2.5,<3" "psycopg[binary]>=3.1"
export CIRC_HMAC_SALT="$(openssl rand -hex 32)"   # dev only; prod resolves from KMS
export RETENTION_POLICY_URL="https://policy.internal/circulation/v2"
export CIRC_DLQ_TOPIC="circ.anonymize.dlq"

Core Implementation

The stage is built from small, typed functions: validate the event, evaluate the retention policy, apply anonymization, and archive idempotently. Composing them keeps every failure attributable to one step and keeps the irreversible transform isolated behind its own function.

Step 1 — Model and validate the circulation event. Validation runs at ingestion and again immediately before anonymization, because a record can sit in retention_pending for years and the schema may tighten in between. Cross-field validators enforce chronological consistency and reject records whose identifiers do not match the institution’s format. Validation failures emit machine-readable error codes and never echo the raw payload into logs — an exception message containing a patron barcode is itself a leak.

from __future__ import annotations

import logging
from datetime import datetime
from typing import Optional

from pydantic import BaseModel, Field, ValidationError, model_validator

logger = logging.getLogger("circulation.anonymize")


class CirculationEvent(BaseModel):
    transaction_id: str = Field(pattern=r"^[A-Z0-9]{10,24}$")
    patron_id: str = Field(min_length=8, max_length=32)
    item_barcode: str = Field(pattern=r"^[0-9]{14}$")
    checkout_ts: datetime
    return_ts: Optional[datetime] = None
    branch_code: str = Field(pattern=r"^[A-Z]{2,4}$")

    @model_validator(mode="after")
    def validate_temporal_order(self) -> "CirculationEvent":
        if self.return_ts is not None and self.return_ts < self.checkout_ts:
            raise ValueError("return_ts must be >= checkout_ts")
        return self


def ingest_and_validate(payload: dict) -> CirculationEvent:
    """Validate an inbound payload. Raises on failure without echoing PII."""
    try:
        return CirculationEvent.model_validate(payload)
    except ValidationError as exc:
        # exc.json() is structured and free of raw field values here; still,
        # log only the error codes and the transaction id, never the payload.
        logger.warning(
            "circ_schema_reject",
            extra={"errors": [e["type"] for e in exc.errors()]},
        )
        raise

Step 2 — Evaluate the retention policy to pick the next state. The policy engine is pure and deterministic: given the event and a policy snapshot, it returns the target state. Keeping it side-effect-free is what makes it testable against known cases and safe to replay. The same idempotency-key discipline used for Async Batch Processing for Catalog Updates applies here — the target state for a given (transaction_id, policy_version) is always the same, so a duplicate delivery cannot double-advance the lifecycle.

from dataclasses import dataclass
from datetime import timedelta, timezone


@dataclass(frozen=True)
class RetentionPolicy:
    version: str
    window: timedelta          # how long a closed loan stays identity-linked
    purge_after: timedelta     # TTL on the anonymized row


def next_state(event: CirculationEvent, policy: RetentionPolicy,
               now: datetime) -> str:
    """Deterministic lifecycle transition for one event."""
    if event.return_ts is None:
        return "active"
    closed_for = now - event.return_ts
    if closed_for < policy.window:
        return "retention_pending"
    if closed_for < policy.window + policy.purge_after:
        return "anonymize"
    return "purge"

Step 3 — Apply irreversible anonymization. The surrogate key is a salted, domain-separated HMAC-SHA-256 of the patron identifier. Domain separation (prefixing the hash input with a context string) guarantees that the same patron produces a different surrogate in circulation than in, say, a holds dataset, so the two cannot be joined to re-identify. The salt lives in an HSM/KMS and rotates on the retention schedule. The transform is deterministic within a salt epoch, which preserves the ability to count “how many distinct borrowers” without ever storing who they were. This is the circulation-specific counterpart to the field obfuscation in PII Masking in Patron Data Exports — the difference is that masking supports reversible joins under key custody, whereas this transform is deliberately one-way.

import hashlib
import hmac


class CirculationAnonymizer:
    """One-way, domain-separated surrogate-key generator for patron ids."""

    def __init__(self, salt: bytes, domain: str = "circulation") -> None:
        if len(salt) < 16:
            raise ValueError("HMAC salt must be at least 16 bytes")
        self._salt = salt
        self._domain = domain.encode("utf-8")

    def surrogate(self, patron_id: str) -> str:
        msg = self._domain + b":" + patron_id.encode("utf-8")
        return hmac.new(self._salt, msg, hashlib.sha256).hexdigest()[:16]

    def anonymize(self, event: CirculationEvent) -> dict:
        """Return an archive row with identity stripped and time coarsened."""
        return {
            "transaction_id": event.transaction_id,
            "patron_surrogate": self.surrogate(event.patron_id),
            "item_barcode": event.item_barcode,
            # Temporal quasi-identifier truncated to the day.
            "checkout_day": event.checkout_ts.date().isoformat(),
            "return_day": event.return_ts.date().isoformat()
            if event.return_ts else None,
            "branch_code": event.branch_code,
            "anonymized_at": datetime.now(timezone.utc).isoformat(),
        }

A common pitfall here is logging the plaintext input or an intermediate hash state for “debugging” — do neither. The only safe artifact to emit is the surrogate, and even that should be truncated in logs.

Step 4 — Persist idempotently to the archive. Archival uses INSERT ... ON CONFLICT DO UPDATE keyed on transaction_id, so an at-least-once redelivery re-writes the same anonymized row rather than creating a duplicate. Consumers acknowledge the broker message only after this commit succeeds, which is what makes at-least-once delivery safe without risking a double anonymization.

import psycopg


def archive_anonymized(dsn: str, rows: list[dict]) -> int:
    """Upsert anonymized rows; returns the count persisted."""
    sql = """
        INSERT INTO circulation_archive
            (transaction_id, patron_surrogate, item_barcode,
             checkout_day, return_day, branch_code, anonymized_at, state)
        VALUES
            (%(transaction_id)s, %(patron_surrogate)s, %(item_barcode)s,
             %(checkout_day)s, %(return_day)s, %(branch_code)s,
             %(anonymized_at)s, 'archived')
        ON CONFLICT (transaction_id) DO UPDATE SET
            anonymized_at = EXCLUDED.anonymized_at,
            state = 'archived'
    """
    with psycopg.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.executemany(sql, rows)
        conn.commit()
    return len(rows)

PII & Compliance Checkpoints

Compliance in this stage is structural: it is expressed in which fields cross the anonymize boundary and when, not in a policy document read after the fact. Three checkpoints must be enforced in code.

First, reject unmasked direct identifiers at ingress. A payload carrying a full SSN, birthdate, or contact field has no business in a circulation-history record; the validator should refuse it before it reaches the retention state machine, exactly as malformed records are turned back at a schema validation quarantine queue. The archive schema itself should have no columns for those fields, so there is nowhere for them to land even if validation is bypassed.

Second, treat the anonymization as irreversible by construction, not by convention. The surrogate function keeps no lookup table mapping surrogate back to patron. If a business requirement genuinely needs reversible pseudonymity (rare, and usually a sign the retention window is wrong), that belongs in the reversible masking layer described in PII Masking in Patron Data Exports, under explicit key custody — not here.

Third, align the deletion triggers with the statutory schedule and with erasure requests. The purge state enforces automated deletion at the TTL, and a right-to-erasure request must be able to force a record from any state directly to purge. Because analytics consumers downstream receive only anonymized, aggregated rows, the re-identification controls in Implementing Differential Privacy for Patron Analytics close the last gap for cohort queries. For academic institutions whose patrons are students, the additional field-suppression rules in Automating FERPA Compliance in Student Patron Records apply on top of everything here. The cross-cutting principle that ties these together is documented in Data Privacy Boundaries in Library Systems.

Error Handling & Quarantine Patterns

Two failure classes dominate: a record that will not validate, and a policy conflict where the record is ambiguous (for example, a return_ts that predates checkout_ts, or a material class with no configured retention window). Neither may be dropped, and neither may be silently anonymized — an incorrectly anonymized record cannot be recovered.

Malformed or conflicting records route to a durable dead-letter destination where the raw payload is stored encrypted-at-rest and only a redacted summary reaches the logs. Retries use exponential backoff with jitter so that a transient policy-service outage does not stampede the rules engine on recovery — the same backoff discipline detailed in ILS REST API Polling & Rate Limiting. A record that fails a bounded number of times is a poison message: it moves to the dead-letter topic so the partition drains, and an operator is alerted with the transaction_id needed to replay it once the root cause is fixed.

class PolicyConflict(Exception):
    """Raised when no deterministic state can be chosen for a record."""


def route_event(payload: dict, policy: RetentionPolicy,
                anonymizer: CirculationAnonymizer, now: datetime,
                dlq) -> Optional[dict]:
    """Validate, route, and anonymize one event; DLQ on any failure."""
    try:
        event = ingest_and_validate(payload)
    except ValidationError:
        dlq.publish(payload, reason="schema_violation")   # encrypted at rest
        return None

    try:
        state = next_state(event, policy, now)
    except Exception as exc:                                # missing window, etc.
        logger.error("policy_conflict", extra={"txn": event.transaction_id})
        dlq.publish(payload, reason="policy_conflict")
        raise PolicyConflict(str(exc)) from exc

    if state == "anonymize":
        return anonymizer.anonymize(event)
    # active / retention_pending / purge are handled by their own workers.
    return None

Publishing the original payload (not a partially transformed one) to the dead-letter queue is deliberate: replay must start from a clean input so a fixed policy re-runs the full transform. Because archival is idempotent, replaying a record that was already anonymized is a harmless no-op.

Performance Considerations

The retention sweep runs over the entire back-catalog of closed loans, which for a mid-sized consortium is tens of millions of rows, so the stage must stream, not materialize. Pull candidates from the archive in keyset-paginated batches and anonymize each batch in place; never load the full result set into a list, which is the classic path to an OOM-killed worker mid-sweep. The streaming and batch-sizing techniques generalize from Optimizing pymarc Performance for Large Record Sets — bounded memory, one batch resident at a time.

Two operation-specific costs dominate. The HMAC itself is cheap, but a naive implementation that calls the KMS to fetch the salt per record will be dominated by network round-trips; resolve the salt once per salt-epoch and cache it in the worker for the batch. Database writes should use batched executemany upserts (as in Step 4) rather than per-row commits, so the archive sees one transaction per batch instead of one per record. Set task-level timeouts and checkpoint the last processed key so a sweep interrupted by an ILS maintenance window resumes from where it stopped rather than reprocessing already-anonymized rows.

Verification & Testing

Anonymization is only trustworthy if it is verified against known-plaintext vectors before every production promotion. Three properties are worth asserting: the transform is deterministic within a salt epoch, it is not reversible, and temporal fields are actually coarsened.

from datetime import datetime, timezone


def test_surrogate_is_deterministic():
    a = CirculationAnonymizer(salt=b"0" * 32)
    assert a.surrogate("P12345678") == a.surrogate("P12345678")


def test_domain_separation_prevents_cross_join():
    salt = b"0" * 32
    circ = CirculationAnonymizer(salt, domain="circulation")
    holds = CirculationAnonymizer(salt, domain="holds")
    # Same patron, different dataset -> non-joinable surrogates.
    assert circ.surrogate("P12345678") != holds.surrogate("P12345678")


def test_direct_identifiers_are_stripped():
    a = CirculationAnonymizer(salt=b"0" * 32)
    event = CirculationEvent(
        transaction_id="TXN0000000001",
        patron_id="P12345678",
        item_barcode="30000000000001",
        checkout_ts=datetime(2026, 1, 2, 14, 33, tzinfo=timezone.utc),
        return_ts=datetime(2026, 1, 9, 9, 5, tzinfo=timezone.utc),
        branch_code="MAIN",
    )
    row = a.anonymize(event)
    assert "patron_id" not in row and "email" not in row
    assert row["checkout_day"] == "2026-01-02"      # time-of-day discarded
    assert row["patron_surrogate"] != event.patron_id

Round out the suite with a mock broker that redelivers the same message twice and asserts the archive holds exactly one row (idempotency), and a temporal-order test that confirms a return_ts < checkout_ts payload lands in the dead-letter queue rather than the archive. Run the whole suite in CI against the pinned Pydantic version so an accidental v1/v3 resolution fails the build before it reaches production.

Troubleshooting & FAQ

Why is the same patron producing two different surrogate keys?

The salt was rotated between the two runs, or the two records were anonymized under different domain strings. Determinism holds only within a salt epoch and a fixed domain. If you need surrogates to remain stable across a rotation for longitudinal counting, retain the previous salt for a grace window and re-key deliberately, rather than treating the drift as a bug.

A record was anonymized too early and we need the original patron back. Can we recover it?

No — that is the intended property. The anonymize transition keeps no reverse-lookup table, so once it runs the linkage is gone. The correct fix is upstream: the record advanced early because the retention window was misconfigured. Widen the window in the policy service and rely on the fact that anything still in retention_pending is fully recoverable.

Memory climbs steadily until the retention-sweep worker is OOM-killed. What is wrong?

The sweep is materializing the candidate set into a list instead of streaming keyset-paginated batches. Consume the archive query in bounded pages, anonymize and upsert each page, and retain nothing across iterations. See the optimization guidance linked under Performance Considerations for the full bounded-memory procedure.

Duplicate rows are appearing in the archive after broker redeliveries. How do I stop it?

The upsert is not keyed on a stable identifier, or the consumer is acknowledging the message before the commit. Ensure the ON CONFLICT (transaction_id) clause is present and that the broker acknowledgement happens strictly after conn.commit(), so an at-least-once redelivery updates the existing row instead of inserting a second one.

The policy service is down and records are piling up. Are we losing anonymizations?

No, provided consumers are not acknowledging un-processed messages. Records accumulate safely in the broker while retries back off with jitter; when the policy service recovers, the sweep drains the backlog. A record that repeatedly fails is moved to the dead-letter topic with its transaction_id for manual replay, never dropped.

Related

• Data Retention Policies for Public Libraries — the jurisdictional windows this state machine enforces
• PII Masking in Patron Data Exports — reversible field obfuscation for exports, the counterpart to this one-way transform
• Threshold Tuning for Identity Validation — how identities are resolved before their history is routed here
• Data Privacy Boundaries in Library Systems — the minimization principle behind the anonymization boundary
• Async Batch Processing for Catalog Updates — the broker and idempotency patterns reused for the retention sweep