Database Scaling

Database scaling is increasing the load and data volume a database deployment can handle while keeping latency acceptable and its correctness and recovery guarantees intact. The central engineering question is not what each technique does in isolation but what it guarantees under concurrency, failure and growth: a replica that returns stale data, a shard that cannot join, or a failover that loses commits has traded correctness for capacity, and that trade has to be explicit.

A useful mental model is a ladder of increasingly expensive rungs. The cheap rungs remove wasted work (a missing index, an N+1 query pattern, an oversized pool). The middle rungs add capacity without changing where data lives: a bigger node, read replicas, caches, partitioned tables. The top rung, sharding, changes where data lives, and every query, transaction, migration and recovery procedure has to learn about it. Climb only as far as the measured bottleneck demands, because each rung is far harder to descend than to climb. In the usual vocabulary, the bigger node is vertical scaling (scale up), while read replicas and sharding are horizontal scaling (scale out). Caches and partitioned tables on a single server are not scale-out by themselves, but they sit on the ladder because they raise capacity without adding database nodes.

This lesson walks the ladder in order and then covers the operational half that throughput benchmarks hide: replication position and failover objectives, online resharding, backups and capacity. Storage-engine and durability basics are in Database Foundations; consensus, quorums and conflict resolution across regions are in Distributed Data.

The same symptom, slow pages, has very different causes with very different remedies. The database may be doing too much work per query (a missing index, a sequential scan, a bad plan), waiting (lock contention, a long transaction, I/O queueing), or being asked too often (an N+1 loop, a cache that misses, too many application instances each holding connections). Adding read replicas to a write-bound or lock-bound workload costs money and leaves latency untouched; sharding a database whose real problem is one unindexed predicate multiplies the problem by the shard count.

Start by decomposing request latency: how much of a slow request is spent in the database at all, how much waiting for a connection from the pool, and how much in the query itself. Then look at the machine and the engine together: CPU, disk throughput and latency, buffer cache hit ratio, lock waits, active versus idle sessions, the slowest and most expensive query plans, and growth projections for rows and storage. The queries below are the PostgreSQL starting kit; other engines expose the same information under different names.

-- PostgreSQL 13+. Needs: shared_preload_libraries = 'pg_stat_statements' and CREATE EXTENSION pg_stat_statements;
-- 1. Where does the database spend its time? (statements ranked by total execution time)
SELECT calls,
       round(total_exec_time::numeric, 0) AS total_ms,
       round(mean_exec_time::numeric, 2)  AS mean_ms,
       rows,
       left(query, 80)                    AS query
  FROM pg_stat_statements
 ORDER BY total_exec_time DESC
 LIMIT 10;

-- 2. What are active sessions waiting on right now? (lock, I/O, client, CPU shows as NULL wait)
SELECT state, wait_event_type, wait_event, count(*)
  FROM pg_stat_activity
 WHERE backend_type = 'client backend'
 GROUP BY 1, 2, 3
 ORDER BY count(*) DESC;

-- 3. Oldest open transactions: long transactions hold locks and block vacuum cleanup.
SELECT pid, now() - xact_start AS xact_age, state, left(query, 60) AS query
  FROM pg_stat_activity
 WHERE xact_start IS NOT NULL
 ORDER BY xact_start
 LIMIT 5;

-- 4. For one suspect statement: real timings and buffer hits versus reads.
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM orders WHERE customer_id = 7731 ORDER BY created_at DESC LIMIT 20;

Read the results as a decision tree. If one or two statements dominate total execution time, fix those first; an index or a rewritten query is usually the cheapest capacity you will ever buy, and the indexing lesson covers how to confirm it from the plan. If many sessions wait on lock events, the fix is shorter transactions or a different access pattern, covered in Transactions and Concurrency, not more hardware. If sessions are mostly idle while requests are slow, the time is going to the network, the pool or the application. Only when CPU, I/O or write throughput is genuinely saturated after the waste is gone does the question of adding capacity become a real one.

