A new Citus worker is only empty capacity. It does not relieve a saturated worker until the cluster transfers real shard ownership—and that transfer competes with the workload it is supposed to protect.

Situation

RetailCo’s order platform has already made the hard architectural decision described in the Citus on EC2 design: tenant-owned tables are distributed by tenant_id, the checkout graph is colocated, and three worker primaries own the existing shards.

The cluster is now approaching its operating limit. One worker has little storage headroom, checkout P99 rises during maintenance, and the next seasonal peak will exceed the current safety margin. RetailCo provisions worker-d and registers it:

SELECT citus_add_node('worker-d.internal', 5432);

The command registers the node and copies reference tables. It does not relocate existing distributed shards. Citus documentation is explicit: old shards stay where they are unless they are redistributed, so adding a worker may not improve performance by itself: Citus cluster management.

flowchart TD
  App[RetailCo checkout traffic] --> Coord[Citus coordinator]
  Coord --> A[worker A — existing shards]
  Coord --> B[worker B — existing shards]
  Coord --> C[worker C — existing shards]
  Coord --> D[worker D — reference data only]
  A --> Limit[storage and write pressure remains]
  B --> Limit
  C --> Limit

RetailCo is a reference architecture, not a claimed customer deployment. Its 40-TB dataset, shard sizes, and traffic shape are scenario inputs. No movement duration, throughput improvement, or application-latency result in this article is presented as measured fact.

Version matters. By this article’s March 31, 2026 cutoff, Citus 13.2 had added snapshot-based node addition and rebalancing performance improvements, while Citus 14.0 added PostgreSQL 18 compatibility. The public stable API manual labels itself 13.0.1, so every production runbook must validate function signatures and behavior against the installed Citus minor before execution: Citus 13.2 release notes and Citus 14.0 release notes.

The Problem

“Run the rebalancer” sounds like one administrative command. Operationally, it is an online data migration across independent PostgreSQL servers.

For each selected shard group, the system must copy the existing state, keep up with concurrent changes, move the ownership boundary, update metadata, and remove the old placement. During that work, the source still serves application traffic while reading data for the copy. The target writes the new placement. Network paths carry the transfer. PostgreSQL logical replication carries concurrent changes. Standbys, WAL archives, monitoring, and backups all observe the additional work.

The word nonblocking therefore needs a precise interpretation. Citus Community Edition 11.0 and later supports nonblocking reads and writes during shard rebalancing. That means the application can continue using a shard during most of the movement. Citus still takes a brief write lock when it updates metadata and promotes the target placement: Citus online rebalancing.

Nonblocking does not mean:

  • zero additional I/O, CPU, WAL, or network traffic;
  • zero latency change for the application;
  • zero locks at cutover;
  • automatic correction of a hot tenant;
  • rollback of all completed moves when the job is stopped;
  • permission to begin with almost no disk headroom.

The production question is not simply, “Can Citus move shards while writes continue?” It is:

Can RetailCo transfer enough ownership before capacity is exhausted, while keeping checkout latency, replication health, recoverability, and abort options inside explicit limits?

Treat Rebalancing as a Controlled Data Migration

The recommended approach is a plan, canary, observe, expand, and validate loop. Do not start with maximum parallelism and wait for graphs to turn red.

flowchart TD
  Baseline[capture workload and capacity baseline] --> Add[add and validate target worker]
  Add --> Plan[generate and review movement plan]
  Plan --> Canary[start with one task per node]
  Canary --> Observe[observe application and database limits]
  Observe --> Expand[increase parallelism only with headroom]
  Observe --> Stop[stop if an abort threshold is crossed]
  Expand --> Validate[validate placements and workload balance]
  Stop --> Replan[inspect state and generate a new plan]

Step 1 — prove the new worker is a safe destination

Before registering the worker, verify the same PostgreSQL major, Citus line, extensions, roles, TLS trust, collation assumptions, storage policy, kernel settings, monitoring, backup agent, and standby topology used by the existing fleet.

