Library Catalog Data Sync Pipelines

Threshold Tuning for Identity Validation

Operating within the broader Patron Validation & Privacy Data Routing architecture, this guide covers the decision layer that every consortial or municipal library hits the moment two source systems disagree about who a patron is: given a computed similarity score between an inbound record and an existing patron row, what does the pipeline do with it? Merge them? Queue them for a human? Reject the match outright? Threshold tuning is where that judgment is encoded — the deterministic gate that turns a floating-point match score into a routing action with legal consequences. Library-tech staff meet this problem during registration deduplication, nightly ILS reconciliation, and every federated-login flow that has to reconcile a campus SIS identity against an existing borrower profile.

This page walks the tuning problem end to end: the scoring contract and the confidence bands that map scores to actions, the environment you need, an annotated Python implementation of the streaming evaluator, the compliance checkpoints that keep identity assertions auditable, the quarantine patterns that keep an ambiguous match from silently corrupting a patron record, how to keep the evaluator inside its memory and query budget, and how to verify a tuning change before it touches production. Because a wrong merge is far more damaging than a missed one — it fuses two people’s circulation histories — this page treats false-positive avoidance as the primary design constraint.

Identity validation threshold decision flow An inbound patron payload is resolved against candidates using a pg_trgm blocking query on institution_id, last_name_hash and birth_year. A probabilistic scorer turns each candidate into a score between 0 and 1, which is multiplied by a source_reliability weight before an adaptive gate applies three confidence bands: a score of 0.85 or above routes to AUTO_MERGE in green, 0.65 to 0.84 routes to the HITL_REVIEW queue in amber, and below 0.65 routes to REJECT with a reason code in red. An unscoreable payload with a missing blocking key branches from the scorer to a quarantine dataset. At the gate, an immutable audit log records the masked ids, computed score, applied thresholds and outcome. RESOLVE SCORE GATE ROUTE Inbound patronpayload Candidate lookuppg_trgm blocking keys Scorerscore ∈ [0, 1] Reliability× source weight Adaptivegate AUTO_MERGE adjusted ≥ 0.85 HITL_REVIEW 0.65 – 0.84 REJECT < 0.65 · reason code missing blocking key Quarantineunscoreable payload logged at gate Immutable audit logmasked ids · computed score · applied thresholds · outcome

Specification & Contract

The contract has one rule everything else follows from: a score is a probability, not a permission. The scorer produces a number in [0.0, 1.0]; the threshold layer — and only the threshold layer — decides what that number authorizes. Keeping scoring and gating separate is what lets you re-tune the bands without touching the matching math, and it is the same separation-of-concerns discipline the ILS Schema Translation Patterns layer applies to vendor payloads.

Scores are computed across a fixed set of patron attributes so that a change in weighting is reproducible and reviewable. Each attribute contributes a normalized sub-score; the aggregate is a weighted sum. The table below is the minimum scoring contract most pipelines start from — treat each row as a rule enforced in tests, not a suggestion.

Attribute Comparison method Weight Notes
last_name trigram similarity (pg_trgm) 0.30 Normalize case and diacritics before hashing; store last_name_hash for the index
birth_year exact / ±1 tolerance 0.20 ±1 absorbs off-by-one data-entry error without collapsing distinct people
barcode exact after check-digit strip 0.25 A barcode match is strong but not decisive — cards get reissued and reused
institution_id exact 0.15 A hard blocking key; candidates must share it before scoring runs
address_token trigram similarity on hashed token 0.10 Lowest weight — addresses churn and are shared by household members

Those sub-scores collapse into three confidence bands, and the bands are the actual contract downstream stages depend on:

Adjusted score Decision Routing action
≥ 0.85 AUTO_MERGE Automated profile update or record merge
0.65 – 0.84 HITL_REVIEW Human-in-the-loop review queue
< 0.65 REJECT Rejection with a structured reason code

Two rules govern the whole contract. First, thresholds are configuration, not code — the cutoffs live in a versioned config object keyed by source system, so a tuning change is a reviewable one-line edit tied to a change ticket, never a scattered hunt through business logic. Second, static cutoffs fail under real conditions: data quality varies across consortium branches, municipal integrations, and third-party identity providers, so the raw score is multiplied by a source_reliability weight and stale records decay before they are gated. A match derived from a freshly refreshed municipal ID registry should outrank one derived from a legacy barcode-only export, and the band logic must express that difference rather than pretend all sources are equal.