Two habits keep this honest. Write down the growth projection (rows, bytes, queries per second in 6 and 18 months) so a scaling decision is made against a number instead of anxiety. And re-measure after every change: a team that adds a replica without a before-and-after comparison cannot tell whether the replica helped or the traffic simply dipped.

Database scalability is usually discussed as a choice between two directions. Vertical scaling (scale up) moves the same database to a bigger machine: more cores, more RAM, or faster storage with higher IOPS. It is simple and keeps the data model, transactions and operations unchanged, but it has a hard ceiling, and the node remains a single failure domain with a single writer. Resizing may need a restart or failover, depending on the platform: some managed services resize with a brief interruption, others with a replica promotion, and self-managed hosts usually need planned downtime. Horizontal scaling (scale out) spreads the work over more nodes: read replicas multiply read capacity, while partitioning across nodes (sharding) also multiplies write capacity and storage. It removes the single-machine ceiling at the cost of coordination: queries need routing, failover has to promote and fence nodes, replicas lag, and anything that spans shards (joins, transactions, aggregates) becomes more expensive.

Database scalability is therefore not one number. It is how well a deployment keeps its targets as demand grows, measured along several axes: throughput (queries or transactions per second), latency percentiles such as p95 and p99 rather than the average, data size, the number of concurrent connections, and whether backup, restore and failover still finish within their recovery objectives. A change that raises throughput while pushing p99 latency or recovery time past its target has not scaled the system. The rest of this section covers the vertical rung and connection limits; the following sections cover replicas, partitioning, sharding and the operational work that horizontal scaling adds.

More CPU, more RAM or faster storage is the least disruptive way to buy headroom: no application change, no data distribution, no new failure modes. It is also frequently enough for years. More RAM keeps more of the working set in the buffer cache, so fewer reads reach disk; faster storage cuts commit and checkpoint latency; more cores help parallel queries and concurrent sessions. Price usually grows faster than capacity at the top of the range.

A pool should protect the database, not maximize connections

A connection pool bounds how much work the database does at once. The instinct to size it to the number of concurrent requests is backwards: a database with 16 cores does not finish 400 simultaneous queries sooner than 20 simultaneous ones. The extra sessions fight for CPU, cache and locks, and context-switching overhead rises, so latency grows for everyone. In PostgreSQL each client connection is also a separate server process with its own memory, which is why the default max_connections is only 100.

Little's law gives a first estimate of how many connections are genuinely busy: average concurrency is arrival rate times time in system. At 2,000 queries per second with a 5 ms average query, that is 2,000 × 0.005 = 10 connections busy on average. A pool of 16 therefore runs at about 62% utilization, which leaves room for bursts. Cut the pool to 8 and utilization exceeds 100%: requests queue faster than they drain. Hand-tuning folklore such as "cores × 2 plus effective spindle count" (popularized by the HikariCP and PostgreSQL community sizing guides) is a reasonable starting point for server-side active connections, but the answer comes from measuring with your workload.

The curve shows why the failure is abrupt. Queueing delay is roughly flat at low utilization and bends upward steeply as utilization approaches 100%, so a 10% traffic increase that is harmless at 50% busy can double latency at 85%. The figure uses the single-server formula for the shape of the knee; a pool with many connections behaves better at the same utilization, but the knee is still there, and a retry storm (see retries with backoff) pushes utilization past it in seconds.

// Node.js, node-postgres. One pool per process; size it for the database, not for the request rate.
import pg from 'pg';

export const pool = new pg.Pool({
  connectionString: process.env.DATABASE_URL,
  max: 10,                        // per process. Total = instances x max, and must fit the server's limit
  idleTimeoutMillis: 30_000,
  connectionTimeoutMillis: 2_000, // fail fast when the pool is exhausted instead of queueing forever
  statement_timeout: 5_000,       // server cancels any statement running longer than 5 s
});

// 40 instances x max 10 = 400 server connections. PostgreSQL's default max_connections is 100.
// Either lower max, run fewer instances, or put a pooler such as PgBouncer in between.