After registration, check cluster metadata and all-node connectivity:

SELECT nodeid,
       nodename,
       nodeport,
       isactive,
       noderole,
       shouldhaveshards
FROM pg_dist_node
ORDER BY nodeid;

SELECT *
FROM citus_check_cluster_node_health();

citus_check_cluster_node_health() checks every node-to-node connection, not only coordinator-to-worker connectivity. That matters because Citus system operations can require workers to connect to one another: Citus utility functions.

Large reference tables can make citus_add_node() slow because they are copied at activation. Citus permits deferring that copy with citus.replicate_reference_tables_on_activate = off, but then reference data is copied when a distributed shard is first created or moved to the node. Deferral moves the cost; it does not remove it: Citus configuration reference.

Do not call the target ready merely because it appears in pg_dist_node. Prove that its standby is streaming, backups include it, alerts are enabled, storage throughput is provisioned, and the coordinator can reach the direct PostgreSQL endpoint. Citus documents that shard rebalancing bypasses pool information such as PgBouncer and uses direct connections: Citus metadata tables.

Step 2 — inventory the movement unit

Citus moves a selected shard together with the shards colocated with it. For RetailCo, moving an orders shard can also move the matching order_items, payments, customers, and tenants shards. That is correct for transaction locality, but it means the operational unit can be much larger than one row in an orders-only size report: Citus shard movement.

Inspect placements and colocation IDs before planning:

SELECT table_name,
       shardid,
       colocation_id,
       nodename,
       shard_size
FROM citus_shards
ORDER BY nodename, colocation_id, shardid, table_name;

The citus_shards view reports table, shard, node, colocation group, and size. Aggregate the entire colocation group when estimating a move. Also identify tables that became implicitly colocated only because their distribution columns share a type and shard count. Current Citus documentation warns that unrelated implicit colocation can create unnecessary movement cascades: Citus create_distributed_table.

Every table in the group must support the selected transfer mode. A forgotten audit table without a primary key or replica identity can block an otherwise well-designed online move.

Step 3 — inspect the strategy Citus will actually use

Current Citus installations ship with by_shard_count and by_disk_size. The current metadata documentation marks by_disk_size as the default. It calculates cost from total relation size, including indexes and colocated shards, and uses a threshold to avoid movement for insignificant disk differences: Citus rebalancer strategy table.

Never infer the active default from memory or a blog post. Query the cluster:

SELECT name,
       default_strategy,
       default_threshold,
       minimum_threshold,
       improvement_threshold
FROM pg_dist_rebalance_strategy
ORDER BY name;

The two built-in strategies answer different questions:

StrategyCost modelUseful whenBlind spot
by_shard_countEach shard group has equal costShards, traffic, and worker capacity are genuinely similarEqual counts can hide large size skew
by_disk_sizeTotal bytes including indexes and colocated shardsStorage balance is the primary constraintEqual bytes do not imply equal CPU, locks, or traffic
Custom strategyOperator-defined shard cost, node capacity, and placement rulesHeterogeneous workers or measured traffic placementMore logic to test and govern

For RetailCo, by_disk_size is the safer starting point because one worker is short on storage. It is not a traffic balancer. A small shard serving a dominant tenant can be hotter than a much larger historical shard. Citus exposes tenant and statement statistics that can inform a custom strategy, but a custom cost function should be tested against skew, reset windows, and missing statistics before it controls placement.

Step 4 — generate the plan before moving bytes

Generate the exact source, destination, shard, and size list:

SELECT table_name,
       shardid,
       pg_size_pretty(shard_size) AS shard_size,
       sourcename,
       sourceport,
       targetname,
       targetport
FROM get_rebalance_table_shards_plan(
    rebalance_strategy := 'by_disk_size'
);

The documented planner can accept controls such as a relation, threshold, maximum move count, excluded shards, and drain-only mode. The plan is not a reservation. Citus warns that the plan produced now can differ from the later execution because cluster facts such as disk space can change between calls: Citus rebalance plan.

