feat(fed): Handle EDUs before PDUs

Aranje needs his crypto keys

.forgejo/actions/setup-rust/action.yml

@@ -19,7 +19,7 @@ inputs:
   rust-version:
     description: 'Rust version to install (e.g. nightly). Defaults to 1.87.0'
     required: false
-    default: '1.87.0'
+    default: ''
   sccache-cache-limit:
     description: 'Maximum size limit for sccache local cache (e.g. 2G, 500M)'
     required: false

.forgejo/regsync/regsync.yml

@@ -46,6 +46,9 @@ creds:
   - registry: ghcr.io
     user: "{{env \"GH_PACKAGES_USER\"}}"
     pass: "{{env \"GH_PACKAGES_TOKEN\"}}"
+  - registry: docker.io
+    user: "{{env \"DOCKER_MIRROR_USER\"}}"
+    pass: "{{env \"DOCKER_MIRROR_TOKEN\"}}"
 
 # Global defaults
 defaults:
@@ -67,3 +70,7 @@ sync:
     target: ghcr.io/continuwuity/continuwuity
     type: repository
     <<: *tags-main
+  - source: *source
+    target: docker.io/jadedblueeyes/continuwuity
+    type: repository
+    <<: *tags-main

.forgejo/workflows/mirror-images.yml

@@ -34,6 +34,8 @@ jobs:
       N7574_GIT_TOKEN: ${{ secrets.N7574_GIT_TOKEN }}
       GH_PACKAGES_USER: ${{ vars.GH_PACKAGES_USER }}
       GH_PACKAGES_TOKEN: ${{ secrets.GH_PACKAGES_TOKEN }}
