Data Consistency and Distributed Transactions Questions

Direct answer

To deduplicate streaming events efficiently you need a compact key, event_id (or a stable natural key), plus enough context to bound how long you must remember it: event_time (when it happened), ingestion_time (when your pipeline saw it), user_id or another partition key, and the source partition offset (to resume correctly after a restart). The retention window for that metadata should track your event-time watermark, typically bounded by the maximum out-of-order lateness you tolerate, not kept forever. Four dedup approaches exist at different points in the pipeline: producer-side dedupe, consumer-side dedupe using state, dedupe tables or upserts, and watermark/window compaction; each trades off where the work happens and how much state it needs.

Structured elaboration

What each metadata field is for:

• event_id: the stable identity of the logical event. This is the dedup key. It must be assigned once, upstream, and preserved through every retry or replay; a UUID generated fresh on each retry defeats the purpose (mirrors the idempotency-key discipline).
• user_id (or another natural partition key): lets you shard dedup state per key instead of maintaining one global structure, and lets you bound state size per partition rather than globally.
• event_time: the business timestamp of when the event actually occurred. This is what watermarks and windowing operate on, and what "same time window" means for compaction.
• ingestion_time: when the pipeline observed the event. Used to detect abnormal ingestion delay and to distinguish "arrived late but is old" from "arrived late and is actually a new, recent event."
• Partition offsets (e.g. Kafka partition + offset): the resume point. After a consumer restart, replaying from the last committed offset can redeliver events already processed; offsets let you reason about exactly which range might be duplicated, narrowing what you need to re-check instead of rescanning everything.

Retention implications. Every dedup approach that keeps state (a seen-IDs set, a dedup table, windowed state) must bound that state, or memory/storage grows without limit. The bound is set by the watermark: once you declare "no event older than event_time T will arrive," you can safely forget dedup state for windows entirely older than T. Choosing too short a window lets late duplicates slip through unnoticed; choosing too long a window wastes memory holding state for windows that will never see another duplicate.

Four deduplication approaches:

Worked example

A user-events stream runs at 20,000 events/sec, and the pipeline tolerates events arriving up to 10 minutes late (the watermark lag). Using consumer-side dedupe with a hash set of event_id keyed per 10-minute window:

$$20{,}000 \text{ events/sec} \times 600 \text{ sec} = 12{,}000{,}000 \text{ events per window}$$

At a 36-byte UUID string per event_id (or 16 bytes if stored as a raw UUID, plus hash-set overhead, call it roughly 50 bytes per entry with typical hash-set bookkeeping overhead):

$$12{,}000{,}000 \times 50 \text{ bytes} = 600{,}000{,}000 \text{ bytes} \approx 572 \text{ MiB per open window}$$

If two windows must be open concurrently (the current one plus a grace-period window for late arrivals), that is roughly 1.1 GiB of dedup state, which is a real, plannable memory budget, not an afterthought. This is exactly why the retention window (the watermark lag, 10 minutes here) is a deliberate engineering choice: doubling it to 20 minutes doubles this number to roughly 2.2 GiB in-flight, while halving it to 5 minutes halves it, at the cost of dropping legitimately late (but real) events as unrecoverable duplicates-of-nothing once the window closes. Consumer-side dedup stores and compacted topics map directly onto this row: a log-compacted topic keyed by event_id is a durable, disk-backed version of the same consumer-side dedup state, trading the in-memory 572 MiB budget above for compaction-cycle latency and broker-side storage, at higher throughput headroom and lower operational risk of losing state on a consumer restart.

Trade-offs and pitfalls

• Layering approaches is normal. Producer-side dedupe removes the cheap, common case (retried sends) before it costs any consumer-side work; consumer-side or sink-side dedupe is still needed as the backstop for the cases producer-side dedupe cannot see (replays, multiple producers, cross-pipeline joins).
• Common mistake: deduplicating on the wrong key. Using a payload hash instead of event_id treats two textually-identical-but-distinct real events (e.g., a user clicking the same button twice, one second apart) as duplicates, silently dropping real data. The dedup key must identify the logical event, not just match its content.
• Common mistake: forgetting that offsets reset on partition reassignment. If consumer group rebalancing moves a partition to a new consumer instance, dedup state that lived only in that consumer's local memory is lost, and a config that resumes "from committed offset" can redeliver a range the old instance already deduplicated. Externalizing dedup state (a shared store or the sink's own upsert) survives rebalances; purely local in-memory state does not.
• Common mistake: sizing retention from "how long till I forget" instead of "how late can a real duplicate legitimately arrive." The watermark should be derived from measured upstream latency distributions (e.g. P99 producer-to-broker delay), not picked as a round number.

