Distributed Database Architecture

Design distributed database systems. Covers CAP theorem, consensus protocols, sharding strategies, replication topologies, conflict resolution, and choosing between consistency and availability.

A distributed database architecture enables a single logical database to span multiple physical nodes across networks, regions, or data centers. It is essential when a single database instance cannot scale horizontally, tolerate node failures, or serve low-latency reads and writes across geographically dispersed clients. This architecture is required when data volume exceeds the capacity of a monolithic database, when availability demands exceed the reliability of a single server, or when latency-sensitive applications must access data close to their users.

Data Distribution and Partitioning

Distributed databases divide data into chunks, called shards, and distribute them across nodes. Each shard contains a subset of the database’s rows, and the distribution is determined by a partition key—a column or set of columns used to compute a hash or range-based layout.

For example, in a user-facing application, a common choice is to partition by user_id, where each shard holds data for a range of user IDs. The partitioning strategy defines how data flows into shards:

  CREATE TABLE orders (
    order_id BIGINT PRIMARY KEY,
    user_id INT NOT NULL,
    order_date TIMESTAMP,
    total DECIMAL(10,2)
  ) PARTITION BY HASH(user_id);
  CREATE TABLE logs (
    log_id BIGINT PRIMARY KEY,
    timestamp TIMESTAMPTZ,
    level TEXT,
    message TEXT
  ) PARTITION BY RANGE (timestamp);

When a new row is inserted, the coordinator node computes the shard ID using a consistent hashing function and forwards the write to the appropriate shard. A failed node may return 503 Service Unavailable, and the client must retry with 429 Too Many Requests if the load exceeds the shard’s capacity.

Shards may be replicated across multiple nodes for fault tolerance. Each shard maintains a primary replica and one or more secondary replicas, with writes directed to the primary and reads optionally served from secondaries. The replication mode affects data consistency:

  {
    "replication": {
      "mode": "sync",
      "quorum": 2,
      "timeout_ms": 3000
    }
  }

A common failure mode occurs when the primary fails and the system promotes a secondary to primary. The new primary may miss writes that arrived at the old primary during the failure window. The client must detect this via timestamp-based staleness or logical timestamps in the replication stream.

Consistency and Conflict Resolution

In a distributed database, consistency is not assumed. Applications must choose a trade-off between consistency, availability, and partition tolerance—the CAP theorem.

The default consistency model is strong consistency (also called linearizability), where all operations appear to occur instantaneously at a single point in time.

To ensure strong consistency across shards, a distributed transaction coordinator manages the commit of operations across multiple nodes. A write that spans two shards requires a two-phase commit (2PC):

  1. Prepare phase: The coordinator sends PREPARE to each shard involved in the transaction.
  2. Commit phase: If all shards reply READY, the coordinator sends COMMIT to all.

If any shard fails during the prepare phase, the coordinator may send ROLLBACK to all participants.

Applications use transaction IDs to track progress. A transaction is in-flight if it has sent PREPARE but not yet received COMMIT or ROLLBACK.

When a node fails during a 2PC, the coordinator must recover the transaction state from persistent logs. The recovery process reads the transaction log, identifies uncommitted transactions, and reattempts the prepare or commit phase.

A common failure mode is split-brain, where two coordinators exist due to network partitioning. Each coordinator commits different sets of transactions. Resolving this requires a conflict resolution mechanism. One approach is version vectors, where each node maintains a vector of timestamps for every other node. When a node receives a write, it updates its vector and merges vectors during read or merge operations.

For example, a write operation returns a version_vector in the response:

{
  "transaction_id": "txn-7a3f2b",
  "status": "committed",
  "version_vector": {
    "shard-1": 142,
    "shard-2": 138,
    "shard-3": 141
  }
}

Upon reading data, clients can compare version vectors to determine whether a read is stale, and apply merge strategies such as last-write-wins or vector clock-based merging.

Topology and Node Discovery

A distributed database maintains a cluster metadata store that tracks the location of shards, their replicas, and the health of nodes. The metadata store is itself distributed and replicated.

Nodes register their presence via heartbeat messages sent at regular intervals. A node sends PING every 10 seconds to the cluster metadata service:

POST /heartbeat
Host: cluster-metadata.example.com
Content-Type: application/json

{
  "node_id": "node-123",
  "address": "10.0.2.15:5432",
  "last_seen": "2024-05-10T12:00:00Z",
  "status": "active",
  "load": 0.72
}

The metadata service responds with 200 OK and may return updated topology information:

{
  "topology": {
    "shards": [
      {
        "id": "shard-01",
        "replicas": [
          { "node_id": "node-123", "role": "primary", "status": "up" },
          { "node_id": "node-124", "role": "replica", "status": "up" }
        ],
        "partition_key": "user_id",
        "range": [1, 10000]
      },
      {
        "id": "shard-02",
        "replicas": [
          { "node_id": "node-125", "role": "primary", "status": "up" },
          { "node_id": "node-126", "role": "replica", "status": "up" }
        ],
        "partition_key": "user_id",
        "range": [10001, 20000]
      }
    ]
  }
}

