Data in Distributed Systems

In a single application with one database, a transaction keeps everything consistent and a join answers any question. Once a system is split into separately deployed services, both of those become hard, and the decisions about where each piece of data lives are among the most expensive to change later. This page builds on the database basics: the CAP theorem, eventual consistency, ACID and database replication are explained there. The examples follow Kestrel Energy, where the customer portal, meter-api and the billing system KBS (Kestrel Billing System) all read and write the same PostgreSQL server, db01.

Database per service

Database per service is the pattern in which each service keeps its persistent data private and other services can reach that data only through the service's API or the events it publishes, never by querying its tables. Microservices.io puts it as: "Keep each microservice's persistent data private to that service and accessible only via its API." It does not have to mean one database server per service. Three common levels of separation:

• Private tables in a shared server: cheap, but only a convention keeps others out.
• A schema per service, with a database role per service that can only use its own schema: the database enforces the boundary.
• A server per service: also separates load, connections, versions and maintenance windows, at the cost of more servers to run.

Kestrel learned why it matters on 6 October 2026. In incident INC-22047, a release of meter-api leaked database connections on db01 (see the worked incident). The leak was a bug in meter-api, but because the portal and KBS used the same server, the portal slowed down too: the shared database widened the blast radius. It also slows down change. Hamid Rahimi, the database administrator, cannot rename a column in the billing tables without asking three teams, because nobody knows which queries in which service read it. The anti-patterns page looks at the shared database as an anti-pattern; here the question is what you give up when you remove it:

Kestrel's target is modest: meter-api owns the readings in its own schema metering with its own database role, KBS owns the billing schema, and the portal reads both through APIs and a read model. All three can stay on db01 for now; separate servers can come later if load requires it. Inside one modular monolith, a schema per module gives most of the same benefit without any network calls.

CAP and PACELC in practice

The CAP theorem describes a choice that a replicated data store must make during a network partition: refuse some requests to stay consistent, or answer them with possibly outdated data. The databases page explains it, including the two common misreadings ("pick any two of three" and "NoSQL is AP, SQL is CP"). Eric Brewer, who proposed CAP, wrote in 2012 that "the '2 of 3' formulation was always misleading because it tended to oversimplify the tensions among properties."

Partitions are rare. The trade-off you pay for on every request is a different one, and PACELC (pronounced "pass-elk") names it. Daniel Abadi's 2012 formulation: "if there is a partition (P), how does the system trade off availability and consistency (A and C); else (E), when the system is running normally in the absence of partitions, how does the system trade off latency (L) and consistency (C)?" A write that must be confirmed by more replicas before it counts is more consistent and slower. Suppose db01's primary had three replicas, one in the same data center, one at a second site and one in another region:

import random

rng = random.Random(7)
# Round-trip times in ms from db01's primary to its replicas:
# one in the same data center, one in a second site 40 km away, one in another region.
def ack_times():
    return sorted([rng.uniform(0.3, 0.9),        # same data center
                   rng.uniform(1.0, 3.0),        # second site
                   rng.uniform(14.0, 30.0)])     # other region

writes = [ack_times() for _ in range(10_000)]

def p(values, pct):
    values = sorted(values)
    return values[round(pct / 100 * (len(values) - 1))]

print("wait for          p50      p99")
for needed, label in ((0, "primary only"), (1, "1 replica"), (2, "2 replicas"), (3, "all 3")):
    lat = [1.0 + (acks[needed - 1] if needed else 0.0) for acks in writes]  # 1 ms local commit
    print(f"{label:13} {p(lat, 50):6.1f} ms {p(lat, 99):6.1f} ms")

wait for          p50      p99
primary only     1.0 ms    1.0 ms
1 replica        1.6 ms    1.9 ms
2 replicas       3.0 ms    4.0 ms
all 3           23.0 ms   30.8 ms

Waiting for the remote region makes every write twenty times slower, all day, to protect against a disaster that happens rarely. Waiting for nobody (asynchronous replication) is fastest, but a replica may lag behind and a failover may lose the last writes. Most databases let you choose per database, per session or even per transaction; PostgreSQL's synchronous_commit setting is one example. The useful question is therefore not "is our database CP or AP?" but "which data needs which choice?":