Three practical rules follow. Cap the pool and make waiting for a connection time out quickly, so overload surfaces as a fast error that a load-shedding or backpressure mechanism can act on, instead of as an unbounded in-memory queue. Remember that the number that matters is total connections across all instances, including autoscaled ones, jobs and admin tools. And if many app instances need to share a few server connections, a pooler such as PgBouncer in transaction-pooling mode multiplexes them, at the price of losing session-level state (session settings, session-scoped advisory locks, LISTEN/NOTIFY), and, depending on version and configuration, restrictions on prepared statements. Use a bulkhead-style split so that reporting queries cannot take the connections that checkout traffic needs.

Database replication keeps a copy of the primary in step by shipping its change log, and a read replica is a copy that continuously applies that log. In PostgreSQL this is streaming replication of the write-ahead log (WAL); MySQL and others have equivalent binary-log mechanisms. Replicas take read traffic off the primary, which helps read-heavy workloads: dashboards, search-like queries, API reads, reporting. They do nothing for write throughput, since every write still happens on the primary and then again on every replica, and a heavy write stream can make replicas the thing that falls behind.

The topology has one writer, N followers and, separately, an archive. Which replica a read goes to matters: each replica has its own lag, and a replica that is busy with a long analytic query can be forced to delay or cancel replay (in PostgreSQL, hot-standby query conflicts governed by settings such as max_standby_streaming_delay), so separate reporting replicas from latency-sensitive ones. Read capacity is also a failover asset, but only if the replicas you promote are healthy and current.

Replica lag is a product behaviour, not a dashboard number

Asynchronous replication means the primary acknowledges a commit before any replica has applied it. Lag is usually milliseconds, and grows with write bursts, large transactions, slow replica disks, network problems, or a replica that is replaying a heavy workload. The user-visible consequence is the classic failure: a client writes to the primary and immediately reads from a replica that has not replayed the write, so it sees its own change missing. Customers see an old order status; an inventory count looks wrong; a user saves a profile and the page shows the old one.

There are three standard answers, and they can be mixed per endpoint. Send consistency-sensitive flows (checkout, account settings, anything just written) to the primary. Pin a session to the primary for a short window after any write by that session, which is simple but only approximate, because the window is a guess about lag. Or track the replication position: remember the log position of the user's last write and only read from a replica that has replayed at least that far, falling back to the primary or waiting briefly otherwise. The figure compares them, including PostgreSQL's synchronous_commit = remote_apply, which makes a commit wait until the named synchronous standby has applied it.

Position tracking is the most precise and is what the code below does. It is also session-scoped: it gives read-your-writes for one user, not a globally consistent view, so two different users can still see different states on different replicas. That is usually the right guarantee for a product. If you need every reader to see every committed write, you are asking for strong consistency, and that means reading from the primary, or the machinery in the distributed data lesson.

// Node.js, node-postgres. Read-your-writes by remembering the WAL position of the user's last write.
import pg from 'pg';
const primary = new pg.Pool({ connectionString: process.env.PRIMARY_URL, max: 10 });
const replica = new pg.Pool({ connectionString: process.env.REPLICA_URL, max: 20 });

export async function updateName(session, userId, name) {
  // A single statement runs in its own transaction and commits before the next query.
  await primary.query('UPDATE users SET name = $1 WHERE id = $2', [name, userId]);
  // The current WAL position is at or after the position of the commit above.
  const { rows } = await primary.query('SELECT pg_current_wal_lsn() AS lsn');
  session.minLsn = rows[0].lsn; // e.g. '0/3000148'; keep it in the session or a signed cookie
}

export async function getUser(session, userId) {
  const sql = 'SELECT id, name FROM users WHERE id = $1';
  if (session.minLsn) {
    const { rows } = await replica.query(
      'SELECT pg_last_wal_replay_lsn() >= $1::pg_lsn AS caught_up',
      [session.minLsn],
    );
    // Replay only moves forward, so once caught_up is true the next read on this replica is fresh enough.
    if (!rows[0].caught_up) return (await primary.query(sql, [userId])).rows[0];
    delete session.minLsn;
  }
  return (await replica.query(sql, [userId])).rows[0];
}

