Distributed Systems Fundamentals Questions

A text editor where users type and delete concurrently without a lock needs a sequence CRDT (a Conflict-free Replicated Data Type: a data structure whose merge operation is commutative, associative, and idempotent, so concurrent replicas always converge to the same state without coordination). Each character gets a globally unique, immutable identifier, deletions mark a tombstone instead of physically removing the character, and a deterministic tie-break rule orders characters that were inserted concurrently at the same position. The design pays for this with permanent per-character metadata and a garbage-collection problem for tombstones.

Data structure

• Represent the document as an ordered sequence of atoms: { id: (site, counter), char, prev_id, tombstone }.
• id is globally unique because it pairs a site identifier (one per editing client) with that site's own monotonically increasing counter.
• prev_id records which atom this one was inserted immediately after, at the moment of insertion. That is what preserves each user's actual intent (insert right after the character I was looking at), even if other edits land nearby before this one is delivered.

Concurrent insert resolution

• Insert(id, prev_id, char) is broadcast to every replica.
• When two inserts share the same prev_id, both intended the same insertion point, that shared prev_id is exactly what concurrency looks like here. A deterministic comparator orders competing children of the same prev_id by site identifier (higher-precedence site placed first), so every replica, regardless of arrival order, produces the same final sequence.

Deletion and tombstones

• Delete(id) sets tombstone=true on that atom. The atom stays in the structure, because its ordering role must persist: a later, concurrently-arriving insert may still reference it as prev_id and needs it to resolve correctly.
• A tombstoned atom cannot be physically removed until every replica has definitely applied it.

Garbage collection of tombstones

• Replicas exchange a version vector (one counter per site, the highest counter from that site each replica has durably applied) during anti-entropy (a periodic background exchange where replicas compare state and reconcile any differences, rather than waiting for a live update to arrive).
• The element-wise minimum across all replicas' version vectors is the causal stability frontier: any tombstone at or below that frontier has been seen everywhere and can be physically purged.
• Cost: computing and propagating that frontier needs periodic all-to-all or gossip-based exchange, and a replica that stays offline indefinitely blocks compaction for everyone unless it is explicitly evicted from the version-vector set.

Worked example: two concurrent inserts at the same position

Both replicas have already converged on a one-character document containing 'X', a single atom with id (A,1). Two users, on different replicas, both position their cursor right after 'X' at the same time:

• Site A's user types 'p': op1 = Insert(id=(A,2), prev=(A,1), char='p')
• Site B's user types 'q': op2 = Insert(id=(B,1), prev=(A,1), char='q')

Both operations reference prev=(A,1): that shared reference is the concurrency. The merge rule for atoms competing for the same prev is order competing children by site identifier, descending, so a child from site B is placed before a child from site A when both point at the same prev.

• At replica A: op1 applies locally first, giving 'Xp'. When op2 arrives, it is spliced into the children of (A,1) and re-sorted by the rule above, giving order [op2, op1], so the document becomes 'Xqp'.
• At replica B: op2 applies locally first, giving 'Xq'. When op1 arrives, the same re-sort rule applies, again giving [op2, op1], so the document becomes 'Xqp'.

Both replicas land on 'Xqp' even though they applied the two operations in opposite order. That is what merging deterministically means in practice: the comparator, not arrival order, decides the final sequence.

graph LR
    X["X (id A,1)"] --> Q["q (id B,1)"]
    Q --> P["p (id A,2)"]

Now say 'q' is later deleted by its author. Atom (B,1) is not removed, only flagged tombstone=true, so the document reads 'Xp', but the underlying structure still holds three atoms, two live and one tombstone, until compaction runs.

Trade-offs & pitfalls

• Metadata growth: every character carries an id pair, a prev pointer, and (once deleted) a tombstone bit; on a long-lived, heavily-edited document the tombstone count can exceed the live character count, so uncompacted storage is proportional to live plus deleted atoms rather than just live ones.
• The same observed-remove technique (adding creates a fresh tag, removing targets only the tags actually observed) applies directly to a shared wishlist service: adding an item is an add-tag, removing it is a remove-tag scoped to the tags a client has actually seen, and the tombstone-compaction and consistent-read story are identical to the text editor's, just at item granularity instead of character granularity.
• Formatting spans (bold, italic) and structural moves (relocate a paragraph) do not compose as cleanly as single-character insert and delete; a naive extension can lose the same intention-preservation guarantee, which is why production collaborative-editing CRDTs spend real engineering effort specifically on this.
• Common wrong turn: implementing deletion by removing the atom outright instead of tombstoning it. That breaks any concurrent insert whose prev_id pointed at the now-missing atom, since there is nothing left to splice after.

