Data Consistency and Distributed Transactions Questions

Direct answer: When reads go through proxies, caches, or CDNs (content delivery networks) that may be stale relative to the origin, read-your-writes needs an explicit mechanism to either route a client's post-write reads to a sufficiently fresh source, or to make the client's next read wait until the cache has caught up; the main approaches are sticky routing to a known-fresh node, a client-carried version token the read path checks against, cache invalidation on write, and directing specific reads to the origin when freshness is required.

Structured elaboration

The core problem. A cache or CDN's whole value proposition is serving reads without hitting the origin every time, which is fundamentally in tension with "always show me my own latest write immediately." Any solution has to selectively bypass or catch up the cache specifically for the writer's own subsequent reads, without giving up caching's benefit for everyone else's reads.

Sticky sessions to the origin (or a fresh-enough replica). For a bounded window after a write, route that specific client's reads directly to the origin (or a replica known to be current), bypassing the cache/CDN entirely. Simple to reason about, but doesn't scale gracefully if a large fraction of traffic is write-then-immediately-read, since it defeats caching exactly when caching would otherwise help most (right after a popular item changes).

Client-side version tokens. The write response includes a version/timestamp; subsequent reads from that client carry that token; the cache or edge layer either serves a cached response only if its own known freshness meets or exceeds the token (otherwise falling through to the origin), or the token is used to select a specific, sufficiently fresh cache tier. This scales better than blanket sticky routing since it only bypasses the cache for reads that actually need to reflect a specific recent write, not for the writer's unrelated reads.

Write-through cache invalidation. On write, actively invalidate (or update) the cached copy synchronously as part of the write path, so the NEXT read from anyone (not just the writer) sees the new value, rather than waiting for a TTL (time-to-live)-based expiry. This helps everyone, not just the writer, but adds latency to the write path (the write isn't "done" until the invalidation propagates) and doesn't fully solve the problem for a globally distributed CDN where invalidation propagation itself takes time.

Directed origin reads. Simplest and most conservative: for a specific, small set of operations where read-your-writes truly matters (e.g. "show me the order I just placed"), always read from the origin for that specific query shape, accepting the cache-bypass cost only where the guarantee is actually needed, rather than building general infrastructure for it.

Trade-offs comparison

Worked example. A social app serves post content through a CDN with a 60-second TTL. A user posts a comment and immediately navigates to the thread view; without any mitigation, they might not see their own comment for up to 60 seconds. The team implements client-side version tokens: the post-comment response includes a token; the thread-view request includes it; the edge layer checks whether its cached copy's version is at least that recent, if not, it fetches from origin (a rare case, since 60 seconds is short and most navigations happen quickly after posting, so this only adds origin load for the specific just-posted case, not broadly).

Trade-offs and pitfalls. Choosing write-through invalidation globally "to be safe" when only a narrow set of reads actually needs read-your-writes is a common over-engineering mistake, it adds write-path latency and complexity system-wide to solve a problem that a narrower, directed-read or token-based approach would have handled with far less blast radius.

Direct answer: At 100M users and 50k updates/sec, you need per-user data partitioned by region with a "home region" per user for writes, asynchronous cross-region replication for eventual global visibility, and a session-scoped mechanism (sticky routing to the user's home region, or a causal token) so each user sees their own updates immediately without requiring global synchronous coordination.

Structured elaboration

Partitioning and write routing. Each user is assigned a home region (based on signup location or explicit preference), and writes for that user's profile are always directed there, giving each user's writes a single, consistent, low-latency write path with no cross-region coordination needed per write. This is what makes 50k writes/sec globally distributed and still individually cheap: it's 50k INDEPENDENT single-region writes, not 50k globally-coordinated ones.

Cross-region replication. Each region's writes replicate asynchronously to every other region (a standard multi-region replication topology, e.g. a change stream fanning out from each region's primary store to the others). This is what provides "eventually visible to other users" within the required bound (here, one minute), the replication pipeline's own throughput and lag characteristics need to comfortably clear that bound under peak load, with monitoring on actual observed lag, not just a theoretical budget.

Read-your-writes without global coordination. Since the user's OWN writes always land in their home region, and their own subsequent reads can be routed to that SAME home region (sticky, based on the user's identity, not their current network location) for some bound (or indefinitely), the user always sees their own latest update, at native single-region latency, no waiting for cross-region replication for their OWN view. Other users reading this profile from a different region see it once replication catches up, within the required window, satisfying "others see it eventually within 1 minute" without that requirement touching the write or same-user-read path at all.

