Distributed Transactions & Cross-Shard Consistency

A single-shard write is an ordinary transaction. A write that spans two shards is a distributed systems problem with no free solution: something must give — atomicity, availability or latency. This topic sits under Cross-Partition Querying & Aggregation Strategies and covers the four mechanisms that are actually used in production, what each costs, and how to decide between them. It is the write-side counterpart to cross-shard aggregation patterns, which handles the read side of the same boundary.

Problem Framing

A payments platform shards by account_id. Transferring money between two accounts on the same shard is a two-row transaction with ordinary ACID guarantees. Transferring between accounts on different shards is the same business operation with none of those guarantees: two independent databases, two independent commits, and a window between them where one has been debited and the other has not.

The naive implementation — commit the debit, then commit the credit — is correct in the happy path and produces missing money whenever the second commit fails. The naive fix — retry the credit — is correct until the retry runs after the credit already applied, producing duplicate money. Both bugs are rare, both are discovered by reconciliation weeks later, and both are the direct consequence of treating a distributed write as a local one.

The Four Mechanisms

Four mechanisms and what each one actually guarantees Two-phase commit gives true atomicity, blocks when any participant is unavailable, and adds a full round trip of latency. A saga gives eventual consistency with compensations, stays available during partial failure, and needs a compensating action written for every step. The transactional outbox gives at-least-once delivery from a local transaction with no coordination. Idempotency keys are not a mechanism on their own but a requirement of the other three. Mechanism Guarantee Availability under failure Cost Two-phase commit Saga + compensation Transactional outbox Idempotency keys atomic across all participants eventually consistent, semantically undoable at-least-once delivery of the side effect exactly-once effect from at-least-once delivery blocks — any participant or coordinator down stops it continues — failed steps compensate rather than block continues — the local commit always succeeds n/a — a property, not a protocol +1 round trip, leaked prepares are dangerous a compensation per step, and visible intermediate states a relay process and an outbox table to prune a key column and a lookup on every write path The last row is not an alternative to the others — it is a prerequisite. Every mechanism above it retries, and a retry without an idempotency key turns a delivery guarantee into a duplication guarantee.

Two-phase commit gives genuine atomicity. A coordinator asks every participant to prepare, and commits only if all of them agree. Its weakness is exactly its strength: after preparing, a participant must hold its locks until it hears back, so a coordinator failure leaves locks held indefinitely.

Sagas replace atomicity with compensation. Each step commits locally and immediately; if a later step fails, earlier steps are undone by explicit compensating actions. The system passes through states an atomic transaction would never expose, and those states must be tolerable to the business.

The transactional outbox solves the narrower problem of “commit locally and reliably tell someone else”. The side effect is written into the same local transaction as the data change, and a relay delivers it afterwards. It cannot make two writes atomic, but it makes the second one inevitable.

Idempotency keys make retries safe. They are not an alternative to the other three; they are what makes the other three usable, because every one of them retries.

Choosing Between Them

The decision follows from three questions about the operation, asked in order.

Three questions that pick the mechanism First, can the operation be redesigned to touch one shard — if so, do that. Otherwise, must intermediate states be invisible to users and reports: if yes, two-phase commit; if no, a saga with compensations. Separately, if the second effect is outside the database, use a transactional outbox regardless of the other answers. can it touchone shard? must partial statesbe invisible? yes → redesign the keycheaper than everything below yes → two-phase commitatomic, blocks on failure no → saga + compensationavailable, eventually consistent separately: is the secondeffect outside the database? → transactional outbox,whichever branch you took The first question is the one teams skip, and it is the only answer that removes work rather than adding it. Idempotency keys are not on this tree because every branch requires them. 
1. Can the operation be redesigned to touch one shard?
   → yes: do that. It is cheaper than every option below.
2. Must intermediate states be invisible to users and reports?
   → yes: two-phase commit, and accept the availability cost.
   → no:  saga with compensations.
3. Is the second effect outside the database (email, webhook, another service)?
   → transactional outbox, always.

The first question is the one teams skip. A transfer between two accounts is only a cross-shard operation because accounts were sharded independently; a design that shards by customer rather than by account keeps most transfers local, and the remaining cross-customer transfers are rare enough for a saga. Choosing the shard key from the transaction boundary rather than from the query pattern is discussed further in partition key selection and design.