Direct answer

Anti-entropy is the background process that repairs replicas which have drifted apart by comparing what each replica actually holds and shipping only the differences. Full Merkle trees, range-based comparison, and delta-based synchronization all solve this, but they trade CPU and memory (building or maintaining a comparison structure) against network bytes (what actually gets transferred) in different proportions, and at terabyte scale that trade-off is what decides which one wins.

Full Merkle trees

A Merkle tree is a hash tree: each leaf is the hash of a key-range's contents (or a single key's version), and each internal node is the hash of its children. Two replicas compare root hashes first; if they match, the whole subtree underneath is known to agree and no further comparison is needed; if they differ, both sides recurse into whichever child subtrees disagree, pruning everything else. This localizes a small number of actually-differing keys very cheaply in network terms, but building the tree from scratch means hashing the entire dataset, which for a terabyte-scale store is a full scan whose CPU and I/O cost is proportional to the dataset size, not to how much has actually drifted.

Range-based comparison

Instead of a tree, the keyspace is partitioned into fixed or size-based ranges up front, and each replica computes one summary per range (a checksum, a record count, a max version) and exchanges those directly. A mismatched range is either subdivided further or transferred whole. This skips the recursive tree bookkeeping, at the cost of coarser precision: if a single key differs inside a large range, the whole range still has to move (or be subdivided) rather than just that one key.

Delta-based synchronization

Each write is stamped with a monotonically increasing per-replica version number, and instead of building any hash structure at all, one replica asks the other for everything with a version greater than its own last-synced position (tailing an operation log). This is by far the cheapest in both CPU and network when it applies, because it transfers exactly the changed records and nothing else, but its correctness depends entirely on both sides retaining a comparable version log; if the divergence is older than the retained log window, delta sync cannot answer the request and the system must fall back to a full Merkle or range comparison instead.

Comparison at terabyte scale

Building a full Merkle tree over a terabyte dataset from scratch on every anti-entropy pass is prohibitively expensive; production systems maintain it incrementally as writes happen rather than rebuilding it each run. Range-based comparison avoids that recursive bookkeeping at the cost of transferring more bytes per mismatch, which can be mitigated with adaptive range sizing (smaller ranges where churn is high, larger ranges where data is cold). Delta-based sync is cheapest whenever the version log covers the actual gap, but has an unbounded-retention cost if you want it to always apply.

Recommendation for a large, skewed keyspace: use adaptive, size- or churn-based ranges as the primary mechanism (isolating hot keys into small ranges and cold data into large ones), maintain shallow per-range Merkle hashes to localize differences quickly within a cold range, and prefer delta-based sync wherever the version log still covers the gap, falling back to the range/Merkle comparison only once the log window has been exceeded.

graph TD
    RootA[Root hash A] --> BL[Branch Left]
    RootA --> BR[Branch Right]
    BL --> L1[Leaf: keys 1-1000]
    BL --> L2[Leaf: keys 1001-2000]
    BR --> L3[Leaf: keys 2001-3000]
    BR --> L4[Leaf: keys 3001-4000]

Read repair: fixing staleness at read time

Anti-entropy runs in the background regardless of what's being read right now; read repair instead fixes staleness the moment a client happens to read a key, without adding latency to that client's own request. When a quorum read is served from multiple replicas, the coordinator compares the versions it gets back: suppose replica R1 returns (value=v2, version=7), R2 returns (value=v1, version=5), and R3 returns (value=v2, version=7). The coordinator returns v2 (the version two of three replicas agree on) to the client immediately, and separately, off the client's response path, pushes a repair write of (K, v2, version=7) to R2 in the background. Because this only happens for keys someone actually reads, read repair does nothing for cold keys nobody queries, so background anti-entropy is still needed as the backstop for data that is never read.

Hinted handoff: staying writable through a node outage

When a target replica for a key is unreachable at write time, for example replica R2 is down, a Dynamo-style store stays always-writable by handing that write to a different, healthy node (a hint holder, say R4) along with a hint recording who the write actually belongs to. R4 stores the write plus the hint. Concretely: the replica set for key K is {R1, R2, R3}; R2 is down; a write W(K=v3) arrives; the coordinator writes directly to R1 and R3, and since R2 is unreachable, hands the write to R4 with hint={target: R2, key: K, value: v3}. R2 recovers ten minutes later; R4's health check or gossip layer learns R2 is back; R4 forwards W(K=v3) to R2 and deletes its own hint copy. All three original replicas now hold v3.

