ci: Use dataaxiom/ghcr-cleanup-action

.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: ''
+    default: '1.87.0'
   sccache-cache-limit:
     description: 'Maximum size limit for sccache local cache (e.g. 2G, 500M)'
     required: false

.forgejo/regsync/regsync.yml

@@ -46,9 +46,6 @@ 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:
@@ -70,7 +67,3 @@ sync:
     target: ghcr.io/continuwuity/continuwuity
     type: repository
     <<: *tags-main
-  - source: *source
-    target: docker.io/jadedblueeyes/continuwuity
-    type: repository
-    <<: *tags-main

.forgejo/workflows/cleanup-registry.yml

@@ -0,0 +1,110 @@
+name: Cleanup Registry Images
+
+on:
+  schedule:
+    # Run daily at 01:30 UTC
+    - cron: '30 1 * * *'
+  workflow_dispatch:
+    inputs:
+      dry_run:
+        description: 'Dry run (check only, no actual deletion)'
+        required: false
+        default: false
+        type: boolean
+  pull_request:
+    types: [closed]
+  delete:
+    # Triggered when branches are deleted
+
+concurrency:
+  group: "cleanup-registry"
+  cancel-in-progress: false
+
+env:
+  BUILTIN_REGISTRY: forgejo.ellis.link
+  IMAGE_PATH: forgejo.ellis.link/continuwuation/continuwuity
+
+jobs:
+  cleanup-pr-images:
+    name: Cleanup PR Images
+    runs-on: ubuntu-latest
+    if: github.event_name == 'pull_request' && github.event.action == 'closed'
+    permissions:
+      packages: write
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v5
+        with:
+          persist-credentials: false
+
+      - name: Delete PR image
+        uses: https://github.com/dataaxiom/ghcr-cleanup-action@v1
+        with:
+          token: ${{ secrets.BUILTIN_REGISTRY_PASSWORD || secrets.GITHUB_TOKEN }}
+          owner: continuwuation
+          repository: continuwuity
+          package: continuwuity
+          registry-url: https://${{ env.BUILTIN_REGISTRY }}
+          delete-tags: pr-${{ github.event.pull_request.number }}
+          dry-run: false
+
+  cleanup-old-commits:
+    name: Cleanup Old Commit Images
+    runs-on: ubuntu-latest
+    if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
+    permissions:
+      packages: write
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v5
+        with:
+          persist-credentials: false
+
+      - name: Set dry-run mode
+        id: params
+        run: |
+          if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
+            echo "dry_run=${{ inputs.dry_run }}" >> $GITHUB_OUTPUT
+          else
+            # Scheduled runs perform actual cleanup
+            echo "dry_run=false" >> $GITHUB_OUTPUT
+          fi
+
+      - name: Cleanup old SHA commit images
+        uses: https://github.com/dataaxiom/ghcr-cleanup-action@v1
+        with:
+          token: ${{ secrets.BUILTIN_REGISTRY_PASSWORD || secrets.GITHUB_TOKEN }}
+          owner: continuwuation
+          repository: continuwuity
+          package: continuwuity
+          registry-url: https://${{ env.BUILTIN_REGISTRY }}
+          delete-tags: sha-*
+          exclude-tags: latest,main,v*.*.*,*.*.*,*-maxperf
+          older-than: 30 days
+          dry-run: ${{ steps.params.outputs.dry_run }}
+          delete-ghost-images: true
+          delete-partial-images: true
+          delete-orphaned-images: true
+
+  cleanup-branch-images:
+    name: Cleanup Deleted Branch Images
+    runs-on: ubuntu-latest
+    if: github.event_name == 'delete' && github.event.ref_type == 'branch'
+    permissions:
+      packages: write
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v5
+        with:
+          persist-credentials: false
+
+      - name: Delete branch image
+        uses: https://github.com/dataaxiom/ghcr-cleanup-action@v1
+        with:
+          token: ${{ secrets.BUILTIN_REGISTRY_PASSWORD || secrets.GITHUB_TOKEN }}
+          owner: continuwuation
+          repository: continuwuity
+          package: continuwuity
+          registry-url: https://${{ env.BUILTIN_REGISTRY }}
+          delete-tags: branch-${{ github.event.ref }}
+          dry-run: false

