Syncing State DOI Rule Updates Incrementally

This page is a focused implementation guide under Regulatory Sync Pipelines: it shows how to re-index only the rules that actually changed since the last poll, using content hashes and per-source high-water marks, so a fifty-jurisdiction sync stays cheap, idempotent, and fully audited.

A carrier that polls every state Department of Insurance on a nightly cadence pulls tens of thousands of rules, of which perhaps a dozen changed since yesterday. The naive approach — re-validate the full feed, rebuild every RuleSet, and re-run shadow evaluation across all jurisdictions — burns compute proportional to the entire rule corpus to absorb a single amended windstorm formula. Worse, it makes the audit log lie: if every rule is re-stamped on every run, the ledger records fifty thousand “changes” a night, and a genuine amendment is buried among tens of thousands of no-op re-writes that an examiner cannot distinguish from real edits.

Three concrete defects follow from full-feed reprocessing. First, non-idempotent application: if a feed is redelivered — a retried poll, a broker replay — the same rule is applied twice and can produce two conflicting staged versions for one (state, rule_id). Second, effective-date smearing: re-ingesting an unchanged rule with today’s date silently moves its effective window, so a claim later resolves against a window that was never published. Third, untraceable change sets: without a per-rule change record, nobody can answer “which exact rules changed in the July sync, and from what to what.” This guide resolves all three: content hashes to detect real change, per-source high-water marks so only new data is fetched, idempotent upserts keyed by (state, rule_id, version), and an explicit audit of every applied change.

This pattern targets Python 3.10+ and builds directly on the RuleSet and registry types defined in the parent Regulatory Sync Pipelines guide. It uses structlog for structured audit lines and the standard library for hashing; pin the dependency so hashing and logging semantics cannot drift:

Copy
python -m venv .venv && source .venv/bin/activate
pip install "structlog==24.*" "orjson==3.*"

Pipeline state required before this stage runs: a persistent store for the per-source high-water mark (the timestamp of the last successfully synced change per state), and a store of the current content hash for every (state, rule_id) already indexed. Both are read at the start of a sync run and written only on success, so a crashed run re-fetches rather than skipping unseen changes.

Fetch only the rules a source reports as changed since the stored high-water mark, then confirm each one genuinely differs by comparing its canonical content hash against the hash already on record. A feed can re-report an unchanged rule — an editorial reformat, a redelivered page — and hashing is what stops that from becoming a spurious version.

Copy
from __future__ import annotations

import hashlib
from dataclasses import dataclass
from datetime import date, datetime, timezone

import orjson
import structlog

log = structlog.get_logger("regulation.sync.incremental")


@dataclass(frozen=True, slots=True)
class IncomingRule:
    """A single rule as pulled from a DOI feed, not yet reconciled."""

    state: str
    rule_id: str
    version: str
    effective_from: date
    effective_to: date | None
    source_bulletin: str
    body: dict[str, str]          # the rule's substantive fields
    source_updated_at: datetime   # the feed's own change timestamp

    def content_hash(self) -> str:
        """Stable SHA-256 over the substantive body only — not metadata."""
        canonical = orjson.dumps(self.body, option=orjson.OPT_SORT_KEYS)
        return hashlib.sha256(canonical).hexdigest()


def changed_rules(
    incoming: list[IncomingRule],
    stored_hashes: dict[tuple[str, str], str],
) -> tuple[list[IncomingRule], int]:
    """Split a fetched batch into genuinely-changed rules and a skip count.

    `stored_hashes` maps (state, rule_id) -> the last indexed content hash.
    """
    changed: list[IncomingRule] = []
    skipped = 0
    for rule in incoming:
        key = (rule.state, rule.rule_id)
        if stored_hashes.get(key) == rule.content_hash():
            skipped += 1
            continue
        changed.append(rule)
    log.info(
        "incremental.diff",
        fetched=len(incoming),
        changed=len(changed),
        skipped=skipped,
    )
    return changed, skipped

Hashing the substantive body rather than the whole record is deliberate: a rule whose only difference is a re-issued bulletin id or a reformatted timestamp must not count as a change, or the effective window smears. The skip count is returned, not discarded — it is a signal, surfaced below in verification.