Trade-offs & pitfalls

Hinted handoff is a latency and availability optimization, not a substitute for anti-entropy: if the hint holder itself crashes before delivering the handoff, that write is only recoverable through whatever the normal replica quorum still retains, so anti-entropy (or read repair) remains the backstop even when hinted handoff is in use. Read repair only heals keys that get read, so a design that relies on it exclusively can leave cold data permanently drifted. Over-aggressive range subdivision in range-based sync adds coordination overhead without meaningfully reducing transferred bytes unless the workload's skew actually justifies the extra ranges.

Direct Answer

Two-phase commit (2PC) is a protocol that lets one coordinator get a group of participants, each owning a different resource such as a database or another service acting as a resource manager, to commit or abort a single transaction atomically. It works by first asking everyone to prepare, and only telling everyone to actually commit once every participant has confirmed it's ready.

The Two Phases

• Prepare (vote) phase: the coordinator sends Prepare to every participant. Each participant does whatever local work is needed to guarantee it can commit if told to, such as writing an undo or redo log entry, acquiring the necessary locks, or checking constraints, then replies Yes or No. Once a participant votes Yes, it must hold its prepared state and locks until it hears the final decision; it can no longer unilaterally change its mind.
• Commit/abort phase: if every participant voted Yes, the coordinator sends Commit to all of them; if any participant voted No, or didn't respond, it sends Abort to all of them. Each participant applies the decision and releases its locks.

The Classic Blocking Failure

If the coordinator crashes after collecting Yes votes from every participant but before sending out the final Commit or Abort, every participant is stuck holding its prepared state and its locks indefinitely. A participant can't safely decide on its own: if it guesses Commit but the coordinator, once recovered, had actually decided Abort because some other participant it hadn't heard from yet said No, that guess would violate atomicity. So each participant has no safe choice but to wait.

This is a real operational problem, not just an inconvenience, because those held locks are on real resources. Any other transaction that touches the same rows, files, or records is blocked too, for as long as the coordinator stays down. A single stuck 2PC transaction can produce an effective outage across everything those locks reach.

A Mitigation: Durable Coordinator Logging and Recovery

The coordinator writes its decision, and the votes it collected, to a durable write-ahead log, a log written to disk before the coordinator acts on it so it survives a crash, before sending Commit or Abort. On restart, or when a replacement coordinator takes over, elected through its own consensus mechanism, it replays that log to learn what it had already decided for any in-flight transaction and resends the correct outcome to whichever participants are still blocked. This doesn't remove the blocking window entirely, but it bounds it to however long recovery takes, instead of leaving participants blocked forever.

Worked Trace

Three participants: a relational database (DB), a document store (Doc), and, for illustration, a payment gateway wrapped with a prepare step (PG). Coordinator C.

1. C sends Prepare to DB. DB validates and locks, replies Yes.
2. C sends Prepare to Doc. Doc validates and locks, replies Yes.
3. C sends Prepare to PG. PG validates, replies Yes.
4. C now has 3 of 3 Yes votes and is about to send Commit, but it crashes before sending anything.
5. DB, Doc, and PG are all blocked, each holding its prepared locks; none of them can safely commit or abort on its own, since each only knows its own vote was Yes, not the others'.
6. A recovered, or replacement, coordinator reads its durable log, sees that all three votes were Yes and that Commit was the decision about to be sent, and resends Commit to DB, Doc, and PG. All three commit and release their locks.

Heterogeneous Resources and a Real-World Wall

2PC's coordinator and participant model works across different kinds of resources in principle: a relational database and a document store can both act as participants as long as each exposes a real prepare step, which is exactly what the XA standard, the X/Open standard defining a two-phase-commit interface for resource managers, formalizes for relational databases. The practical wall most teams hit is that an external payment gateway typically doesn't expose anything like a prepare/commit interface at all; it's a single, irrevocable HTTP call. That's one concrete reason a workflow like reserve inventory, charge card, and ship is usually built as a saga rather than a literal distributed transaction: at least one of the participants can't be plugged into 2PC as a real participant.

When to Reach for a Saga Instead

Reach for a saga instead of 2PC when at least one step can't participate in a real prepare/commit handshake, such as a third-party API with only a single irreversible call, when holding locks for the full duration of the workflow is unacceptable for availability or latency, or when some steps are long-running, such as waiting on a person, a batch job, or a slow downstream service, and you can't justify holding resources locked that long.

