Data Consistency and Distributed Transactions Questions

Direct answer: To guarantee exactly-once external side effects (like sending an email or charging a card) despite retries and orchestrator crashes, the orchestrator needs a durable command log recording every side-effecting call it has made, keyed so a retried or resumed saga recognizes "I already issued this command" and skips re-issuing it, combined with idempotency tokens the external system itself can use to deduplicate on its side.

Structured elaboration

Durable command log. Before calling an external system (charge card, send email), the orchestrator writes a durable record: (saga_id, step, command_id, status=pending). It only sends the actual external call after that write succeeds. If the orchestrator crashes after the write but before getting a response, on recovery it finds the pending record and must resolve the ambiguity, not by blindly retrying (the original call may have already succeeded on the external system's side), but by checking status first where possible, or retrying with the SAME command_id as an idempotency token so the external system can recognize the duplicate.

Idempotency tokens for external systems. Any external side effect worth protecting this way needs to accept an idempotency key from the caller (most payment gateways and transactional email providers do): the orchestrator generates a stable command_id once per logical command and reuses it on every retry of that same command, so the external system either applies it once and returns the same result on retries, or explicitly rejects a duplicate.

Deduplication of outbound commands. Before issuing any command, the orchestrator checks its own log for an existing record with that command_id, if status=applied already, it skips the call entirely (no network round-trip needed, we already know the outcome); if status=pending, it either polls the external system's status endpoint (if one exists) or safely retries using the same idempotency token.

Transactional outbox pattern for writing the command log. When the decision to issue a side-effecting command is itself derived from a change the orchestrator is making to its own database (e.g. "saga step N just completed, therefore issue the notification-email command"), writing the (command_id, status=pending) row to the command log and committing the saga's own state change must happen atomically, otherwise you get the same dual-write problem the command log exists to prevent, just one level up: the saga state could commit while the command-log row is lost, or vice versa. The transactional outbox pattern is how this is made atomic: the pending command-log row is inserted in the SAME local database transaction as the saga's state update, and a separate relay (polling or change-data-capture based) is what actually issues the external call and marks the row applied, rather than the orchestrator calling the external system synchronously inline. This decouples "decide to send the command, durably" from "actually send it," and gives the orchestrator a natural resume point after a crash: any outbox row still pending is a command the relay has not yet confirmed was issued.

Compensating actions and reconciliation. If a step later in the saga fails and this side effect needs to be "undone" (refund a charge, can't literally un-send an email), the compensating action is itself logged and issued the same way, with its own idempotency token. A periodic reconciliation job cross-checks the command log against the external system's own record of what was applied (where the external system offers a lookup API), to catch and alert on any command stuck in pending for longer than expected, evidence of a bug or an external outage rather than a normal race.

Worked example. Saga step "charge card" generates command_id = S-991-charge. Orchestrator logs (S-991, charge, S-991-charge, pending), then calls the payment gateway's charge API with Idempotency-Key: S-991-charge. The call times out with no response (ambiguous: could have succeeded server-side). Orchestrator does NOT immediately retry blindly assuming failure; it either calls the gateway's "look up by idempotency key" endpoint if available, or safely retries the SAME charge call with the same key, the gateway either recognizes the duplicate and returns the original result, or, if the first attempt genuinely never reached it, applies it now. Either way exactly one real charge happens, and once the orchestrator gets a definitive applied response it updates the log to status=applied and proceeds.

Trade-offs and pitfalls. The recurring failure mode is generating a NEW command_id on every retry instead of reusing the same one, that defeats the whole point, since the external system can no longer tell a retry from a genuinely new request, and you're back to the double-charge/double-email risk this design exists to prevent. The command_id must be stable per LOGICAL command (tied to the saga+step), generated once and reused for every retry attempt of that specific command.

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: 2PC can block indefinitely because the commit decision lives in exactly one place, the coordinator's durable log. If the coordinator crashes after collecting votes but before every participant has received the decision, a participant that already voted yes cannot safely guess the outcome, so it must sit holding its locks until the coordinator (or someone with equivalent information) comes back. Three-phase commit (3PC) tries to fix this by adding an extra round, but it depends on assumptions that don't hold in real networks, so it's essentially never used.

Structured elaboration

Why 2PC blocks. The failure scenario is specific: all participants voted yes, so none of them may unilaterally abort (that would break atomicity if the coordinator had already decided to commit). But without the coordinator's decision, a participant also doesn't know it's safe to commit. It's stuck in an "in-doubt" state that only resolves once it learns the real outcome, either the coordinator recovers, or another participant that happens to already know the answer tells it (this only works if such a participant exists and can be reached).

What 3PC changes. 3PC inserts a "pre-commit" phase between prepare and commit: after everyone votes yes, the coordinator broadcasts PRE-COMMIT and waits for acknowledgments before sending the final COMMIT. The idea is that once a majority of participants have seen PRE-COMMIT, they know a commit decision was reached and can safely commit even without hearing directly from the coordinator, because a pre-commit message could only have been sent after unanimous yes votes.

Why 3PC still doesn't solve it in practice. The non-blocking property of 3PC relies on synchronous system assumptions: a known upper bound on message delay and processing time, so that a timeout reliably distinguishes "the coordinator crashed" from "the coordinator is just slow." Real networks are asynchronous: you cannot tell a slow coordinator from a dead one purely by waiting. If participants time out and elect a new coordinator while the old one is actually still alive but partitioned, you can get two coordinators making conflicting decisions, a split-brain that can violate atomicity, exactly the thing the protocol exists to prevent. 3PC also costs an extra network round-trip on every transaction, for a safety property it only delivers under an assumption that doesn't hold in production.

Non-blocking alternatives actually used in practice

Worked example of the blocking window. Coordinator collects yes votes from participants P1 and P2, durably logs "commit", sends COMMIT to P1 (which applies it and moves on), then crashes before the message to P2 goes out. P2 is now holding its locks with no way to know the transaction committed. If P2 tries to reach P1, P1 can honestly tell it "I got COMMIT", which lets P2 also commit safely, that's the one case where a peer can rescue an in-doubt participant. If P1 is unreachable too, P2 has no choice but to keep waiting for the coordinator to restart.

Trade-offs and pitfalls. The most common mistake is treating "we compared timeouts and picked a value" as if it solves the blocking problem, it only bounds the WORST-case wait, it doesn't remove the possibility that the guess made after the timeout is wrong. Anyone proposing 3PC in an interview should be able to name the synchrony assumption it needs and explain why that's the actual reason it isn't deployed, not just "it's more complex."

Direct answer: For a sharded counter that needs to keep incrementing correctly under high concurrency without coordination, a CRDT counter (a PN-Counter, sharded by key with one counter object per key) gives exact eventual convergence with no central bottleneck; per-shard counters with periodic aggregation trade a small aggregation delay for simpler infrastructure; a central counter service gives immediate global consistency at the cost of becoming a write bottleneck; and optimistic CAS (compare-and-swap)-based increments work for moderate contention but degrade under very high concurrent write rates on the same key.

Structured elaboration

CRDT counter (PN-Counter per key). Each region/replica maintains its own local increment count for a given counter key; merging (element-wise max plus sum) converges to the exact correct total with no coordination required at write time. Accuracy is eventually exact (once all increments have propagated and merged, the total is precisely correct, nothing is approximated or lost). Storage overhead is one entry per replica per counter key, and merge/read cost scales with the number of replicas, not the number of increments, since increments to the same replica's entry just add to that entry rather than each needing its own record.

Per-shard counters with periodic aggregation. Instead of a CRDT's continuous merge, each shard just accumulates its own local count independently, and a periodic batch job sums all shards' counts to produce the displayed total. Simpler to build (no CRDT library or merge logic needed), but the displayed total is only as fresh as the last aggregation run, real-time reads see a stale, under-counted total between aggregation cycles, an explicit trade-off, not a bug, if the use case tolerates it.

Central counter service. A single service (backed by its own strongly-consistent store) owns the counter and serializes all increments through it. Gives immediate, always-accurate reads with no aggregation delay, but throughput is capped by that single service's write capacity, and it becomes a single point of contention (and, without careful design, a single point of failure) as write volume grows, exactly the scaling problem sharding elsewhere in the system was meant to avoid.

Optimistic CAS-based increments. Each increment reads the current value, computes the new value, and writes it back with a compare-and-swap (only succeeding if the value hasn't changed since the read); a losing CAS retries. Works well at moderate contention (few concurrent writers per key), but retry rate grows sharply as concurrent writers to the SAME key increase, under very high contention (many writers hammering one popular key, a "hot key"), this can degrade into a large fraction of writes retrying repeatedly, hurting both latency and throughput.

Comparison

Worked example. A "post likes" counter under a viral post might receive thousands of concurrent increments per second from a single popular key. A central counter service or CAS-based approach would see heavy contention specifically on that one hot key; a PN-Counter sharded across regions handles it gracefully since each region's writers only contend with OTHER writers in the SAME region (much lower local contention), converging the true global total asynchronously with no single bottleneck.

Trade-offs and pitfalls. Teams sometimes default to a central counter service for simplicity and only discover the bottleneck once a specific key goes viral or otherwise becomes hot, worth explicitly asking during design "what's the expected max increment rate on a SINGLE key," not just the aggregate system-wide rate, since that's what determines whether central/CAS approaches will hold up.

Direct answer: For a create-with-side-effects operation like create-order (which triggers inventory reservation and payment), the API contract needs a client-generated idempotency key that scopes the ENTIRE multi-step operation, not just the first HTTP call, so a client retry after a timeout re-resolves to the same underlying order and side effects rather than triggering a second one; the server tracks the operation's progress against that key and returns a consistent result regardless of how many times the client retries.

Structured elaboration

Idempotency key structure. The client generates a key once per logical intent (e.g. a UUID generated when the "place order" button is clicked, reused on every retry of that same click, never regenerated on retry). The server stores this key alongside the operation's current state and final result once known, exactly the shape used for a single-endpoint idempotent write, but here the "operation" spans multiple internal steps (order creation, inventory reservation, payment) rather than a single database write.

Client responsibilities. Generate the key once and persist it (e.g. in local state) before the first attempt, reuse the SAME key on every retry of the same logical action, and never reuse a key for a genuinely different order (e.g. a second, separate purchase needs its own key).

Server guarantees: idempotent-create vs at-most-once. The contract promises idempotent-create semantics: retrying with the same key is guaranteed to return the result of the FIRST successful attempt (or the current in-progress status), never trigger a second inventory reservation or a second charge, this is stronger than plain at-most-once (which would just refuse to retry at all) and is what actually lets clients retry safely after an ambiguous failure like a timeout.

Visibility of side effects to users. Because the operation spans multiple steps that don't all complete instantly, the API returns a status field alongside the order/operation ID: pending (steps still in flight), completed (all steps succeeded), or failed (a step failed and compensations, if any, have run). A client polling on the same idempotency key (or the returned operation ID) gets a consistent, current view rather than each poll re-triggering work.

Error semantics. A definitive rejection (e.g. inventory genuinely unavailable) returns a failed status with a reason, and the idempotency key is now permanently associated with that failed outcome, a retry with the SAME key returns the same failure rather than re-attempting (since re-attempting the same operation with the same inputs would fail again anyway); a genuinely NEW attempt requires a new key, communicating the client's intent to try again as a fresh action, which matters if the failure was due to a transient condition the client wants to retry past.

Worked example. Client calls POST /orders with Idempotency-Key: 8f3c... and the order payload. The server hasn't seen this key before, so it starts the underlying saga (create order, reserve inventory, charge payment), stores status: pending against the key, and returns 202 Accepted with an operation ID. A network blip means the client never sees this response, and retries the SAME POST /orders call with the SAME idempotency key. The server recognizes the key, sees the saga is still pending, and returns the current status (again 202, same operation ID) WITHOUT starting a second saga. The client can now poll GET /orders/{operation_id} (itself naturally idempotent, a read) until it sees completed or failed.

Trade-offs and pitfalls. A common design mistake is treating the idempotency key as scoping only the first HTTP request rather than the whole underlying operation, if the key is discarded once the initial 202 response is sent, a later retry re-triggers the ENTIRE saga from scratch, defeating the purpose. The key must remain valid and checked for the full lifetime of the operation it protects, not just the initial acknowledgment.