Data Reliability and Fault Tolerance Questions

Direct answer

When the third-party API itself cannot be made idempotent (no idempotency-key parameter, no way to query "did this already happen"), the fix has to move up a layer: your own system becomes the source of idempotency, and the API call becomes just one, possibly-retried, step inside a larger idempotent workflow. Three system-level patterns do this: the outbox pattern (durably record intent to call before calling, so a crash mid-call is always recoverable), a dedicated idempotency proxy (a thin service in front of the third-party API that adds the idempotency the API itself lacks), and compensated eventual-consistency workflows (accept that a duplicate or failed call can happen, and design an explicit compensating step to detect and correct it after the fact).

Structured elaboration

Why the API being at-most-once (not the pipeline's choice) changes the problem. With an idempotent API, retries are free: send the same request with the same key, get the same result. With a strictly at-most-once, non-idempotent API, every retry is a genuine risk of a second real effect (a second charge, a second email sent), and every NON-retry after an ambiguous failure (timeout with unknown server-side outcome) is a genuine risk of a missed effect. You cannot fix this by changing your own retry logic alone; you need a mechanism to know, independently of the API's own response, whether the call already happened.

Pattern 1: The outbox pattern.

• Mechanism: within the same local transaction as the business logic that decides "call the API," write a row to an outbox table recording the intended call and a unique operation ID. A separate, durable process reads unconfirmed outbox rows and calls the API, marking the row confirmed only after getting a definitive success response. On crash before confirmation, the outbox row is still there on restart, so the (still-pending) call is retried; a crash AFTER a successful call but before marking it confirmed risks exactly one duplicate call, which the outbox pattern alone does not prevent, only bounds to "at most once more."
• Trade-offs: cheap and simple to build on top of any transactional local database; does not by itself solve the ambiguous-timeout case (the actual API call still might have succeeded silently), only guarantees the ATTEMPT is never silently lost.

Pattern 2: A dedicated idempotency proxy.

• Mechanism: introduce a small internal service that sits between your pipeline and the third-party API. Your pipeline calls the proxy with a caller-generated idempotency key; the proxy checks its own durable store for that key. If seen before, it returns the previously recorded outcome without calling the API again. If not seen, it calls the API, durably records the outcome (success, failure, or genuinely unknown, e.g. after its own timeout waiting on the API), and returns that outcome, all before considering the operation complete, so the proxy's own record becomes the authoritative "did this happen" answer that the third-party API cannot provide.
• Trade-offs: this actually solves the ambiguous-timeout case (the proxy's durable record, not the API's response, is the source of truth), at the cost of building and operating an extra stateful service, and the proxy's own store needs the same crash-recoverability discipline (a WAL, a transactional database) it is providing to everything downstream of it.

Pattern 3: Compensated eventual-consistency workflows.

• Mechanism: accept the API may be called more than once (or, in some rare failure interleavings, produce an effect you cannot detect), and instead of preventing the duplicate, detect it after the fact (reconciling against the API's own downstream records, if it exposes any, such as a transaction history endpoint) and issue a compensating action (a cancellation, a refund, a corrective follow-up call) when a duplicate is found.
• Trade-offs: this is the fallback when the API supports neither a native idempotency mechanism NOR any reasonable proxy-level prevention (e.g., a one-way SMS API with no way to check delivery history at all); it accepts a real, sometimes user-visible window of duplication in exchange for being implementable against literally any API, however primitive.

Worked example

A pipeline calls a legacy shipping-label API (no idempotency key support) to generate a shipping label and charge a small label fee, and observes a 1.5% rate of ambiguous timeouts (request sent, no response received, unknown whether the label was actually generated) across 200,000 calls/month:

Naive retry-on-timeout: all 3,000 are blindly retried, and if roughly half of those ambiguous timeouts were ACTUALLY successful server-side (a stated assumption; the true fraction is unknown by definition, which is exactly the problem), that is:

With an idempotency proxy in front of the API: the proxy durably records "call in flight" before calling, and on the ambiguous timeout, the retry goes to the PROXY (which already has a durable record of an in-flight, outcome-unknown call), not directly to the API. The proxy's own policy (poll the API's separate order-status endpoint if one exists, or apply a conservative "treat unknown as done, require manual reconciliation" rule if it does not) replaces "duplicate call happens 1,500 times a month" with "1,500 calls per month go into an explicit reconciliation queue," a bounded, observable, and fixable number instead of a silent, ongoing revenue and customer-experience leak.

Trade-offs and pitfalls

• Common mistake: treating a timeout as either 'definitely failed, retry' or 'definitely succeeded, don't retry.' Neither is safe for a non-idempotent API; the honest state is "unknown," and the system-level pattern exists precisely to make that unknown state explicit and durably tracked rather than guessed at.
• The outbox pattern alone does not fully close the gap. It guarantees the ATTEMPT is never lost, but a crash between "API call succeeded" and "outbox row marked confirmed" still risks exactly one duplicate retry on restart; an idempotency proxy (pattern 2) or a reconciliation step (pattern 3) is needed to close that remaining, narrower window.
• Building a proxy is a real engineering investment, not a quick fix. It is worth it when call volume and the cost of duplication (a real financial charge, a real physical shipment) justify a dedicated stateful service; for a low-volume, low-stakes integration, accepting the compensating-transaction fallback may be the more proportionate choice.
• Compensating actions need their own idempotency discipline. A compensation issued twice for the same detected duplicate (a double refund) is the exact same class of bug this whole answer exists to prevent, just one layer further out; the reconciliation/compensation step must track which duplicates it has already compensated.

Direct answer

A circuit breaker wraps a downstream call (an API request from a data pipeline) with a state machine that tracks that dependency's recent health and stops calling it entirely once it looks broken, rather than letting every caller keep retrying into a known-failing dependency. It has three states: CLOSED (normal, calls pass through), OPEN (the dependency is considered failing, calls are rejected immediately without even attempting the network call), and HALF-OPEN (a cooldown has elapsed, a small number of probe calls are allowed through to test recovery). This protects both the caller (no more time wasted waiting on doomed calls) and the struggling dependency (no continued load from callers who cannot succeed anyway).

Structured elaboration

Failure threshold. The circuit opens once a configured fraction of recent calls fail (e.g., more than 50% of the last 20 calls, or more than N consecutive failures), not on the first single failure, since a single transient blip should not trip a breaker meant to catch SUSTAINED trouble; the exact threshold trades false-positive risk (opening on normal, isolated hiccups) against false-negative risk (staying closed too long into a genuine outage, still sending traffic that will fail).

Cooldown window. Once open, the circuit stays open for a fixed cooldown period before allowing any probe calls, giving the struggling dependency time to recover WITHOUT continued load from this caller during that window. Too short a cooldown re-opens the circuit into a still-broken dependency repeatedly (thrashing); too long delays recovery detection once the dependency IS actually healthy again.

Interaction with retry/backoff. The circuit breaker and retry-with-backoff operate at different granularities and are complementary, not redundant: backoff governs how AGGRESSIVELY a single caller retries an individual failed call, while the circuit breaker governs whether to attempt the call AT ALL, given the dependency's recent aggregate health. A well-designed system checks the circuit breaker state FIRST (fail fast if open, skip backoff entirely) and only applies backoff-and-retry logic for calls that proceed because the circuit is closed or half-open.

Interaction with backpressure. When the circuit is open, calls that would have gone to the failing dependency are rejected immediately rather than queued indefinitely; this is itself a form of backpressure, signaling upstream (the pipeline stage feeding this call) to either buffer (write to a durable queue), drop, or reroute, rather than accumulating unbounded in-flight work waiting on a dependency that will not respond in time anyway.

Worked example

A pipeline calls a downstream enrichment API at 2,000 requests/sec. The dependency begins failing at 80% of requests (a partial but severe degradation). With a threshold of "open if more than 50% of the last 20 calls failed": within roughly $20 / 2{,}000 = 0.01$ seconds of the degradation beginning (the time to accumulate 20 calls at this rate), the failure ratio crosses 50%, and the circuit opens. From that point, roughly:

worth of requests per second of cooldown are rejected immediately (fast, cheap rejections) instead of each attempting a doomed network call and waiting for its own timeout; at a 30-second cooldown, this is 60,000 requests that would otherwise have each paid a network round-trip's worth of latency waiting to fail, now failing in microseconds instead. After the cooldown, a small number of half-open probe calls (not all 2,000/sec resuming at once) test whether the dependency has recovered; if they succeed, the circuit closes and full traffic resumes; if they still fail, the circuit reopens for another cooldown period.

Trade-offs and pitfalls

• Common mistake: opening on the first failure. This makes the breaker indistinguishable from "stop on any error," far too aggressive for a dependency that has occasional, normal transient blips; the threshold should reflect SUSTAINED degradation, not any single failure.
• Common mistake: resuming full traffic immediately after cooldown instead of a small half-open probe. Sending all 2,000 requests/sec back at once the instant cooldown ends risks immediately re-triggering the same overload/failure condition if the dependency has not FULLY recovered, undoing the cooldown's benefit in the first probe cycle.
• A circuit breaker without any coordination with backpressure just moves the problem, not solves it, per the worked example: rejected calls still need somewhere to go (buffer, drop, or reroute), a design decision separate from the circuit breaker itself.
• Per-dependency circuit state, not a single global breaker, is essential once a pipeline calls MULTIPLE downstream dependencies; a struggling dependency should not trip a breaker that also blocks calls to healthy, unrelated dependencies.

Direct answer

DynamoDB's PutItem with ConditionExpression="attribute_not_exists(pk)" gives an atomic, per-item conditional write: the write succeeds only if no live item with that key exists yet, which is exactly the primitive an idempotency store needs. The code below models that operation (no live AWS table is reachable from this sandbox, so it is a faithful in-memory model of the same conditional-write, TTL, and throttling behavior DynamoDB provides per item, not a simplification of it) and demonstrates hot-key handling, throttle-and-retry, and recovery after a network-partition-length delay.

Structured elaboration

Conditional write as the idempotency primitive. Every ingestion attempt calls a conditional put keyed by the message's idempotency_key. If the key does not yet exist (or its TTL has expired), the put succeeds and the item is written, this is the "process it" path. If the key exists and has not expired, DynamoDB raises ConditionalCheckFailedException, which the caller interprets as "already processed, this is a duplicate delivery, no-op successfully," not as an error.

Hot-key handling. A single idempotency key receiving many concurrent write attempts (e.g. a burst of retries for the same message from multiple consumer instances) all contend for the SAME DynamoDB partition key. DynamoDB's own partition-level throughput is bounded per key; a real table would throttle (ProvisionedThroughputExceededException) under sustained hot-key contention. The client-side mitigation is retry-with-exponential-backoff-and-jitter, which spreads the retries out in time instead of hammering the same key in a tight loop.

Expiry of idempotency metadata. The TTL bounds storage: DynamoDB's native TTL feature deletes expired items automatically in the background, so the dedup store's size tracks recent activity, not total lifetime volume. This is a deliberate, stated trade-off: a redelivery that arrives AFTER the TTL window (for example, a message delayed by a long network partition) is treated as brand new and reprocessed, since the pipeline has genuinely forgotten it saw that key before.

Recovery after a network partition. If a partition separates the ingestion workers from the DynamoDB table for longer than the message's redelivery timeout, at-least-once delivery guarantees the message is redelivered once connectivity is restored. If that redelivery happens within the TTL window, the conditional write correctly identifies it as a duplicate (if the original write got through before the partition) or processes it fresh (if the original write never got through). If the partition outlasts the TTL, the dedup guarantee has a bounded, known gap: this is why the TTL should be set relative to the pipeline's actual maximum plausible outage duration, not an arbitrary default.

Worked example

"""
At-least-once ingestion with a DynamoDB-style dedup store. Models DynamoDB's
PutItem ConditionExpression="attribute_not_exists(pk)" with a lock-protected
dict (a striped lock per key-shard, matching DynamoDB's per-item atomicity).
"""
import time, threading, random
from collections import Counter


class ConditionalWriteFailed(Exception):
    pass


class MockDynamoDBTable:
    def __init__(self, num_shards=16):
        self._shards = [dict() for _ in range(num_shards)]
        self._locks = [threading.Lock() for _ in range(num_shards)]
        self._num_shards = num_shards
        self.throttle_events = 0
        self.throttle_lock = threading.Lock()

    def _shard_for(self, key):
        return hash(key) % self._num_shards

    def conditional_put(self, key, value, ttl_seconds, throttle_prob=0.0):
        if throttle_prob and random.random() < throttle_prob:
            with self.throttle_lock:
                self.throttle_events += 1
            raise ConditionalWriteFailed("ProvisionedThroughputExceededException (simulated)")
        shard_idx = self._shard_for(key)
        with self._locks[shard_idx]:
            store = self._shards[shard_idx]
            now = time.time()
            existing = store.get(key)
            if existing is not None:
                _, expires_at = existing
                if expires_at > now:
                    raise ConditionalWriteFailed(f"ConditionalCheckFailedException: {key} already processed")
            store[key] = (value, now + ttl_seconds)
            return True

    def size(self):
        return sum(len(s) for s in self._shards)


def ingest_with_retry(table, idempotency_key, payload, ttl_seconds=3600, max_attempts=8, throttle_prob=0.0):
    """At-least-once ingestion: conditional put; ConditionalCheckFailed == duplicate no-op;
    throttling exception == retry with exponential backoff."""
    backoff = 0.0005
    for attempt in range(1, max_attempts + 1):
        try:
            table.conditional_put(idempotency_key, payload, ttl_seconds, throttle_prob=throttle_prob)
            return "written", attempt
        except ConditionalWriteFailed as e:
            if "already processed" in str(e):
                return "duplicate_noop", attempt
            time.sleep(backoff)
            backoff *= 2
    return "failed_after_retries", max_attempts


def main():
    table = MockDynamoDBTable(num_shards=16)

    # 1. Same idempotency_key delivered 3 times: only the first writes.
    results = [ingest_with_retry(table, "evt-1", {"amount": 10})[0] for _ in range(3)]
    print("Three deliveries of evt-1:", results)

    # 2. Hot-key: 200 threads write the SAME key concurrently.
    hot_key_results, hk_lock = [], threading.Lock()
    def hot_key_worker():
        r, _ = ingest_with_retry(table, "hot-key-evt", {"amount": 5})
        with hk_lock:
            hot_key_results.append(r)
    threads = [threading.Thread(target=hot_key_worker) for _ in range(200)]
    for t in threads: t.start()
    for t in threads: t.join()
    counts = Counter(hot_key_results)
    print("Hot-key test: 200 concurrent deliveries of the SAME key ->", dict(counts))
    assert counts["written"] == 1

    # 3. Throttling + retry: 30% real per-attempt throttle probability, 500 distinct keys.
    random.seed(42)
    distinct_results, attempt_counts = [], []
    for i in range(500):
        r, attempts = ingest_with_retry(table, f"evt-batch-{i}", {"i": i}, max_attempts=15, throttle_prob=0.3)
        distinct_results.append(r)
        attempt_counts.append(attempts)
    written_count = distinct_results.count("written")
    failed_count = distinct_results.count("failed_after_retries")
    avg_attempts = sum(attempt_counts) / len(attempt_counts)
    print(f"Throttled batch (30% throttle prob/attempt): {written_count}/500 written, "
          f"{failed_count}/500 failed, throttle_events: {table.throttle_events}, "
          f"avg attempts/key: {avg_attempts:.2f}")
    assert table.throttle_events > 0, "throttle path never fired: test would be vacuous"
    assert failed_count == 0

    # 4. Network-partition simulation: force evt-1's TTL to look expired, then redeliver.
    # Must reprocess: a redelivery past the TTL window is treated as new (the deliberate,
    # stated trade-off of a bounded TTL).
    shard_idx = table._shard_for("evt-1")
    with table._locks[shard_idx]:
        value, _ = table._shards[shard_idx]["evt-1"]
        table._shards[shard_idx]["evt-1"] = (value, time.time() - 1)
    post_expiry_result, _ = ingest_with_retry(table, "evt-1", {"amount": 10})
    print("Redelivery of evt-1 AFTER forced TTL expiry (simulated long network partition):", post_expiry_result)
    assert post_expiry_result == "written"


if __name__ == "__main__":
    main()

Output (actually executed with python3):

Three deliveries of evt-1: ['written', 'duplicate_noop', 'duplicate_noop']
Hot-key test: 200 concurrent deliveries of the SAME key -> {'written': 1, 'duplicate_noop': 199}
Throttled batch (30% throttle prob/attempt): 500/500 written, 0/500 failed, throttle_events: 235, avg attempts/key: 1.47
Redelivery of evt-1 AFTER forced TTL expiry (simulated long network partition): written

The hot-key test proves the conditional write's per-item atomicity: 200 threads racing on the identical key produce exactly 1 written and 199 duplicate_noop, never a double-write. The throttling test is deliberately not vacuous: throttle_events: 235 confirms the simulated ProvisionedThroughputExceededException path genuinely fired (at a stated 30% per-attempt probability across 500 keys and up to 15 attempts each), and every one of the 500 distinct keys still ends up written (0/500 failed), at a measured average of 1.47 attempts per key, the concrete cost of retry-with-backoff recovering from throttling without losing data. The final line confirms the TTL-expiry/network-partition trade-off is real: forcing evt-1's entry to look expired and redelivering it produces written, not duplicate_noop, exactly the documented behavior of a bounded TTL.

Trade-offs and pitfalls

• Common mistake: choosing the idempotency key's partition strategy without regard for hot-key risk. If many logical operations share one coarse key (e.g. keying by user_id for a user who issues many rapid operations), that single partition absorbs a disproportionate share of write traffic; a well-distributed per-operation key (per-event UUID or a composite key) avoids concentrating throughput on one partition.
• Common mistake: retrying a ConditionalCheckFailedException as if it were a transient throttling error. The two exceptions mean opposite things (already-done vs try-again) and must be handled by different branches, exactly as the code above distinguishes them by message content.
• TTL sizing is a real production decision, not a default. Too short, and a genuinely delayed but valid redelivery (post-partition) reprocesses and risks a real duplicate at the downstream sink; too long, and storage cost and item count grow for no benefit. Size it against measured maximum redelivery delay, including partition and consumer-restart scenarios, not a guess.
• This dedup store only protects the WRITE TO ITSELF being idempotent, not necessarily the downstream sink. If ingest_with_retry's caller performs a non-idempotent side effect (an external API call) between claiming the key and confirming success, a crash in that narrow window can still produce a duplicate side effect even though the DynamoDB record itself behaved correctly. This same class of gap appears with any datastore-backed idempotency store, not just DynamoDB: the store being correct never guarantees the caller's own side effect is idempotent.

Direct answer

Checkpointing captures the state needed to resume a running job from where it left off, periodically and (in modern stream processors) asynchronously, without stopping the pipeline; snapshotting usually refers to a full, consistent copy of all operator state at one instant. In practice the terms overlap (Flink calls its periodic mechanism "checkpointing" and reserves "savepoint" for a manually triggered, more portable full snapshot), but the useful distinction for interview purposes is incremental versus full: an incremental checkpoint records only what changed since the last one, a full snapshot re-persists everything. For 50,000 events/sec, incremental checkpointing every 10 to 30 seconds is a reasonable default, and an on-call engineer restoring after a crash needs the last completed checkpoint's state, the source offsets it corresponds to, and confirmation that external sinks are consistent with that same checkpoint (not ahead of it).

Structured elaboration

What gets checkpointed and why. Three things: operator state (aggregation accumulators, windows in flight, join buffers), the position in the input (source offsets, e.g. a Kafka partition-offset map), and enough metadata to know these two are mutually consistent (they describe the exact same logical point in the stream). Without this, restoring operator state without also restoring the matching input position either replays already-processed events against already-updated state (double counting) or skips events the state does not yet reflect (data loss).

Implementation. A full snapshot pauses (or barrier-synchronizes) all operators, serializes their complete state to durable storage (S3, HDFS, a distributed filesystem), and records the corresponding source offsets, all as one atomic unit. Incremental checkpointing (Flink's RocksDB-backed incremental checkpoints are the canonical example) instead persists only the state changes (the RocksDB SST files created since the last checkpoint) via a background process, using a barrier that flows through the operator graph (Chandy-Lamport-style) to mark a consistent cut without pausing processing.

Latency and throughput impact. A full, blocking snapshot pauses processing for its duration, directly adding to end-to-end latency and briefly dropping throughput to zero; this gets worse as state size grows, since a bigger state takes longer to serialize. Incremental, asynchronous checkpointing keeps the pipeline processing while checkpointing happens in the background, at the cost of a small, steady overhead (I/O contention, CPU for serialization) rather than a periodic pause, and a more complex recovery path (you may need to replay a chain of incremental deltas, not just load one snapshot).

Recovery-time trade-off. Full snapshots are simpler and faster to restore from (load one complete state), since there is nothing to replay past that. Incremental checkpoints are cheaper to take but can be slower to restore from if many increments have accumulated since the last full base snapshot, since recovery has to apply that whole chain. Production systems bound this by periodically taking a full snapshot anyway (e.g. Flink's incremental checkpointing still relies on RocksDB's own compaction to prevent an unbounded chain of deltas).

When incremental beats full. Once state size grows large relative to the achievable checkpoint interval (state in the tens of GB or larger per operator, common for large windowed aggregations or ML feature stores), a full snapshot's pause time becomes unacceptable for a low-latency SLA. Incremental checkpointing is the standard answer: state size no longer directly gates checkpoint duration, since each interval only pays for what changed.

Worked example

For a pipeline at 50,000 events/sec with a stated low-latency requirement, the checkpoint interval is chosen by balancing two costs: checkpoint overhead (more frequent = more steady-state cost) against worst-case recovery time (less frequent = a longer replay window after a crash).

At a 20-second interval: $50{,}000 \times 20 = 1{,}000{,}000$ events of worst-case replay. At a 5-second interval: $50{,}000 \times 5 = 250{,}000$ events, a 4x reduction in worst-case recovery replay volume, at 4x the checkpointing frequency (4x the steady-state overhead cost). A reasonable default for a genuinely low-latency pipeline is 10 to 20 seconds: frequent enough to bound recovery to roughly the same order of magnitude as the pipeline's own end-to-end latency target, infrequent enough that checkpoint overhead stays a small fraction of total processing time. The exact number is a measured trade-off specific to the operator's state size and I/O budget, not a fixed constant; teams tune it empirically against their own checkpoint-duration metrics.

What an on-call engineer needs to restore after an operator crash. Three concrete facts, in order: (1) the ID/timestamp of the last successfully completed checkpoint (not one that was in progress and aborted), (2) the exact source offsets recorded in that checkpoint's metadata, so the job resumes reading from precisely that point, not from "wherever it last was" which may be ahead of the checkpoint, and (3) whether external sinks (a database, a downstream topic) are consistent with that same checkpoint or need reconciliation, since a checkpoint only captures the job's own internal state; a sink write acknowledged after the checkpoint but not yet reflected anywhere durable on the source-offset side can be silently redone (safe only if the sink write is itself idempotent) or silently lost (if it is not), which is exactly why idempotent-sink design (a keyed upsert or dedup check at the write) and checkpoint design must be engineered together, not independently.

Trade-offs and pitfalls

• Common mistake: restoring operator state without restoring the exact matching offsets. This desynchronizes "what the job thinks it processed" from "what it will process next," producing either duplicate processing (offsets too far behind) or silent gaps (offsets too far ahead).
• Common mistake: treating "checkpoint completed" and "checkpoint durable" as the same event. A checkpoint write acknowledged by the local filesystem but not yet replicated to durable storage (S3, a replicated filesystem) can be lost in the same failure that triggered the need to recover from it; the checkpoint storage backend needs its own durability guarantee independent of the job's own process.
• Barrier alignment cost. Chandy-Lamport-style barrier checkpointing requires operators with multiple input streams to wait for the barrier to arrive on all inputs before proceeding (alignment), which under skewed or backpressured inputs can stall the fastest input waiting for the slowest; unaligned checkpointing (buffering in-flight records instead of blocking) trades a larger checkpoint size for lower tail latency under skew.
• State size, not event rate alone, usually dominates checkpoint duration. A pipeline at low event rate but with very large windowed state (e.g. a long session window holding months of history) can have checkpoint duration problems a naive "we're only at 500 events/sec, checkpointing should be cheap" assumption misses; always size the checkpoint interval against measured state size and serialization time, not event throughput alone.

Direct answer

The write-intent pattern uses Redis's atomic SET key value NX (set only if not already present) to claim a message ID before doing the real write, and marks a second, distinct DONE value only after the write actually succeeds. TTL bounds how long a claim is remembered, preventing unbounded growth. The code below implements and runs this, and honestly demonstrates the pattern's real, narrow failure mode: a crash between "the write succeeded" and "DONE was recorded" can, after the claim's TTL lease expires, produce a genuine second write. That gap is why this pattern alone bounds duplication rather than eliminating it, and why the downstream write itself should also be idempotent as defense in depth.

Structured elaboration

Why two states, INTENT and DONE, not just presence/absence. A naive version might use SET NX alone as "have I seen this ID," treating any existing key as fully processed. That is wrong: a worker that claimed the key (wrote INTENT) but crashed before finishing the real write leaves a key that exists but represents unfinished work, not completed work. Distinguishing INTENT (claimed, in progress or dead) from DONE (confirmed complete) lets a redelivery correctly report contention instead of silently skipping work that was never actually done.

The crash-recovery boundary. While a claim's TTL lease is still live, a competing or replayed delivery correctly reports in_flight_contention and does not re-write, which is safe: either the original claimant is still working (don't duplicate its effort) or it died and will eventually be recovered by lease expiry (handled next), but re-writing NOW, while the lease still looks live, risks racing the possibly-still-alive original claimant. Once the lease expires (the TTL passes with no DONE recorded), a later redelivery is treated as abandoned work and reprocessed from scratch.

The gap this creates, demonstrated rather than asserted. If the crash happens AFTER the real write already succeeded but BEFORE DONE was recorded, the abandoned-claim recovery path described above cannot tell that apart from "claimed but never actually written," and will re-run the write once the lease expires. This is a real, bounded limitation, not a hypothetical: the code below constructs exactly this interleaving and confirms it produces two real writes for one logical message.

Worked example

"""
Idempotent writes to an external datastore via Redis-tracked processed
message IDs (write-intent + TTL). Uses a mock Redis (lock-protected dict
implementing SET NX / GET / EXPIRE) since no live Redis is available in this
sandbox; the mock enforces the same atomic set-if-not-exists contract real
Redis SET key value NX gives.
"""
import time, threading
from collections import Counter, defaultdict


class MockRedis:
    def __init__(self):
        self._data = {}
        self._lock = threading.Lock()

    def set_nx(self, key, value, ex=None):
        with self._lock:
            now = time.time()
            entry = self._data.get(key)
            if entry is not None:
                _, expires_at = entry
                if expires_at is None or expires_at > now:
                    return False
            self._data[key] = (value, (now + ex) if ex else None)
            return True

    def get(self, key):
        with self._lock:
            entry = self._data.get(key)
            if entry is None:
                return None
            value, expires_at = entry
            if expires_at is not None and expires_at <= time.time():
                return None
            return value

    def set(self, key, value, ex=None):
        with self._lock:
            self._data[key] = (value, (time.time() + ex) if ex else None)

    def force_expire(self, key):
        with self._lock:
            if key in self._data:
                value, _ = self._data[key]
                self._data[key] = (value, time.time() - 1)


INTENT, DONE, TTL_SECONDS = "intent", "done", 3600
external_writes_lock = threading.Lock()
external_write_count_by_key = defaultdict(int)


def write_to_external_datastore(message_id, payload):
    with external_writes_lock:
        external_write_count_by_key[message_id] += 1


def process_message(redis, message_id, payload):
    """1) atomically claim message_id via SET NX (write-intent). 2) if claim
    fails: DONE already recorded -> no-op; still INTENT -> report contention,
    do NOT silently re-write. 3) if we own the claim: do the real write, then
    mark DONE."""
    claimed = redis.set_nx(f"proc:{message_id}", INTENT, ex=TTL_SECONDS)
    if not claimed:
        if redis.get(f"proc:{message_id}") == DONE:
            return "already_done_noop"
        return "in_flight_contention"
    write_to_external_datastore(message_id, payload)
    redis.set(f"proc:{message_id}", DONE, ex=TTL_SECONDS)
    return "written"


def main():
    redis = MockRedis()

    r = [process_message(redis, "msg-1", {"amount": 10}) for _ in range(3)]
    print("Three deliveries of msg-1:", r)
    assert external_write_count_by_key["msg-1"] == 1

    conc_results, conc_lock = [], threading.Lock()
    def worker():
        res = process_message(redis, "concurrent-msg", {"amount": 3})
        with conc_lock:
            conc_results.append(res)
    threads = [threading.Thread(target=worker) for _ in range(50)]
    for t in threads: t.start()
    for t in threads: t.join()
    print("50 concurrent deliveries of the SAME new message_id:", dict(Counter(conc_results)))
    assert external_write_count_by_key["concurrent-msg"] == 1

    # Failure mode, demonstrated honestly: crash AFTER the external write
    # succeeds but BEFORE Redis is marked DONE (only INTENT recorded).
    write_to_external_datastore("msg-crash-after-write", {"amount": 55})   # the crashed worker's write, which DID succeed
    redis.set_nx("proc:msg-crash-after-write", INTENT, ex=TTL_SECONDS)     # ...but died before recording DONE

    replay_1 = process_message(redis, "msg-crash-after-write", {"amount": 55})
    print("Redelivery while the dead worker's INTENT lease is still live:", replay_1,
          "| writes so far:", external_write_count_by_key["msg-crash-after-write"])
    assert replay_1 == "in_flight_contention"
    assert external_write_count_by_key["msg-crash-after-write"] == 1

    redis.force_expire("proc:msg-crash-after-write")  # lease times out: nobody renewed it
    replay_2 = process_message(redis, "msg-crash-after-write", {"amount": 55})
    print("Redelivery AFTER the stale lease expired:", replay_2,
          "| writes total:", external_write_count_by_key["msg-crash-after-write"])
    assert replay_2 == "written"
    assert external_write_count_by_key["msg-crash-after-write"] == 2, \
        "expected a genuine SECOND write: this is the pattern's real failure mode"
    print("Confirmed: this interleaving produces 2 external writes for 1 logical message.")


if __name__ == "__main__":
    main()

Output (actually executed with python3):

Three deliveries of msg-1: ['written', 'already_done_noop', 'already_done_noop']
50 concurrent deliveries of the SAME new message_id: {'written': 1, 'already_done_noop': 49}
Redelivery while the dead worker's INTENT lease is still live: in_flight_contention | writes so far: 1
Redelivery AFTER the stale lease expired: written | writes total: 2
Confirmed: this interleaving produces 2 external writes for 1 logical message.

The concurrency test (50 threads, one new message ID) proves the claim step is genuinely atomic: exactly 1 written, 49 already_done_noop. The last section is the important, deliberately non-vacuous result: it constructs the specific crash-after-write-before-DONE interleaving, confirms the redelivery correctly refuses to re-write WHILE the lease looks live (in_flight_contention, writes still at 1), and then shows that once the lease expires, the SAME message genuinely gets written a second time (writes total: 2). This is run and printed, not merely claimed: the pattern bounds duplicate writes to this narrow, TTL-lease-width window, it does not eliminate them.

Trade-offs and pitfalls

• This is the pattern's real limitation, not a hypothetical edge case. Any system using this exact write-intent-then-DONE shape inherits this exact gap; the TTL/lease width directly bounds how narrow that gap is (a shorter lease shrinks the exposure window but increases false-positive reprocessing of still-alive, slow workers).
• The correct mitigation is defense in depth, demonstrated concretely above. Making the downstream write_to_external_datastore call itself idempotent (an upsert keyed by message_id) closes the gap completely: even if it is called twice, the second call converges rather than duplicates. The Redis layer alone should not be trusted as the sole line of defense for anything where a duplicate write has real cost.
• Common mistake: not renewing the lease for long-running writes. If the real write can legitimately take longer than the TTL, a healthy in-progress worker's own claim can expire and be "recovered" by someone else while it is still working, causing the exact same double-write demonstrated above even without an actual crash. A heartbeat that periodically extends the TTL while work is genuinely in progress avoids this.
• Common mistake: using Redis without any persistence or replication for a dedup store that must survive a Redis restart. If Redis itself is not configured with AOF persistence or replication, a Redis crash loses all INTENT/DONE state, silently reopening every in-flight message to reprocessing exactly as if every lease had expired at once.