Checkpoint and State Recovery › Exactly-Once Reconciliation with Idempotent Sinks

# Exactly-Once Reconciliation with Idempotent Sinks

This page answers a question every crash-safe streaming reconciler eventually hits: given that the transport delivers at least once and recovery replays from the last checkpoint, how do you guarantee that each discrepancy is recorded and each remediation applied exactly once? It sits under the checkpoint and state recovery reference and is the property that makes both routine restarts and recovering from checkpoint corruption safe. The insight is that you do not need distributed transactions across the source, the log, and the sink — you need a deterministic key and an idempotent write, which together make replay a no-op.

# Problem Framing

Your reconciler consumes a change stream, emits a discrepancy record when source and target disagree, and applies a remediation. After a crash it replays the last thirty seconds from the checkpoint. Without idempotency, that replay re-emits every discrepancy in the window and re-applies every remediation: discrepancy counts inflate, a re-sync runs twice, and an aggregate the remediation touched is now wrong. Exactly-once processing across independent systems is impossible in general, but exactly-once effect is achievable: assign each unit of work a deterministic identity derived from its inputs, and make the sink write conditional on that identity being new. Replay then produces the same keys, and the conditional write silently drops the duplicates.

# Implementation

Each discrepancy gets a deterministic id computed from the reconciliation inputs — key, event time, and sealing watermark — never from wall clock or a random value. The sink upserts on that id, and a bounded dedup store rejects a replayed id in the hot path so downstream effects run once.

import hashlib
import logging
import time
from dataclasses import dataclass
from typing import Dict, Optional

logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", level=logging.INFO
)
logger = logging.getLogger("recon.streaming.exactly_once")


@dataclass(frozen=True)
class Discrepancy:
    key: str
    event_time_ms: int
    sealing_watermark_ms: int
    source_hash: str
    target_hash: Optional[str]

    def dedup_id(self) -> str:
        """Deterministic identity: identical inputs always produce the same id."""
        material = f"{self.key}|{self.event_time_ms}|{self.sealing_watermark_ms}|{self.source_hash}|{self.target_hash}"
        return hashlib.sha256(material.encode("utf-8")).hexdigest()


@dataclass
class IdempotentSink:
    ttl_ms: int
    _seen: Dict[str, int] = None                # dedup_id -> first-seen wall clock

    def __post_init__(self) -> None:
        self._seen = {}

    def _evict(self, now: int) -> None:
        expired = [k for k, ts in self._seen.items() if now - ts > self.ttl_ms]
        for k in expired:
            del self._seen[k]

    def apply(self, d: Discrepancy, upsert) -> bool:
        """Apply a discrepancy exactly once; a replayed id is a no-op. Returns True if applied."""
        now = int(time.time() * 1000)
        self._evict(now)
        did = d.dedup_id()
        if did in self._seen:
            logger.info("duplicate suppressed id=%s key=%s", did[:12], d.key)
            return False
        # Conditional upsert: the storage layer must also be keyed on did so a
        # dedup-store miss after a long gap still cannot double-write.
        upsert(did, d)
        self._seen[did] = now
        return True

# Exactly-Once Strategy Trade-Off

The dedup key plus an idempotent write is the portable option; a transactional sink is stronger where the target supports it.

# Key Implementation Notes

• The id must be a pure function of inputs. Deriving dedup_id from wall clock, a UUID, or an attempt counter defeats the whole design, because a replay would produce a different id and write again. Only inputs that are identical on replay — key, event time, sealing watermark, and value hashes — may feed it.
• The durable sink must also be keyed on the id. The in-memory dedup store is a hot-path optimization with a TTL; after a long gap its entry may be evicted, so the storage layer itself must reject a duplicate id (an upsert or a unique constraint). The cache accelerates; the keyed write guarantees.
• Bound the dedup store or it becomes the leak. Size the TTL to the maximum replay window — a small multiple of the checkpoint interval — so the cache covers every realistic replay while staying bounded. An unbounded _seen map trades a duplication bug for a memory-exhaustion bug.
• Idempotency is what licenses walk-back recovery. Because replay is a no-op, a reconciler can safely resume from an older clean checkpoint after corruption, re-applying the intervening window without inflating any count.

# Verification

Assert that a replayed discrepancy is suppressed and that distinct discrepancies still apply.

applied = {}
sink = IdempotentSink(ttl_ms=60_000)
d = Discrepancy("k1", 1000, 1200, "abc", None)

assert sink.apply(d, lambda did, disc: applied.__setitem__(did, disc)) is True
assert sink.apply(d, lambda did, disc: applied.__setitem__(did, disc)) is False  # replay = no-op
assert len(applied) == 1

d2 = Discrepancy("k1", 1000, 1200, "abc", "xyz")   # target changed -> new identity
assert sink.apply(d2, lambda did, disc: applied.__setitem__(did, disc)) is True
assert len(applied) == 2
logger.info("exactly-once idempotent sink verified")

# Operational Considerations

Expose duplicates_suppressed_total so a healthy replay after a restart is visible and expected rather than alarming, and alert only when suppression stays elevated in steady state — that signals a redelivery storm upstream, not a benign recovery. Keep the dedup store’s TTL tied to the checkpoint interval so the two evolve together, and make the durable sink’s keyed constraint the source of truth for uniqueness, treating the cache as disposable. Because a discrepancy record can carry divergent regulated values, apply the same masking and retention the security boundaries for reconciliation reference defines, and record each suppressed duplicate so an auditor can confirm that a replayed window produced no double-counted effect.

# Related

• Checkpoint and state recovery — the parent reference whose replay this property makes safe.
• Recovering from checkpoint corruption — the walk-back procedure that depends on idempotent replay.
• Column-level checksum generation — the value hashes that feed the deterministic dedup id.
• Discrepancy routing and remediation — the downstream path whose effects must be applied exactly once.