Turn the output into an operator-reviewed movement ledger:

GateRequired evidence
Destination capacityPlanned bytes plus index growth, WAL, temporary transfer state, autovacuum, and safety reserve fit
Source capacitySource can serve traffic and scan the selected shard group without crossing latency limits
NetworkSource-to-target transfer has measured headroom and does not starve replication or application paths
Replica identityEvery updated or deleted table in the group has a supported replica identity
Standby healthSource and target standbys are caught up before the move
RecoveryBackup and WAL retention cover the entire maintenance window
ApplicationAbort thresholds exist for error rate, P95, P99, timeout rate, and checkout throughput
OperationsAn owner can stop, inspect, replan, and communicate the change

For a rough lower bound—not a promise—estimate:

copy-time floor = total bytes in the colocation group
                  ÷ measured end-to-end copy throughput

Actual elapsed time includes indexes, concurrent change catch-up, contention, cutover, retries, and sequential scheduling. A 1-TB shard has no credible duration estimate until the target environment has measured the full path under representative load.

Step 5 — understand the transfer modes

Citus exposes three shard transfer modes:

  • auto requires replica identity when logical replication is available and is the default;
  • force_logical uses logical replication without replica identity, but concurrent UPDATE and DELETE statements can fail;
  • block_writes uses a cross-worker copy while blocking modifications for tables without a primary key or replica identity.

These behaviors are documented for citus_move_shard_placement() and node draining, and all modes are available in Citus Community Edition from version 11.0: Citus utility functions.

For RetailCo’s checkout tables, auto is the recommended production posture. The schema already includes tenant-scoped primary keys. force_logical is not a harmless workaround: a payment or order update failing during movement is an application incident. block_writes is appropriate only when the business can tolerate and bound the write pause for that shard group.

The online path works like this:

flowchart TD
  Source[source shard group serves traffic] --> Copy[initial copy to target]
  Source --> Changes[logical stream captures concurrent changes]
  Copy --> Catchup[target applies accumulated changes]
  Changes --> Catchup
  Catchup --> Lock[brief write lock for cutover]
  Lock --> Metadata[Citus promotes target placement]
  Metadata --> Cleanup[source placement removed after success]

PostgreSQL needs a replica identity to apply UPDATE and DELETE changes to the subscriber. A primary key supplies it by default; a qualifying unique index can be selected explicitly. Citus advises against REPLICA IDENTITY FULL because subscriber-side row lookup can become very expensive: Citus replica identity guidance.

Step 6 — start conservatively and watch the right signals

Start the background job only after saving the approved plan and baseline:

SELECT citus_rebalance_start(
    rebalance_strategy := 'by_disk_size'
);

The function returns immediately. Monitor both the job summary and per-shard progress:

SELECT job_id,
       state,
       description,
       started_at,
       finished_at,
       details
FROM citus_rebalance_status();

SELECT table_name,
       shardid,
       sourcename,
       targetname,
       progress,
       source_shard_size,
       target_shard_size
FROM get_rebalance_progress();

citus_rebalance_status() is the clearer current job view; get_rebalance_progress() exposes per-shard states and source and target sizes: Citus rebalance monitoring.

The default citus.max_background_task_executors_per_node is 1. Raising it permits parallel moves but may also require changes to citus.max_background_task_executors and PostgreSQL max_worker_processes. Shards in the same colocation group still move sequentially, so more executors do not guarantee proportional speedup: Citus parallel rebalancing and Citus configuration.

RetailCo should begin at the default of one task per node. Increase to two only after a canary shows adequate application latency, storage latency, network, WAL, standby lag, and target headroom. Parallelism is a risk budget, not a performance setting to maximize.

Monitor four layers together:

LayerSignals
Applicationcheckout TPS, P50, P95, P99, timeouts, retries, failed orders and payments
Coordinatorrebalance state, distributed activity, worker connections, locks, planning latency
Source and targetCPU, disk read and write latency, throughput, queue depth, free space, autovacuum, checkpoints
Replication and recoveryWAL generation, physical standby lag, logical catch-up, archive health, backup success