Every changed rule is written through an upsert keyed by (state, rule_id, version). Because the key includes the version, redelivering the same change is a no-op that returns the existing row rather than creating a duplicate, which is what makes the sync safe to retry or replay off the broker.

Copy
class IncrementalSyncError(Exception):
    """Raised when a change cannot be applied without ambiguity."""


@dataclass(frozen=True, slots=True)
class StoredRule:
    state: str
    rule_id: str
    version: str
    content_hash: str
    effective_from: date
    effective_to: date | None
    source_bulletin: str


class RuleStore:
    """Idempotent, version-keyed store for reconciled rules."""

    def __init__(self) -> None:
        self._rows: dict[tuple[str, str, str], StoredRule] = {}
        self._current_hash: dict[tuple[str, str], str] = {}

    def upsert(self, rule: IncomingRule) -> tuple[StoredRule, bool]:
        """Insert a new version or return the existing one unchanged.

        Returns (row, created). `created` is False on a replayed change, so the
        caller writes an audit record only for genuinely new versions.
        """
        key = (rule.state, rule.rule_id, rule.version)
        new_hash = rule.content_hash()
        existing = self._rows.get(key)
        if existing is not None:
            if existing.content_hash != new_hash:
                # Same version string, different body — a publisher error we must
                # never silently overwrite.
                raise IncrementalSyncError(
                    f"version {rule.version} of {rule.state}/{rule.rule_id} "
                    f"already stored with a different content hash"
                )
            return existing, False

        row = StoredRule(
            state=rule.state,
            rule_id=rule.rule_id,
            version=rule.version,
            content_hash=new_hash,
            effective_from=rule.effective_from,
            effective_to=rule.effective_to,
            source_bulletin=rule.source_bulletin,
        )
        self._rows[key] = row
        self._current_hash[(rule.state, rule.rule_id)] = new_hash
        return row, True

    def current_hashes(self) -> dict[tuple[str, str], str]:
        return dict(self._current_hash)

The guard against a re-used version string carrying a different body is the crux of idempotency: (state, rule_id, version) must be an immutable identity. If a publisher ships two different rule bodies under one version, that is a data fault to surface loudly, not a silent last-write-wins overwrite that corrupts replay.

Each genuinely new version produces one change record — old hash, new hash, effective window, source bulletin — written to the append-only audit store before the high-water mark advances. The mark moves only after the whole batch is durably applied, so a mid-run crash re-fetches the unfinished tail instead of skipping it.

Copy
@dataclass(frozen=True, slots=True)
class RuleChangeRecord:
    state: str
    rule_id: str
    from_hash: str | None
    to_hash: str
    version: str
    effective_from: str
    source_bulletin: str
    applied_at: str


def apply_batch(
    changed: list[IncomingRule],
    store: RuleStore,
    prior_hashes: dict[tuple[str, str], str],
) -> list[RuleChangeRecord]:
    """Upsert each changed rule and emit an audit record for the new versions."""
    records: list[RuleChangeRecord] = []
    for rule in changed:
        _, created = store.upsert(rule)
        if not created:
            log.info("incremental.replayed", state=rule.state, rule_id=rule.rule_id,
                     version=rule.version)
            continue
        record = RuleChangeRecord(
            state=rule.state,
            rule_id=rule.rule_id,
            from_hash=prior_hashes.get((rule.state, rule.rule_id)),
            to_hash=rule.content_hash(),
            version=rule.version,
            effective_from=rule.effective_from.isoformat(),
            source_bulletin=rule.source_bulletin,
            applied_at=datetime.now(timezone.utc).isoformat(),
        )
        records.append(record)
        log.info("incremental.applied", state=rule.state, rule_id=rule.rule_id,
                 version=rule.version, source_bulletin=rule.source_bulletin)
    return records


def next_high_water_mark(applied: list[IncomingRule], previous: datetime) -> datetime:
    """Advance the mark to the newest source timestamp actually applied."""
    if not applied:
        return previous
    return max(r.source_updated_at for r in applied)

The mark is derived from source_updated_at — the feed’s own change clock — not from wall-clock “now”, so clock skew between the platform and a DOI publisher can never cause the next run to skip a change that landed during processing.

