Conversation

This PR implements a new outgoing federation queue. The end goal is to create a queue that can scale to reddit scale (that is 100-1000 activities per second, sent to each federated instance).

The basic idea is to change the primary division of the federation queue to be of the target instance. Federation to each instance is mostly handled separately.

The queue works as follows:

The main lemmy_server process with its send_lemmy_activity function only stores the sent_activity in the db (like currently), with an addition of the send targets.
There is a new table federation_queue_state (domain, last_successful_id, retries) that tracks the state of outgoing federation per instance
One or more lemmy_federate processes pick up the activities from the db and send them out. lemmy_federate works as follows:
1. All known allow/non-blocklisted instances are read from the database every 60s
2. A worker tokio task is started / stopped per federated instance. It does the following
  1. It fetches a list of local communities that at least one person on the remote instance is subscribe to (refreshed every 10-60 seconds)
  2. It loops through the activities table starting from last_successful_id up until the max(id). For every activity:
    1. It checks if this activity should be sent to this instance using (a) the list of communities from above and (b) a list of other inboxes stored in the activities tab
    2. It sends out the activity to the correct shared or individual inbox. If there's a retryable failure, it waits with exponential back off and tries again. I had to expose a raw non-retrying send_activity function from the activitypub-federation crate since the state of the retry has to be in the db. The activity_queue from there is not used at all.
  3. The updated federation_queue_state is stored to the database.
3. A separate task logs the current progress of each domain once per minute. Example output:
```
 2023-07-12T21:43:15.642275Z  INFO lemmy_federate::worker: https://lemmy.phiresky.xyz/activities/announce/c06d2a17-4df9-43d3-b142-7a596b310d74 to https://toad.work/inbox failed: Queueing activity for retry after failure with status 502 Bad Gateway: <html><head><title>502 Bad Gateway</title></head>...
2023-07-12T21:43:15.016121Z  INFO lemmy_federate: Federation state as of 2023-07-12T21:43:15+00:00:
2023-07-12T21:43:15.024684Z  INFO lemmy_federate: lemmy.serverfail.party: Ok. 6 behind
2023-07-12T21:43:15.025516Z  INFO lemmy_federate: toad.work: Warning. 3437 behind, 5 consecutive fails, current retry delay 320.00s
2023-07-12T21:43:15.016220Z  INFO lemmy_federate: lemmy.byteunion.com: Ok. 0 behind
2023-07-12T21:43:15.016228Z  INFO lemmy_federate: swiss-talk.net: Ok. 0 behind
...
```
4. If a signal is received (ctrl+c, SIGINT, SIGTERM), all the workers are gracefully stopped, their most current state stored in the db.

This implementation has the following advantages:

The queue is persistent and reliable - if any or all lemmy processes are stopped, the federation will continue without activity loss afterwards. If any process is killed or crashes, worst case is 100 activities or 10s of activity is resent. No losses.
Low memory usage: The memory usage scales linearily with the number of instances but is constant for each instance and independent of the number of activities per second. It is also mostly uncoupled from the reliability of the other servers. No more unbounded memory use.
The queue is performant - Each domain has a separate serialized queue, which means there is only ever a single request waiting for a server response per federated instance. Performance of each instance doesn't affect the others.
Horizontally scalable - this outgoing federation can run on multiple separate processes (split by outgoing domain) or multiple servers - they just need access to the same PG database.
No more DOSing other instances. Activities are sent as fast as the receiving end can take them or as fast as we can send them, whichever is lower.
If a remote instance goes down, activities will be replayed reliably and in order from the time it goes down.
It's easier to find out what the state of federation is and where bottlenecks are.

It has the following disadvantages:

It assumes that every inbox in the same instance/domain has the same reliability. If the instance fails to respond to one activity, it will not receive any other activities either. (failure is decided the same as before, http 2xx and 4xx count as success, everything else as retryable). This is not relevant to lemmy instances since they use shared inbox for everything.
It's optimized for large instances and has more overhead if there's very little activity.
- Also I inverted the logic for figuring out which remote instance care about which local community. Instead of getting the remote inboxes for one activity every remote inbox has a set of communities it cares about. The reason is that this exact thing is the most expensive federation activity type so optimizing it is important. The inversion optimizes for most instances caring about many communities. This is not intrinsic to the general approach though and could be changed.
There's a time delay of up to 10 seconds for outgoing activities and follower changes and up to 60 seconds for instance blacklist changes.
Right now, every http request waits for the response to arrive before the next one is sent (per instance). this could be changed by adding a FuturesOrdered or similar limited to N=e.g. 10 concurrent in-flight requests and adding more complicated retry / backoff logic but right now I don't think it's necessary.
There's overhead in the way I load every activity for every remote instance. Caching is used to make sure this doesn't really cause redundant DB queries, just CPU work. This is a tradeoff to allow for different remote instances to be in different positions in time.

The approach of one worker per remote instance should scale to reddit scale imo (~100 - 1000 activities per second). The details will of course need tweaking in the future when bottlenecks become clearer.

I've tested this so far only with my own very low activity instance and the basics work as expected.

Here's an example of how the federation_queue_state table looks:

domain	last_successful_id	fail_count	last_retry
toad.work	6832351	14	2023-07-12 21:42:22.642379+00
lemmy.deltaa.xyz	6837196	0	1970-01-01 00:00:00+00
battleangels.net	6837196	0	1970-01-01 00:00:00+00
social.fbxl.net	6837196	0	1970-01-01 00:00:00+00
mastodon.coloradocrest.net	6837196	0	1970-01-01 00:00:00+00

And here's an example of how the activity table looks (for sendable activities):

id            | 6817007
data          | {"cc": ["https://lemmy.phiresky.xyz/c/localtest/followers"], "id": "https://lemmy.phiresky.xyz/activities/announce/c06d2a17-4df9-43d3-b142-7a596b310d74", "to": ["https://www.w3.org/ns/activitystreams#Public"], "type": "Announce", ...
local         | t
published     | 2023-07-12 21:36:31.749541
updated       |
ap_id         | https://lemmy.phiresky.xyz/activities/announce/c06d2a17-4df9-43d3-b142-7a596b310d74
sensitive     | f
send_targets  | {"all_instances": false, "inboxes": [], "community_followers_of": [54]}
actor_type    | community
actor_apub_id | https://lemmy.phiresky.xyz/c/localtest

Merged

Closed

4 tasks

		about = "A link aggregator for the fediverse",
		long_about = "A link aggregator for the fediverse.\n\nThis is the Lemmy backend API server. This will connect to a PostgreSQL database, run any pending migrations and start accepting API requests."
		)]
		pub struct CmdArgs {

		fetcher: () => Promise<T>,
		checker: (t: T) => boolean,
		retries = 10,
		delaySeconds = 2,

		federation_queue_state (id) {
		id -> Int4,
		#[max_length = 255]
		domain -> Varchar,

Merged

Closed

4 tasks

Closed

4 tasks

Closed

Merged

Labels

None yet