Measure the thing you are depending on. On the primary, pg_stat_replication reports per-replica write, flush and replay lag; a replica-side timestamp comparison is cheap but misleading on an idle system, because with no new transactions the apparent delay keeps growing even though the replica is fully caught up. Alert on replay lag in both bytes and seconds, and on a replica that has stopped replaying altogether.

-- On the primary (PostgreSQL 10+): per-replica lag as measured by the primary.
SELECT application_name, state, sync_state,
       pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn) AS replay_lag_bytes,
       write_lag, flush_lag, replay_lag
  FROM pg_stat_replication;

-- On a replica: time since the last replayed transaction.
-- Misleading on an idle primary: with no writes this keeps growing although the replica is fully caught up.
SELECT now() - pg_last_xact_replay_timestamp() AS replay_delay;

Sharding means splitting one dataset across independent database nodes or clusters, each owning a subset of the data, usually chosen by a shard key. It differs from table partitioning, which divides one logical table inside a single database deployment.

The two terms are often used interchangeably, and they should not be. Table partitioning divides one logical table into partitions, usually inside one database deployment. The query planner can skip partitions that cannot match (partition pruning), and lifecycle operations become cheap: dropping last year's partition is a metadata change, not a multi-hour DELETE that bloats the table and floods WAL. Sharding distributes data across independent database nodes or clusters, each owning a subset. It addresses limits partitioning cannot: write throughput beyond one primary, data beyond one machine, blast-radius and tenant isolation.

Partitioning is therefore a single-node optimization with an operational benefit. It only helps queries that include the partition key. A query that does not constrain the key visits every partition and can be slower than the unpartitioned table, because the planner and executor handle many more relations.

-- PostgreSQL 11+ (DETACH ... CONCURRENTLY needs 14+). Range partitioning by month.
CREATE TABLE events (
  id         bigserial,
  tenant_id  bigint      NOT NULL,
  created_at timestamptz NOT NULL,
  payload    jsonb       NOT NULL,
  PRIMARY KEY (id, created_at)           -- a unique key on a partitioned table must include the partition key
) PARTITION BY RANGE (created_at);

-- Bounds are half-open; spell the offset out so the session time zone cannot shift them.
CREATE TABLE events_2026_10 PARTITION OF events
  FOR VALUES FROM ('2026-10-01 00:00:00+00') TO ('2026-11-01 00:00:00+00');
CREATE TABLE events_2026_11 PARTITION OF events
  FOR VALUES FROM ('2026-11-01 00:00:00+00') TO ('2026-12-01 00:00:00+00');

CREATE INDEX ON events (tenant_id, created_at);  -- creates a matching index on every partition

-- Pruning: only events_2026_10 should appear in this plan.
EXPLAIN SELECT count(*) FROM events
 WHERE created_at >= '2026-10-10 00:00:00+00' AND created_at < '2026-10-11 00:00:00+00';

-- Retention without a bulk DELETE: detach, archive if needed, drop. (Not allowed inside a transaction block.)
ALTER TABLE events DETACH PARTITION events_2026_10 CONCURRENTLY;
DROP TABLE events_2026_10;

That last constraint is the first taste of the cost. A partitioned table cannot enforce a global unique key on id alone; you keep uniqueness by construction (a sequence or other generator that never repeats) rather than by index. Foreign keys into partitioned tables and cross-partition updates that move a row between partitions each have their own rules, so review the documentation for your version.

Both techniques require a deliberate key, and the key shapes the failure mode. Time partitions simplify retention and make recent-data queries fast, but concentrate all inserts on the newest partition, a built-in hotspot for write-heavy tables. Tenant-based sharding gives locality and isolation, but a single large tenant can overwhelm its shard while the others idle. Neither is a mistake; each is a trade-off to choose knowingly. A common combination partitions inside each shard by time, so both levels do what they are good at.