.forgejo/workflows/mirror-images.yml

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

Cargo.toml

@@ -166,8 +166,8 @@ default-features = false
 features = ["raw_value"]
 
 # Used for appservice registration files
-[workspace.dependencies.serde-saphyr]
-version = "0.0.7"
+[workspace.dependencies.serde_yml]
+version = "0.0.12"
 
 # Used to load forbidden room/user regex from config
 [workspace.dependencies.serde_regex]
@@ -210,13 +210,13 @@ default-features = false
 version = "0.1.41"
 default-features = false
 [workspace.dependencies.tracing-subscriber]
-version = "0.3.20"
+version = "0.3.19"
 default-features = false
 features = ["env-filter", "std", "tracing", "tracing-log", "ansi", "fmt"]
 [workspace.dependencies.tracing-journald]
 version = "0.3.1"
 [workspace.dependencies.tracing-core]
-version = "0.1.34"
+version = "0.1.33"
 default-features = false
 
 # for URL previews
@@ -286,7 +286,7 @@ features = [
 ]
 
 [workspace.dependencies.hyper-util]
-version = "=0.1.17"
+version = "0.1.11"
 default-features = false
 features = [
 	"server-auto",
@@ -412,27 +412,28 @@ default-features = false
 
 # optional opentelemetry, performance measurements, flamegraphs, etc for performance measurements and monitoring
 [workspace.dependencies.opentelemetry]
-version = "0.31.0"
+version = "0.30.0"
 
 [workspace.dependencies.tracing-flame]
 version = "0.2.0"
 
 [workspace.dependencies.tracing-opentelemetry]
-version = "0.32.0"
+version = "0.31.0"
 
 [workspace.dependencies.opentelemetry_sdk]
-version = "0.31.0"
+version = "0.30.0"
 features = ["rt-tokio"]
 
 [workspace.dependencies.opentelemetry-otlp]
-version = "0.31.0"
+version = "0.30.0"
 features = ["http", "trace", "logs", "metrics"]
 
-
+[workspace.dependencies.opentelemetry-jaeger-propagator]
+version = "0.30.0"
 
 # optional sentry metrics for crash/panic reporting
 [workspace.dependencies.sentry]
-version = "0.45.0"
+version = "0.42.0"
 default-features = false
 features = [
     "backtrace",
@@ -448,9 +449,9 @@ features = [
 ]
 
 [workspace.dependencies.sentry-tracing]
-version = "0.45.0"
+version = "0.42.0"
 [workspace.dependencies.sentry-tower]
-version = "0.45.0"
+version = "0.42.0"
 
 # jemalloc usage
 [workspace.dependencies.tikv-jemalloc-sys]
@@ -476,7 +477,7 @@ default-features = false
 features = ["use_std"]
 
 [workspace.dependencies.console-subscriber]
-version = "0.5"
+version = "0.4"
 
 [workspace.dependencies.nix]
 version = "0.30.1"
@@ -563,7 +564,19 @@ version = "0.7.5"
 
 # backport of [https://github.com/tokio-rs/tracing/pull/2956] to the 0.1.x branch of tracing.
 # we can switch back to upstream if #2956 is merged and backported in the upstream repo.
-
+# https://forgejo.ellis.link/continuwuation/tracing/commit/b348dca742af641c47bc390261f60711c2af573c
+[patch.crates-io.tracing-subscriber]
+git = "https://forgejo.ellis.link/continuwuation/tracing"
+rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
+[patch.crates-io.tracing]
+git = "https://forgejo.ellis.link/continuwuation/tracing"
+rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
+[patch.crates-io.tracing-core]
+git = "https://forgejo.ellis.link/continuwuation/tracing"
+rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
+[patch.crates-io.tracing-log]
+git = "https://forgejo.ellis.link/continuwuation/tracing"
+rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
 
 # adds a tab completion callback: https://forgejo.ellis.link/continuwuation/rustyline-async/src/branch/main/.patchy/0002-add-tab-completion-callback.patch
 # adds event for CTRL+\: https://forgejo.ellis.link/continuwuation/rustyline-async/src/branch/main/.patchy/0001-add-event-for-ctrl.patch
@@ -587,7 +600,7 @@ rev = "9c8e51510c35077df888ee72a36b4b05637147da"
 # reverts hyperium#148 conflicting with our delicate federation resolver hooks
 [patch.crates-io.hyper-util]
 git = "https://forgejo.ellis.link/continuwuation/hyper-util"
-rev = "5886d5292bf704c246206ad72d010d674a7b77d0"
+rev = "e4ae7628fe4fcdacef9788c4c8415317a4489941"
 
 #
 # Our crates
@@ -947,7 +960,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.11
+ENV BINSTALL_VERSION=1.15.7
 # 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.11
+ENV BINSTALL_VERSION=1.15.7
 # 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-SJwZ8g0zF2WrKDVmHrVG3pD2RGoQeo24MEXnNx5FyuI=";
+            sha256 = "sha256-+9FmLhAOezBZCOziO0Qct1NOrfpjNsXxc/8I0c7BdKE=";
           };
         in
         {

renovate.json

@@ -18,6 +18,14 @@
         "tikv-jemallocator",
         "tikv-jemalloc-sys",
         "tikv-jemalloc-ctl",
+        "opentelemetry",
+        "opentelemetry_sdk",
+        "opentelemetry-jaeger",
+        "tracing-opentelemetry",
+        "tracing-subscriber",
+        "tracing",
+        "tracing-core",
+        "tracing-log",
         "rustyline-async",
         "event-listener",
         "async-channel",

rust-toolchain.toml

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

src/admin/Cargo.toml

@@ -85,7 +85,7 @@ futures.workspace = true
 log.workspace = true
 ruma.workspace = true
 serde_json.workspace = true
-serde-saphyr.workspace = true
+serde_yml.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true

src/admin/appservice/commands.rs

@@ -16,7 +16,7 @@ pub(super) async fn register(&self) -> Result {
 
 	let range = 1..checked!(body_len - 1)?;
 	let appservice_config_body = body[range].join("\n");
-	let parsed_config = serde_saphyr::from_str(&appservice_config_body);
+	let parsed_config = serde_yml::from_str(&appservice_config_body);
 	match parsed_config {
 		| Err(e) => return Err!("Could not parse appservice config as YAML: {e}"),
 		| Ok(registration) => match self
@@ -57,7 +57,7 @@ pub(super) async fn show_appservice_config(&self, appservice_identifier: String)
 	{
 		| None => return Err!("Appservice does not exist."),
 		| Some(config) => {
-			let config_str = serde_saphyr::to_string(&config)?;
+			let config_str = serde_yml::to_string(&config)?;
 			write!(self, "Config for {appservice_identifier}:\n\n```yaml\n{config_str}\n```")
 		},
 	}

src/api/client/well_known.rs

@@ -78,7 +78,7 @@ pub(crate) async fn well_known_support(
 			while let Some(user_id) = stream.next().await {
 				// Skip server user
 				if *user_id == services.globals.server_user {
-					continue;
+					break;
 				}
 				contacts.push(Contact {
 					role: role_value.clone(),

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, info,
+	debug_warn, err, error,
 	result::LogErr,
 	trace,
 	utils::{
@@ -79,11 +79,13 @@ pub(crate) async fn send_transaction_message_route(
 	}
 
 	let txn_start_time = Instant::now();
-	info!(
+	trace!(
 		pdus = body.pdus.len(),
 		edus = body.edus.len(),
+		elapsed = ?txn_start_time.elapsed(),
 		id = ?body.transaction_id,
-		"Processing transaction",
+		origin =?body.origin(),
+		"Starting txn",
 	);
 
 	let pdus = body
@@ -102,21 +104,14 @@ pub(crate) async fn send_transaction_message_route(
 		.filter_map(Result::ok)
 		.stream();
 
-	info!(
-		pdus = body.pdus.len(),
-		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!(
+	debug!(
 		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 {
@@ -143,10 +138,6 @@ 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()
@@ -163,8 +154,7 @@ async fn handle(
 		.into_iter()
 		.try_stream()
 		.broad_and_then(|(room_id, pdus): (_, Vec<_>)| {
-			let count = pdus.len();
-			handle_room(services, client, origin, started, room_id, pdus.into_iter(), count)
+			handle_room(services, client, origin, started, room_id, pdus.into_iter())
 				.map_ok(Vec::into_iter)
 				.map_ok(IterStream::try_stream)
 		})
@@ -173,6 +163,11 @@ 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)
 }
 
@@ -183,7 +178,6 @@ 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
@@ -193,20 +187,10 @@ 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
@@ -214,16 +198,11 @@ async fn handle_room(
 				.await
 				.map(|_| ());
 
-			info!(
-				%room_id,
-				%event_id,
-				pdu = n + 1,
-				total = count,
+			debug!(
 				pdu_elapsed = ?pdu_start_time.elapsed(),
 				txn_elapsed = ?txn_start_time.elapsed(),
-				"Finished handling PDU {event_id}",
+				"Finished PDU {event_id}",
 			);
-			n += 1;
 
 			Ok((event_id, result))
 		})

src/core/Cargo.toml

@@ -92,7 +92,7 @@ ruma.workspace = true
 sanitize-filename.workspace = true
 serde_json.workspace = true
 serde_regex.workspace = true
-serde-saphyr.workspace = true
+serde_yml.workspace = true
 serde.workspace = true
 smallvec.workspace = true
 smallstr.workspace = true

src/core/error/mod.rs

@@ -83,9 +83,7 @@ pub enum Error {
 	#[error(transparent)]
 	TypedHeader(#[from] axum_extra::typed_header::TypedHeaderRejection),
 	#[error(transparent)]
-	YamlDe(#[from] serde_saphyr::Error),
-	#[error(transparent)]
-	YamlSer(#[from] serde_saphyr::ser_error::Error),
+	Yaml(#[from] serde_yml::Error),
 
 	// ruma/conduwuit
 	#[error("Arithmetic operation failed: {0}")]

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, S>(
-	graph: &HashMap<Id, HashSet<Id, Hasher>, S>,
+pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher>(
+	graph: &HashMap<Id, HashSet<Id, Hasher>>,
 	key_fn: &F,
 ) -> Result<Vec<Id>>
 where
@@ -430,7 +430,6 @@ 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::handle_from;
+use super::get::{cached_handle_from, handle_from};
 use crate::Handle;
 
 pub trait Get<'a, K, S>
@@ -58,6 +58,20 @@ 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) {
-		Pool::worker_handle(cmd);
+		self.worker_handle(cmd);
 	}
 }
 
@@ -331,11 +331,11 @@ fn worker_wait(self: &Arc<Self>, recv: &Receiver<Cmd>) -> Result<Cmd, RecvError>
 }
 
 #[implement(Pool)]
-fn worker_handle(cmd: Cmd) {
+fn worker_handle(self: &Arc<Self>, cmd: Cmd) {
 	match cmd {
-		| Cmd::Get(cmd) if cmd.key.len() == 1 => Pool::handle_get(cmd),
-		| Cmd::Get(cmd) => Pool::handle_batch(cmd),
-		| Cmd::Iter(cmd) => Pool::handle_iter(cmd),
+		| Cmd::Get(cmd) if cmd.key.len() == 1 => self.handle_get(cmd),
+		| Cmd::Get(cmd) => self.handle_batch(cmd),
+		| Cmd::Iter(cmd) => self.handle_iter(cmd),
 	}
 }
 
@@ -346,7 +346,7 @@ fn worker_handle(cmd: Cmd) {
 	skip_all,
 	fields(%cmd.map),
 )]
-fn handle_iter(mut cmd: Seek) {
+fn handle_iter(&self, mut cmd: Seek) {
 	let chan = cmd.res.take().expect("missing result channel");
 
 	if chan.is_canceled() {
@@ -375,7 +375,7 @@ fn handle_iter(mut cmd: Seek) {
 		keys = %cmd.key.len(),
 	),
 )]
-fn handle_batch(mut cmd: Get) {
+fn handle_batch(self: &Arc<Self>, 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(mut cmd: Get) {
 	skip_all,
 	fields(%cmd.map),
 )]
-fn handle_get(mut cmd: Get) {
+fn handle_get(&self, 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 = Pat::Ident(PatIdent {
+		*pat = Box::new(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/main/Cargo.toml

@@ -62,8 +62,7 @@ standard = [
 	"media_thumbnail",
 	"systemd",
 	"url_preview",
-	"zstd_compression",
-    "sentry_telemetry"
+	"zstd_compression"
 ]
 full = [
     "standard",
@@ -130,6 +129,7 @@ perf_measurements = [
 	"dep:tracing-opentelemetry",
 	"dep:opentelemetry_sdk",
 	"dep:opentelemetry-otlp",
+	"dep:opentelemetry-jaeger-propagator",
 	"conduwuit-core/perf_measurements",
 	"conduwuit-core/sentry_telemetry",
 ]
@@ -201,7 +201,6 @@ conduwuit-core.workspace = true
 conduwuit-database.workspace = true
 conduwuit-router.workspace = true
 conduwuit-service.workspace = true
-conduwuit-build-metadata.workspace = true
 
 clap.workspace = true
 console-subscriber.optional = true
@@ -213,6 +212,8 @@ opentelemetry.optional = true
 opentelemetry.workspace = true
 opentelemetry-otlp.optional = true
 opentelemetry-otlp.workspace = true
+opentelemetry-jaeger-propagator.optional = true
+opentelemetry-jaeger-propagator.workspace = true
 opentelemetry_sdk.optional = true
 opentelemetry_sdk.workspace = true
 sentry-tower.optional = true

src/main/logging.rs

@@ -94,7 +94,7 @@ pub(crate) fn init(
 
 		let otlp_layer = config.allow_otlp.then(|| {
 			opentelemetry::global::set_text_map_propagator(
-				opentelemetry_sdk::propagation::TraceContextPropagator::new(),
+				opentelemetry_jaeger_propagator::Propagator::new(),
 			);
 
 			let exporter = opentelemetry_otlp::SpanExporter::builder()

src/main/sentry.rs

@@ -1,12 +1,10 @@
 #![cfg(feature = "sentry_telemetry")]
 
 use std::{
-	borrow::Cow,
 	str::FromStr,
 	sync::{Arc, OnceLock},
 };
 
-use conduwuit_build_metadata as build;
 use conduwuit_core::{config::Config, debug, trace};
 use sentry::{
 	Breadcrumb, ClientOptions, Level,
@@ -46,7 +44,7 @@ fn options(config: &Config) -> ClientOptions {
 		server_name,
 		traces_sample_rate: config.sentry_traces_sample_rate,
 		debug: cfg!(debug_assertions),
-		release: release_name(),
+		release: sentry::release_name!(),
 		user_agent: conduwuit_core::version::user_agent().into(),
 		attach_stacktrace: config.sentry_attach_stacktrace,
 		before_send: Some(Arc::new(before_send)),
@@ -93,21 +91,3 @@ fn before_breadcrumb(crumb: Breadcrumb) -> Option<Breadcrumb> {
 	trace!("Sentry breadcrumb: {crumb:?}");
 	Some(crumb)
 }
-
-fn release_name() -> Option<Cow<'static, str>> {
-	static RELEASE: OnceLock<Option<String>> = OnceLock::new();
-
-	RELEASE
-		.get_or_init(|| {
-			let pkg_name = env!("CARGO_PKG_NAME");
-			let pkg_version = env!("CARGO_PKG_VERSION");
-
-			if let Some(commit_short) = build::GIT_COMMIT_HASH_SHORT {
-				Some(format!("{pkg_name}@{pkg_version}+{commit_short}"))
-			} else {
-				Some(format!("{pkg_name}@{pkg_version}"))
-			}
-		})
-		.as_ref()
-		.map(|s| Cow::Borrowed(s.as_str()))
-}

src/service/Cargo.toml

@@ -108,7 +108,7 @@ rustyline-async.workspace = true
 rustyline-async.optional = true
 serde_json.workspace = true
 serde.workspace = true
-serde-saphyr.workspace = true
+serde_yml.workspace = true
 sha2.workspace = true
 termimad.workspace = true
 termimad.optional = true

src/service/appservice/mod.rs

@@ -271,7 +271,7 @@ impl Service {
 			.id_appserviceregistrations
 			.get(id)
 			.await
-			.and_then(|ref bytes| serde_saphyr::from_slice(bytes).map_err(Into::into))
+			.and_then(|ref bytes| serde_yml::from_slice(bytes).map_err(Into::into))
 			.map_err(|e| err!(Database("Invalid appservice {id:?} registration: {e:?}")))
 	}
 

src/service/federation/execute.rs

@@ -1,13 +1,9 @@
-use std::{
-	error::Error as _,
-	fmt::{Debug, Write},
-	mem,
-};
+use std::{fmt::Debug, 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, warn,
+	error::inspect_debug_log, implement, trace, utils::string::EMPTY,
 };
 use http::{HeaderValue, header::AUTHORIZATION};
 use ipaddress::IPAddress;
@@ -197,9 +193,9 @@ fn handle_error(
 ) -> Result {
 	if e.is_timeout() || e.is_connect() {
 		e = e.without_url();
-		warn!(?url, "network error while sending federation request: {e:?}");
+		debug_warn!("{e:?}");
 	} else if e.is_redirect() {
-		warn!(
+		debug_error!(
 			method = ?method,
 			url = ?url,
 			final_url = ?e.url(),
@@ -208,17 +204,9 @@ fn handle_error(
 			e,
 		);
 	} else {
-		warn!(?url, "failed to send federation request: {e:?}");
+		debug_error!("{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,7 +188,9 @@ impl Service {
 	}
 
 	#[tracing::instrument(skip(self), level = "debug")]
-	pub fn all_local_aliases(&self) -> impl Stream<Item = (&RoomId, &str)> + Send + '_ {
+	pub fn all_local_aliases<'a>(
+		&'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, info,
+	Error, Event, Result, debug, err, error,
 	result::LogErr,
 	trace,
 	utils::{
@@ -142,7 +142,7 @@ impl Service {
 	}
 
 	fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, e: &Error) {
-		debug!(dest = ?dest, "error response: {e:?}");
+		debug!(dest = ?dest, "{e:?}");
 		statuses.entry(dest).and_modify(|e| {
 			*e = match e {
 				| TransactionStatus::Running => TransactionStatus::Failed(1, Instant::now()),
@@ -177,21 +177,7 @@ impl Service {
 		if !new_events.is_empty() {
 			self.db.mark_as_active(new_events.iter());
 
-			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);
-				}
-			}
-
+			let new_events_vec = new_events.into_iter().map(|(_, event)| event).collect();
 			futures.push(self.send_events(dest.clone(), new_events_vec));
 		} else {
 			statuses.remove(dest);
@@ -873,20 +859,12 @@ 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
-			.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);
-			});
+			.await;
 
 		for (event_id, result) in result.iter().flat_map(|resp| resp.pdus.iter()) {
 			if let Err(e) = result {