Prerequisites & Environment Setup

The evaluator is pure Python plus a PostgreSQL candidate store. Pin versions so a coercion or similarity-operator change never silently alters which records pass the gate.

Install and freeze in one shot:

bash
python -m venv .venv
source .venv/bin/activate
pip install "structlog>=24.1" "psycopg[binary]>=3.1"
pip freeze > requirements.txt

A common early pitfall: comparing scores as float. Floating-point rounding means a record scoring exactly on a band edge (0.6500000001) can route inconsistently across runs. Use decimal.Decimal for every band comparison so 0.65 means 0.65, and reserve float for the intermediate similarity math only.

Core Implementation

The evaluator has three labeled stages: resolve candidates for an inbound record, score each candidate against it, then gate the best score into a routing decision. Each stage is independently testable, and the gate is a pure function so it scales horizontally during peak reconciliation windows.

Step 1 — The typed records and the masking primitive

Define the data the evaluator moves and the one function that keeps identifiers out of logs. Decimal on the score field is deliberate — it makes band comparison exact.

python
from __future__ import annotations

import hashlib
import hmac
import os
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Iterator, Literal

Decision = Literal["AUTO_MERGE", "HITL_REVIEW", "REJECT"]


def _load_salt() -> bytes:
    """Fetch the HMAC salt at startup; fail fast if it is missing."""
    salt = os.environ.get("PII_MASKING_SALT")
    if not salt:
        raise RuntimeError("PII_MASKING_SALT is not set; refusing to start")
    return salt.encode("utf-8")


_SALT = _load_salt()


def mask_id(raw_value: str) -> str:
    """Deterministic, salted correlation token for audit-safe logging.

    HMAC-SHA256 (not a bare hash) so the token cannot be reversed with a
    precomputed rainbow table of known barcodes or patron ids.
    """
    return hmac.new(_SALT, raw_value.encode("utf-8"), hashlib.sha256).hexdigest()[:16]


@dataclass(frozen=True)
class PatronRecord:
    patron_id: str
    institution_id: str
    last_name_hash: str
    birth_year: int
    barcode: str


@dataclass(frozen=True)
class MatchResult:
    candidate_id: str
    score: Decimal
    decision: Decision
    reason: str
    audit_payload: dict = field(default_factory=dict)

Step 2 — The candidate query

Never score against the whole patron table. Block on institution_id and the hashed name, then let pg_trgm rank the shortlist. The parameterized query below returns only pre-filtered candidates, keeping the per-record comparison count bounded regardless of table size.

python
import psycopg

CANDIDATE_SQL = """
    SELECT patron_id,
           similarity(last_name_hash, %(name)s) AS name_sim,
           birth_year,
           barcode
    FROM patron
    WHERE institution_id = %(inst)s
      AND last_name_hash %% %(name)s          -- % is the pg_trgm similarity operator
      AND abs(birth_year - %(birth)s) <= 1
    ORDER BY name_sim DESC
    LIMIT 25;
"""


def fetch_candidates(conn: psycopg.Connection, record: PatronRecord) -> list[dict]:
    """Return at most 25 blocking-key candidates for one inbound record."""
    with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
        cur.execute(
            CANDIDATE_SQL,
            {"inst": record.institution_id, "name": record.last_name_hash,
             "birth": record.birth_year},
        )
        return cur.fetchall()

The LIMIT 25 is a guardrail, not a tuning parameter: if a single blocking key returns more than 25 near-identical names, the data is too coarse to auto-merge safely and the record belongs in review regardless of score.

Step 3 — Scoring and the adaptive gate

Scoring produces the raw probability; the gate applies the source-reliability weight and the configured bands. Keep both pure — no I/O, no logging — so they are trivial to unit-test at the band edges.

python
def score_candidate(record: PatronRecord, candidate: dict) -> Decimal:
    """Weighted sum of per-attribute sub-scores in [0.0, 1.0]."""
    name_sim = Decimal(str(candidate["name_sim"]))          # from pg_trgm
    birth_match = Decimal("1") if candidate["birth_year"] == record.birth_year else Decimal("0.5")
    barcode_match = Decimal("1") if candidate["barcode"] == record.barcode else Decimal("0")

    return (name_sim * Decimal("0.30")
            + birth_match * Decimal("0.20")
            + barcode_match * Decimal("0.25")
            + Decimal("0.15")                                # institution_id already matched
            + Decimal("0")).quantize(Decimal("0.0001"))


