Distributed Systems Fundamentals Questions

A CRDT-based counter (CRDT: Conflict-free Replicated Data Type, a data structure whose replicas can be updated independently and merged with a deterministic rule that always converges to the same value) lets every region increment locally with no coordination, so writes never block and never fail because of a remote outage. A consensus-backed counter (built on Raft or Paxos, where a majority of replicas must durably agree on each state transition before it counts as committed) gives you a single, linearizable value where every acknowledged read reflects every previously acknowledged write, at the cost of needing a live majority and at least one round trip per operation. The concrete thing the CRDT approach gives up is that read guarantee: a client can observe a stale, undercounted value during the window before a remote increment has propagated, with no signal that the value is incomplete, whereas consensus makes that undercounting structurally impossible.

What each approach actually guarantees

Worked example: where the freshness guarantee actually breaks

Take a G-Counter (a grow-only counter CRDT) tracking a global increment total across three regions, A, B, and C, each holding its own local counter; a read sums the counters a replica currently knows about.

1. Start: A = 0, B = 0, C = 0.
2. Region A processes one local increment: A = 1.
3. Region C processes two local increments: C = 2.
4. Before anti-entropy has propagated A's and C's updates to B, a client reads the counter at region B. B's local view is still A = 0, B = 0, C = 0, so the read returns 0, even though three increments are already durably accepted somewhere in the system.
5. An anti-entropy round runs: B receives the vectors {A: 1} and {C: 2}, merges by taking the elementwise maximum, and B's state becomes A = 1, B = 0, C = 2. A read at B now returns 3, the correct eventual total.

No increment was lost between steps 2 and 5, which is the CRDT's core promise. But the read at step 4 was not just a little stale, it was materially undercounted with no indication of that to the caller. A consensus-backed counter cannot produce this outcome, because an increment is not considered committed until a majority has durably logged it, and any successful read after that point is defined to include it.

A related but distinct choice: automatic CRDT merge vs. application-level conflict resolution

The same coordination-free idea shows up one level up in shared-document collaboration, where two users edit the same paragraph while offline. There a third option exists beyond CRDT and consensus: application-level conflict resolution, where conflicting edits are detected and handled by explicit business logic (show both versions to the user, prefer the longer edit, run a custom three-way merge) instead of a mathematically guaranteed merge function. A CRDT gives the same coordination-free availability as the counter case, with a merge that is provably correct for that specific data type; application-level resolution can encode arbitrary rules a generic CRDT cannot express, such as preferring the document owner's edit, but only for data types someone is willing to hand-write a merge function for, and it carries no formal convergence guarantee if that logic has a bug.

Trade-offs and pitfalls

This is not a universal ranking of CRDT versus consensus, it is a question of which correctness property a specific use case cannot live without. A monitoring signal or a rolling feature aggregate feeding a dashboard, or a non-blocking model feature, can absorb a bounded, self-correcting undercount, so a CRDT is the right default: no leader, no quorum loss during a regional outage, cheaper per write. A signal that gates an irreversible action, such as a billing or quota threshold, a fraud rule, or an inventory decrement, cannot absorb that undercount, because the moment a threshold decision fires is exactly the moment staleness turns from cosmetic delay into a correctness bug, so it belongs on a linearizable path even though every write costs more there. A common mistake is defaulting to CRDTs everywhere for their operational simplicity and only discovering the freshness gap when a threshold check fires on stale data; the fix is rarely to abandon the CRDT for everything, but to route the one decision that needs freshness through a linearizable read, or a periodic consensus-backed reconciliation snapshot, while leaving the bulk of the aggregation coordination-free.

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 CAP theorem says a distributed system that can be split by a network partition can only guarantee two of three properties at once: Consistency, Availability, and Partition tolerance. Because real networks do partition (links fail, messages get delayed or dropped), partition tolerance isn't really an optional design choice, so the actual trade-off every replicated system makes, and only makes while a partition is actually happening, is between Consistency and Availability.

What Each Property Guarantees

• Consistency (C): every read returns the result of the most recent completed write, as if there were only one copy of the data (this is the strong, linearizable notion of consistency).
• Availability (A): every request that reaches a non-failed node gets a response, without a guarantee that the response reflects the latest write.
• Partition tolerance (P): the system keeps operating even when the network drops or delays messages between nodes, splitting them into groups that can't talk to each other.