A good shard key spreads write load evenly, supports the dominant queries by landing them on one shard, and avoids frequent cross-shard joins. Those goals conflict, which is why the choice needs workload data. Hashing a customer id spreads load evenly but turns range queries ("all orders this week") into all-shard queries. Routing by region keeps data close to users but complicates global users and cross-region transfers. A low-cardinality or skewed key, such as the first letter of a name where perhaps 40% of users start with A–C, creates a hot shard that saturates while the others idle, which defeats the purpose of sharding.

Route through an indirection rather than computing the shard directly from the key. The figure hashes the key into one of a fixed number of buckets (1,024 here) and consults a versioned map from bucket to shard. The hash never changes, so a key keeps its bucket forever; resharding is an edit to the map. The consistent hashing lesson covers the ring-based variant of the same idea, which reaches the same goal (few keys move when capacity changes) without an explicit map.

// Node.js (ESM). Stable key -> bucket -> shard routing. Resharding edits the map, never the hash.
import { createHash } from 'node:crypto';

const BUCKETS = 1024;

export function bucketOf(key) {
  // First 4 bytes of a cryptographic hash; any well-mixed, stable hash works. Never use Math.random or Object hashes.
  return createHash('sha1').update(String(key)).digest().readUInt32BE(0) % BUCKETS;
}

// Version 7 of the map. In production this lives in a durable store and routers cache it by version.
const mapV7 = { version: 7, ranges: [
  { from: 0,   to: 340,  shard: 'shard-a' },
  { from: 341, to: 681,  shard: 'shard-b' },
  { from: 682, to: 1023, shard: 'shard-c' },
] };

export function shardFor(key, map = mapV7) {
  const b = bucketOf(key);
  const r = map.ranges.find((x) => b >= x.from && b <= x.to);
  if (!r) throw new Error('bucket ' + b + ' unmapped in map v' + map.version);
  return { bucket: b, shard: r.shard, mapVersion: map.version };
}

// Moving buckets 300-340 to a new shard publishes map v8 with those ranges reassigned:
// 41 of 1024 buckets move (about 4% of keys); every other key keeps its shard.

Why not simply hash(key) mod shard_count? Because the shard count is the thing that changes. A key stays on the same shard only if its remainder is identical under the old and new divisor. Going from 3 to 4 shards, that holds for just 3 of every 12 consecutive hash values, so 75% of keys move. With a fixed bucket map and an even rebalance, adding a fourth shard to three moves only the buckets needed to give it a quarter: about 25% of keys, the theoretical minimum for an even spread.

Pick a bucket count much larger than any plausible shard count (1,024 or 4,096 is common), so you can rebalance in small steps and move load without changing the hash. The map itself is critical metadata: keep it in a durable, versioned store, treat updates as a change process with review, and make routers fail closed when they cannot tell which version is current.

Sharding introduces routing, cross-shard queries, global uniqueness challenges, rebalancing and partial failure. Each deserves a design decision before the first row is moved.

Cross-shard queries and tail latency

A query that carries the shard key goes to one shard and behaves like it did before. A query that does not must scatter to every shard and gather the answers, and its latency is that of the slowest shard. If each shard is slow on 1% of requests, independently, a 3-shard fan-out is slow on 1 − 0.99³ ≈ 3.0% of requests and a 100-shard fan-out on 1 − 0.99¹⁰⁰ ≈ 63%. Real slowness is correlated, so treat the formula as the shape of the problem, not a forecast.

Joins, aggregations and sorting across shards move to the application or a query layer, where they are expensive and easy to get subtly wrong. The usual responses are denormalizing so that the dominant query lives in one shard, keeping small reference tables replicated to every shard, and feeding global reports from a separate analytical copy built from change data capture instead of live scatter-gather. Treat any endpoint that fans out to all shards as a deliberate, rate-limited exception.

Global uniqueness and identifiers