Direct answer: For a collaborative real-time document editor, I'd lean toward a CRDT-based approach over Operational Transformation for most new systems: CRDTs converge correctly with a simpler, more composable merge model and support offline/peer-to-peer editing naturally, while OT requires a central transformation server and more intricate correctness proofs, but OT has a longer production track record and can be more metadata-efficient for simple linear text.

Structured elaboration

Convergence guarantees. Both approaches aim for the same end goal, all replicas converge to the same document state despite concurrent edits, but they get there differently. CRDTs achieve this through mathematically provable merge properties (commutativity, associativity, idempotence) baked into the data structure itself; OT achieves it by transforming each incoming operation against every concurrently-applied operation it wasn't aware of, correctness depends on the TRANSFORMATION FUNCTIONS being correctly implemented for every pair of operation types, historically a significant source of subtle bugs (Google Wave's OT implementation, for instance, took considerable engineering effort to get right).

Metadata growth. Sequence CRDTs for text (like RGA or Logoot) typically attach a unique, ordered identifier to every character (or character-run), plus tombstones for deleted characters (needed so a concurrent operation referencing a deleted character's position can still be correctly placed), this metadata can significantly exceed the actual document content's size over a long editing history unless actively compacted. OT, by contrast, doesn't need to retain this kind of per-character identity metadata long-term, since transformation happens against a linear operation history rather than persistent structural identifiers, generally lighter-weight for the steady state, though OT servers do need to retain enough recent operation history to transform late-arriving concurrent edits.

Undo/redo complexity. Both are genuinely hard here, but for different reasons. In OT, undo means transforming the undo operation against everything that has happened since the original edit, which is exactly the same transformation-composition machinery already prone to subtle bugs elsewhere in the system. In CRDTs, undo of a specific edit typically means marking that edit's contribution for removal in a way that itself must merge correctly with everyone else's concurrent state, "selective undo" (undo MY edit specifically, not everyone's edits since) is a known hard problem in both models, without a clean, universally agreed-on solution in either.

Latency of local edits. Both support immediate LOCAL application of a user's own edit (optimistic local rendering, before confirmation from anywhere else), that's table stakes for a responsive editor and isn't a meaningful differentiator between the two; the difference shows up in how quickly and cheaply REMOTE edits can be correctly incorporated, CRDTs merge remote state/ops directly; OT typically routes through a central server that serializes and transforms operations, adding a network hop most implementations rely on for correctness.

Offline editing support. This is where CRDTs have a structural advantage: because merges are commutative and don't depend on a central authority to serialize operations, two devices can edit fully offline and merge directly with each other (or via any intermediary) whenever they reconnect, in any order, with the same guaranteed-correct result. Most production OT implementations assume a central server is reachable to transform operations against, making genuinely peer-to-peer, server-optional offline editing harder to support cleanly.

Recommendation and reasoning. A concrete version of this decision: building a Google-Docs-scale collaborative editor where low-latency local editing is the critical requirement and the team has limited SRE staffing to operate additional centralized infrastructure tips the recommendation further toward CRDTs specifically, since a CmRDT design avoids the central transformation server OT would otherwise require the team to build, harden, and keep available. For a NEW system generally, especially one that needs to support offline editing or a less centralized architecture, CRDTs are the stronger default: the correctness story is more composable and easier to reason about incrementally (add a new CRDT type without re-deriving transformation functions against every existing operation type), at the cost of needing active tombstone/metadata management. OT remains a reasonable choice specifically when the system is already centralized (a single authoritative server is a given architectural constraint anyway) and the team has deep existing OT expertise, or needs the tighter metadata footprint of OT for a very large, long-lived, heavily-edited document.

Trade-offs and pitfalls. A common mistake is treating this as purely a technical choice without weighing team expertise, a team with deep OT experience and a centralized architecture already in place may ship a correct OT-based editor faster and with fewer subtle bugs than the SAME team's first attempt at a CRDT-based one, "which is theoretically better" and "which will actually be correct in production, on this team's timeline" are different questions.

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:

$$\text{key\_expiry} = \text{event\_time} + \text{lateness\_tolerance} = 12{:}00{:}00 + 10 \text{ min} = 12{:}10{:}00$$

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: Distributed deadlocks between multiple 2PC coordinators happen the same way single-node deadlocks do, two or more transactions each hold a lock the other needs, and neither will release until it gets the other, except now the "who's waiting for whom" graph spans coordinators and machines, so no single node can see the whole picture. The standard approaches are a distributed wait-for graph (or timeouts as a cheaper approximation), and lock ordering or timeouts to break the cycle once detected.

Structured elaboration

How it arises. Suppose transaction T1, coordinated by coordinator A, has prepared on resource manager R1 (holding R1's lock) and is now trying to prepare on R2, which is currently locked by transaction T2, coordinated by coordinator B, which is in turn waiting to prepare on R1. Neither T1 nor T2 can proceed: T1 waits on R2 (held by T2), T2 waits on R1 (held by T1). This is a classic deadlock cycle, but no single resource manager, and no single coordinator, can see the whole cycle: R1 only knows it's locked by T1 and that T2 is asking for it; R2 only knows the symmetric fact. You need a mechanism that can assemble the global picture.

Distributed wait-for graphs. Each resource manager reports its local "who's waiting for whom" edges (transaction X is waiting for a lock held by transaction Y) to a detector, which can be centralized or run as a periodic gossip/aggregation protocol across nodes. The detector unions these local edges into a global graph and looks for cycles. When it finds one, it picks a victim (commonly, the transaction that started most recently, or the one that's done the least work, to minimize wasted effort) and forces that transaction to abort, breaking the cycle. This gives precise detection but costs continuous communication overhead and adds complexity: the graph itself can be stale or incomplete under network partitions, occasionally causing false negatives (a real deadlock not detected promptly) or requiring careful handling to avoid false positives.

Timeout-based detection (used far more often in practice). Instead of building and analyzing a global graph, each participant gives up and aborts if it hasn't gotten a lock, or hasn't heard back from the coordinator, within some bounded time. This doesn't PROVE a deadlock exists, a legitimately slow network could trigger the same timeout, but it's vastly simpler to build and operate, and the cost of a false-positive abort (retrying a transaction that wasn't actually deadlocked) is usually much lower than the cost of running a distributed graph-detection protocol continuously.

Lease renewal and preemption. For locks held across a network round-trip (as 2PC's prepared-but-not-yet-decided locks are), attaching a lease (a time-bounded validity) to each lock means a coordinator that's crashed or partitioned automatically loses its hold once the lease expires, rather than requiring active deadlock detection at all; a competing transaction can then acquire the lock once the lease lapses. This trades a small window of "the original holder might still be alive and get preempted" for avoiding the deadlock-detection machinery entirely. Some designs go further with explicit preemption: a higher-priority transaction can force a lower-priority one to release, provided the loser's work can be safely rolled back (which prepared-but-uncommitted 2PC state generally supports, since nothing has been externally observed yet).

Worked example. Two account-transfer transactions run concurrently: T1 debits account A (shard 1) then credits account B (shard 2); T2 debits account B (shard 2) then credits account A (shard 1). If both prepare their first step before attempting their second, T1 holds shard 1's lock on A and waits for shard 2's lock on B; T2 holds shard 2's lock on B and waits for shard 1's lock on A. Classic deadlock. A wait-for-graph detector spanning shard 1 and shard 2's lock managers would see both edges and abort one transaction (say T2, as the later starter), letting T1 complete and then retrying T2. The same outcome is reached more simply if both locks were leased for, say, 5 seconds: whichever transaction's lease expires first gets preempted, and the other proceeds.

Trade-offs and pitfalls. Lock ORDERING (always acquire locks in a consistent global order, e.g. by account ID) prevents this class of deadlock entirely rather than detecting and recovering from it, and is usually the first thing to check for before reaching for a detection mechanism: in the worked example, if both transactions always locked the lower account ID first, the cycle could never form. Detection-based approaches are for cases where a consistent lock order genuinely isn't achievable (e.g. the resources being locked aren't known upfront).

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.