Replication, Partitioning, and Sharding Questions

Direct answer

A single-shard failure in a sharded SQL database only affects the subset of users whose data lives on that shard, which is both the good news (the blast radius is bounded) and the trap (it is easy to under-react to an outage that "only" affects a fraction of users but is, for those users, a full outage). The playbook is: detect which shard and confirm the primary is actually unreachable (not just slow), fence it so it cannot accept writes if it comes back unexpectedly, promote the most caught-up in-sync replica, repoint the routing layer's shard map, then run data-integrity checks before declaring the incident over.

The playbook

flowchart TD
  A[Alert: elevated errors on shard 7] --> B{Is shard 7 primary reachable?}
  B -->|no| C[Fence old primary: revoke credentials, block writes]
  C --> D[Promote most-caught-up in-sync replica to primary for shard 7]
  D --> E[Repoint routing layer: shard map entry for 7 -> new primary]
  B -->|yes, but degraded| F[Diagnose: disk, connections, replication lag on shard 7]
  F --> G{Root cause fixable in place?}
  G -->|yes| H[Mitigate in place, no promotion]
  G -->|no| C
  E --> I[Run data-integrity checks: row counts, checksums vs other shards' expectations]
  H --> I
  I --> J[Validate application-level reads/writes on shard 7 succeed]
  J --> K[Post-recovery: reconcile any writes lost in the gap, document timeline]

1. Detection. The alert should identify the specific shard, not just "database errors," which means shard-aware monitoring (per-shard error rate, per-shard connection health, per-shard replication lag, meaning how far behind the primary this shard's replica has fallen) has to exist before the incident, not be built during it. Confirm scope immediately: query the shard-to-tenant or shard-to-user-range mapping to know exactly which users are affected, since this drives both the urgency (how many users, and are any high-priority accounts among them) and the communication (what to tell support or status pages).

2. Immediate mitigation. First check whether the primary is truly unreachable versus merely degraded (high load, a stuck query, disk pressure); a degraded-but-reachable primary may be fixable in place (kill a runaway query, add I/O capacity) without the risk of a promotion. If it is genuinely unreachable, fence it (revoke its credentials or otherwise ensure it cannot accept writes if it unexpectedly comes back mid-recovery, which prevents a split-brain where two nodes for the same shard both think they are primary), then promote the most caught-up in-sync replica for that shard specifically, and update the routing layer's shard map so application traffic for that shard's key range goes to the new primary.

3. Data-integrity checks. Before declaring the shard healthy, verify: row counts and, where feasible, checksums on recently-written tables against expectations (comparing against a recent backup or against another shard's equivalent tables if the schema allows structural comparison); confirm the promoted replica's applied position (its last-replayed log sequence number) to bound how many, if any, in-flight writes from the old primary may not have made it across, which tells you the actual recovery point objective (RPO) hit for this specific incident, not a theoretical one; and run a small set of representative application-level read/write operations end-to-end against the shard, not just database-level health checks, since a shard can be database-healthy while still broken from the application's perspective (wrong connection string, stale cached routing entry).

4. Post-recovery validation. Reconcile any writes that were in flight but unconfirmed at the moment of failure (check application-level idempotency logs or an outbox table if one exists, to identify anything that needs replaying or compensating). Document the actual timeline (detection time, fencing time, promotion time, full recovery time) against the shard's specific RTO target, and capture the RPO actually incurred (how many seconds, or how many specific rows, of writes were lost, if any) for the post-mortem, since "we recovered" and "we recovered with zero data loss" are different claims that need different evidence.

Trade-offs and pitfalls

The most common mistake is skipping the fencing step under time pressure ("it's clearly dead, let's just promote"), which is exactly the scenario that produces split-brain if the old primary was actually just network-partitioned rather than dead, and comes back mid-incident still believing it is primary. The second common mistake is declaring victory once the promoted replica is serving traffic without running the data-integrity checks, which defers discovering data loss or corruption from "during the incident, with full attention on it" to "days later, when a user reports a missing record," at which point root-causing it is far harder.

Direct answer

Roll the change across shards one at a time (or in small waves), never all at once, and design the change so the application code works correctly whether it is talking to an already-migrated shard or one still waiting its turn: that "mixed schema version" window is not an edge case to tolerate briefly, it is the normal operating state for most of the rollout's duration across 200 shards. A simple nullable-column add is close to free on modern engines and barely needs this discipline; a column TYPE change is a fundamentally harder problem that needs a shadow-table strategy, and treating the two as the same kind of migration is the most common way this goes wrong.

Adding a nullable column: close to free, but still roll it shard by shard

On PostgreSQL, adding a column with no default (or a NULL default) has always been metadata-only, since there is nothing to backfill: existing rows simply return NULL for the new column. Since PostgreSQL 11, adding a column with a constant DEFAULT is ALSO metadata-only (the default is stored in the catalog and applied lazily on read, not written into every existing row), which removed the old "this locks the whole table while it rewrites every row" cost. On MySQL 8.0.12 and later, ADD COLUMN can use the INSTANT algorithm for the same reason: a metadata-only change with no table rewrite. In both cases the DDL (data definition language, the ALTER/CREATE-style statement that changes a table's schema) itself is fast and low-risk per shard, so the real work is orchestration, not the ALTER statement:

1. Run the ADD COLUMN against one shard, verify it applied cleanly and did not cause unexpected lock contention.
2. Roll to the rest in small batches (not all 200 at once), watching replication lag and error rates between batches so a bad interaction surfaces on a handful of shards before it has touched everyone.
3. Application code must tolerate the column being absent on any shard it might still hit mid-rollout: read paths treat a missing column the same as a NULL value, write paths that need the new column simply do not populate it yet on unmigrated shards.
4. Only after every shard has the column does application code start relying on it being present everywhere.

Pitfall worth naming even here: MySQL's INSTANT algorithm has a real limit, a maximum of 64 row-format versions per table (MySQL keeps one internal row layout, or "version," for each instant column change so it can still read older rows written before that change; every instant ADD/DROP COLUMN uses up one of those 64 slots) before instant ADD/DROP COLUMN operations start failing outright. A table that has been through many rounds of instant schema changes over its life can hit this ceiling, at which point the next "trivial" nullable column add unexpectedly requires a full rebuild. Track how many instant changes a hot table has accumulated, it is not an infinite resource.

Changing a column's type: the actually hard case

A type change, for example widening INT to BIGINT, is not metadata-only on either engine, because every existing row's on-disk representation has to change. This forces a full table rewrite, which on a large, live shard means either a long lock (unacceptable) or an online shadow-table approach:

1. Create a shadow table with the new column type.
2. Backfill existing rows into the shadow table in throttled batches.
3. Capture ongoing writes to the original table during the backfill (a trigger-based approach like pt-online-schema-change, or binlog-tailing (reading MySQL's binlog, the append-only log of every row change the database writes for replication) like gh-ost on MySQL) so the shadow table stays current.
4. Once the shadow table has caught up, cut over: atomically rename the shadow table into the original table's place.
5. Repeat independently on every shard, since each shard's cutover is a separate, shard-local operation with its own timing.

The distinct hard part for a SHARDED cluster specifically: the application's cross-shard query layer has to handle shards that have already cut over serving the new type alongside shards that have not, for the entire multi-day (or longer) window it takes to work through 200 shards one at a time. That usually means the application reads the column as whichever type is wider until every shard has migrated, rather than assuming a single type cluster-wide.

Rollback

For the nullable-add case, rollback is close to free: drop the column on the shards that got it, since nothing downstream should have taken a hard dependency on it yet if the rollout followed the tolerate-absence discipline above. For the type-change case, rollback means keeping the OLD shadow (or not dropping the original table until the new type has been running cleanly in production for a defined bake period), since a shadow-table cutover is not trivially reversible once the old table has been dropped and traffic has been writing to the new one.

Minimizing downtime and avoiding cross-shard inconsistency

• Never gate the whole rollout on a single all-shards-at-once cutover; the failure blast radius of a wave-based rollout is one wave, not the whole cluster.
• Make the application layer explicitly version-aware for the schema, rather than assuming every shard is in the same state at every point in time. This single discipline is what actually prevents cross-shard inconsistency: an inconsistency only becomes a bug if code somewhere assumes uniformity that does not hold yet.
• Watch replication lag and lock wait time during each wave, and have an explicit pause/abort trigger (not just a plan to "keep an eye on it") if a wave shows problems before continuing to the next one.

Trade-offs and pitfalls

• Treating a type change like a nullable add is the single most common mistake here: teams that have gotten comfortable with instant column adds sometimes assume all schema changes are similarly cheap, and get surprised by the lock (or the shadow-table complexity) a type change actually requires.
• A multi-day mixed-schema-version window across 200 shards is not a brief edge case, it is most of the rollout. Code that only gets tested against "all shards migrated" or "no shards migrated" will have untested bugs in the state that actually matters most.
• Shadow-table migrations need their own monitoring separate from the DDL itself: the backfill's progress, its lag behind live writes, and its resource consumption (it is competing with production traffic for IO) all need visibility, not just "did the ALTER succeed."

Direct answer

An automated failover system for primary-replica pairs needs four pieces working together to minimize split-brain risk: a consensus-based health check (a majority, not a single observer, must agree the primary is actually down), fencing that guarantees the old primary cannot keep accepting writes once it loses leadership, a promotion-safety check that refuses to promote a replica too far behind, and an audit trail proving what happened and when. This is the same design pattern Patroni implements for PostgreSQL clusters, and it generalizes to any primary-replica system.

Architecture

flowchart TD
  A[Health check: primary missed N heartbeats] --> B{DCS quorum reachable?}
  B -->|no quorum| C[Do not promote: minority side, refuse writes]
  B -->|quorum reached| D[Elect candidate: most caught-up in-sync replica]
  D --> E[Fence old primary via watchdog/STONITH]
  E --> F{Fencing confirmed?}
  F -->|no| G[Abort promotion, alert on-call: cannot guarantee safety]
  F -->|yes| H[Promote candidate to primary]
  H --> I[Update DNS/service discovery to new primary]
  I --> J[Audit log: who/what promoted, at what LSN, data-loss window]
  J --> K[Old primary rejoins later as a follower once fencing is lifted]

Health checks. A single node observing "the primary didn't respond" is not enough evidence, since that node itself could be the one that is network-partitioned. Route the decision through a distributed configuration store (DCS: a small, separately-replicated coordination service such as etcd, Consul, or ZooKeeper) that requires a majority, or quorum, of its own members to agree before any leadership change proceeds; this is the same mechanism Patroni uses for PostgreSQL. If the surviving side of a partition cannot reach a majority of the DCS, it must not promote, even if it genuinely believes the primary is down, because it cannot distinguish "the primary is dead" from "I am the one who is isolated."

Fencing mechanisms. Before any promotion, the old primary must be guaranteed unable to keep accepting writes. Two layers, used together: revoke its ability to renew its leadership lease in the DCS (a soft fence: it should demote itself once it notices the lease expired), and a hardware or OS-level watchdog that forcibly resets the host if the soft fence has not been confirmed within a bounded time (a hard fence, sometimes called STONITH, "shoot the other node in the head"). The hard fence exists specifically for the case where the old primary is unresponsive rather than well-behaved, since a well-behaved node would have already demoted itself.

Promotion safety checks. Never promote a replica that is more than a defined, bounded amount behind the primary's last known position (measured by comparing log sequence numbers, LSNs). If the most caught-up available replica is still too far behind, the safer action is to alert and wait (accepting downtime) rather than promote and silently accept a larger-than-intended data loss; that threshold (how far behind is "too far") is a deliberate, pre-agreed trade-off between availability and data loss, not a default left unexamined.

Validating and auditing automatic promotions. Every automatic promotion writes an audit record: which node was promoted, what LSN it was at, what the old primary's last known LSN was (which bounds the data-loss window), and what triggered the decision. This is what turns "the system did something" into something a human can verify after the fact, and it is what a chaos-testing exercise (deliberately triggering failures in a controlled way, including in production during low-traffic windows, specifically to verify this pipeline works under realistic conditions rather than only in a lab) should be validating end-to-end, including checking that dependent services handling the failover (connection pools reconnecting, in-flight requests retried safely) do not cascade the failure outward.

Manual versus automatic failover: what data can be lost, and the RPO/RTO difference

Automatic failover is the right default for a well-tested, generic primary-down scenario, because RTO matters more than shaving the last few seconds of RPO for most services. Manual (or manual-confirmed) failover is worth keeping as an option, or even the default, for a scenario where the failure mode is ambiguous (a regional network partition isolating a meaningful fraction of followers, rather than a clean single-node crash), where an automated system cannot reliably distinguish "promote now" from "wait, this might resolve itself" and a wrong automatic decision (over-eager promotion during a transient blip) is worse than the extra minute a human takes to confirm.

Trade-offs and pitfalls

The most common design flaw is building the health check and the fencing mechanism as two independently-tuned systems that can disagree: a health check that decides to promote faster than the fencing mechanism can guarantee the old primary is actually stopped creates exactly the split-brain window the whole design exists to prevent. The second common flaw is never testing the failover path against a real, partial-partition scenario (only testing against a clean node crash), since a partition that isolates a meaningful fraction, not all, of the followers behaves very differently (some followers see a different quorum than others) than a simple full-node failure, and is the scenario most likely to expose a subtle bug in the promotion-safety logic.

Direct answer

Reshard with a six-phase pipeline: provision the target topology, dual-write new traffic to both old and new shard layouts, backfill historical data with a resumable watermark, verify old and new are byte-for-byte equivalent, cut traffic over atomically behind a versioned routing pointer, and only then decommission the old layout after a bake period. Consistency during the window comes from making every step reversible and idempotent rather than from any single "big atomic moment": the dual-write phase means both layouts are always current, so the cutover itself is just a routing-table flip, not a data-moving event, which is what makes it possible to roll back by flipping the pointer back.

Structured elaboration

Picking a target shard count

Do not reshard to "twice as many shards" by feel; size the target against an explicit operability budget. One reasonable budget: keep any single shard's full backfill or restore time under a fixed ceiling.

import math

TARGET_TOTAL_BYTES = 800 * (1000**4)      # 800 TB, decimal
COPY_THROUGHPUT_BPS = 1 * (1000**3)       # 1 GB/s sustained per-shard backfill/restore throughput (assumption)
MAX_BACKFILL_MINUTES = 30                 # operability budget: any single shard restores in <=30 min (assumption)
max_bytes_per_shard = COPY_THROUGHPUT_BPS * MAX_BACKFILL_MINUTES * 60
min_shards = math.ceil(TARGET_TOTAL_BYTES / max_bytes_per_shard)
print(f"max shard size for a {MAX_BACKFILL_MINUTES}-min backfill budget at {COPY_THROUGHPUT_BPS/1e9:.0f}GB/s: "
      f"{max_bytes_per_shard/1e12:.2f} TB/shard")
print(f"shards needed at 800TB total: {TARGET_TOTAL_BYTES/1e12:.0f}TB / {max_bytes_per_shard/1e12:.2f}TB = "
      f"{TARGET_TOTAL_BYTES/max_bytes_per_shard:.2f} -> {min_shards} shards (rounded up)")
CURRENT_SHARDS, CURRENT_BYTES = 64, 100 * (1000**4)
print(f"at 100TB today on {CURRENT_SHARDS} shards: {CURRENT_BYTES/CURRENT_SHARDS/1e12:.2f} TB/shard (under budget)")
print(f"at 800TB on {CURRENT_SHARDS} shards, unchanged: {TARGET_TOTAL_BYTES/CURRENT_SHARDS/1e12:.2f} TB/shard "
      f"-> {TARGET_TOTAL_BYTES/CURRENT_SHARDS/max_bytes_per_shard:.1f}x over budget")

Output:

max shard size for a 30-min backfill budget at 1GB/s: 1.80 TB/shard
shards needed at 800TB total: 800TB / 1.80TB = 444.44 -> 445 shards (rounded up)
at 100TB today on 64 shards: 1.56 TB/shard (under budget)
at 800TB on 64 shards, unchanged: 12.50 TB/shard -> 6.9x over budget

64 shards is fine today (1.56 TB/shard, under the 1.80 TB ceiling); left unchanged at 800 TB it would sit at 6.9x over budget, confirming a reshard is required, not optional. Round the 445-shard minimum up to a convenient power of two, 512, since binary range-splitting and consistent-hash ring math both work cleanly against powers of two.

The six phases

1. Provision: stand up the new (512-shard) topology, empty, with its own routing metadata entry (version N+1, inactive).
2. Dual-write: every write path starts writing to both the old shard (source of truth, synchronous) and the new shard (asynchronous, via the application's write path or a change-data-capture stream reading the old shard's write-ahead log). Writes carry an idempotency key (for example a UUID generated by the writer) so a retried write is a no-op on replay, not a duplicate. This is the step with a real latency cost: see Trade-offs below.
3. Backfill: bulk-copy historical rows in key order, tracking a resumable watermark (last key copied) so an interrupted backfill resumes instead of restarting; throttle the copy rate against a live I/O budget so it does not starve production traffic.
4. Verify: compare old and new for equivalence before trusting the new layout. Bucket rows by a stable key range, compute a per-bucket row count and content checksum on each side, and diff the buckets; a mismatched bucket isolates exactly which range needs re-copying without a full-table diff.
5. Cutover: flip the routing pointer from version N to N+1 as a single atomic metadata write (a compare-and-swap on a versioned config record), then have every router or client refresh and start reading and writing exclusively against the new layout.
6. Bake and decommission: keep dual-writing to the old layout for a defined bake period (long enough to catch anything the verification step missed) before stopping writes there and reclaiming it.

Consistency guarantees during cutover

The cutover itself moves zero bytes, it only flips a pointer, because phases 2 to 4 already made the new layout current and verified. The remaining risk is a request that started routing under version N completing its write after the pointer flips to N+1: closing that window means the router holds a brief write-barrier (reject or queue new writes for a bounded number of milliseconds) while in-flight version-N writes drain, confirmed by each shard acking that its write queue is empty, before flipping the pointer live for reads.

Automated rebalancing

Once cut over, the new, larger shard count still needs ongoing balance as data grows unevenly across it; that is a separate, continuous concern (splitting hot ranges, merging cold ones) rather than a one-time step of this migration.

Rollback

Because the old layout kept receiving writes through the bake period, rollback is the same pointer flip in reverse, no data movement, as long as it happens before the bake period ends and the old layout is decommissioned. Rolling back after decommission means re-running the backfill in reverse, which is why the bake period length is a real risk decision, not a formality.

Worked example

See the shard-count derivation above. The per-bucket checksum comparison from the Verify phase is the concrete way to confirm cutover correctness: bucket rows by key range, compute a checksum per bucket on both layouts, and diff them; run against a real database and deliberately corrupt one row, and the mismatched bucket isolates the problem immediately, without a full-table diff.

Operational concerns

• Migration bandwidth: the backfill and dual-write catch-up share the same network and disk I/O budget as production traffic; throttle the bulk-copy stream and prefer reading the source's replication log (WAL/CDC) for the dual-write's asynchronous side over re-querying the base table, since a log read is a small, steady stream instead of a query load on the primary.
• Latency impact: synchronous dual-write to two targets adds the slower of the two write latencies to every request; the fix is to make only the OLD write synchronous (correctness-critical, unchanged) and the NEW write asynchronous with its own retry queue, accepting that the new layout can lag by seconds during backfill without affecting user-facing latency, since it is not yet the source of truth.
• Verification of correctness: bucket-checksum comparison (above) over full re-diffing, because it isolates a mismatch to a specific key range instead of forcing a byte-for-byte scan of 800 TB on every check.

Trade-offs and pitfalls

• Choosing a target shard count that is too conservative (for example, doubling instead of sizing against a real growth and operability budget) means repeating this entire six-phase project again soon; size against the actual growth target, not the next obvious power of two.
• Synchronous dual-write to both layouts is the simplest correctness story but pays the tax described above on every write; most production migrations use synchronous-old/asynchronous-new plus CDC catch-up instead, trading a small, bounded staleness window on the new layout (acceptable, since it is not yet serving reads) for materially better write latency during the migration.
• Skipping or shortening the bake period to "move faster" removes your only cheap rollback path; a verification pass that looked clean can still miss an edge case that only a few days of real production traffic surfaces.
• A common mistake is treating the backfill as the hard part and the cutover as trivial; in a well-run migration it is the reverse; nearly all of the engineering risk lives in making the cutover pointer flip truly atomic and the write barrier truly bounded, not in the bulk copy.

Direct answer

Replication lag breaks read-after-write consistency (the guarantee that a user who just wrote something sees that write on their very next read) whenever the read happens to land on a follower that has not yet applied it. There are four standard patterns to guarantee a user sees their own recent writes despite this: sticky sessions (route a user's reads to the same node their write went to, for a window of time), session-scoped read-your-writes tracking (remember the position a user's last write reached, such as an LSN, log sequence number: a monotonically increasing marker of a database's write history, and only serve their reads from a replica that has caught up to at least that position), forcing reads-after-writes to the primary for a short window, and, less commonly, bounded-staleness reads that block briefly until a replica catches up rather than either redirecting or serving stale data outright.

The four patterns, and their trade-offs

Worked example: a social-network timeline

A user posts a comment, then immediately reloads their own timeline. If the reload hits a replica still 800 milliseconds behind (a fully normal amount of replication lag under moderate load), the comment appears to have vanished, which reads as a bug even though nothing is actually wrong. A concrete implementation: on write, the API server captures the WAL position (or an equivalent monotonic write marker) the write reached and returns it to the client as an opaque read-your-writes token embedded in the response (client-side technique); the client attaches that token to its next read request; the read-routing layer (server-side technique) checks each candidate replica's current applied position and only sends the request to one that has caught up, falling back to the primary if no replica has caught up within a short timeout (for example 50 milliseconds), rather than blocking indefinitely. This combination avoids sticky-session's node-affinity cost and avoids sending every read to the primary, while still guaranteeing the user sees their own comment.

Implementation details worth naming explicitly

A connection pooler alone does not solve this: PgBouncer, the common PostgreSQL pooler, is a pure TCP-level connection pooler with no query inspection and no awareness of a replica's replication position, so it cannot itself implement token-aware read/write splitting, and pointing it at a random or round-robin pool of replicas will pick one that has not caught up. The token-aware routing logic has to live in a layer that actually understands queries and replica state: application-level routing that picks a connection target per request based on the token, or a proxy built for query routing rather than plain pooling (for example Pgpool-II's load-balancing mode, which does inspect queries to split reads from writes) sitting in front of, or instead of, a plain pooler like PgBouncer, which continues to do only what it is built for: pooling connections to whichever destination it is told to use. A retry-on-lag pattern (catch the case where no replica is caught up, retry against the primary once, rather than retrying the same lagging replica) keeps tail latency bounded. Session-affinity state (which token belongs to which in-flight session) needs to live somewhere the load balancer or API layer can cheaply access it on every read, typically in the request context itself (the token travels with the request) rather than in a separate lookup service, to avoid adding a new dependency to every read's critical path.

Trade-offs and pitfalls

Sticky sessions are the easiest pattern to ship first, but teams often leave them in place permanently rather than migrating to a token-based approach, which silently caps how much read-scaling benefit replicas actually provide as traffic grows, since sticky users never spread across the replica fleet. Forcing all writes' subsequent reads to the primary "to be safe" is the opposite mistake: it looks correct but reintroduces exactly the primary-load problem replicas exist to solve, for reads that, in most product flows, do not actually need same-millisecond freshness. The right default is to reserve the primary-forcing pattern for the specific, small set of user actions where staleness is visibly wrong (their own just-written content), and use the token-based approach for everything else.