Consistency models

A consistency model is the set of rules a data store promises about which values a read may return after writes, especially when several clients and replicas are involved. "Consistent" in CAP means one specific, strong model; "eventually consistent" means one of the weakest promises. Between them lies a whole range. The Jepsen project, which tests databases against their claims, maps the models as a hierarchy in which each stronger model implies the weaker ones below it. A simplified part of it, for single objects:

"Sticky available" means a client can keep working during a partition as long as it stays connected to the same server. Jepsen states it as: "All models at or stronger than read your writes can be at most sticky available." Eventual consistency does not appear on the Jepsen map at all: it only promises that replicas converge once writes stop, and says nothing about what a read returns in the meantime.

Linearizability

Linearizability is one of the strongest single-object consistency models: every operation appears to take effect at one instant between its start and its end, so once a write has completed, every later read by any client returns that value or a newer one. Jepsen's definition: "every operation appears to take place atomically, in some order, consistent with the real-time ordering of those operations." This is the C that Seth Gilbert and Nancy Lynch used when they proved the CAP theorem. You need it for things such as a lock, a leader election or a unique username: two clients must never both believe they won. It costs a round trip to a majority of replicas on every operation and refuses requests on the minority side of a partition.

Read-your-writes consistency

Read-your-writes consistency is the guarantee that after a client writes a value, its own later reads always reflect that write, even if other clients may still see older data. It sounds obvious, and it is exactly what breaks when an application writes to a primary and reads from an asynchronous read replica.

Kestrel's portal does this to take load off db01. A customer types in a meter reading by hand, the page reloads from the replica, which is two seconds behind, and the reading is gone. The customer enters it again, or calls Daniel Mensah's customer service. One common fix is to read from the primary for a short time after a user's own write:

LAG = 2  # seconds before the replica has applied a write

class Database:
    """A primary and one asynchronous read replica."""
    def __init__(self):
        self.primary, self.replica, self.pending = {}, {}, []

    def write(self, now, key, value):
        self.primary[key] = value
        self.pending.append((now + LAG, key, value))

    def tick(self, now):            # the replica catches up
        for due, key, value in [w for w in self.pending if w[0] <= now]:
            self.replica[key] = value
            self.pending.remove((due, key, value))

class Portal:
    def __init__(self, db, sticky_seconds=0):
        self.db, self.sticky = db, sticky_seconds
        self.last_write = {}        # customer -> time of their last write

    def submit_reading(self, now, customer, kwh):
        self.db.write(now, customer, kwh)
        self.last_write[customer] = now

    def show_reading(self, now, customer):
        recent = now - self.last_write.get(customer, -999) < self.sticky
        source = self.db.primary if recent else self.db.replica
        return source.get(customer, "no reading yet"), "primary" if recent else "replica"

for sticky in (0, 5):
    db = Database()
    portal = Portal(db, sticky_seconds=sticky)
    print(f"read from primary for {sticky} s after a write:")
    portal.submit_reading(0, "K-1042", 18_745)
    for now in (0, 1, 3, 6):
        db.tick(now)
        value, source = portal.show_reading(now, "K-1042")
        print(f"  t={now}s  {source:7}  {value}")

read from primary for 0 s after a write:
  t=0s  replica  no reading yet
  t=1s  replica  no reading yet
  t=3s  replica  18745
  t=6s  replica  18745
read from primary for 5 s after a write:
  t=0s  primary  18745
  t=1s  primary  18745
  t=3s  primary  18745
  t=6s  replica  18745

Only the customer who wrote is sent to the primary, and only briefly, so the replica still carries almost all of the load. Other fixes: show the submitted value from the write's own response without reading it back, or remember the position in the transaction log of the write and read from a replica only once it has caught up to that position. A fixed sticky period is a guess: if the replica ever lags more than five seconds, the problem returns, so monitor the replication lag.

Two-phase commit and why it is avoided

If one transaction could span the metering database and the billing database, consistency across services would be easy. Two-phase commit (2PC) is the classic protocol for that: a coordinator first asks every participant to prepare, which means doing the work, writing it durably and promising to commit; only when every participant has voted yes does it tell them all to commit, and if any votes no, it tells them all to roll back. Database support for it exists, for example the XA standard and PostgreSQL's PREPARE TRANSACTION.