The goal is not the fastest possible copy. The goal is the highest movement rate that preserves the application and recovery safety margins.

Step 7 — stop safely, then inspect rather than assume

If an abort threshold is crossed:

SELECT citus_rebalance_stop();

The API documents that this cancels a rebalance in progress. It does not document a cluster-wide undo of moves already completed. Citus also documents that the rebalancer processes placements one by one. The safe operational inference is therefore to inspect current placements and generate a fresh plan after a stop; do not assume the cluster returned to its initial layout.

For an individual citus_move_shard_placement() call, Citus states that failure leaves the source and target unchanged for that move. That guarantee should not be generalized into an all-job rollback: Citus shard movement failure behavior.

After a stop or failure:

  1. capture citus_rebalance_status() and database logs before restarting anything;
  2. verify application errors and locks have recovered;
  3. inspect citus_shards and pg_dist_placement for the actual ownership state;
  4. check source and target disk space, standbys, WAL archives, and logical replication artifacts;
  5. generate a new plan from current state;
  6. resume only after the original trigger is understood.

Do not combine a worker failover, Citus version change, storage resize, and rebalance restart into one recovery action. Preserve enough evidence to identify which operation changed the system.

Step 8 — validate the result, not just the job state

A completed background job is necessary evidence, not acceptance evidence.

Validate placement and size by node:

SELECT nodename,
       count(DISTINCT shardid) AS shard_count,
       pg_size_pretty(sum(shard_size)) AS reported_shard_bytes
FROM citus_shards
WHERE citus_table_type = 'distributed'
GROUP BY nodename
ORDER BY nodename;

Then validate what the rebalancer cannot prove:

  • checkout latency and errors returned to baseline;
  • the new worker receives tenant-local reads and writes;
  • no worker or standby has unsafe storage, WAL, or lag growth;
  • colocated tables remain colocated;
  • the routing and constraint invariants still pass;
  • backups and monitoring include the new topology;
  • the next plan contains no unexpected moves.

Disk balance alone is not success. If one tenant still drives most writes on worker-b, the cluster remains traffic-skewed even when every worker stores the same number of terabytes. Use citus_stat_tenants, application traces, and per-worker database metrics to distinguish storage balance from workload balance.

On Citus 13.2 and later, consider adding a clone instead of copying shards

Everything above assumes the only way to fill a new worker is to copy shard groups into it over the network while the source serves traffic. Citus 13.2 added a second path that changes the economics for exactly the situation in this article: a source worker that is short on storage and cannot comfortably absorb the read amplification of a large outbound copy.

Snapshot-based node addition registers an existing PostgreSQL streaming replica of a busy worker as a clone, promotes it to a full worker, and then splits the shard range between the original and the promotion — instead of copying rows between two live workers:

SELECT citus_add_clone_node('worker-b-replica.internal', 5432,
                            'worker-b.internal', 5432);

The clone is caught up by streaming replication, which RetailCo already operates for worker high availability, rather than by a foreground logical-replication copy competing with checkout. After promotion, Citus splits shards between the source worker and the newly promoted node, so the write-blocking window is bounded by the split rather than by the size of the data.

This does not remove the discipline in this runbook. The clone still needs the storage, monitoring, backup, and standby validation of Step 1; the split is still a data-reorganization operation with abort thresholds; and promoting a clone consumes the standby that was providing that worker’s HA, so the redundancy plan has to say how and when a replacement standby is rebuilt. Verify function signatures and behavior against the installed minor before use — Citus 13.2 release notes.

Choose between the two paths on evidence:

SituationPreferred path
Source has healthy storage and I/O headroomcitus_rebalance_start() — simpler, no standby consumed
Source is storage-constrained or copy would breach latency gatesClone promotion and split, if the Citus line supports it
Cluster is below the supported versioncitus_rebalance_start() with the canary and gate discipline above
Redundancy cannot be reduced even brieflyBuild a fresh standby first, or rebalance conventionally