Per-shard sequences collide across shards. Use identifiers that are unique by construction: a UUID or time-ordered ID, a Snowflake-style ID that embeds a node or shard number, or a sequence with a per-shard offset. Uniqueness of a business key such as an email address across all shards cannot be enforced by a local unique index; it needs a dedicated lookup table on one shard, a routing rule that places every email on a deterministic shard, or acceptance of eventual detection and repair.

Transactions and partial failure

A transaction that touches two shards is no longer one local transaction. A distributed commit protocol such as two-phase commit keeps locks held across the network round trips and blocks when the coordinator fails at the wrong moment; most systems avoid it by choosing a key that keeps a business transaction on one shard, and by expressing the remaining multi-shard workflows as a saga with idempotent steps and compensations. Publishing events alongside a write belongs in a transactional outbox in the same shard as the data. Make every cross-shard write idempotent, since retries after a partial failure are certain.

Finally, one shard being down is now a normal condition, not an outage. The application must decide per request whether a failed shard means an error, a partial result, or a degraded answer, and monitoring must be per shard, because a healthy average hides one saturated group.

Resharding means moving live data, under traffic, with nothing lost and a path to undo it. A robust procedure has stages: a snapshot copy at a known log position, incremental change capture (CDC) that replays every change committed after that position, validation, shadow traffic, a controlled cutover and a rollback plan. Each stage exists because a naive version of it fails.

The snapshot takes time, and the source keeps accepting writes during it, so the snapshot alone is stale the moment it finishes. CDC closes that gap: the copy and the change stream start from the same consistent point, and the target applies changes until it is within a fraction of a second of the source. Validation then proves the copy is faithful; shadow reads route some read traffic to the target in parallel, compare the result to the source and discard it, which finds semantic differences, such as a missed column or a different collation, that counts cannot. Cutover is the only moment with a visible blip: writes to the moving buckets are paused, the target replays the last changes up to the source's final position, the new map version is published, and traffic resumes on the target. Keep the source read-only, not deleted, until the rollback window expires.

-- PostgreSQL 15+ (row filters on publications). Moves buckets 300-340 of orders from SOURCE to TARGET.
-- Prerequisites: the same schema already exists on TARGET (DDL is not replicated);
-- "bucket" is part of the table's replica identity (e.g. in the primary key) so UPDATE/DELETE can be filtered;
-- sequences are not replicated; wal_level = logical on SOURCE.

-- On SOURCE
CREATE PUBLICATION move_300_340 FOR TABLE orders WHERE (bucket BETWEEN 300 AND 340);

-- On TARGET: copy_data = true takes the initial snapshot, then the same slot streams later changes.
CREATE SUBSCRIPTION move_300_340
  CONNECTION 'host=source-db dbname=shop user=replicator'
  PUBLICATION move_300_340
  WITH (copy_data = true);

-- On SOURCE: how far behind is the subscriber? Cutover waits for this to reach (near) zero.
SELECT slot_name,
       pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes
  FROM pg_replication_slots
 WHERE slot_name = 'move_300_340';

Why dual writes alone are unsafe

The tempting shortcut is to have the application write to both the old and new databases and switch reads later. Without an explicit strategy this fails in two ways. Two concurrent writers can interleave differently in the two databases, so both copies "succeed" and end up with different final values; and one of the two writes can fail after the other committed, leaving a row in only one place with no record of which rows are affected. There is no ordering and no recovery.

Dual writing can be made safe, but only with machinery: a single authoritative writer (usually the old database, with CDC as the copy mechanism, which is the pattern above), or a durable record of the intent written in the same local transaction and replayed to the second store with ordering and idempotency, as in an outbox. If you cannot name which store is authoritative for every write during the migration and how disagreements are repaired, you are not ready to cut over.

Validate semantics, not just counts

Matching row counts prove only that the same number of rows exist. Two rows with swapped values, a column truncated at a different length or an encoding change all pass a count check. Compare per-bucket checksums of the full row contents, compare samples semantically through the application, and confirm that constraints and indexes exist and behave on the target. Even checksums have limits: they validate the data you copied, not behaviour that depends on collations, sequences, triggers or time zones.