def gate(
    candidate_id: str,
    raw_score: Decimal,
    source_reliability: Decimal = Decimal("1.0"),
    auto_cutoff: Decimal = Decimal("0.85"),
    review_cutoff: Decimal = Decimal("0.65"),
) -> MatchResult:
    """Apply source weighting and return the routing decision."""
    adjusted = (raw_score * source_reliability).quantize(Decimal("0.0001"))

    if adjusted >= auto_cutoff:
        decision, reason = "AUTO_MERGE", "high-confidence match; automated reconciliation"
    elif adjusted >= review_cutoff:
        decision, reason = "HITL_REVIEW", "moderate confidence; staff verification required"
    else:
        decision, reason = "REJECT", "low confidence; potential false positive"

    return MatchResult(
        candidate_id=candidate_id,
        score=adjusted,
        decision=decision,
        reason=reason,
        audit_payload={"raw_score": str(raw_score), "reliability_weight": str(source_reliability)},
    )

Step 4 — The streaming evaluation loop

Nightly reconciliation can process millions of records; materializing them into a list exhausts the heap on a memory-constrained container. Drive the evaluator from a generator so records are scored and released one at a time, and let structured logging emit a masked audit event per decision. This backpressure-friendly shape mirrors the streaming discipline used for bulk record movement in Async Batch Processing for Catalog Updates.

python
import structlog

logger = structlog.get_logger()


def evaluate_stream(
    conn: psycopg.Connection,
    records: Iterator[PatronRecord],
    source_reliability: Decimal = Decimal("1.0"),
) -> Iterator[MatchResult]:
    """Score each inbound record against its candidates, yielding one result each."""
    for record in records:
        candidates = fetch_candidates(conn, record)
        if not candidates:
            yield MatchResult(candidate_id="", score=Decimal("0"),
                              decision="REJECT", reason="no candidate in blocking window")
            continue

        best = max(
            (gate(c["patron_id"], score_candidate(record, c), source_reliability)
             for c in candidates),
            key=lambda r: r.score,
        )

        logger.info(
            "patron_identity_decision",
            patron_id=mask_id(record.patron_id),
            candidate_id=mask_id(best.candidate_id) if best.candidate_id else None,
            score=str(best.score),
            decision=best.decision,
            reason=best.reason,
            **best.audit_payload,
        )
        yield best

Note the two failure exits: a record with no blocking-window candidate is rejected with an explicit reason rather than silently dropped, and only the highest-scoring candidate is gated so one inbound record can never merge into two profiles.

PII & Compliance Checkpoints

Threshold tuning touches the most sensitive fields a library holds — names, birth years, addresses, and circulation-linked barcodes — so every stage carries a compliance obligation, not just the export at the end.

Error Handling & Quarantine Patterns

An identity pipeline fails in two distinct ways, and they need different handling. A transient failure — the candidate database is briefly unreachable — should retry. A permanent failure — an inbound payload is unscoreable because a blocking key is missing or malformed — must never be retried into a loop; it belongs in quarantine for inspection, exactly as malformed bibliographic records route to a schema validation quarantine queue.

python
import time


class UnscoreableRecord(Exception):
    """Raised when a payload cannot be gated (missing blocking key, bad type)."""


def evaluate_with_quarantine(
    conn: psycopg.Connection,
    record: PatronRecord,
    quarantine,                     # object exposing .write(dict)
    max_retries: int = 3,
) -> MatchResult | None:
    """Gate one record; retry transient DB errors, quarantine permanent ones."""
    if not record.institution_id or record.birth_year <= 0:
        quarantine.write({"patron_id": mask_id(record.patron_id),
                          "reason": "missing or invalid blocking key"})
        return None

    for attempt in range(1, max_retries + 1):
        try:
            candidates = fetch_candidates(conn, record)
            if not candidates:
                return MatchResult("", Decimal("0"), "REJECT", "no candidate in window")
            return max(
                (gate(c["patron_id"], score_candidate(record, c)) for c in candidates),
                key=lambda r: r.score,
            )
        except psycopg.OperationalError as exc:
            wait = 2 ** attempt
            logger.warning("candidate_db_retry", patron_id=mask_id(record.patron_id),
                           attempt=attempt, wait_s=wait, error=str(exc))
            conn.rollback()
            time.sleep(wait)

    quarantine.write({"patron_id": mask_id(record.patron_id),
                      "reason": "candidate store unreachable after retries"})
    return None