Draining a worker is the same problem in reverse

Removing a worker requires moving every placement off it first. citus_remove_node() rejects a node that still holds shard placements: Citus node removal.

For one worker, Citus provides citus_drain_node():

SELECT citus_drain_node('worker-c.internal', 5432);

-- Verify no distributed shard placements remain first.
SELECT citus_remove_node('worker-c.internal', 5432);

For several workers, mark them as ineligible and run one drain-only plan so Citus can minimize unnecessary moves:

SELECT citus_set_node_property(
    'worker-c.internal',
    5432,
    'shouldhaveshards',
    false
);

SELECT *
FROM get_rebalance_table_shards_plan(drain_only := true);

SELECT citus_rebalance_start(drain_only := true);

flowchart TD
  Mark[mark worker as not eligible for shards] --> Plan[review drain-only plan]
  Plan --> Move[move colocated shard groups]
  Move --> Verify[verify zero distributed placements]
  Verify --> Remove[remove node from Citus metadata]
  Remove --> Retire[retire standby storage and monitoring]

shouldhaveshards = false prevents normal new placement and tells the rebalancer to move shards away. It does not itself move data: Citus worker node metadata.

In Practice

Context: Citus documents that citus_add_node() registers a worker and copies reference tables, while existing distributed shards remain on their current workers.

Action: Record a before-and-after placement report from citus_shards, then add the target without starting a rebalance.

Result: The test demonstrates the exact separation between provisioning capacity and assigning existing ownership. If checkout throughput changes before shards move, investigate caching, routing, or test noise rather than crediting the empty worker.

Learning: Capacity exists only where the workload’s data lives.

Context: Citus’s default current by_disk_size strategy includes index and colocated-shard bytes but not application traffic.

Action: Review the disk-size plan alongside tenant request count, CPU time, lock waits, and write rate per worker. Flag any planned move that improves bytes while leaving the dominant tenant on the same saturated worker.

Result: The approval separates storage balance from workload balance and identifies hot-tenant isolation or a custom strategy as a different operation.

Learning: A balanced cost function is only as useful as the resource it models.

Context: Citus’s online move uses logical replication and a brief write lock at cutover.

Action: Run a canary movement under a controlled checkout workload. Capture application P95 and P99, database locks, source and target I/O, network, WAL, physical standby lag, logical catch-up, and the cutover interval.

Result: The lab produces an evidence-based maintenance envelope and abort thresholds for the next move. Until raw results exist, no duration or latency claim is approved.

Learning: “Online” is a concurrency capability, not a service-level guarantee.

Proposed rebalance experiment

The lab should compare four states with the same dataset, offered workload, durability settings, worker shapes, and client hosts:

RunCluster actionQuestion
AThree workers, no new nodeWhat is the steady-state baseline?
BAdd fourth worker, no rebalanceDoes empty capacity change anything?
CRebalance with one executor per nodeWhat is the safe movement rate and application impact?
DRebalance with two, then four executors if gates passWhere does parallelism move the bottleneck?

Use a tenant-local checkout script and a deliberately hot-tenant mix. Capture:

  • planned and actual shard-group bytes moved;
  • elapsed time per move and for the whole job;
  • application TPS, P50, P95, P99, timeouts, retries, and errors;
  • coordinator and per-worker CPU, connections, locks, and memory;
  • storage latency, IOPS, throughput, queue depth, and free space;
  • network throughput and packet loss between every source and target;
  • WAL rate, archive delay, physical standby lag, and logical catch-up;
  • autovacuum and checkpoint activity;
  • final bytes, shard groups, tenant traffic, and CPU per worker.

Inject failures one at a time:

FailureRequired evidence
Target PostgreSQL stops during copyJob state, source integrity, target cleanup, replan procedure
Source PostgreSQL failsAffected tenants, worker promotion, movement state, restart decision
Coordinator restartsBackground-job state, application behavior, operator recovery steps
Target disk approaches abort thresholdStop latency, remaining temporary state, free-space recovery
Standby lag crosses limitWhether stopping the rebalance restores lag before WAL safety is threatened
Checkout P99 crosses limitWhether movement stops and latency returns to baseline

Do not run every fault in one experiment. A combined failure may be useful later, but first isolate the behavior and recovery contract of each component.

Where It Breaks

ConditionWhy rebalancing is unsafe or insufficientRecommended action
Source is already at emergency capacityCopy work competes with the traffic that created the emergencyAdd capacity earlier, shed noncritical work, or migrate a bounded canary
Target has little temporary headroomCopy, indexes, WAL, and normal growth can exhaust storageResize before activation and enforce a free-space abort threshold
Tables lack replica identityOnline UPDATE and DELETE replication cannot identify rows safelyAdd a supported key or schedule bounded blocking mode
Shard groups are enormousOne movement unit takes too long and concentrates failure riskRedesign shard count before the emergency; measure real transfer time
Reference tables are largeNode activation or first shard placement carries extra copy costMeasure separately; defer only with an explicit later-copy plan
Physical standbys already lagRebalance-related work reduces recovery marginRestore replication health before moving data
Equal disk hides hot trafficby_disk_size solves the wrong imbalanceIsolate a tenant, use measured custom cost, subdivide ownership, or decompose
One tenant exceeds one workerMoving the tenant changes location, not its write ceilingSplit the tenant by another stable key or give it a separate architecture
Peak traffic is activeApplication and movement contend when the least headroom existsRebalance before peak and retain an abort window
Stop and resume are untestedOperators may mistake cancellation for rollbackRehearse stop, inspect, replan, and resume in the lab

RetailCo should not start the production job if any publication or change ticket still says “nonblocking, so there is no impact.” That wording hides the very measurements the operation requires.

It should also reject a plan that moves the largest shard first merely because it delivers the most bytes to the new worker. A better canary is a representative colocation group large enough to exercise the full logical-replication and cutover path but small enough to keep the first abort recoverable.

What to Do Next

For this RetailCo scenario, the recommendation is to add worker-d, validate it completely, use the cluster’s verified by_disk_size strategy, and begin with the default one background task executor per node. The first movement should be a representative canary colocation group outside peak traffic. Parallelism can increase only after application latency, source and target storage, network, WAL, standby lag, and recovery gates remain healthy.

The conclusion is deliberately conditional:

Good fit: many movable tenant-local shard groups, adequate source and target headroom, supported replica identity, healthy standbys, and operators who can stop and replan.

Poor fit: emergency-low storage, giant movement units, missing replica identity, lagging recovery systems, or one hot tenant that already exceeds one worker.

Main advantage: Citus can transfer colocated PostgreSQL ownership while most application reads and writes continue.

Main limitation: the rebalancer moves the units the schema created; it cannot make an oversized shard or indivisible hot tenant smaller.

Operational complexity: high. Rebalancing is a distributed PostgreSQL migration involving application SLOs, metadata, direct node connections, logical replication, physical standbys, storage, WAL, network, and recovery.

Migration complexity: proportional to real colocation-group size and change rate, not the number of SQL commands.

  • Problem: a fourth worker adds empty capacity while RetailCo’s existing shards and write pressure remain on three workers.
  • Solution: inspect the full colocation movement unit, review the active strategy and plan, canary at one executor per node, monitor the application and recovery path, and replan from actual state after any stop.
  • Proof: Citus documentation shows that adding a node does not move old shards, online rebalance uses logical replication plus a brief cutover lock, citus_rebalance_status() exposes the background job, and the current default disk strategy balances bytes rather than traffic. The proposed lab must measure the scenario-specific safety envelope.
  • Action: build the before-state placement report and abort-threshold sheet before calling citus_rebalance_start(). If the cluster lacks enough headroom to survive one representative shard-group move, do not use production as the benchmark environment.