That waiting is the core problem. Between the two phases, the participants hold locks on the rows involved, and if the coordinator fails at that moment, they stay blocked until it recovers. Every transaction also needs several round trips between services, and all participants must be reachable at once, so the availability of the whole is the product of the parts. Finally, most message brokers and many NoSQL databases do not support it at all. Microservices.io's outbox pattern page sums up why it is rarely used between services: the database and the broker "might not support 2PC. Also, it's often undesirable to couple the service to both the database and the message broker."

Two-phase commit still has its place inside one product, for example between the nodes of one distributed database, where the vendor has engineered recovery for it. Between independently deployed services, the usual alternatives are local transactions plus messages: the transactional outbox below and sagas, which replace one big transaction with a sequence of local ones and compensations.

Keeping data in sync

When meter-api stores a reading, KBS needs to know about it. The obvious code commits the reading to the database and then publishes a ReadingReceived message to the broker. That is a dual write: one operation that updates two separate systems, such as a database and a message broker, in two independent steps without a shared transaction, so that a crash or error between the steps leaves them disagreeing. The order of the steps only decides which way it goes wrong:

import sqlite3

class Crash(Exception):
    pass

def setup():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE readings (ean TEXT, ts TEXT, kwh REAL)")
    db.execute("CREATE TABLE outbox (id INTEGER PRIMARY KEY, event TEXT)")
    return db, []                     # the list stands in for the broker

def db_then_broker(db, broker, crash):
    with db:                          # commit
        db.execute("INSERT INTO readings VALUES ('871687400000000017', '02:30', 0.42)")
    if crash:
        raise Crash
    broker.append("ReadingReceived")

def broker_then_db(db, broker, crash):
    broker.append("ReadingReceived")
    if crash:
        raise Crash
    with db:
        db.execute("INSERT INTO readings VALUES ('871687400000000017', '02:30', 0.42)")

def outbox(db, broker, crash):
    with db:                          # one local transaction: both rows or neither
        db.execute("INSERT INTO readings VALUES ('871687400000000017', '02:30', 0.42)")
        db.execute("INSERT INTO outbox (event) VALUES ('ReadingReceived')")
    if crash:
        raise Crash
    # a relay process would publish the outbox rows later, also after a restart

for write in (db_then_broker, broker_then_db, outbox):
    db, broker = setup()
    try:
        write(db, broker, crash=True)  # the process dies between the two steps
    except Crash:
        pass
    rows = db.execute("SELECT count(*) FROM readings").fetchone()[0]
    queued = db.execute("SELECT count(*) FROM outbox").fetchone()[0]
    print(f"{write.__name__:15} readings={rows}  events on broker={len(broker)}"
          f"  in outbox={queued}")

db_then_broker  readings=1  events on broker=0  in outbox=0
broker_then_db  readings=0  events on broker=1  in outbox=0
outbox          readings=1  events on broker=0  in outbox=1

Database first: the reading is stored, but KBS never hears of it and the customer is not billed for it. Broker first: KBS bills a reading that does not exist. With the outbox, the reading and the message are committed in one local transaction, and a relay publishes the message afterward, even if it has to wait for a restart. This transactional outbox, its relays and the reason consumers must then handle duplicates are explained on Event-Driven Architecture and Service Communication. Kestrel has proposed this choice in ADR-0007, still under review (see Documenting Architecture).

Three common ways to keep copies of data in sync:

• Outbox and events. The owning service publishes what changed; others update their own copy. Best when the owner knows what others need.
• Change data capture (CDC). A tool reads the owning database's transaction log and turns committed changes into messages, without changing the application. Powerful for legacy systems such as KBS, but it publishes table rows, so the consumers become coupled to the owner's internal schema unless the stream is translated.
• Asking when needed. No copy at all: call the owner's API at the moment the data is needed. Always current, but it makes the caller depend on the owner being available and fast.

CQRS: separate models for writing and reading

CQRS (Command Query Responsibility Segregation) is a pattern in which the part of a system that changes data (commands) uses a different model, and often a different store, than the part that answers questions (queries), with the read side kept up to date from the write side. Martin Fowler's 2011 description credits Greg Young with it. The Azure Architecture Center summarizes it as: "Segregate the read and write operations for a data store into separate data models."

