Worker Sharding: The Right Way to Split Work

First, stop: sharding, or a message broker?

A message broker: every worker pulls from the same queue, whoever grabs a job first owns it (one RabbitMQ queue + N consumers, Redis BRPOP, a standard SQS queue). Kafka is not on that list: the moment you choose a partition key you are already sharding. In most cases a broker is enough, and sharding gets written for nothing:

The key sentence: the real purpose of sharding is not speed, it is removing collisions. With locks everyone runs to the same door and waits in line; with sharding they were already entering through different doors.

The starting point: modulo (everyone starts here, nobody should stay)

Divide the key's hash by the worker count; the remainder (7 % 4 = 3) says which worker it goes to:

shard = hash(user_id) % WORKER_COUNT

# worker 2 only processes (4 workers, a power of two -> & 3):
SELECT * FROM jobs
 WHERE status = 'pending'
   AND (hashtext(user_id::text) & 3) = 2   -- full scan every time; see below
 ORDER BY created
 LIMIT 100;

-- the shard is computed at write time, stored in a column, indexed
ALTER TABLE jobs ADD COLUMN shard smallint NOT NULL;      -- app side: murmur3(user_id) & 63
CREATE INDEX jobs_pending ON jobs (shard, created) WHERE status = 'pending';

SELECT * FROM jobs WHERE status = 'pending' AND shard IN (16,17,...,31)   -- my shards
 ORDER BY created LIMIT 100 FOR UPDATE SKIP LOCKED;
-- note: SKIP LOCKED + LIMIT can return fewer rows than asked (locked rows are dropped after the limit)

Modulo's flaw is that the key → worker mapping depends on the worker count. Both methods below cut that dependency, from different directions.

Method 1: Static Partitioning (fixed shards)

You split the work not by the worker count but by a fixed number of shards decided in advance: 64, 128 or 256. There are two separate mappings, and keeping them apart is the whole point:

• Key → Shard: murmur3(user_id) & 63. This mapping is fixed for the lifetime of the system; it never changes — which is why the hash must be stable too (the box above).
• Shard → Worker: the 64 shards are shared out among the workers that are up right now; with 4 workers, 16 each. This is the only thing that changes.
• How the distribution is done: no ring needed. A coordinator sits behind it — etcd, Consul, ZooKeeper, or a simple lease table — and when a worker comes up it checks out the unowned shards from there.
• Where it is used: Kafka (this is its partition logic — with one catch: increasing the partition count breaks the key → partition mapping, and decreasing it is impossible, so in Kafka you must choose the partition count generously on day one), Redis Cluster (16384 hash slots), database sharding.

SHARD_COUNT = 64          # never changes
shard = hash(user_id) & 63

# with 4 workers:
worker 0 -> shards 0..15
worker 1 -> shards 16..31
worker 2 -> shards 32..47
worker 3 -> shards 48..63

# when a 5th worker is added the shards are re-shared,
# but a key's SHARD does not change. Only its owner does.

Because key → shard stays fixed, per-shard state, cache and progress become portable. One rule for choosing the shard count: several times the worker count you can imagine reaching — because the shard count is the ceiling on parallelism: with 64 shards and 100 workers, 36 sit idle. And "never changes" has one escape hatch: if you chose a power of two, 64 → 128 is harmless; a key's new shard is either its old one or old+64, so the data splits at most once and predictably.

Who hands a shard to a worker: the coordinator

"Who owns shard 17 right now?" must be answered by one place; if every worker computes it on its own, two workers will believe they own the same shard during a transition. The simplest coordinator is the database you already have. The hard part of the design is not claiming but each worker knowing its share: so the first worker to start does not grab all 64, count the live workers first, then the share.

CREATE TABLE shard_ownership (
  shard          int PRIMARY KEY,   -- 0..63
  owner          text,
  lease_expires  timestamptz,
  claim_version  bigint NOT NULL DEFAULT 0   -- fencing token, see below
);

-- 1) work out my share: live workers = those with a heartbeat in the last 30 s
SELECT ceil(64.0 / count(DISTINCT owner)) AS share
  FROM shard_ownership WHERE lease_expires > clock_timestamp();   -- + myself
-- clock_timestamp(): now() returns the transaction start time; in a long transaction you misread the lease

-- 2) if I hold fewer than my share: grab unowned shards or ones whose lease expired
UPDATE shard_ownership
   SET owner = :me,
       lease_expires = now() + interval '30 seconds',
       claim_version = claim_version + 1
 WHERE shard IN (
       SELECT shard FROM shard_ownership
        WHERE owner IS NULL OR lease_expires < now()
        LIMIT :share - :held
        FOR UPDATE SKIP LOCKED           -- two workers never wait on the same row
 )
RETURNING shard, claim_version;          -- each shard has ITS OWN version