When a node fails to respond to three consecutive heartbeats, it is marked down. The metadata service triggers a failover process: it promotes a replica to primary, updates the topology, and notifies all clients.

Clients discover the cluster by querying a seed list of initial nodes. Upon startup, a client connects to one seed node and retrieves the full topology. The client caches the topology and periodically refreshes it. If the seed node becomes unreachable, the client falls back to another seed node.

A subtle failure arises when the metadata service itself is distributed. If the metadata store is partitioned, clients may receive inconsistent views of the cluster. A write to a shard may succeed, but the metadata service does not update the shard’s location until a later time, leading to stale metadata.

Data Movement and Load Balancing

As data grows or access patterns shift, data must be rebalanced across nodes. The process of moving data from one node to another is called data migration.

Rebalancing is triggered by:

When a node is overloaded, the system initiates a split operation: a shard is divided into two, and the data is moved to two different nodes. The split is atomic: during the split, the original shard serves both old and new data, and reads are served from both.

A split is coordinated via a split coordinator that manages the following steps:

  1. Create new shard: A new shard is created on a target node.
  2. Copy data: The coordinator copies data from the source shard to the new shard.
  3. Update metadata: The metadata store is updated to reflect the new shard boundaries.
  4. Switch routing: Clients begin routing reads and writes to the new shard, with a dual-write phase where both old and new shards receive writes.
  5. Finalize: After a stabilization period, the old shard is marked inactive, and data is moved to the new shard.

During the split, a client may read from the old shard and write to the new one, leading to write skew. A write to a user with user_id=5000 may arrive at the old shard, but the next read may find the data in the new shard.

Load balancing is achieved through routing tables that map partition keys to shard IDs. The routing table is maintained in a distributed key-value store. When a client sends a query, the routing layer computes the target shard using a hash function:

SELECT * FROM orders WHERE user_id = 12345;

The routing layer resolves user_id = 12345 to shard-03, and forwards the query to the node hosting that shard.

If the routing table is stale, a client may send a query to a node that no longer hosts the shard. The node responds with a 307 Temporary Redirect:

HTTP/1.1 307 Temporary Redirect
Location: http://node-125:5432
Content-Type: application/json

{
  "error": "shard_moved",
  "new_shard": "shard-04",
  "new_location": "node-127"
}

Clients must follow redirects and cache the mapping to avoid repeated lookups.

Failure Modes and Recovery

Distributed databases face several persistent failure modes:

Recovery from node failure involves:

  1. Failover: Promoting a secondary replica to primary.
  2. Reconnection: The new primary sends REPLICA_SYNC messages to catch up with the previous primary’s write-ahead log.
  3. Client reconfiguration: Clients update their routing tables and begin routing to the new primary.

A common recovery pitfall is double-write during failover: during the time the primary is down, the secondary receives writes, but the old primary also continues writing. When the old primary comes back, it may have a conflict in the write-ahead log.

The system resolves this by replaying the log from the old primary and applying it to the new primary. The replay process returns a log_applied response:

{
  "replay_status": "complete",
  "applied_count": 1423,
  "conflicts": [
    {
      "transaction_id": "txn-123",
      "conflict_type": "write_conflict",
      "old_value": "2024-05-10T11:30:00Z",
      "new_value": "2024-05-10T11:35:00Z"
    }
  ],
  "conflict_resolution": "last_write_wins"
}

The client may then query the database with consistency_level = "strong" to ensure that all writes are visible.

Topology Management and Configuration

Distributed databases are configured via a set of topology parameters that govern how nodes interact, how data is partitioned, and how failures are handled.

Key configuration keys:

A configuration file might look like:

cluster:
  name: "user-data-prod"
  seed_nodes:
    - "10.0.2.15:5432"
    - "10.0.2.16:5432"
  metadata_store:
    type: "distributed_kv"
    endpoint: "etcd://etcd-1:2379"
    consistency_level: "strong"
  replication:
    mode: "sync"
    quorum: 2
    timeout_ms: 5000
  partitioning:
    strategy: "hash"
    key_columns:
      - "user_id"
      - "region"
  load_balancing:
    algorithm: "least_connections"
    min_load: 0.3
    max_load: 0.8

The system applies configuration changes dynamically. A change to replication.quorum takes effect immediately, and the system adjusts the replication protocol accordingly.

When a configuration change is applied, the system logs a config_update event:

{
  "event": "config_update",
  "timestamp": "2024-05-10T12:05:30Z",
  "config": {
    "replication.quorum": 3,
    "load_balancing.algorithm": "consistent_hash"
  },
  "applied_by": "[email protected]",
  "notes": "Increased quorum to 3 for higher durability"
}

Clients may subscribe to configuration updates via a watch mechanism:

GET /config/watch?version=12345
Host: cluster-metadata.example.com
Accept: application/json

// Response:
{
  "version": 12346,
  "changes": [
    {
      "field": "replication.quorum",
      "old": 2,
      "new": 3
    },
    {
      "field": "load_balancing.algorithm",
      "old": "least_connections",
      "new": "consistent_hash"
    }
  ]
}

This enables clients to adapt to configuration changes without restarting.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.