Online dynamic scaling of shards for federation sender and event persistence

This issue has been migrated from [#12604](https://github.com/matrix-org/synapse/issues/12604).

---

Currently if you want to add or remove federation senders or event persisters you need to take down the entire service¹, change the config, and bring everything back online. This makes it hard to scale up the service when under load (as restarting things is costly due to all the caches being dropped), and makes it impossible to dynamically scale these workers.

## Current algorithm

Currently, the way we balance work between the different workers is to *shard*. For federation senders we shard by remote destination, and for event persisters by room ID. Each process knows which process is responsible for a given shard by looking at the relevant config.

The way that federation senders and event persisters actually operate have some important differences:

1. Each federation senders works through the event stream, handling each event one by one and recording its position in the DB. On start up the process reads from the DB where it got up to and continues from there. No other worker cares how the federation senders divvy up the work.
2. Event persisters are send events to persist over HTTP replication from other workers, so all other workers that can persist events care which event persister to talk to for any given room. The event persisters are otherwise stateless, in the sense that they don't persist any state to the DB across restarts.

When adding or removing federation senders or event persisters a rebalancing operation has to happen. For event persisters, this is simply that all workers need to be restarted to pick up the new config. For federation senders it is more complicated, as each shard persists state across restarts. In that case, the position in the events stream of all the federation sender workers is set to the *minimum* stream position across all sender workers (this works as processing the same event twice is handled correctly by the federation senders, and by taking the minimum we ensure that all events are correctly processed by all federation senders).

## Desired new features

* Able to dynamically spin up and down event persisters and federation senders without restarting any worker.
* Rebalancing after such a change does causes a minimal change in what each shard is handling, i.e. for event persisters the set of rooms a given shard is handling doesn't *completely* change after a rebalance (for cache coherence purposes).

## Potential new algorithm

Instead of splitting up the space of room IDs/hosts into N shards for N workers, we instead split it up into e.g. 1000 buckets², with the buckets assigned uniformly to each of the N workers. This makes rebalancing easy to do with minimal changes, as each bucket can be reassigned independently without moving other buckets. For example, if we add a new worker it would take a fraction of the buckets from each existing worker.

How we otherwise handle the rebalancing depends on the worker type.

#### Event persisters

The important thing here is to ensure that two different event persisters are *not* handling the same room concurrently, as each worker assumes that it is the one solely persisting to the room. Therefore, when a bucket moves from one worker to another, the new worker has to wait until it knows the other worker has finished processing all events in its queue for that room.

This can be achieved roughly by:

1. Which worker to send events to events to is stored in Redis.
2. When an event persister receives an event it takes a lock out in the DB, if a lock already exists then it waits.
3. When a bucket is moved between workers the mapping of bucket to worker in Redis is immediately updated. This is safe as the lock in the previous step ensures that the new worker only starts processing events once the previous worker has finished.

A rebalance is done by:

1. **Adding** an event persister: On startup the new worker:
1. Takes out a lock (to ensure only one rebalance happens at a time)
2. Gets the number of existing workers and bucket assignments, and calculates which buckets from each worker it will assign itself to
3. Update the bucket to worker map on redis
4. (Optionally?) Send out the updates over replication so all workers know about the new assignments
2. **Removing** an event persister: A worker needs to (and tbd which one):
1. Takes out a lock (to ensure only one rebalance happens at a time)
2. Gets the assignments for the worker that is being removed
3. (Optionally?) Send out the updates over replication so all workers know about the new assignments
4. Update the bucket to worker map on redis

#### Federation senders

The important thing is that all events are processed by federation senders whose set of buckets sum up to the full set of all buckets (otherwise we may fail to send an event out to a remote destination).

This can be done by:

1. Changing the stream position table in the DB to map from *bucket* to stream position rather than from *worker* to stream position. The table also tracks which worker is assigned to which bucket.
2. Every time a server finishes processing an event it updates the stream position for all of its currently assigned buckets (exactly how this is done in an efficient manner with large numbers of buckets is TBD).
3. Whenever a federation sender is handling stuff for a remote host (i.e. in the process of sending a transaction), it takes out a lock for that host in the DB.

A rebalance is done by:
1. **Addng** a federation sender: On startup the new worker:
1. Takes out a lock (to ensure only one rebalance happens at a time)
2. Gets the number of existing workers and bucket assignments, and calculates which buckets from each worker it will assign itself to
3. Fetches the current position for the new workers and starts processing.
4. Assigns those buckets to itself in the DB.
5. Send out the updates over replication so all workers know about the new assignments. Other workers stop handling hosts in buckets that the new worker has taken ownership of.
6. The new worker continues normal start up (e.g. calls `_wake_destinations_needing_catchup` to handle destinations with pending outbound stuff).
2. **Removing** a federation sender: A worker needs to (and tbd which one):
1. Takes out a lock (to ensure only one rebalance happens at a time)
2. Gets the assignments for the worker that is being removed
3. Assigns those workers to other federation senders
4. Send out the updates over replication so all workers know about the new assignments. Federation senders with new assignments may need to rewind their positions if the newly assigned buckets have older stream positions.
5. Checks if any of the hosts in the newly assigned buckets have pending data to be sent (i.e. same as calling `_wake_destinations_needing_catchup` on startup)

---
¹ Technically it's possible to only take down *some* of the workers when doing this, but the point still stands
² We want to choose a sufficiently large number that even if we spin up many workers the variance of the number of buckets assigned to each worker is small.

No implementation file or test is named in the issue. Start by locating the federation sender and event persister sharding, worker-assignment, and stream-position code, including `_wake_destinations_needing_catchup`; then trace how replication and Redis mappings are used. Done means workers can be added or removed without restarting, with safe rebalancing and complete event processing.

Written by the indexing model from the issue text.