Two rules make this safe. First, quarantine is not a dead end — a record that lands there is triaged, its blocking key repaired upstream, and it is replayed, never discarded. Second, transient retries use exponential backoff so a flapping database is not hammered; the same backoff discipline the ingestion layer applies to vendor APIs under ILS REST API Polling & Rate Limiting.

Performance Considerations

Two resources bound the evaluator: heap and the candidate query.

Verification & Testing

Tune against a labeled fixture set, never against production data blind. The most important tests exercise the band edges, because that is where a one-line cutoff change flips real decisions.

python
from decimal import Decimal


def test_band_edges():
    # Exactly on the auto cutoff -> AUTO_MERGE
    assert gate("c1", Decimal("0.85")).decision == "AUTO_MERGE"
    # One tick below -> HITL_REVIEW, not a merge
    assert gate("c1", Decimal("0.8499")).decision == "HITL_REVIEW"
    # Exactly on the review cutoff -> HITL_REVIEW
    assert gate("c1", Decimal("0.65")).decision == "HITL_REVIEW"
    # Below review -> REJECT with a reason
    r = gate("c1", Decimal("0.6499"))
    assert r.decision == "REJECT" and r.reason


def test_source_reliability_demotes_stale_source():
    # A 0.90 raw score from a source weighted 0.9 -> 0.81 -> falls to review
    assert gate("c1", Decimal("0.90"), source_reliability=Decimal("0.9")).decision == "HITL_REVIEW"


def test_no_candidate_rejects_not_crashes():
    result = next(evaluate_stream(FakeConn(rows=[]), iter([SAMPLE_RECORD])))
    assert result.decision == "REJECT"

Beyond unit tests, measure two operational signals before promoting a tuning change: the precision of AUTO_MERGE against a hand-labeled sample (target: near-zero false merges), and the AUTO_MERGE : HITL_REVIEW ratio, which sizes the human review queue. A tuning change that halves the review queue but doubles false merges is a regression, not an improvement.

Troubleshooting & Frequently Asked Questions

The AUTO_MERGE rate suddenly spiked overnight. What happened?

Almost always an upstream data change, not a tuning bug: a source system started emitting a new default value (blank barcodes, a constant birth year) that inflates similarity across unrelated patrons. Alert on the AUTO_MERGE : HITL_REVIEW ratio and treat a sudden shift as a data-quality incident. Lower that source’s source_reliability weight until the upstream issue is fixed, so borderline matches demote to review instead of auto-merging.

Two clearly different patrons keep merging. How do I stop it?

A false merge means the band is too loose for that source’s data quality. Do not merely raise auto_cutoff globally — that pushes good matches into the review queue everywhere. Instead lower the offending source’s reliability weight, and check whether a single attribute (usually a reused barcode) is carrying too much weight. Re-run the labeled fixture set after any weight change and confirm AUTO_MERGE precision before promoting.

Candidate resolution is slow and the sync window is blowing out. Why?

Run EXPLAIN (ANALYZE, BUFFERS) on the candidate query. A Seq Scan on patron means the composite blocking index or the GIN trigram index is missing or not selectable — usually because last_name_hash is normalized differently at write time than at query time, so the values never match. Normalize case and diacritics identically on both paths, then confirm an Index Scan in the plan.

Records fail with UnscoreableRecord for patrons that clearly exist. What is wrong?

The inbound payload is missing a blocking key (institution_id) or carries an invalid birth_year (0, or a string that coerced badly upstream). That is intended — the record cannot be safely gated without a blocking key, so it quarantines for repair rather than scoring against the wrong window. Fix the extraction upstream and replay the quarantined record; do not loosen the blocking check.

How do I safely roll out a new threshold?

Never flip the cutoff in place. Run the new bands in shadow mode: gate each record with both the current and candidate thresholds, log both decisions (masked), and diff them over a full cycle. Promote only when the diff shows the intended movement and AUTO_MERGE precision holds on the labeled sample. Tie the change to a change-management ticket so the audit trail records who approved it.