Shuffle-shard everything

I’ve administered a lot of system design interviews over my career, and I’ve never had anyone use shuffle-sharding.

I think this is a real shame, because shuffle-sharding is one of the vanishingly few free lunches available in distributed systems. Almost any multitenant system with contention can use shuffle-sharding to reduce the blast radius from noisy neighbors.

Shuffle-sharding 101

Consider a basic system where all tenants share a single worker:

If a few tenants hit the system hard enough to cause service disruptions, then it will impact every other tenant consuming the service because they’re all sharing a single worker. Admission control techniques like rate limiting help mitigate this risk, but even so all tenants are still equally impacted by availability problems with our single worker.

If we spend a bit of money, we can reduce the blast radius by partitioning tenants across different workers, like so:

An incident impacting a single shard now only impacts one third of the system’s tenants on average. This is a nice improvement, but it’s still not ideal that a poorly-behaved tenant in a single shard could cause issues for 33% of our customers!

Adding more shards to spread tenants across continues to yield linear blast radius reductions, provided you are willing to pay for more hardware. But there are some pretty steep downsides to this:

1. Past a certain point, adding more shards becomes cost-prohibitive. You might not be able to afford to push the blast radius down to <1% by simply throwing more hardware at the problem!

2. Spreading tenants more thinly across shards makes it harder to fully utilize your infrastructure, which makes the system even less cost-efficient.

The fundamental problem at play is that within each shard, tenants share all of their workers with each other—just like in the case with a single monolithic worker. Shuffle-sharding is how you can reduce the overlap between tenants without necessarily having to go out and buy more nodes.

The key idea is to assign tenants to individual workers, rather than assigning them to predefined groups of workers. Here are our same four tenants shuffle-sharded across six instances:

Here are the worker assignments in tabular form. It’s valid for two tenants to share a pair of workers (and unavoidable at scale), but in this example no two tenants happen to share the same pair of workers:

Suppose tenant B misbehaves and causes an incident with all of its assigned workers. Previously this would have caused a total outage for tenant A, but under this shuffle-sharding assignment tenant A still has one worker available to serve its traffic. Provided one worker is enough to serve tenant A’s traffic at this point in time, tenant A can continue using the system.

The probability of two given tenants being assigned the same pair of workers is only 1 / C(6, 2) = 6.67%¹. This is about five times better than the initial sharded design, without needing to buy additional hardware!

The tradeoff made here is the probability that two tenants have any overlapping worker assignments is significantly higher. At scale, this approach of selecting two workers from a pool of six for each tenant results in a 1 - C(4, 2) / C(6, 2) = 60% chance that two arbitrary tenants will have some overlapping worker assignments, which is higher than the 33% chance we had under the initial sharded design.

The intuition is simple: any given pool size contains far more overlapping subsets than it does disjoint groups. Our pool of six workers contains fifteen distinct pairs, but only three disjoint pairs.

Provided that tenants can handle partial degradation of their assigned worker nodes, shuffle-sharding significantly reduces blast radius with no incremental hardware cost compared to a “normal” sharding strategy. Assigning three workers to each tenant from a pool of ten means two tenants have only a 1-in-120 chance of sharing the same set of workers, a result which ordinarily would have required 120 shards to achieve.

Most discussions on shuffle-sharding end there, but the technique is actually far more general and you can use it in almost any situation where you have contention for a shared resource. Let’s look at some examples.

Advanced shuffle-sharding

Rate limits

At Rye, we leveraged browser use to order from long-tail merchants and preferred direct API integrations with major retailers where they were available. For Amazon, we integrated with the Amazon Business Ordering API.

Out of the box, this API has a 5 requests/second rate limit which we needed to share across all of our tenants. Internally it’s easy to avoid going beyond Amazon’s rate limits by using a token bucket shared across all tenants, but this doesn’t prevent individual tenants from consuming all of the upstream request budget.

Per-tenant rate limits at the API layer help, but sizing them can be awkward. In our use case we didn’t want tenants to have to think about which retailer they were ordering from, so Amazon orders were intermingled with other retailers. Maintaining per-tenant buckets deeper in the call stack also works, but requires a lot of extra state—especially when we had many direct retailer integrations.

One way to improve the situation is to apply shuffle-sharding! Instead of a single token bucket refilling at 5 requests/second, we can define five token buckets which each refill at 1 request/second and then assign tenants in the system to a subset of two token buckets.

When we need to make a request to Amazon on behalf of a tenant, a token is consumed from one of their assigned buckets. Noisy tenants can exhaust their assigned buckets, but cannot exhaust the buckets they haven’t been assigned to. Each tenant can sustain at most 2 requests/second, but we only need to track five shared token buckets instead of a separate bucket for every tenant.

Recursive shuffle-sharding

A common system design question is to build a webhook service responsible for delivering webhook events to tenants. A typical minimal design involves pushing webhooks to a queue for delivery later on by a serverless function:

But if one tenant produces a lot of webhooks and backs up the queue, then delivery of webhooks to other tenants can end up delayed. Ideally you’d use fair queuing to avoid this problem, but this isn’t always easy to do. If you’re deployed to GCP, for instance, then neither Cloud Tasks nor Pub/Sub support fair queuing out of the box.

Shuffle-sharding can help here. You introduce additional queues, assign tenants to a subset of queues, enforce a concurrency limit on each queue, and then when a webhook comes in you add it to one of the tenant’s assigned queues randomly:

This does, however, introduce undesirable behavior around webhook ordering. If two webhooks for the same resource end up being added to different queues then it’s possible for the webhooks to get delivered out of order!

We can preserve ordering of a resource’s events by leveraging recursive shuffle-sharding. The idea is to apply multiple levels of sharding, for example:

1. The tenant is initially assigned a subset of queues.

2. Individual webhooks are added to a queue from the subset in (1) based on a consistent hashing strategy, e.g. using the ID of the resource the webhook is for.

This way all webhooks for resource sub_3F06NpJ8Y will end up in the same queue for the same tenant, where an ordered consumer can process them sequentially.

Recursive shuffle-sharding can be used in many other circumstances. At Rye, our API tenants themselves serve many users. In theory, we could allow our tenants to pass along their own user’s ID and use that to recursively shuffle shard traffic coming from each consumer, so that the impact of a noisy user within one of our tenant’s applications is contained!