+      DOCKER_MIRROR_USER: ${{ vars.DOCKER_MIRROR_USER }}
+      DOCKER_MIRROR_TOKEN: ${{ secrets.DOCKER_MIRROR_TOKEN }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@v5

.forgejo/workflows/update-flake-hashes.yml

@@ -20,7 +20,8 @@ jobs:
           fetch-tags: false
           fetch-single-branch: true
           submodules: false
-          persist-credentials: false
+          persist-credentials: true
+          token: ${{ secrets.FORGEJO_TOKEN }}
 
       - uses: https://github.com/cachix/install-nix-action@7ab6e7fd29da88e74b1e314a4ae9ac6b5cda3801 # v31.8.0
         with:

Cargo.lock

@@ -4901,9 +4901,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
 
 [[package]]
 name = "syn"
-version = "2.0.109"
+version = "2.0.110"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2f17c7e013e88258aa9543dcbe81aca68a667a9ac37cd69c9fbc07858bfe0e2f"
+checksum = "a99801b5bd34ede4cf3fc688c5919368fea4e4814a4664359503e6015b280aea"
 dependencies = [
  "proc-macro2",
  "quote",

Cargo.toml

@@ -947,7 +947,7 @@ semicolon_outside_block = "warn"
 str_to_string = "warn"
 string_lit_chars_any = "warn"
 string_slice = "warn"
-string_to_string = "warn"
+
 suspicious_xor_used_as_pow = "warn"
 tests_outside_test_module = "warn"
 try_err = "warn"

docker/Dockerfile

@@ -48,7 +48,7 @@ EOF
 
 # Developer tool versions
 # renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
-ENV BINSTALL_VERSION=1.15.10
+ENV BINSTALL_VERSION=1.15.11
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree

docker/musl.Dockerfile

@@ -18,7 +18,7 @@ RUN --mount=type=cache,target=/etc/apk/cache apk add \
 
 # Developer tool versions
 # renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
-ENV BINSTALL_VERSION=1.15.10
+ENV BINSTALL_VERSION=1.15.11
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree

nix/packages/rust.nix

@@ -15,7 +15,7 @@
             file = inputs.self + "/rust-toolchain.toml";
 
             # See also `rust-toolchain.toml`
-            sha256 = "sha256-+9FmLhAOezBZCOziO0Qct1NOrfpjNsXxc/8I0c7BdKE=";
+            sha256 = "sha256-SJwZ8g0zF2WrKDVmHrVG3pD2RGoQeo24MEXnNx5FyuI=";
           };
         in
         {

rust-toolchain.toml

@@ -10,7 +10,7 @@
 
 [toolchain]
 profile = "minimal"
-channel = "1.89.0"
+channel = "1.90.0"
 components = [
     # For rust-analyzer
     "rust-src",

src/api/server/send.rs

@@ -5,7 +5,7 @@ use axum_client_ip::InsecureClientIp;
 use conduwuit::{
 	Err, Error, Result, debug,
 	debug::INFO_SPAN_LEVEL,
-	debug_warn, err, error,
+	debug_warn, err, error, info,
 	result::LogErr,
 	trace,
 	utils::{
@@ -79,13 +79,11 @@ pub(crate) async fn send_transaction_message_route(
 	}
 
 	let txn_start_time = Instant::now();
-	trace!(
+	info!(
 		pdus = body.pdus.len(),
 		edus = body.edus.len(),
-		elapsed = ?txn_start_time.elapsed(),
 		id = ?body.transaction_id,
-		origin =?body.origin(),
+		"Processing transaction",
-		"Starting txn",
 	);
 
 	let pdus = body
@@ -104,14 +102,21 @@ pub(crate) async fn send_transaction_message_route(
 		.filter_map(Result::ok)
 		.stream();
 
-	let results = handle(&services, &client, body.origin(), txn_start_time, pdus, edus).await?;
+	info!(
-
+		pdus = body.pdus.len(),
-	debug!(
+		edus = body.edus.len(),
+		elapsed = ?txn_start_time.elapsed(),
+		id = ?body.transaction_id,
+		"Validated transaction",
+	);
+
+	let results = handle(&services, &client, body.origin(), txn_start_time, pdus, edus).await?;
+
+	info!(
 		pdus = body.pdus.len(),
 		edus = body.edus.len(),
 		elapsed = ?txn_start_time.elapsed(),
 		id = ?body.transaction_id,
-		origin =?body.origin(),
 		"Finished txn",
 	);
 	for (id, result) in &results {
@@ -138,6 +143,10 @@ async fn handle(
 	pdus: impl Stream<Item = Pdu> + Send,
 	edus: impl Stream<Item = Edu> + Send,
 ) -> Result<ResolvedMap> {
+	edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
+		.boxed()
+		.await;
+
 	// group pdus by room
 	let pdus = pdus
 		.collect()
@@ -154,7 +163,8 @@ async fn handle(
 		.into_iter()
 		.try_stream()
 		.broad_and_then(|(room_id, pdus): (_, Vec<_>)| {
-			handle_room(services, client, origin, started, room_id, pdus.into_iter())
+			let count = pdus.len();
+			handle_room(services, client, origin, started, room_id, pdus.into_iter(), count)
 				.map_ok(Vec::into_iter)
 				.map_ok(IterStream::try_stream)
 		})
@@ -163,11 +173,6 @@ async fn handle(
 		.boxed()
 		.await?;
 
-	// evaluate edus after pdus, at least for now.
-	edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
-		.boxed()
-		.await;
-
 	Ok(results)
 }
 
@@ -178,6 +183,7 @@ async fn handle_room(
 	txn_start_time: Instant,
 	room_id: OwnedRoomId,
 	pdus: impl Iterator<Item = Pdu> + Send,
+	count: usize,
 ) -> Result<Vec<(OwnedEventId, Result)>> {
 	let _room_lock = services
 		.rooms
@@ -187,10 +193,20 @@ async fn handle_room(
 		.await;
 
 	let room_id = &room_id;
+	let mut n = 0;
 	pdus.try_stream()
 		.and_then(|(_, event_id, value)| async move {
 			services.server.check_running()?;
 			let pdu_start_time = Instant::now();
+			info!(
+				%room_id,
+				%event_id,
+				pdu = n + 1,
+				total = count,
+				pdu_elapsed = ?pdu_start_time.elapsed(),
+				txn_elapsed = ?txn_start_time.elapsed(),
+				"Handling PDU",
+			);
 			let result = services
 				.rooms
 				.event_handler
@@ -198,11 +214,16 @@ async fn handle_room(
 				.await
 				.map(|_| ());
 
-			debug!(
+			info!(
+				%room_id,
+				%event_id,
+				pdu = n + 1,
+				total = count,
 				pdu_elapsed = ?pdu_start_time.elapsed(),
 				txn_elapsed = ?txn_start_time.elapsed(),
-				"Finished PDU {event_id}",
+				"Finished handling PDU {event_id}",
 			);
+			n += 1;
 
 			Ok((event_id, result))
 		})

src/core/matrix/state_res/mod.rs

@@ -421,8 +421,8 @@ where
 /// `key_fn` is used as to obtain the power level and age of an event for
 /// breaking ties (together with the event ID).
 #[tracing::instrument(level = "debug", skip_all)]
-pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher>(
+pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher, S>(
-	graph: &HashMap<Id, HashSet<Id, Hasher>>,
+	graph: &HashMap<Id, HashSet<Id, Hasher>, S>,
 	key_fn: &F,
 ) -> Result<Vec<Id>>
 where
@@ -430,6 +430,7 @@ where
 	Fut: Future<Output = Result<(Int, MilliSecondsSinceUnixEpoch)>> + Send,
 	Id: Borrow<EventId> + Clone + Eq + Hash + Ord + Send + Sync,
 	Hasher: BuildHasher + Default + Clone + Send + Sync,
+	S: BuildHasher + Clone + Send + Sync,
 {
 	#[derive(PartialEq, Eq)]
 	struct TieBreaker<'a, Id> {

src/database/map/get_batch.rs

@@ -10,7 +10,7 @@ use conduwuit::{
 use futures::{Stream, StreamExt, TryStreamExt};
 use rocksdb::{DBPinnableSlice, ReadOptions};
 
-use super::get::{cached_handle_from, handle_from};
+use super::get::handle_from;
 use crate::Handle;
 
 pub trait Get<'a, K, S>
@@ -58,20 +58,6 @@ where
 		.try_flatten()
 }
 
-#[implement(super::Map)]
-#[tracing::instrument(name = "batch_cached", level = "trace", skip_all)]
-pub(crate) fn get_batch_cached<'a, I, K>(
-	&self,
-	keys: I,
-) -> impl Iterator<Item = Result<Option<Handle<'_>>>> + Send + use<'_, I, K>
-where
-	I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
-	K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
-{
-	self.get_batch_blocking_opts(keys, &self.cache_read_options)
-		.map(cached_handle_from)
-}
-
 #[implement(super::Map)]
 #[tracing::instrument(name = "batch_blocking", level = "trace", skip_all)]
 pub(crate) fn get_batch_blocking<'a, I, K>(

src/database/pool.rs

@@ -184,7 +184,7 @@ fn spawn_one(
 	let handle = thread::Builder::new()
 		.name(WORKER_NAME.into())
 		.stack_size(WORKER_STACK_SIZE)
-		.spawn(move || self.worker(id, recv))?;
+		.spawn(move || self.worker(id, &recv))?;
 
 	workers.push(handle);
 
@@ -260,9 +260,9 @@ async fn execute(&self, queue: &Sender<Cmd>, cmd: Cmd) -> Result {
 		tid = ?thread::current().id(),
 	),
 )]
-fn worker(self: Arc<Self>, id: usize, recv: Receiver<Cmd>) {
+fn worker(self: Arc<Self>, id: usize, recv: &Receiver<Cmd>) {
 	self.worker_init(id);
-	self.worker_loop(&recv);
+	self.worker_loop(recv);
 }
 
 #[implement(Pool)]
@@ -309,7 +309,7 @@ fn worker_loop(self: &Arc<Self>, recv: &Receiver<Cmd>) {
 	self.busy.fetch_add(1, Ordering::Relaxed);
 
 	while let Ok(cmd) = self.worker_wait(recv) {
-		self.worker_handle(cmd);
+		Pool::worker_handle(cmd);
 	}
 }
 
@@ -331,11 +331,11 @@ fn worker_wait(self: &Arc<Self>, recv: &Receiver<Cmd>) -> Result<Cmd, RecvError>
 }
 
 #[implement(Pool)]
-fn worker_handle(self: &Arc<Self>, cmd: Cmd) {
+fn worker_handle(cmd: Cmd) {
 	match cmd {
-		| Cmd::Get(cmd) if cmd.key.len() == 1 => self.handle_get(cmd),
+		| Cmd::Get(cmd) if cmd.key.len() == 1 => Pool::handle_get(cmd),
-		| Cmd::Get(cmd) => self.handle_batch(cmd),
+		| Cmd::Get(cmd) => Pool::handle_batch(cmd),
-		| Cmd::Iter(cmd) => self.handle_iter(cmd),
+		| Cmd::Iter(cmd) => Pool::handle_iter(cmd),
 	}
 }
 
@@ -346,7 +346,7 @@ fn worker_handle(self: &Arc<Self>, cmd: Cmd) {
 	skip_all,
 	fields(%cmd.map),
 )]
-fn handle_iter(&self, mut cmd: Seek) {
+fn handle_iter(mut cmd: Seek) {
 	let chan = cmd.res.take().expect("missing result channel");
 
 	if chan.is_canceled() {
@@ -375,7 +375,7 @@ fn handle_iter(&self, mut cmd: Seek) {
 		keys = %cmd.key.len(),
 	),
 )]
-fn handle_batch(self: &Arc<Self>, mut cmd: Get) {
+fn handle_batch(mut cmd: Get) {
 	debug_assert!(cmd.key.len() > 1, "should have more than one key");
 	debug_assert!(!cmd.key.iter().any(SmallVec::is_empty), "querying for empty key");
 
@@ -401,7 +401,7 @@ fn handle_batch(self: &Arc<Self>, mut cmd: Get) {
 	skip_all,
 	fields(%cmd.map),
 )]
-fn handle_get(&self, mut cmd: Get) {
+fn handle_get(mut cmd: Get) {
 	debug_assert!(!cmd.key[0].is_empty(), "querying for empty key");
 
 	// Obtain the result channel.

src/macros/refutable.rs

@@ -25,13 +25,13 @@ pub(super) fn refutable(mut item: ItemFn, _args: &[Meta]) -> Result<TokenStream>
 		};
 
 		let name = format!("_args_{i}");
-		*pat = Box::new(Pat::Ident(PatIdent {
+		**pat = Pat::Ident(PatIdent {
 			ident: Ident::new(&name, Span::call_site().into()),
 			attrs: Vec::new(),
 			by_ref: None,
 			mutability: None,
 			subpat: None,
-		}));
+		});
 
 		let field = fields.iter();
 		let refute = quote! {

src/service/federation/execute.rs

@@ -1,9 +1,13 @@
-use std::{fmt::Debug, mem};
+use std::{
+	error::Error as _,
+	fmt::{Debug, Write},
+	mem,
+};
 
 use bytes::Bytes;
 use conduwuit::{
 	Err, Error, Result, debug, debug::INFO_SPAN_LEVEL, debug_error, debug_warn, err,
-	error::inspect_debug_log, implement, trace, utils::string::EMPTY,
+	error::inspect_debug_log, implement, trace, utils::string::EMPTY, warn,
 };
 use http::{HeaderValue, header::AUTHORIZATION};
 use ipaddress::IPAddress;
@@ -193,9 +197,9 @@ fn handle_error(
 ) -> Result {
 	if e.is_timeout() || e.is_connect() {
 		e = e.without_url();
-		debug_warn!("{e:?}");
+		warn!(?url, "network error while sending federation request: {e:?}");
 	} else if e.is_redirect() {
-		debug_error!(
+		warn!(
 			method = ?method,
 			url = ?url,
 			final_url = ?e.url(),
@@ -204,9 +208,17 @@ fn handle_error(
 			e,
 		);
 	} else {
-		debug_error!("{e:?}");
+		warn!(?url, "failed to send federation request: {e:?}");
 	}
 
+	let mut nice_error = "Request failed".to_owned();
+	let mut src = e.source();
+	while let Some(source) = src {
+		write!(nice_error, ": {source:?}").expect("writing to string should not fail");
+		src = source.source();
+	}
+	warn!(nice_error, "Federation request error");
+
 	Err(e.into())
 }
 

src/service/rooms/alias/mod.rs

@@ -188,9 +188,7 @@ impl Service {
 	}
 
 	#[tracing::instrument(skip(self), level = "debug")]
-	pub fn all_local_aliases<'a>(
+	pub fn all_local_aliases(&self) -> impl Stream<Item = (&RoomId, &str)> + Send + '_ {
-		&'a self,
-	) -> impl Stream<Item = (&'a RoomId, &'a str)> + Send + 'a {
 		self.db
 			.alias_roomid
 			.stream()

src/service/sending/sender.rs

@@ -10,7 +10,7 @@ use std::{
 
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use conduwuit_core::{
-	Error, Event, Result, debug, err, error,
+	Error, Event, Result, debug, err, error, info,
 	result::LogErr,
 	trace,
 	utils::{
@@ -142,7 +142,7 @@ impl Service {
 	}
 
 	fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, e: &Error) {
-		debug!(dest = ?dest, "{e:?}");
+		debug!(dest = ?dest, "error response: {e:?}");
 		statuses.entry(dest).and_modify(|e| {
 			*e = match e {
 				| TransactionStatus::Running => TransactionStatus::Failed(1, Instant::now()),
@@ -177,7 +177,21 @@ impl Service {
 		if !new_events.is_empty() {
 			self.db.mark_as_active(new_events.iter());
 
-			let new_events_vec = new_events.into_iter().map(|(_, event)| event).collect();
+			let new_events_vec: Vec<SendingEvent> =
+				new_events.into_iter().map(|(_, event)| event).collect();
+
+			if let Some(status) = statuses.get(&dest.clone()) {
+				if matches!(status, TransactionStatus::Running) {
+					// If the server is in backoff, clear it
+					info!(
+						?dest,
+						"Catching up previously failed destination with {}+ new events",
+						new_events_vec.len()
+					);
+					statuses.insert(dest.clone(), TransactionStatus::Running);
+				}
+			}
+
 			futures.push(self.send_events(dest.clone(), new_events_vec));
 		} else {
 			statuses.remove(dest);
@@ -859,12 +873,20 @@ impl Service {
 			pdus,
 			edus,
 		};
+		let pdu_count = request.pdus.len();
+		let edu_count = request.edus.len();
 
 		let result = self
 			.services
 			.federation
 			.execute_on(&self.services.client.sender, &server, request)
-			.await;
+			.await
+			.inspect(|_| {
+				info!(%txn_id, %server, "Sent {} PDUs, {} EDUs", pdu_count, edu_count);
+			})
+			.inspect_err(|e| {
+				info!(%txn_id, %server, "Failed to send transaction ({} PDUs, {} EDUs): {e:?}", pdu_count, edu_count);
+			});
 
 		for (event_id, result) in result.iter().flat_map(|resp| resp.pdus.iter()) {
 			if let Err(e) = result {