Incremental change data capture

A grant office does not re-key its entire award portfolio every morning, and its automation should not re-ingest it either. Yet the naive sync does exactly that: it pulls every record a sponsor portal or ERP feed will hand over, rewrites every row, and stamps a fresh audit entry against records nobody touched. The result is a ledger swollen with no-op churn, reconciliation jobs that grind through hundreds of thousands of unchanged rows, and an audit trail where a genuine amendment is indistinguishable from a routine full reload. Change data capture closes that gap: it identifies the small set of records that genuinely changed since the last run and moves only those. This guide is one of the ingestion layers anchored to the parent guide on Automated Ingestion & Data Sync Workflows, and it depends on the acquisition mechanics of API Polling & Portal Integration to fetch candidate records before it decides which ones matter.

Two mechanisms do the work together, and they answer different questions. A high-water-mark cursor answers which records to even ask for — it narrows the pull to rows a source claims are new or modified since a saved position. A canonical content hash answers which of those actually changed — it fingerprints the meaningful content of each record so that a bumped server timestamp or a re-serialized field never masquerades as a real edit. Only records that survive both filters take a write, and only a real write appends to the audit ledger. This page frames those mechanisms, shows the schema that records them, and gives the idempotent upsert that ties a delta to a single ledger entry. Its downstream contracts on structure and reconciliation are shared with Schema Validation Pipelines, and its state-transition semantics inherit from Grant Lifecycle Architecture Design.

Problem framing #

Full reload is seductive because it is simple: truncate, re-fetch, re-insert, and the store is guaranteed consistent with the source. It also destroys the very thing a research office needs most, which is an honest record of when something changed. Under a nightly full reload, an award whose end date was extended in March looks identical in the audit log to an award that has sat untouched since the grant was issued — every row carries the same “last written” stamp, the moment of the reload. When a sponsor later disputes when an amendment was recognized, or an A-133 auditor asks the office to demonstrate change control over federal award data, “we rewrite everything every night” is not an answer.

Incremental capture inverts the default. Its job is to write as little as possible while proving that everything it did not write was genuinely unchanged. Three ideas make that provable:

• A high-water-mark cursor stores the last position the office successfully consumed — a last-modified timestamp, a monotonically increasing sequence number, or a sponsor-issued change token. The next run asks the source only for records past that mark, so a 400,000-row portfolio yields a fetch of a few dozen candidates on a quiet day.
• Canonical content hashing computes a SHA-256 digest over a normalized serialization of each record, with volatile fields excluded. Two records with the same business content produce the same hash even if the source re-ordered keys or refreshed a synced_at timestamp, so the pipeline never mistakes churn for change.
• Idempotent upsert commits a record only when its new hash differs from the stored one, and only that write appends to the audit ledger. Re-running the same batch is a no-op: zero writes, zero new ledger rows, no duplicated history.

The contrast with full reload is not a performance footnote; it is a compliance posture. A full reload cannot distinguish a no-op from an edit, so its audit trail is noise. Change data capture makes every ledger entry mean exactly one thing: this field, on this record, actually changed at this time.

Policy constraints #

Incremental capture touches federal award data, so what it may skip and what it must record are bounded by regulation, not by convenience. The controlling rules are the same matrix maintained in the University Policy Mapping Frameworks, applied here to the specific question of change detection and retention.

Operational boundary. Policy dictates that a real change must be captured and retained; it does not dictate that unchanged rows be rewritten. Skipping a no-op is not “missing” a record — the ledger’s continuity proves the record was seen and found identical. What the layer must never do is silently lose a change: a hash collision risk, an unstable canonicalization, or a mishandled delete each turn a genuine edit into a silent skip, which is the one failure this design treats as unacceptable. Credential scope and network isolation for the capture workers follow the Security Boundary Configuration; when a source is unreachable and the cursor cannot advance, the run defers rather than reloading, per the Fallback Routing Protocols.

Data schema & field mapping #

Change data capture needs a small state table alongside the records themselves — the place where the last-seen fingerprint and cursor position for each record live between runs. That state is what turns a stateless fetch into an incremental one. The canonical fields below are the minimum the layer persists; the record’s business columns sit in the domain table and are referenced by record_id.

The pair of content_hash and prev_hash is what makes a delta self-describing: an auditor reading a ledger row sees not just that a record changed but the exact fingerprint it moved from and to. last_seen_at is stamped on every run including no-ops, which is how the system proves a skipped record was actually observed and compared rather than simply missed by the fetch. change_type of unchanged is recorded in the state table but does not append to the audit ledger — only new, changed, and deleted do.

Implementation #

The layer has two composable pieces: a canonical hash function that produces a stable fingerprint regardless of source-side serialization noise, and an idempotent upsert that writes and ledgers a record only when its fingerprint actually moved. Heavy candidate batches are decoupled from the commit path through Async Processing & Queue Management, so a slow store write never stalls the fetch loop.

Canonical hashing and delta classification #

The fingerprint must depend only on business content. That means a fixed key order, exclusion of volatile server-owned fields, and normalized dates — otherwise the hash captures serialization accidents and every record looks changed on every run.

import hashlib
import json
from datetime import datetime, timezone
from typing import Any

# Fields the source rewrites on every export but that carry no business meaning.
# Excluding them is what separates a genuine change from no-op churn.
VOLATILE_FIELDS = frozenset({"synced_at", "etag", "_row_version", "server_timestamp"})


