federation: parallel sending per instance

Cargo.toml

@@ -139,7 +139,7 @@ anyhow = { version = "1.0.81", features = [
 diesel_ltree = "0.3.1"
 typed-builder = "0.18.1"
 serial_test = "2.0.0"
-tokio = { version = "1.36.0", features = ["full"] }
+tokio = { version = "1.37.0", features = ["full"] }
 regex = "1.10.3"
 once_cell = "1.19.0"
 diesel-derive-newtype = "2.1.0"

config/defaults.hjson

@@ -106,10 +106,11 @@
   port: 8536
   # Whether the site is available over TLS. Needs to be true for federation to work.
   tls_enabled: true
-  # The number of activitypub federation workers that can be in-flight concurrently
-  worker_count: 0
-  # The number of activitypub federation retry workers that can be in-flight concurrently
-  retry_count: 0
+  federation: {
+    # Limit to the number of concurrent outgoing federation requests per target instance.
+    # Set this to a higher value than 1 (e.g. 6) only if you have a huge instance (>10 activities per second) and if a receiving instance is not keeping up.
+    concurrent_sends_per_instance: 1
+  }
   prometheus: {
     bind: "127.0.0.1"
     port: 10002

crates/federate/src/inboxes.rs

@@ -0,0 +1,149 @@
+use crate::util::LEMMY_TEST_FAST_FEDERATION;
+use anyhow::Result;
+use chrono::{DateTime, TimeZone, Utc};
+use lemmy_db_schema::{
+  newtypes::{CommunityId, InstanceId},
+  source::{activity::SentActivity, site::Site},
+  utils::{ActualDbPool, DbPool},
+};
+use lemmy_db_views_actor::structs::CommunityFollowerView;
+use once_cell::sync::Lazy;
+use reqwest::Url;
+use std::collections::{HashMap, HashSet};
+
+/// interval with which new additions to community_followers are queried.
+///
+/// The first time some user on an instance follows a specific remote community (or, more precisely: the first time a (followed_community_id, follower_inbox_url) tuple appears),
+/// this delay limits the maximum time until the follow actually results in activities from that community id being sent to that inbox url.
+/// This delay currently needs to not be too small because the DB load is currently fairly high because of the current structure of storing inboxes for every person, not having a separate list of shared_inboxes, and the architecture of having every instance queue be fully separate.
+/// (see https://github.com/LemmyNet/lemmy/issues/3958)
+static FOLLOW_ADDITIONS_RECHECK_DELAY: Lazy<chrono::TimeDelta> = Lazy::new(|| {
+  if *LEMMY_TEST_FAST_FEDERATION {
+    chrono::TimeDelta::try_seconds(1).expect("TimeDelta out of bounds")
+  } else {
+    chrono::TimeDelta::try_minutes(2).expect("TimeDelta out of bounds")
+  }
+});
+/// The same as FOLLOW_ADDITIONS_RECHECK_DELAY, but triggering when the last person on an instance unfollows a specific remote community.
+/// This is expected to happen pretty rarely and updating it in a timely manner is not too important.
+static FOLLOW_REMOVALS_RECHECK_DELAY: Lazy<chrono::TimeDelta> =
+  Lazy::new(|| chrono::TimeDelta::try_hours(1).expect("TimeDelta out of bounds"));
+
+pub(crate) struct CommunityInboxCollector {
+  // load site lazily because if an instance is first seen due to being on allowlist,
+  // the corresponding row in `site` may not exist yet since that is only added once
+  // `fetch_instance_actor_for_object` is called.
+  // (this should be unlikely to be relevant outside of the federation tests)
+  site_loaded: bool,
+  site: Option<Site>,
+  followed_communities: HashMap<CommunityId, HashSet<Url>>,
+  last_full_communities_fetch: DateTime<Utc>,
+  last_incremental_communities_fetch: DateTime<Utc>,
+  instance_id: InstanceId,
+  domain: String,
+  pool: ActualDbPool,
+}
+impl CommunityInboxCollector {
+  pub fn new(
+    pool: ActualDbPool,
+    instance_id: InstanceId,
+    domain: String,
+  ) -> CommunityInboxCollector {
+    CommunityInboxCollector {
+      pool,
+      site_loaded: false,
+      site: None,
+      followed_communities: HashMap::new(),
+      last_full_communities_fetch: Utc.timestamp_nanos(0),
+      last_incremental_communities_fetch: Utc.timestamp_nanos(0),
+      instance_id,
+      domain,
+    }
+  }
+  /// get inbox urls of sending the given activity to the given instance
+  /// most often this will return 0 values (if instance doesn't care about the activity)
+  /// or 1 value (the shared inbox)
+  /// > 1 values only happens for non-lemmy software
+  pub async fn get_inbox_urls(&mut self, activity: &SentActivity) -> Result<Vec<Url>> {
+    let mut inbox_urls: HashSet<Url> = HashSet::new();
+
+    if activity.send_all_instances {
+      if !self.site_loaded {
+        self.site = Site::read_from_instance_id(&mut self.pool(), self.instance_id).await?;
+        self.site_loaded = true;
+      }
+      if let Some(site) = &self.site {
+        // Nutomic: Most non-lemmy software wont have a site row. That means it cant handle these activities. So handling it like this is fine.
+        inbox_urls.insert(site.inbox_url.inner().clone());
+      }
+    }
+    if let Some(t) = &activity.send_community_followers_of {
+      if let Some(urls) = self.followed_communities.get(t) {
+        inbox_urls.extend(urls.iter().cloned());
+      }
+    }
+    inbox_urls.extend(
+      activity
+        .send_inboxes
+        .iter()
+        .filter_map(std::option::Option::as_ref)
+        .filter(|&u| (u.domain() == Some(&self.domain)))
+        .map(|u| u.inner().clone()),
+    );
+    Ok(inbox_urls.into_iter().collect())
+  }
+
+  pub async fn update_communities(&mut self) -> Result<()> {
+    if (Utc::now() - self.last_full_communities_fetch) > *FOLLOW_REMOVALS_RECHECK_DELAY {
+      tracing::debug!("{}: fetching full list of communities", self.domain);
+      // process removals every hour
+      (self.followed_communities, self.last_full_communities_fetch) = self
+        .get_communities(self.instance_id, Utc.timestamp_nanos(0))
+        .await?;
+      self.last_incremental_communities_fetch = self.last_full_communities_fetch;
+    }
+    if (Utc::now() - self.last_incremental_communities_fetch) > *FOLLOW_ADDITIONS_RECHECK_DELAY {
+      // process additions every minute
+      let (news, time) = self
+        .get_communities(self.instance_id, self.last_incremental_communities_fetch)
+        .await?;
+      if !news.is_empty() {
+        tracing::debug!(
+          "{}: fetched {} incremental new followed communities",
+          self.domain,
+          news.len()
+        );
+      }
+      self.followed_communities.extend(news);
+      self.last_incremental_communities_fetch = time;
+    }
+    Ok(())
+  }
+
+  /// get a list of local communities with the remote inboxes on the given instance that cares about them
+  async fn get_communities(
+    &mut self,
+    instance_id: InstanceId,
+    last_fetch: DateTime<Utc>,
+  ) -> Result<(HashMap<CommunityId, HashSet<Url>>, DateTime<Utc>)> {
+    let new_last_fetch =
+      Utc::now() - chrono::TimeDelta::try_seconds(10).expect("TimeDelta out of bounds"); // update to time before fetch to ensure overlap. subtract 10s to ensure overlap even if published date is not exact
+    Ok((
+      CommunityFollowerView::get_instance_followed_community_inboxes(
+        &mut self.pool(),
+        instance_id,
+        last_fetch,
+      )
+      .await?
+      .into_iter()
+      .fold(HashMap::new(), |mut map, (c, u)| {
+        map.entry(c).or_default().insert(u.into());
+        map
+      }),
+      new_last_fetch,
+    ))
+  }
+  fn pool(&self) -> DbPool<'_> {
+    DbPool::Pool(&self.pool)
+  }
+}

crates/federate/src/lib.rs

@@ -1,7 +1,11 @@
 use crate::{util::CancellableTask, worker::InstanceWorker};
 use activitypub_federation::config::FederationConfig;
 use chrono::{Local, Timelike};
-use lemmy_api_common::{context::LemmyContext, federate_retry_sleep_duration};
+use lemmy_api_common::{
+  context::LemmyContext,
+  federate_retry_sleep_duration,
+  lemmy_utils::settings::structs::FederationWorkerConfig,
+};
 use lemmy_db_schema::{
   newtypes::InstanceId,
   source::{federation_queue_state::FederationQueueState, instance::Instance},
@@ -14,6 +18,8 @@ use tokio::{
 };
 use tokio_util::sync::CancellationToken;
 
+mod inboxes;
+mod send;
 mod util;
 mod worker;
 
@@ -34,7 +40,8 @@ pub struct Opts {
 async fn start_stop_federation_workers(
   opts: Opts,
   pool: ActualDbPool,
-  federation_config: FederationConfig<LemmyContext>,
+  federation_lib_config: FederationConfig<LemmyContext>,
+  federation_worker_config: FederationWorkerConfig,
   cancel: CancellationToken,
 ) -> anyhow::Result<()> {
   let mut workers = HashMap::<InstanceId, CancellableTask>::new();
@@ -43,7 +50,9 @@ async fn start_stop_federation_workers(
   let exit_print = tokio::spawn(receive_print_stats(pool.clone(), stats_receiver));
   let pool2 = &mut DbPool::Pool(&pool);
   let process_index = opts.process_index - 1;
-  let local_domain = federation_config.settings().get_hostname_without_port()?;
+  let local_domain = federation_lib_config
+    .settings()
+    .get_hostname_without_port()?;
   loop {
     let mut total_count = 0;
     let mut dead_count = 0;
@@ -71,26 +80,19 @@ async fn start_stop_federation_workers(
           continue;
         }
         // create new worker
-        let config = federation_config.clone();
+        let config = federation_lib_config.clone();
         let stats_sender = stats_sender.clone();
-        let pool = pool.clone();
+        let federation_worker_config = federation_worker_config.clone();
         workers.insert(
           instance.id,
           CancellableTask::spawn(WORKER_EXIT_TIMEOUT, move |stop| {
-            let instance = instance.clone();
-            let req_data = config.clone().to_request_data();
-            let stats_sender = stats_sender.clone();
-            let pool = pool.clone();
-            async move {
-              InstanceWorker::init_and_loop(
-                instance,
-                req_data,
-                &mut DbPool::Pool(&pool),
-                stop,
-                stats_sender,
-              )
-              .await
-            }
+            InstanceWorker::init_and_loop(
+              instance.clone(),
+              config.clone(),
+              federation_worker_config.clone(),
+              stop,
+              stats_sender.clone(),
+            )
           }),
         );
       } else if !should_federate {
@@ -126,12 +128,16 @@ pub fn start_stop_federation_workers_cancellable(
   opts: Opts,
   pool: ActualDbPool,
   config: FederationConfig<LemmyContext>,
+  federation_worker_config: FederationWorkerConfig,
 ) -> CancellableTask {
   CancellableTask::spawn(WORKER_EXIT_TIMEOUT, move |stop| {
-    let opts = opts.clone();
-    let pool = pool.clone();
-    let config = config.clone();
-    async move { start_stop_federation_workers(opts, pool, config, stop).await }
+    start_stop_federation_workers(
+      opts.clone(),
+      pool.clone(),
+      config.clone(),
+      federation_worker_config.clone(),
+      stop,
+    )
   })
 }