KBS stores invoices and payments in normalized tables, which is right for making changes correctly. The portal's statement page needs a per-customer summary, and computing it with joins and sums for every page view, on the busy db01, is wasteful. A read model holds the answer already prepared, a deliberate form of denormalization. In SQL:

-- Write model (KBS): normalized, one row per fact
CREATE TABLE invoices (id TEXT PRIMARY KEY, customer TEXT, period TEXT, amount_cents INTEGER);
CREATE TABLE payments (invoice_id TEXT, paid_cents INTEGER);
INSERT INTO invoices VALUES
  ('INV-5501', 'K-1042', '2026-08', 11250),
  ('INV-5602', 'K-1042', '2026-09', 11250),
  ('INV-5603', 'K-2077', '2026-09',  8400);
INSERT INTO payments VALUES ('INV-5501', 11250), ('INV-5603', 8400);

-- Read model (portal): one denormalized row per customer, ready to show
CREATE TABLE statement_view (customer TEXT PRIMARY KEY, invoices INTEGER,
                             billed_cents INTEGER, open_cents INTEGER);

-- The projection: rebuilds the read model from the write model
INSERT OR REPLACE INTO statement_view
SELECT i.customer, count(*), sum(i.amount_cents),
       sum(i.amount_cents - coalesce(p.paid_cents, 0))
FROM invoices AS i LEFT JOIN payments AS p ON p.invoice_id = i.id
GROUP BY i.customer;

-- K-1042 pays the September invoice: the write model changes at once...
INSERT INTO payments VALUES ('INV-5602', 11250);
SELECT * FROM statement_view WHERE customer = 'K-1042';

-- ...the read model only after the projection runs again
INSERT OR REPLACE INTO statement_view
SELECT i.customer, count(*), sum(i.amount_cents),
       sum(i.amount_cents - coalesce(p.paid_cents, 0))
FROM invoices AS i LEFT JOIN payments AS p ON p.invoice_id = i.id
GROUP BY i.customer;
SELECT * FROM statement_view WHERE customer = 'K-1042';

customer  invoices  billed_cents  open_cents
--------  --------  ------------  ----------
K-1042    2         22500         11250

customer  invoices  billed_cents  open_cents
--------  --------  ------------  ----------
K-1042    2         22500         0

The first query still shows 11,250 cents open after the payment: the read model is eventually consistent. In a real system the projection runs on each InvoicePaid event, so the delay is seconds rather than a nightly rebuild, and the portal shows "payment received, processing" for the customer's own payment (the read-your-writes problem again). Within one database, a view or a materialized view can be a simple first step; a separate store for the read side only pays off when the reads need a different technology, such as a search index, or a different scale.

Fowler's own warning is part of the definition: "beware that for most systems CQRS adds risky complexity." Use it for the few places where reads and writes really differ in shape or load, not for a whole system.

Event sourcing

Most systems store the current state and overwrite it. Event sourcing is a pattern in which a system stores every change to its data as an immutable event in an append-only log, and derives the current state by replaying those events instead of storing and overwriting it. Fowler (2005): "Capture all changes to an application state as a sequence of events." A bank account ledger works this way, and so does Git: the history is the truth, and the current state can always be rebuilt from it.

Meter readings are a natural fit, because they are facts about the past. A wrong reading is not overwritten; a correction is a new event. That keeps the history auditable, and it lets you answer "what did we believe on 1 October?", which matters when an invoice from that date is disputed:

from collections import defaultdict
from dataclasses import dataclass
from decimal import Decimal
from functools import reduce

@dataclass(frozen=True)
class ReadingReceived:
    ean: str
    day: str
    kwh: Decimal

@dataclass(frozen=True)
class ReadingCorrected:          # a correction is a new fact, not an UPDATE
    ean: str
    day: str
    kwh: Decimal

