Online resharding, 2PC, and the operational cost of sharding

A Citus cluster needs to add a worker node to handle capacity growth. The old approach required a maintenance window: halt writes, copy data, restart. Citus 11.1 changed this with an online rebalancer that moves shards while the cluster stays fully operational, using the same logical replication mechanism that Postgres uses for upgrades.

Online resharding: how it works

Citus 11.1 (September 2022) introduced non-disruptive online resharding. The mechanism:

1. Start rebalancer: SELECT citus_rebalance_start();
2. Planner picks: which shards move from worker A to worker B for balance
3. For each shard being moved:
   a. Create the shard on the target worker (empty)
   b. Start logical replication slot on source for that shard's rows
   c. Logical replication copies existing rows to target, then streams changes
   d. Once lag is near zero: pause writes to shard (~sub-second)
   e. Atomically update shard-map metadata: shard now lives on target
   f. Drop the source shard
   g. Resume writes — now routed to target
4. Next shard begins (pipeline of moves in parallel, configurable concurrency)

Performance profile:

• Per-shard write pause: under 1 second
• Total rebalance time: scales with data volume (~10–100 MB/s network throughput)
• Application impact: brief connection stall during the per-shard pause; retried transparently by the coordinator

The rebalancer is idempotent and resumable — if it crashes mid-move, the next citus_rebalance_start() picks up from where it left off using pg_stat_subscription state.

Senior teams rehearse the rebalancer in staging before running in production. The first execution should not be an emergency.

Distributed transactions and 2PC

A transaction touching rows on multiple shards requires two-phase commit (2PC):

-- Single-shard: safe, normal Postgres transaction
BEGIN;
UPDATE orders SET status = 'shipped' WHERE tenant_id = 42 AND order_id = 99;
UPDATE inventory SET qty = qty - 1 WHERE tenant_id = 42 AND sku = 'ABC';
COMMIT;
-- Both tables are co-located on the same shard → single-node transaction

-- Cross-shard: requires 2PC
BEGIN;
UPDATE orders  SET status = 'refunded' WHERE tenant_id = 42 AND order_id = 99;
UPDATE ledger  SET balance = balance + 100 WHERE account_id = 999;  -- NOT tenant-scoped
COMMIT;
-- Citus uses PREPARE TRANSACTION + COMMIT PREPARED across both shards

2PC cost:

• Latency: minimum 2 round-trips (prepare phase + commit phase); often more with clock skew handling.
• Throughput: each shard holds its connection open during the prepare phase — contention under high concurrency.
• In-doubt transactions: if the coordinator crashes between PREPARE TRANSACTION and COMMIT PREPARED, the prepared transactions sit in pg_prepared_xacts holding locks until manually resolved. This is the failure mode to monitor and have a runbook for.

Best practices:

1. Design transactions to be single-shard: co-located tables make this natural for tenant-scoped operations.
2. Prefer eventual consistency (outbox pattern, saga) where the business logic permits rather than distributed 2PC.
3. Monitor pg_prepared_xacts on every worker; alert on any row older than 5 minutes; have a recovery runbook (COMMIT PREPARED / ROLLBACK PREPARED).

The N× operational multiplier

Senior teams automate everything via per-shard orchestration — Ansible with a loop over worker hosts, Terraform for_each, Kubernetes operators — before going live. Without automation, ops staff scales linearly with shard count, which is the operational tax of sharding made concrete.

Resharding the key: the expensive case

Sometimes the original shard key turns out wrong (sharded by user_id but 90% of queries need org_id). Re-sharding by a different key requires rewriting every row to its new location:

1. Dual-write + backfill: app writes to both old and new sharding schemes; backfill historical data; cut over reads; drop old. Months of work.
2. New parallel cluster: stand up a new cluster with the correct key, replicate via CDC, cut over. Most flexible; most operational work.
3. Accept the cost: in practice, teams endure the wrong shard key rather than pay the migration cost — a strong argument for choosing carefully from the start.

The senior takeaway: shard-key change is among the most expensive operations in databases. The design decision made at scale-out time is lived with for years.