Achieving sub-50ms P99 read latency per region. Reads (for other users viewing a profile, not the owner's own reads) are served from the LOCAL region's replica, so they never cross a region boundary, keeping latency to local-network/local-disk numbers rather than being bounded by cross-region round-trip time (which alone would often exceed 50ms). This is only possible because the read doesn't need to be perfectly fresh, it needs to be fresh within a minute, which the async replication path already provides.

Handling the home-region-unavailable case. If a user's home region has an outage, either the user experiences degraded write availability (a real, honest trade-off of this design, since the write path is intentionally NOT cross-region-coordinated) or the system fails over the user's home-region assignment to another region (a more complex, operationally significant decision that itself needs careful handling to avoid conflicting writes from the old and new home regions during the transition).

Worked example. User in Region A updates their bio. The write commits to Region A's primary (their home region) as a purely local operation, with no cross-region network hop on the critical path, so its latency is bounded by local disk/network characteristics rather than by inter-region round-trip time, the same reasoning that keeps the P99 read latency for other users' LOCAL reads low. The response confirms success, and if the user immediately reloads their own profile, that read is also routed to Region A (sticky by user identity), showing the update instantly, own-write visibility achieved with zero cross-region dependency. Meanwhile the change replicates asynchronously to Regions B and C; a different user in Region B viewing this profile sees the OLD bio for however long replication takes (comfortably under the 1-minute bound under normal load, monitored explicitly for tail cases during regional replication backlogs).

Trade-offs and pitfalls. A common design mistake at this scale is routing the OWNER's reads based on their CURRENT network location rather than their home region (e.g. "nearest region" routing applied uniformly to all reads), which breaks read-your-writes the moment a user travels or is routed to a different region than the one their write landed in, own-write reads need identity-based (not network-proximity-based) routing specifically for this reason.

Direct answer

Safe garbage collection of a dedup store's idempotency keys means the TTL for a key must always exceed the maximum time a legitimately late or retried event for that key could still arrive, which in a watermarked streaming system means TTL is set RELATIVE TO THE WATERMARK (how far behind real time the system's own lateness tolerance extends), not to wall-clock time alone, and every node performing GC must agree on a CONSISTENT watermark position before reclaiming any key, or a node with a lagging watermark view can reclaim a key another node still needs.

Structured elaboration

TTL calculation relative to watermarks. The watermark represents "no event with a timestamp earlier than this will legitimately arrive going forward" (a streaming-lateness concept that is load-bearing here for GC correctness specifically). A dedup key for an event with timestamp $t$ is safe to reclaim only once the watermark has advanced PAST $t$ by the pipeline's full lateness tolerance, i.e., key_expiry = event_time + max_lateness_tolerance, not key_expiry = key_creation_time + fixed_TTL; the latter, wall-clock-anchored version can reclaim a key too early if the pipeline experiences a temporary processing delay (the key's CREATION was on time, but its RELEVANCE window, tied to the event's own timestamp and the lateness tolerance, is a different clock than "how long ago did we create this key").

Watermark alignment across nodes. In a distributed dedup store (sharded across multiple nodes, or with multiple consumer instances each independently able to trigger GC), each node's LOCAL view of the current watermark can differ (a node processing a slower-arriving partition has a further-behind watermark than one processing a fast partition). GC must use the MINIMUM watermark across all nodes/partitions relevant to a given key's correctness, a global, coordinated view, not each node's own local, potentially-ahead watermark, or a fast node could reclaim a key while a slow node's still-in-flight, legitimately-late event for that same logical entity is still on its way.

Coordinating GC in a distributed deployment. A coordinator (or a shared, externally-tracked global watermark, updated by every node reporting its own local watermark and the coordinator publishing the MINIMUM) gates GC: no node reclaims any key with an expiry timestamp later than the currently-published global-minimum watermark. This is directly analogous to (and can reuse) the SAME checkpoint-barrier coordination infrastructure many streaming systems already have for exactly this kind of cross-node, cross-partition synchronization.

Why unsafe GC specifically risks re-accepting duplicates, not just losing state. If a key is reclaimed too early (before a legitimately late or retried event for it has arrived), that late arrival's idempotency check finds NO existing record, and INCORRECTLY treats it as new, applying it a second time, exactly reopening the duplicate-application risk the dedup store exists to prevent. This is the specific, concrete failure mode that makes GC timing a genuine correctness question, not just a storage-cost optimization.

Worked example

A pipeline's lateness tolerance is 10 minutes (the pipeline accepts events up to 10 minutes late relative to its watermark). An idempotency key for event evt-500 (event_time = 12:00:00) is created when the event is first processed at 12:00:02. Using watermark-relative TTL:

This key is safe to GC only once the (globally-coordinated, minimum-across-nodes) watermark passes 12:10:00, regardless of when the key was actually CREATED (12:00:02) or how long ago that was in wall-clock terms. Contrast with a naive wall-clock TTL of, say, "expire 5 minutes after creation": the key would expire at 12:05:02, well BEFORE the legitimate 10-minute lateness window closes at 12:10:00, so a genuinely legitimate retry or late arrival of evt-500 between 12:05:02 and 12:10:00 would incorrectly find no record and be double-applied, a concrete, quantified illustration of exactly the risk the question asks to avoid.

Trade-offs and pitfalls

• Common mistake: a fixed, wall-clock-anchored TTL disconnected from the pipeline's actual lateness tolerance. As the worked example shows precisely, this can expire keys BEFORE the legitimate late-arrival window closes, directly causing the re-acceptance risk the question specifically asks to be avoided; TTL must be a function of the SAME lateness tolerance the pipeline's own watermark logic uses, not an independently-chosen storage-convenience number.
• Common mistake: each node performing GC based on its own local watermark without coordination. As covered above, a fast node's locally-advanced watermark can reclaim a key a slow node's still-in-flight data still needs; the MINIMUM-across-nodes coordination is not optional caution, it is the specific mechanism that prevents this class of bug.
• Alternative idempotency-enforcement techniques (outbox, offsets-alongside-writes, two-phase application logic): these are all DIFFERENT idempotency-ENFORCEMENT mechanisms (how the write itself is made safe), distinct from the dedup-STORE-GC question this answer addresses (how the metadata TRACKING that enforcement eventually gets cleaned up safely); they matter here only as context for WHY a dedup store with metadata needing GC exists in the first place, versus an approach (like idempotent upserts alone) that might not need a separate, explicitly-GC'd metadata store at all.
• A longer lateness tolerance directly costs more dedup-store storage, since keys live longer before becoming GC-eligible; this is a real, quantifiable trade-off (per the worked example's own arithmetic pattern) between correctness margin and storage cost, worth stating explicitly rather than treating lateness tolerance as a free parameter.

Direct answer: Two-phase commit (2PC) is a protocol that lets a coordinator get a set of participants to agree on committing or aborting a single transaction atomically, even though each participant only controls its own local resource. It works in two rounds: first the coordinator asks everyone if they're ready (prepare), then it tells everyone the outcome (commit or abort).

Structured elaboration

Roles. One process is the coordinator (initiates and drives the protocol); the rest are participants (each owns one local resource manager, e.g. a database shard or a service's own datastore).

Phase 1: Prepare. The coordinator sends a PREPARE message to every participant. Each participant does whatever work is needed to be able to commit (validates constraints, acquires locks, writes its intended changes to a durable, not-yet-visible log record) and replies VOTE_YES if it can guarantee it will be able to commit later, or VOTE_NO if it can't. Voting yes is a promise: from that point the participant may not unilaterally abort.

Phase 2: Commit/Abort. If every participant voted yes, the coordinator durably logs "commit" and sends COMMIT to everyone; each participant applies its prepared changes and acknowledges. If any participant voted no (or timed out), the coordinator durably logs "abort" and sends ABORT; participants roll back and release the locks they took in phase 1.

Durable logging. Both the coordinator and each participant write their decision to a durable log before sending the next message. This is what makes the protocol able to survive a crash and resume where it left off, rather than just picking an arbitrary outcome.

Worked example. A transfer needs to debit account A on shard 1 and credit account B on shard 2.

1. Coordinator sends PREPARE to shard 1 and shard 2.
2. Shard 1 checks A has sufficient balance, takes a row lock on A, writes an intent record ("debit A by $50, pending"), and votes yes. Shard 2 does the analogous check for the credit and votes yes.
3. Coordinator sees two yes votes, writes "transaction T commits" to its own durable log, then sends COMMIT to both shards.
4. Each shard applies its prepared change and releases the lock, then acknowledges. The coordinator can now discard its log record for T.

If shard 2 had instead found a constraint violation and voted no, the coordinator would log "abort" and tell shard 1 to roll back its debit, leaving both accounts unchanged.

sequenceDiagram
    participant C as Coordinator
    participant P1 as Participant 1
    participant P2 as Participant 2

    Note over C,P2: Phase 1: Prepare
    C->>P1: PREPARE
    C->>P2: PREPARE
    P1-->>C: VOTE_YES
    P2-->>C: VOTE_YES

    Note over C: Durably log "commit"
    Note over C,P2: Phase 2: Commit
    C->>P1: COMMIT
    C->>P2: COMMIT
    P1-->>C: ACK
    P2-->>C: ACK
    Note over C: Discard log record

Failure modes and typical mitigations

• Participant crash before voting: the coordinator times out waiting for that vote and aborts the whole transaction (safe: nothing was promised yet).
• Participant crash after voting yes, before receiving the decision: on recovery the participant reads its own log, sees it voted yes but doesn't know the outcome, and must ask the coordinator (or another participant) what was decided. Until it gets an answer it is stuck holding its locks: this is the core "in-doubt" state.
• Coordinator crash after collecting votes but before all participants get the decision: the surviving participants that already voted yes are now blocked indefinitely, holding their locks, because only the coordinator's log has the real decision. In production this is mitigated by making the coordinator itself durable and quickly recoverable (its log survives the crash and a restarted coordinator resumes from it), and sometimes by coordinator replication so a standby can take over without waiting for the original to come back.
• Network partition between coordinator and a participant: looks identical to a crash from the participant's point of view, so the same in-doubt blocking applies until connectivity (or an operator) resolves it.

Direct answer: An LWW merge function compares the timestamps of two candidate versions of a value and returns the one with the later timestamp, breaking exact ties with a secondary, deterministic tiebreaker (commonly the node/replica ID) so that both replicas resolve a genuine tie identically rather than each arbitrarily picking a different winner.

Structured elaboration. The subtlety worth calling out explicitly: a tie on timestamp ALONE isn't safely resolvable arbitrarily, if replica A and replica B both see a tie and each picks differently (e.g. "keep my own local value on a tie"), they'd DIVERGE, exactly what LWW is supposed to prevent. The tiebreaker has to be something both replicas can compute identically and agree on, most simply, comparing node IDs directly.

from dataclasses import dataclass

@dataclass(frozen=True)
class LWWValue:
    value: object
    timestamp: float   # wall-clock or logical timestamp
    node_id: str        # tiebreaker; must be a stable, comparable, unique ID

def lww_merge(a: LWWValue, b: LWWValue) -> LWWValue:
    """Deterministic: given the same (a, b) pair, always returns the
    same winner, regardless of which replica is doing the comparing,
    and regardless of argument order."""
    if a.timestamp != b.timestamp:
        return a if a.timestamp > b.timestamp else b
    # Exact timestamp tie: fall back to node_id, a stable, globally
    # comparable value both replicas can compute identically.
    return a if a.node_id > b.node_id else b

Handling ties and clock skew. The node_id tiebreaker resolves EXACT ties deterministically and symmetrically (lww_merge(a,b) == lww_merge(b,a) always holds, verified by running both orderings on the same pair and confirming identical output). It does NOT resolve clock skew, if replica A's clock is running fast, its writes might incorrectly "win" against replica B's genuinely-more-recent writes simply because A's (skewed) timestamp reads later, this is a real correctness gap, not something the tiebreaker logic can fix, since from the merge function's point of view, a skewed timestamp is indistinguishable from a genuinely later one.

Drawbacks for counters and sets. Applying this SAME LWW merge function to a counter (treating the counter's current total as a single LWWValue) silently drops one of two concurrent increments entirely, rather than correctly summing them, exactly the counter problem discussed elsewhere in this topic, which is why counters need a purpose-built structure (a G-Counter or PN-Counter) rather than a generic LWW register. Applying it to a set (treating the whole set as one LWWValue) has the same shopping-cart-style data-loss problem: one replica's entire set of additions is discarded if the OTHER replica's write happens to have the later timestamp, an OR-Set is the correct structure when both concurrent contributions need to survive.

Worked example. Replica A writes LWWValue(value="blue", timestamp=100.0, node_id="A"). Replica B, concurrently, writes LWWValue(value="green", timestamp=100.0, node_id="B"), an EXACT timestamp tie (a real, if less common, scenario, especially with coarse-grained or synchronized clocks). lww_merge compares timestamps (equal), falls to node_id, "B" > "A" lexicographically, so it returns B's value, "green". Running lww_merge(b_value, a_value) (arguments swapped) gives the identical result, "green", confirming the merge is order-independent, both replicas converge to the same answer regardless of which one happens to see the comparison first.

Trade-offs and pitfalls. Omitting the tiebreaker (or using a non-deterministic one, like "keep whichever value I currently have locally on a tie") is a subtle but real correctness bug: it looks fine in every test that doesn't specifically construct an exact-timestamp tie, and then causes silent, hard-to-reproduce divergence in production the first time two writes genuinely land in the same timestamp bucket, more common than it sounds with millisecond-or-coarser timestamp resolution under real concurrent load.