What Consistency Actually Means Here

“Consistent” is doing a lot of work in most design documents. Three separate properties get conflated, and separating them makes the trade-offs decidable:

Property Question it answers Mechanism that provides it
Atomicity can an observer see one write applied and not the other? 2PC; or a saga if the intermediate state is acceptable
Durability of intent if we commit locally, is the remote effect guaranteed to happen eventually? transactional outbox with a relay
Idempotence if we deliver twice, does the effect apply once? idempotency key stored with the effect
Isolation can a reader see a partially applied multi-shard write? only 2PC with the right isolation level; sagas explicitly cannot

Most production systems need durability of intent and idempotence, tolerate a lack of atomicity, and never actually needed cross-shard isolation. Recognising that is what allows a saga rather than 2PC, and the difference in operational burden between the two is large.

The same transfer under 2PC and under a saga Under two-phase commit both shards prepare, hold locks, and commit together, so no observer ever sees a partial transfer, but the accounts are locked for the duration and a coordinator failure leaves them locked. Under a saga the debit commits immediately, the credit is attempted, and a failure triggers a compensating credit on the source, so an observer can briefly see money that has left one account and not yet arrived in the other. Two-phase commit — no partial state is ever visible coordinator shard A (debit) shard B (credit) PREPARE prepared — locks held COMMIT PREPARED locks are held on both accounts from PREPARE until COMMIT — a coordinator crash here holds them indefinitely Saga — the partial state is visible and is compensated debit — commits immediately credit attempted credit rejected (account frozen) compensating credit on shard A between the debit and the compensation, a balance query on shard A shows money already gone — the business must accept this window

Operational Requirements

Whichever mechanism is chosen, three operational things must exist before it goes to production.

A monitor for stuck state. For 2PC that is pg_prepared_xacts with an alert on any entry older than a few minutes. For sagas it is a count of instances stuck in a non-terminal state. For outboxes it is the age of the oldest undelivered row. All three are the same alert in different clothes: something started and did not finish.

-- prepared transactions older than five minutes are an emergency, not a warning
SELECT gid, prepared, owner, database,
       now() - prepared AS age
FROM   pg_prepared_xacts
WHERE  now() - prepared > interval '5 minutes'
ORDER  BY prepared;

A reconciliation job. Cross-shard invariants — total debits equal total credits, every order has its line items — must be checked on a schedule, because the mechanisms above reduce the probability of violation rather than eliminating it. A daily reconciliation that finds zero discrepancies is the evidence that the design works.

A tested failure path. The compensating action, the coordinator recovery and the outbox relay restart must each have been exercised deliberately. Every one of them is code that runs only during failures, which means it is code that is never tested by normal traffic.

Designing the Transaction Boundary Before the Shard Key

Every mechanism on this page is a workaround for a schema whose transaction boundary crosses its sharding boundary. That framing is useful because it turns an ongoing operational burden into a one-time design decision, and the decision is usually still open earlier than teams assume.

Start by listing the operations that must be atomic — not the queries, the operations: place an order, transfer funds, provision a tenant, cancel a subscription. For each, list the entities it writes. Then ask whether a single sharding dimension exists under which each list falls entirely inside one shard.

For a business-to-business SaaS product, sharding by tenant almost always achieves this: everything an operation touches belongs to one customer, and the handful of genuinely global objects — plans, feature flags, currency tables — are small enough to replicate to every shard as reference data. For a consumer marketplace it usually does not, because an order joins a buyer and a seller who are independent entities, and no key puts both on the same shard.

That distinction predicts how much of this page a system needs. Tenant-sharded systems can often run for years with zero distributed transactions and a single outbox for external effects. Marketplace-shaped systems need sagas from the first month, and should build the saga infrastructure before the first cross-shard operation rather than after the first reconciliation discrepancy.

The cost of finding out late

Retrofitting is expensive in a specific way: the mechanisms are not hard to build, but making existing writes idempotent is invasive. Every write path needs a deterministic key, every table needs a uniqueness constraint that did not exist, and every caller needs to pass something it never had. That work touches more code than the saga engine itself.

The mitigation is cheap and worth doing even when no cross-shard operation exists yet. Give every externally-triggered write an operation id from the start, store it, and constrain it. If the system never shards, the column is a small overhead and a useful audit trail. If it does, the hardest part of the migration is already finished.