Why You Only Get Two, and Only During a Partition

When there is no partition, a well-built system can offer both C and A: every node can talk to every other node, so it can confirm it has the latest data before answering. The theorem only bites once a partition actually separates the cluster into two or more groups. At that point, a node in the minority (or either side, in a symmetric split) that receives a request has exactly two choices:

• Answer immediately with whatever data it has locally. That satisfies Availability, but the data might be stale relative to a write that landed on the other side of the partition, so it does not satisfy strong Consistency.
• Refuse to answer (return an error or block) until it can confirm it isn't giving out stale data, typically by waiting for the partition to heal or for enough of the cluster to be reachable. That satisfies Consistency, but it fails Availability for that request.

There is no third option that gives both while the partition is open. That is the entire content of the theorem: it's about behavior during the partition window, not a permanent label on a system.

CP and AP Examples

• A CP-leaning example: a consensus-backed coordination store, such as etcd (a distributed key-value store built on the Raft consensus protocol). If a partition isolates a minority of nodes from the quorum, that minority stops serving both reads and writes rather than risk returning stale or conflicting data. It gives up availability on the minority side to preserve strong consistency everywhere it does respond.
• An AP-leaning example: a Dynamo-style, eventually-consistent key-value store. During a partition, every reachable node keeps accepting reads and writes on both sides, so the system stays available, but the two sides can accumulate divergent writes that must be reconciled once the partition heals (via version vectors, last-write-wins, or application-level merge logic). It gives up guaranteed-fresh reads to preserve availability.

CAP's "Consistency" vs. ACID's "Consistency"

These are two different axes, and conflating them is a common interview trap. ACID (atomicity, consistency, isolation, durability) describes properties of a single transaction, typically on one database: its "C" means a transaction only ever moves the database from one state that satisfies its own defined invariants (foreign keys, uniqueness constraints, application-level rules) to another such state. It says nothing about how fresh a read on a different replica is.

CAP's "C" is about replication: whether a read anywhere in the system reflects the most recent completed write, regardless of which physical replica served it. A system can be perfectly ACID-consistent (every transaction respects its constraints) on every individual replica while still being CAP-inconsistent overall, because a stale replica can return an old value that was, at the time it was written, a perfectly valid state.

Trade-offs and Common Pitfalls

• Treating CAP as a fixed label for an entire system is a common misreading. The choice is scoped to a partition and can even be scoped per operation: a single system can serve some requests (say, checkout) with a CP posture and others (say, product-view counts) with an AP posture.
• Don't assume "P" is a design choice you can decline. Every distributed system that spans more than one process over a real network needs to survive partial network failure, so the honest framing is which of C or A you give up when partitioned, not whether to support partition tolerance.
• A frequent good follow-up is PACELC, which asks what you trade off between latency and consistency even when there is no partition happening, since CAP alone is silent about that normal-operation case.

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.

At a handful of nodes a fixed heartbeat timeout works fine: miss N beats, declare the node down. At several thousand nodes spread across racks and regions, the same fixed threshold produces both false positives (a node under scheduling pressure or behind a congested link gets marked down while still alive) and slow true detections (a threshold loose enough to avoid those false positives takes too long to catch a real failure), because network and scheduling jitter is not uniform across the cluster. A Phi Accrual failure detector replaces the single global timeout with a statistical model of each node's own recent heartbeat behavior, so the question becomes how improbable this silence is given what has actually been observed from that specific node, rather than whether more than T seconds have passed, and it outputs a continuously increasing suspicion level, phi, instead of a binary up-or-down verdict.

Membership and probing, before the suspicion model