-- 3) every 10 s: renew the lease on the (shard, version) pairs I hold (heartbeat)
UPDATE shard_ownership SET lease_expires = clock_timestamp() + interval '30 seconds'
 WHERE owner = :me
   AND (shard, claim_version) IN ((17, 7), (18, 7), (31, 9))    -- what I hold
RETURNING shard;
-- sent 16, got 12 back -> I LOST 4 shards: drop their jobs IMMEDIATELY (self-fencing)
-- same if I cannot reach the DB at all: a worker that cannot renew its lease stops working

-- 4) if I hold more than my share: release the excess (a new worker has arrived)

-- on the jobs table: fence bigint NOT NULL DEFAULT 0
UPDATE jobs
   SET status = 'done', result = :result, fence = :my_version
 WHERE id = :id
   AND fence <= :my_version;   -- A arrives with v7, B has already written v8 -> 0 rows, ROLL BACK

If you need no ordering guarantee, skip the shard layer entirely: one table, FOR UPDATE SKIP LOCKED, every worker grabs jobs one at a time, there is no such thing as a rebalance, and autoscaling comes for free. Retry counts, backoff and dead-letter queues belong there too; that is a separate post.

Method 2: Ring Hashing (Consistent Hashing)

There is no "fixed 64 shards" layer in between. You place the keys and the workers on the same virtual ring (a circle from 0 to 2^32):

• Key → Worker: hash(user_id) lands on a point of the ring; walk clockwise and the first worker you meet takes the job. Nothing is read from a record; it is computed.
• It is a dynamic design: when the worker count goes from 4 to 5, a new point is added to the ring; only the keys next to that point (≈ 1 / new worker count) move to the new worker. There is no logical shard layer and no coordinator to manage.
• Where it is used: Memcached, DynamoDB, Envoy/Nginx load balancers, streaming and WebSocket node distribution.

Ring (0 .. 2^32):

   worker-A ●          ● worker-B
       ↘  keys 7,12,31  ↙
            ●
        worker-C

# When worker-D is added:
# Only the keys of D's NEIGHBOUR on the ring move.
# Everything else stays untouched.

owner(key) = the first worker clockwise from hash(key)

# Single-point simplification. In reality each worker sits at 100-200 virtual points;
# when D is added it enters the ring in 150 places and takes keys from EVERY worker,
# proportionally. The total is still ~1/N, but "only one neighbour is affected"
# is true only in this drawing.

A practical detail: place each worker on the ring at 100–200 virtual points, not one; with a single point the distribution is left to chance and one worker can get three times another's share. Most libraries use virtual nodes by default.

Why it matters: the reconnect storm

Rebalancing: hand over gradually

The ring reduces how many keys move but does not make it zero. Hand the movers over one piece at a time, not all at once — exactly the same for static partitioning:

for (piece : piecesToMove) {          // a shard or a key range
    oldOwner.stop(piece);        // stop taking new work, finish what you hold
    oldOwner.confirmDone(piece); // ← wait for this
    newOwner.start(piece);
    Thread.sleep(200);           // ← breathe before the next piece
}

Which one: static partitioning or ring hashing?

Do not mix them — we did for a while: we had built a ring on top of fixed shards. If you have a shard layer, you do not need the ring; distributing 64 shards through a table is both simpler and visible. The ring is the answer for systems without a shard layer. In production we stayed with static partitioning.

From the field: one customer, four workers, a hash that did nothing

A notification system was split by customer_id % 4. One enterprise customer alone produced about 40% of the daily notifications. Result: the worker that customer hashed to was permanently at 100% CPU while the other three sat idle. We raised the worker count to 8 — nothing changed, because that customer still landed on exactly one worker.

-- before
shard = hash(customer_id) & 63

-- after: spread across all shards
shard = hash(customer_id || ':' || notification_id) & 63

-- middle ground (salting): spread across 8 shards, keep partial locality
shard = hash(customer_id || ':' || (notification_id % 8)) & 63

In a system where order did matter we saw the reverse: order statuses (preparing → shipped → delivered) split across two workers, and some orders went back from "delivered" to "shipped". We sharded by order id and made the database protect itself as well: UPDATE ... WHERE status_rank < :new_rank — a late old message cannot write. The same idea as the fencing token: entrust correctness to order, not to timing.

Four things worth measuring

That third row matters: do not look at the average. With three of four workers idle, average latency looks great; the customer on the congested shard gets their notifications half an hour later.

Checklist

Conclusion

There are two methods and the choice is simple: if you can build a shard layer, static partitioning plus a table; if you cannot, the ring. Back to the opening picture: the reason those three workers were idle was not capacity, it was the choice of key — and the right key is always chosen by looking at the real distribution of the data, never by guessing.