Prove the three defect classes named above are closed: an unchanged rule is skipped, a redelivered change is idempotent, and a version collision is rejected rather than silently overwritten.

Copy
from datetime import date, datetime, timezone


def _rule(body: dict[str, str], *, version: str = "2026.07.1") -> IncomingRule:
    return IncomingRule(
        state="FL", rule_id="FL-HURR-DED", version=version,
        effective_from=date(2026, 7, 1), effective_to=None,
        source_bulletin="OIR-2026-07", body=body,
        source_updated_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
    )


def test_unchanged_rule_is_skipped() -> None:
    rule = _rule({"min_pct": "2.0"})
    stored = {("FL", "FL-HURR-DED"): rule.content_hash()}
    changed, skipped = changed_rules([rule], stored)
    assert changed == [] and skipped == 1


def test_redelivered_change_is_idempotent() -> None:
    store = RuleStore()
    rule = _rule({"min_pct": "2.0"})
    _, created_first = store.upsert(rule)
    _, created_second = store.upsert(rule)          # broker replay
    assert created_first is True and created_second is False


def test_version_collision_is_rejected() -> None:
    store = RuleStore()
    store.upsert(_rule({"min_pct": "2.0"}))
    try:
        store.upsert(_rule({"min_pct": "5.0"}))     # same version, new body
    except IncrementalSyncError:
        pass
    else:
        raise AssertionError("expected IncrementalSyncError on version collision")

Run the suite under python -m pytest -q. The decisive signal is that the skip count and the created flag are exact: a correct incremental run reports precisely the number of rules that changed, and re-running it against the same feed applies zero new versions.

Because every applied change writes a RuleChangeRecord carrying the prior and new content hashes, the effective window, and the source bulletin, the incremental sync produces a per-rule change history that reads as a diff, not a full re-dump. When a market-conduct examiner asks which rules changed in a given sync and what each changed from, the answer comes directly from the append-only ledger designed in Audit Log Schema Design — the from_hash/to_hash pair reconstructs the exact transition, and the effective window proves the rule governed the correct claims. This satisfies the NAIC and state-DOI expectation that a carrier can demonstrate, rule by rule, that each claim was adjudicated under the regulation in force on its date of loss, the same traceability the State Regulation Mapping engine relies on when it defends an individual routing decision.

• Every rule re-applies on every run. Symptom: the change count equals the full corpus each night and the audit log floods. Cause: the content hash covers volatile metadata (a re-issued bulletin id or fetch timestamp) rather than the substantive body. Fix: hash only the rule body with sorted keys, as in IncomingRule.content_hash, so cosmetic feed churn does not register as change.
• A retried poll double-applies a change. Symptom: two staged versions for one (state, rule_id). Cause: the upsert key omits the version, so a redelivery inserts a duplicate. Fix: key the upsert on (state, rule_id, version) and return the existing row when created is False.
• A genuine change is skipped after a crash. Symptom: an amendment never lands in the store. Cause: the high-water mark advanced before the batch was durably applied, so the crashed tail is now behind the mark. Fix: advance the mark only after the batch is committed, and derive it from source_updated_at so re-fetching is safe.
• Effective window shifts on an unchanged rule. Symptom: a claim resolves against a window that was never published. Cause: an unchanged rule was re-ingested with today’s date. Fix: skip unchanged rules by hash before upsert so the stored effective_from is never rewritten.
• A version collision passes silently. Symptom: two different rule bodies share one version string and replay is no longer deterministic. Cause: the store overwrites on key match instead of validating the hash. Fix: raise IncrementalSyncError when a stored version’s hash differs from the incoming one, and treat it as a publisher data fault.

• Regulatory Sync Pipelines — the parent guide whose RuleSet and effective-date machinery this incremental loop extends
• Audit Log Schema Design — the append-only ledger that stores every per-rule change record
• State Regulation Mapping — the engine that evaluates claims against the versions this sync keeps current
• Dynamic Threshold Tuning — a downstream consumer whose thresholds track the rule updates applied here
• Coverage Validation Rules — the eligibility gate that reads the effective-dated rules this pipeline maintains