• Partial views and gossip: each node keeps a small, mostly random subset of the membership rather than the full list, and periodically gossips membership updates and heartbeats to a few random peers, so state propagation stays cheap instead of all-to-all.
• Direct and indirect probing (SWIM-style; SWIM: a gossip-based membership protocol combining direct pings with an indirect fallback): a node pings a random peer directly, and if it gets no acknowledgment, asks a handful of other random members to ping that peer on its behalf, since one lossy path between two specific nodes should not be read as evidence the target itself is down.
• A minimal membership-manager state machine per monitored node: ALIVE, SUSPECT, DEAD. Failed direct-plus-indirect probes move a node from ALIVE to SUSPECT rather than straight to DEAD, and the suspicion is gossiped with an incarnation number, so a node that is actually alive can refute it by re-announcing itself with a higher incarnation before it is declared DEAD and dropped from membership.

From a binary probe result to a continuous suspicion level

Probing only tells you whether a heartbeat arrived, which is still binary. Phi accrual sits on top of that: instead of one global timeout, each node keeps a short sliding window of the actual inter-arrival times it has observed from a given peer, uses that history to estimate how spread out that peer's heartbeats normally are, and converts the elapsed time since the last heartbeat into a suspicion level using roughly:

where $P_{later}(t)$ is the estimated probability, given that peer's own recent history, that a heartbeat still arrives more than $t$ time units after the previous one. As $t$ grows past what the peer's history says is normal, $P_{later}(t)$ shrinks and phi grows smoothly and without bound, instead of snapping from 0 to 1 at one fixed cutoff.

Why this matters at thousands of nodes: a peer on a congested cross-region link naturally has wider jitter in its own history, so it takes a longer silence to reach the same phi value as a peer on a quiet local link, without anyone hand-tuning a per-peer timeout. It also lets different consumers of the same signal act at different confidence levels: something that only needs to stop routing traffic to a possibly-dead node can act at a low phi and cheaply reverse course if the node turns out fine, while something that removes a node from a lock's quorum permanently can wait for a much higher phi, all from the same underlying measurement.

Worked example: computing phi from a pinned heartbeat history

Take a 6-sample sliding window of observed inter-arrival times, in milliseconds, for one peer: [1000, 1300, 700, 1100, 900, 1200].

Mean $\mu = 1033.33$ ms, population standard deviation $\sigma = 197.2$ ms.

Approximating the interval distribution as Normal($\mu, \sigma^2$), so that $P_{later}(t) = \tfrac{1}{2}\,\mathrm{erfc}\!\left(\tfrac{t-\mu}{\sigma\sqrt{2}}\right)$ (erfc is the complementary error function, a standard tool for turning a distance from the mean into the probability mass remaining in the distribution's tail beyond that point), gives:

Tracing the 1500 ms row end to end: standardize the elapsed time against this peer's own mean and spread, $z = (1500 - 1033.33) / (197.2 \times \sqrt{2}) = 466.67 / 278.88 \approx 1.673$; convert that into a tail probability, $P_{later}(1500) = \tfrac{1}{2}\,\mathrm{erfc}(1.673) \approx \tfrac{1}{2} \times 0.0180 \approx 0.00898$; then take $-\log_{10}$ of that probability, $\phi(1500) = -\log_{10}(0.00898) \approx 2.05$. The other three rows fall out of the same three steps with a different elapsed time plugged in.

At t = 1000 ms, right around this peer's own mean, phi stays near zero: nothing unusual. At t = 1500 ms, phi crosses 2, a mild anomaly worth noting but not worth acting on for most purposes. By t = 2500 ms, phi is in the double digits, meaning the observed silence is astronomically improbable given this specific peer's recent behavior, which is enough statistical confidence for even a conservative consumer to move the node to SUSPECT, or to DEAD if it was already SUSPECT and never refuted with a higher incarnation.

Trade-offs and pitfalls

Phi is not a probability itself, it is a log-scaled confidence score, and reading a given phi value as a direct percentage chance of failure is a common misunderstanding; the number only means something relative to whatever threshold a given consumer chooses. A second pitfall is picking one global phi threshold across a heterogeneous fleet, which reintroduces the exact problem this design was meant to solve, since the entire benefit comes from calibrating to each peer's own recent history rather than from a smarter constant. Third, phi accrual detects the absence of a heartbeat; it cannot distinguish a genuinely dead node from a network partition dropping every packet between two specific nodes while both are alive and healthy elsewhere, so membership decisions that need to be safe, like removing a node from a lock's quorum, still need a consensus-backed or fencing mechanism layered on top of the suspicion signal, not phi alone.