Trade-offs and Pitfalls

• A common wrong turn is assuming a participant can unilaterally abort on its own after a timeout once it has already voted Yes. It can't, safely, because it doesn't know whether the coordinator already told everyone else to commit; that's exactly why 2PC is called blocking.
• Presumed-commit and presumed-abort optimizations reduce how much needs to be logged in the common case, but they don't remove the fundamental blocking window; they just make the typical path cheaper.
• Three-phase commit adds an extra round specifically to shrink this blocking window, but it doesn't fully eliminate it, and it's rarely deployed in practice because of the added message and latency cost for a problem that a saga usually sidesteps entirely.

Direct answer

PACELC, short for "if Partition, Availability vs. Consistency; Else, Latency vs. Consistency", says that CAP's dilemma, choose Consistency or Availability when a network Partition is happening, is only half the story. Even when there is no partition at all, a system still has to choose between Latency and Consistency for every write it replicates, because making a write durable on every replica before acknowledging it takes longer than acknowledging it once it's durable on a single node. PACELC packages this as: if Partition occurs, trade off Availability against Consistency (exactly what CAP already says); Else, meaning no partition, trade off Latency against Consistency.

Restating CAP precisely first

CAP says that during an actual network partition, a distributed system can guarantee only one of Consistency (every read sees the latest completed write) or Availability (every request gets a non-error response) for the nodes on either side of the split, not both. A common misreading treats CAP as "pick two of three, always"; it isn't. CAP's teeth are specifically about behavior during a partition. Most systems are both consistent and available almost all of the time, precisely because a true network partition is a rare event relative to total uptime, not something happening continuously.

flowchart TD
    Start[Write occurs] --> P{Partition active?}
    P -->|Yes| AC[Choose Availability or Consistency]
    P -->|No| LC[Choose Latency or Consistency]

What PACELC adds

PACELC names the trade-off CAP is silent about: during normal operation, with no partition, you still choose between Latency (L) and Consistency (C), because synchronous replication that waits for a majority of replicas costs a round trip before it can acknowledge a write, while asynchronous or single-node-acknowledged replication returns faster but risks a reader seeing stale data, or the acknowledged write being lost outright if that one node fails before it propagates. Systems are commonly labeled by both branches together, for example PA/EL (favor Availability under partition, favor Latency otherwise, the Cassandra/Dynamo-style default) or PC/EC (favor Consistency in both cases, the HBase-style default).

Worked example: a decision with no partition occurring

A write to a piece of user data must be replicated to three nodes: R1 in the local region, and R2, R3 in two remote regions. All three are reachable; no partition is happening anywhere in this example.

• Favor consistency (the "C" side of the Else branch): the write path waits for acknowledgment from a majority, at least two of the three replicas, say R1 and R2, before returning success to the caller. Any subsequent read from a majority quorum is now guaranteed to see this write. Cost: the caller's write waits on the round trip to R2, a remote replica, even though R1, the local one, already has it durably.
• Favor latency (the "L" side of the Else branch): the write path acknowledges as soon as R1 has it durably, and replicates to R2 and R3 asynchronously in the background. Cost: the caller gets a fast, local acknowledgment, but a read served from R2 immediately afterward, before the async replication catches up, will not see the write yet. If R1 crashes before that background replication completes, the already-acknowledged write can be lost entirely, with zero partition ever occurring.

This decision, wait for two of three versus acknowledge on one, is made on every single write regardless of whether any partition is happening, which is exactly the trade-off PACELC's Else branch names and CAP alone has nothing to say about, since CAP only speaks to a system that is not fully connected.

Trade-offs & pitfalls

A common misconception is treating a database's PACELC label as a fixed law of the software rather than a description of its typical default: most systems let you tune the replication wait per request (via quorum size), so "Cassandra is PA/EL" describes its usual configuration, not something it's incapable of changing. It's also easy to blur this Else-branch trade-off with an availability discussion; in the worked example above, no node was ever unreachable, so the trade being made is purely about how long the write path waits before acknowledging, not about surviving an outage, which is a separate concern belonging to the partition branch of the theorem.

Direct answer

The transactional outbox pattern gets atomicity between a local database update and publishing an event by writing the event as a plain row in the SAME database transaction as the business change, instead of trying to atomically commit across two separate systems (the database and the message broker). A separate background publisher then reads that outbox table and delivers the events, so the hard part of the problem (getting a message onto a broker) is pushed into an at-least-once delivery loop the consumer can absorb, rather than solved by a distributed commit protocol.