-- Run on SOURCE and on TARGET and compare per bucket. Row-to-text requires identical column order and types.
-- For big buckets do this in id ranges; string_agg builds the whole string in memory.
SELECT bucket,
       count(*)                                        AS n,
       md5(string_agg(md5(o::text), '' ORDER BY o.id)) AS checksum
  FROM orders AS o
 WHERE bucket BETWEEN 300 AND 340
 GROUP BY bucket
 ORDER BY bucket;

Plan the way back before the way forward. Until the rollback window closes, keep the source intact and consider reverse replication from target to source so that a rollback does not discard writes accepted after cutover. Record the map version, LSN or position at each step, and rehearse the whole sequence, including the abort path, on a production-sized copy.

Two numbers frame every availability design. The recovery point objective (RPO) is the amount of data loss you can accept, measured in time or transactions. The recovery time objective (RTO) is the downtime you can accept. They are requirements to be written down per system, and then checked against what the replication setup can actually deliver.

With asynchronous replication, the primary acknowledged commits that the replica may not yet have, so a failover loses whatever was in the lag window: at 500 commits per second and 1 second of lag, roughly 500 commits exist only on the failed primary. Synchronous replication closes that window by making a commit wait for a standby, at the price of higher commit latency (one network round trip plus the standby's flush) and of availability: if the required standbys are unreachable, commits block. Configurations such as PostgreSQL's quorum form, ANY 1 (a, b), keep writes available when one standby is down. Even then, the guarantee covers acknowledged commits only while the surviving standby is part of the promotion decision; it is not a blanket guarantee against every failure combination.

RTO is the sum of the phases the figure splits it into: detecting that the primary is really gone, promoting a replica, and getting every client to talk to the new primary (DNS or proxy changes, connection pools reconnecting, cached endpoints). Detection is a trade-off: a short threshold fails over on a network blip, a long one extends downtime. The numbers in the figure are not benchmarks; measure your own with game days, because the long pole is usually client reconnection, not promotion.

# postgresql.conf on the primary (PostgreSQL 10+ syntax). Commit waits for ONE of the two named standbys.
synchronous_standby_names = 'ANY 1 (replica_a, replica_b)'
synchronous_commit        = on          # flush on the standby; remote_apply also waits for replay (read-your-writes)

# Promotion of a standby, PostgreSQL 12+:   SELECT pg_promote();
# The former primary has diverged. Rewind it with pg_rewind or rebuild it before letting it rejoin as a standby.

Preventing split brain

A successful promotion is not enough; clients must stop writing to the old primary. If the manager cannot reach the old primary, it cannot know whether it is dead or merely partitioned or paused. Promoting a replica without removing the old primary's ability to accept writes produces two primaries and two diverging histories, and reconciling them later is manual and lossy. Split-brain prevention needs an authoritative leadership mechanism and fencing: a lease or consensus-based leader election, an epoch number that storage or proxies check on every write, and the ability to isolate the old node (power off, network isolation, revoked credentials). The distributed locks lesson describes the same fencing-token idea for stale workers.

Rejoining the old primary is its own step. It may hold commits that never reached the new primary, so it has diverged; rewind it (for example with pg_rewind) or rebuild it from a backup, and treat the lost commits as an incident to reconcile with the business, not something to silently discard. Managed services and orchestration tools automate most of this, but the objectives and the fencing assumptions are still yours to verify.

A replica is a live copy, which is exactly why it is not a backup. An accidental DELETE, a bad migration or logical corruption replicates to every replica within moments, and it will also replicate to a synchronous standby. A backup is a point-in-time copy you can return to. For databases that matters in two parts: a periodic base backup, and a continuous archive of the log so that you can replay forward to any moment, which is point-in-time recovery (PITR).

The restore, not the backup, is the thing to test. Restore into an isolated environment on a schedule, time how long it takes (that duration is your actual RTO for data corruption, and it grows with data size), run integrity checks and a handful of application queries against the result, and practise the one operation you will be doing in a panic: restoring to just before a specific event and extracting the affected rows. Keep at least one copy in a different account or region, protected from deletion by the same credentials that run the database.

# On the primary: continuous WAL archiving. This archive_command is the documentation's illustrative example;
# production setups normally use pgBackRest, WAL-G or Barman, which also handle checksums, retries and retention.
archive_mode    = on
archive_command = 'test ! -f /mnt/wal_archive/%f && cp %p /mnt/wal_archive/%f'

# On an ISOLATED restore host: restore the latest base backup into an empty data directory, then set
restore_command          = 'cp /mnt/wal_archive/%f %p'
recovery_target_time     = '2026-10-29 09:13:59+00'   # just before the bad DELETE
recovery_target_inclusive = false
recovery_target_action   = 'pause'                    # inspect the data before promoting
# and create an empty file named recovery.signal in the data directory (PostgreSQL 12+), then start the server.

Capacity is an operational metric, not a benchmark

Track storage growth, WAL or redo retention, vacuum or compaction pressure, index bloat and backup duration alongside throughput. They fail differently from CPU: slowly, then all at once. Steady growth is a planning problem you can chart weeks ahead. A stuck replication slot or CDC consumer is an incident: the primary keeps every WAL segment the slot has not confirmed, and a disk that was fine yesterday fills in hours. Setting max_slot_wal_keep_size (PostgreSQL 13+) trades that risk for a slot that is invalidated, which a CDC pipeline then has to rebuild from a snapshot.

-- PostgreSQL. Capacity and housekeeping signals worth graphing.
SELECT pg_size_pretty(pg_database_size(current_database())) AS database_size;

-- WAL held back by replication slots: an inactive slot (a dead consumer or CDC job) retains WAL without limit
-- unless max_slot_wal_keep_size (PostgreSQL 13+) is set.
SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
  FROM pg_replication_slots;

-- Dead-row pressure and vacuum recency.
SELECT relname, n_live_tup, n_dead_tup, last_autovacuum
  FROM pg_stat_user_tables
 ORDER BY n_dead_tup DESC
 LIMIT 10;

Alert on thresholds that leave time to act: free disk against growth rate (not a fixed percentage alone), retained WAL per slot, replication lag, longest open transaction, dead-tuple ratio, backup age and the age of the last successful restore test. A scaling plan is finished only when it includes the operational recovery path for the larger system: backup and restore time at the new data size, failover for each shard, and the runbook for losing one of them.

For a typical growing system the sequence is: remove wasted work through indexes and query fixes; cap concurrency with a pool and add headroom vertically; add caching and read replicas for read-dominated load, with explicit freshness rules; partition large append-heavy tables for pruning and retention; and shard only when a single primary's write capacity or data size is the measured limit. Each step has a signal that tells you to take it and a signal that tells you it worked:

• Fix queries and indexes. Signal: a few statements dominate total execution time. Success: those statements' mean and p99 times drop and CPU or I/O falls with them.
• Pool and scale up. Signal: connection waits, high utilization, buffer cache misses. Success: pool wait time falls, p99 stays flat as load rises.
• Replicas and caching. Signal: read CPU or I/O saturation on the primary with writes comfortable. Success: primary read load falls; stale-read incidents stay at zero because the freshness rules were written first.
• Partitioning. Signal: large tables where most queries touch recent data, or retention deletes are painful. Success: plans show pruning, and retention runs in seconds.
• Sharding. Signal: the single primary is write-bound or too big to operate, after the rungs above. Success: write throughput scales with shards, and the number of cross-shard queries stays small and deliberate.

Premature sharding is a common and expensive mistake: a team shards at 50,000 users because it has heard that large systems do, and acquires cross-shard queries, distributed transactions and migration work years before it has the load that justifies them. The opposite mistake, scaling without identifying the bottleneck (adding replicas to fix a missing index), wastes money and leaves the problem in place. The sequence above is not dogma; the rule behind it is. Measure first, change one thing, verify with the same measurement, and keep a tested recovery path as the system gets more complicated.