Reads have a boundary too

One more decision belongs here rather than on the read-side pages: whether a reader may observe a partially applied cross-shard operation. Sagas guarantee it will happen, so any report, export or webhook that consumes both halves needs a rule. The usual answer is that consumers filter on the saga’s terminal state rather than on the underlying rows, which means the state must be queryable from wherever those consumers run — one more reason to keep saga state in a durable store rather than in the process that drives it.

A last practical note on sequencing: build the reconciliation query before building the mechanism. It is a few lines of SQL, it works against the current single-shard system unchanged, and it establishes the baseline that proves the invariant held before any distributed write existed. Teams that add it afterwards can never tell whether an early discrepancy came from the new mechanism or was always there.

Failure Modes

Failure mode Root cause Detection Mitigation
Prepared transaction never resolved coordinator crashed between prepare and commit; nobody owns the recovery pg_prepared_xacts entry with a growing age; vacuum unable to advance alert at five minutes; run a recovery process that resolves orphans from the coordinator’s log; never enable 2PC without this
Compensation itself fails the compensating write hit the same failure as the original, or the state changed underneath it saga instances stuck in compensating make compensations idempotent and retryable forever; escalate to a human queue after N attempts
Duplicate effect after retry the retried operation had no idempotency key, or the key was scoped per attempt instead of per operation reconciliation finds doubled amounts key on the business operation id, store it with the effect under a unique constraint
Outbox grows without bound the relay stopped and nobody noticed; or delivered rows are never pruned oldest undelivered row age; table size alert on age not on count; prune delivered rows on a schedule

Common Mistakes

  • Reaching for 2PC first. It is the most familiar mechanism and the least often appropriate, because it converts a partial failure into total unavailability.
  • Writing compensations that assume the world has not moved. A compensating credit must work even if the account has since been closed — which usually means it becomes a ledger entry rather than a balance update.
  • Treating the outbox as a queue. It is a durability mechanism for the local commit; ordering and fan-out belong in a real broker downstream of the relay.
  • Scoping idempotency keys to the attempt. A key generated per retry is not an idempotency key. It must be generated once, by the caller, for the business operation.
  • Skipping reconciliation because the mechanism is correct. The mechanism is correct; the deployment, the network and the operator are not.

FAQ

When is two-phase commit the right answer across shards?

When the operation must be atomic, the participants are few, they are all databases you control, and the operation is rare enough that its availability cost is acceptable. 2PC gives real atomicity and pays for it by making the transaction unavailable whenever any participant or the coordinator is down. For high-volume operations on the request path, a saga with compensations is almost always the better trade.

What is a prepared transaction leak and why does it matter?

A PREPARE TRANSACTION that is never committed or rolled back holds its locks and pins the transaction id horizon forever. Vacuum cannot advance past it, so bloat grows on every table in the database and the server eventually approaches transaction id wraparound. Monitoring pg_prepared_xacts and alerting on any entry older than a few minutes is mandatory if 2PC is enabled at all.

How does a saga differ from just retrying the second write?

A retry assumes the second write will eventually succeed. A saga assumes it might never succeed and defines what to do about the first write in that case — a compensating action that semantically undoes it. Retries handle transient failures; compensations handle permanent ones, such as the target shard rejecting the write on a business rule.

Do I need distributed transactions if my shard key is chosen well?

Usually not, and that is the point of choosing the key from the transaction boundary rather than from the query pattern alone. If every business operation touches exactly one tenant and tenants map to shards, the local transaction is sufficient. Distributed transactions are the price of operations that genuinely span the sharding dimension, such as transfers between accounts on different shards.

Articles in This Section

Idempotency Keys for Safe Cross-Shard Retries Every reliable cross-shard mechanism retries, and every retry is a duplicate waiting to happen. Idempotency keys are...
Implementing Two-Phase Commit Across PostgreSQL Shards Two-phase commit is the only mechanism that makes a multi-shard write genuinely atomic, and the only one that can take a...
Saga Pattern for Cross-Shard Business Transactions A saga trades atomicity for availability: each step commits immediately on its own shard, and failure is handled by...
Transactional Outbox for Cross-Shard Event Publishing When a write on one shard must cause an effect somewhere else — another shard, a search index, a webhook — the outbox...