def canonical_hash(record: dict[str, Any]) -> str:
    """Deterministic SHA-256 over business content only.

    Sorted keys make the digest independent of source key order; dropping
    volatile fields keeps a refreshed server timestamp from faking a change.
    """
    stable = {k: v for k, v in record.items() if k not in VOLATILE_FIELDS}
    canonical = json.dumps(stable, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def classify(record_id: str, new_hash: str, stored_hash: str | None) -> str:
    """Return the change_type for a single candidate record."""
    if stored_hash is None:
        return "new"
    if stored_hash != new_hash:
        return "changed"
    return "unchanged"

Idempotent upsert with a single ledger entry per delta #

A record is committed on a stable record_id and appends to the ledger only when its hash differs from what is stored. The upsert is guarded by a content_hash comparison, so a re-run of the identical batch performs zero writes.

from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session


def apply_delta(
    session: Session,
    record_id: str,
    payload: dict[str, Any],
    cursor: str,
) -> str:
    """Upsert one record and append at most one ledger entry. Idempotent:
    re-running with the same payload writes nothing and ledgers nothing."""
    new_hash = canonical_hash(payload)
    prior = session.get(CaptureState, record_id)
    change_type = classify(record_id, new_hash, prior.content_hash if prior else None)

    if change_type == "unchanged":
        # Prove the record was seen and compared, without a ledger entry.
        prior.last_seen_at = datetime.now(timezone.utc)
        session.commit()
        return "unchanged"

    stmt = pg_insert(CaptureState).values(
        record_id=record_id,
        source_cursor=cursor,
        content_hash=new_hash,
        prev_hash=prior.content_hash if prior else None,
        last_seen_at=datetime.now(timezone.utc),
        change_type=change_type,
    )
    # Guard the write on a genuine hash difference — the database-level
    # idempotency contract for this layer.
    stmt = stmt.on_conflict_do_update(
        index_elements=["record_id"],
        set_={
            "source_cursor": cursor,
            "prev_hash": CaptureState.content_hash,
            "content_hash": new_hash,
            "last_seen_at": datetime.now(timezone.utc),
            "change_type": change_type,
        },
        where=(CaptureState.content_hash != new_hash),
    )
    session.execute(stmt)
    # One delta, one immutable ledger row.
    session.add(AuditLedger(
        record_id=record_id,
        content_hash=new_hash,
        prev_hash=prior.content_hash if prior else None,
        change_type=change_type,
        captured_at=datetime.now(timezone.utc),
    ))
    session.commit()
    return change_type

The where=(CaptureState.content_hash != new_hash) clause is the load-bearing line: it makes the write conditional on a real change at the database layer, so even a concurrent duplicate run cannot append a second ledger row for the same fingerprint. Tombstones — records the source stops returning — are handled by a separate reconciliation pass that classifies a record_id present in state but absent from the source as deleted, appending a soft-delete ledger entry rather than physically removing the row.

Integration points #

Capture workers sit between acquisition and the validated store; they never write to production ERP or LIMS tables directly. Each boundary has an explicit contract:

• Sponsor portals and ERP feeds. Candidate records arrive from API Polling & Portal Integration, which passes the source cursor it advanced. Capture decides which of those candidates are real deltas.
• Schema validation. A record classified new or changed is handed to Schema Validation Pipelines before it reaches a domain table, so a delta is still policy-checked — capture decides whether to write, validation decides whether it may.
• Downstream reporting. The deleted and changed classifications feed reconciliation and compliance reporting, where a change to a federal award field must surface within the cycle it occurred.

A capture-state row emitted for a genuine budget revision:

{
  "record_id": "IC-2025-004417",
  "source_cursor": "2026-07-15T04:12:09Z",
  "content_hash": "b71d…9a",
  "prev_hash": "3e0c…44",
  "change_type": "changed",
  "last_seen_at": "2026-07-15T05:00:03Z"
}

Verification & audit #

The audit ledger is append-only and hash-linked: each row carries the fingerprint it moved to and the one it moved from, so an auditor can walk a record’s entire history from first sight to current state. To confirm a run behaved:

1. Delta accounting. new + changed + deleted must equal the count of ledger rows appended this run; unchanged records update last_seen_at but must add zero ledger rows. A ledger row without a corresponding classification is a defect.
2. Re-run is a no-op. Execute the same batch against the current state. Every record must classify unchanged, apply_delta must return unchanged for all of them, and the ledger row count must not move. A non-zero delta on a second identical pass means the canonicalization is unstable.
3. Hash chain continuity. For any record, each ledger row’s prev_hash must equal the content_hash of the row before it. A break in the chain means a write bypassed the capture path.

from sqlalchemy import select, func


def verify_run(session: Session, run_start: datetime) -> dict[str, int]:
    rows = session.execute(
        select(AuditLedger.change_type, func.count())
        .where(AuditLedger.captured_at >= run_start)
        .group_by(AuditLedger.change_type)
    ).all()
    return {change_type: n for change_type, n in rows}

Because the ledger is never updated in place, its retention satisfies the 2 CFR 200.333 window directly: the office keeps the ledger, not a nightly snapshot, and the ledger is the change history.

Failure modes & recovery #

Every recovery here is idempotent-safe — re-running it cannot duplicate a delta or a ledger row.

Frequently asked questions #

Related #

• Parent guide: Automated Ingestion & Data Sync Workflows
• Detecting Sponsor Portal Record Deltas with Hashing — the step-by-step Python how-to under this guide
• API Polling & Portal Integration — the acquisition layer that advances the source cursor
• Schema Validation Pipelines — validates each delta before it reaches a domain table
• Grant Lifecycle Architecture Design — the state-transition semantics this layer inherits