Why not just use a distributed transaction directly

A two-phase commit (2PC) across the database and the broker would need the broker to act as a participant in the same commit protocol as the database: it would have to accept a "prepare" call, hold the message uncommitted, and only make it visible once a coordinator later sends "commit". Kafka's and SQS's client APIs do not expose that kind of prepare/commit participant interface (Kafka has its own separate transactional-producer API, not an XA participant interface, XA being the X/Open standard for coordinating a transaction manager with multiple resource managers), so 2PC across a database and a managed broker directly is not something you can wire up against most production message buses. Even where a broker does support it, the coordinator becomes a blocking point: each participant holds its local lock from "prepare" until it hears back, so a coordinator crash between phases can leave a participant blocked indefinitely.

Outbox table schema

Writing the event

Inside the same database transaction that updates the business tables (e.g. orders), the application also inserts a row into outbox. Both inserts commit together or not at all, so there is no window where the business change exists without a corresponding outbox row, or vice versa.

Background publisher

A worker process polls with something like SELECT * FROM outbox WHERE status = 'pending' ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 100. FOR UPDATE SKIP LOCKED lets multiple publisher instances run concurrently without two of them claiming the same row: each instance simply skips rows another instance already has locked. The worker publishes each claimed row to the broker, and only after the broker acknowledges does it mark the row sent. As an alternative to polling, a change-data-capture (CDC) tool such as Debezium can tail the database's write-ahead log (WAL, the durability log the database already writes before committing) and stream outbox inserts to the broker with lower latency and no polling interval, without any application code change.

Avoiding duplicates and lost events across a crash

Trace through a concrete run for outbox row id=482:

1. Application transaction commits: orders row and outbox row id=482 (status=pending) both durable together.
2. Publisher polls, claims row 482 via SELECT ... FOR UPDATE SKIP LOCKED.
3. Publisher sends row 482's payload to the broker; broker acknowledges receipt.
4. Publisher crashes before executing UPDATE outbox SET status='sent' WHERE id=482.
5. Publisher restarts, polls again; row 482 is still status='pending', so it gets re-claimed and re-published. The broker now has two copies of the same event.
6. Because row 482's id travels with the payload as a dedup key, the consumer (or the broker's own dedup mechanism) recognizes the second delivery as a repeat and drops or no-ops it.

If the crash instead happens before step 3 (before the publish call), nothing has reached the broker at all: row 482 simply stays pending and gets picked up on the next poll, with no data loss. The pattern never produces a business change with a missing event, only occasional duplicate deliveries, which is why the consumer side still needs to be idempotent or dedup-aware; the outbox guarantees "at least once", not "exactly once", on its own.

sequenceDiagram
    participant App
    participant DB as Database
    participant Pub as Publisher
    participant Bus as Message Bus
    App->>DB: BEGIN TXN
    App->>DB: INSERT orders row
    App->>DB: INSERT outbox row (pending)
    App->>DB: COMMIT
    Pub->>DB: SELECT pending FOR UPDATE SKIP LOCKED
    DB-->>Pub: outbox row id=482
    Pub->>Bus: publish(id=482)
    Bus-->>Pub: ack
    Pub->>DB: UPDATE outbox SET status=sent WHERE id=482

Production checklist

• Alert on outbox backlog depth (rows still pending beyond an expected age), not just publisher liveness, since a stuck publisher looks alive but stops draining the table.
• Retention job to delete or archive sent rows so the table doesn't grow unbounded.
• Partition/order guarantee: publish using aggregate_id as the broker partition key so events about the same entity are delivered in the order they were written.
• Version the payload schema so a newer producer and an older consumer can coexist during a rollout.

Trade-offs & pitfalls

Outbox trades immediate, synchronous cross-system consistency for simple local atomicity plus eventual, at-least-once delivery: the business transaction commits instantly (no waiting on the broker), but there is a real (if usually short) window between the commit and the event actually reaching the broker. A common mistake is treating "broker acknowledged" as "consumer processed": the outbox only guarantees the event left the outbox table, not that the ultimate side effect happened, so end-to-end correctness still depends on consumer idempotency. Another common mistake is forgetting to key on aggregate_id for ordering, or running multiple publisher instances without SKIP LOCKED-style claiming, which produces either out-of-order delivery or duplicate publishes from two workers claiming the same row simultaneously.