EAN = "871687400000000017"
events = [                        # the append-only event store
    ReadingReceived(EAN, "2026-09-29", Decimal("9.8")),
    ReadingReceived(EAN, "2026-09-30", Decimal("41.0")),   # meter glitch
    ReadingReceived(EAN, "2026-10-01", Decimal("10.4")),
    ReadingCorrected(EAN, "2026-09-30", Decimal("10.1")),  # Lucía fixes it
    ReadingReceived(EAN, "2026-10-02", Decimal("11.2")),
]

def apply(state, event):          # current state = fold over all events
    days = dict(state)
    days[event.day] = event.kwh   # received and corrected: the newest fact wins
    return days

current = reduce(apply, events, {})
print("total now:      ", sum(current.values()), "kWh")

as_of_oct_1 = reduce(apply, events[:3], {})   # replay only part of the history
print("total on Oct 1: ", sum(as_of_oct_1.values()), "kWh (before the correction)")

per_month = defaultdict(Decimal)  # a projection: a read model for the portal
for day, kwh in current.items():
    per_month[day[:7]] += kwh
for month in sorted(per_month):
    print(f"{month}: {per_month[month]} kWh")

total now:       41.5 kWh
total on Oct 1:  61.2 kWh (before the correction)
2026-09: 19.9 kWh
2026-10: 21.6 kWh

Each event type is a separate class, and replaying must handle every one of them. A language with sealed types makes the compiler check that: in Java 25, a sealed interface lists the permitted event records, and a switch without a default stops compiling when someone adds a new event type and forgets the replay:

import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

sealed interface MeterEvent permits ReadingReceived, ReadingCorrected {}
record ReadingReceived(String day, BigDecimal kwh) implements MeterEvent {}
record ReadingCorrected(String day, BigDecimal kwh, String reason) implements MeterEvent {}

public class Main {
    // Replaying the events rebuilds the state. The switch must handle every
    // permitted event type: add a third record and this no longer compiles.
    static Map<String, BigDecimal> replay(List<MeterEvent> events) {
        Map<String, BigDecimal> days = new TreeMap<>();
        for (MeterEvent event : events) {
            switch (event) {
                case ReadingReceived r -> days.put(r.day(), r.kwh());
                case ReadingCorrected c -> days.put(c.day(), c.kwh());
            }
        }
        return days;
    }

    public static void main(String[] args) {
        List<MeterEvent> events = List.of(
            new ReadingReceived("2026-09-29", new BigDecimal("9.8")),
            new ReadingReceived("2026-09-30", new BigDecimal("41.0")),
            new ReadingReceived("2026-10-01", new BigDecimal("10.4")),
            new ReadingCorrected("2026-09-30", new BigDecimal("10.1"), "meter glitch"),
            new ReadingReceived("2026-10-02", new BigDecimal("11.2")));

        Map<String, BigDecimal> now = replay(events);
        BigDecimal total = now.values().stream().reduce(BigDecimal.ZERO, BigDecimal::add);
        System.out.println("days:  " + now);
        System.out.println("total: " + total + " kWh");
        System.out.println("total on Oct 1: "
            + replay(events.subList(0, 3)).values().stream()
                  .reduce(BigDecimal.ZERO, BigDecimal::add) + " kWh");
    }
}

days:  {2026-09-29=9.8, 2026-09-30=10.1, 2026-10-01=10.4, 2026-10-02=11.2}
total: 41.5 kWh
total on Oct 1: 61.2 kWh

Replaying millions of events for every request would be slow, so event-sourced systems keep snapshots (the state at event number N, so only later events need replaying) and projections (read models built from the events, which is where CQRS often comes in). The costs are real:

• Old events live forever. Code written in 2030 must still understand an event stored in 2026, so event formats need the same care as public API contracts (see API Design Between Services).
• Deleting is hard. When a customer has the right to have personal data erased, an append-only log of immutable events is in the way. Keep personal data out of events where possible, or design for erasure from the start.
• A different way of thinking. Queries on current state go through projections, and fixing a bug in a projection means rebuilding it from the full history.

Choosing per piece of data

None of these patterns is right for a whole system. The useful unit of decision is the piece of data: who owns it, how fresh other services need it, and what happens if two copies disagree for a minute. Joost Bakker and Ravi Kumar's working list for Kestrel:

Most rows in that table are plain, current-state relational tables. That is normal: the special patterns are for the places where an ordinary database transaction cannot do the job.