fix(complement): Fix complement conflicting with first-run

- Disabled first-run mode when running Complement tests
- Updated logging config under complement to be a bit less verbose
- Changed test result and log output locations

Cargo.toml

@@ -68,7 +68,7 @@ default-features = false
 version = "0.1.3"
 
 [workspace.dependencies.rand]
-version = "0.8.5"
+version = "0.10.0"
 
 # Used for the http request / response body type for Ruma endpoints used with reqwest
 [workspace.dependencies.bytes]
@@ -97,7 +97,7 @@ features = [
 ]
 
 [workspace.dependencies.axum-extra]
-version = "0.10.1"
+version = "0.12.0"
 default-features = false
 features = ["typed-header", "tracing"]
 
@@ -253,7 +253,7 @@ features = [
 version = "0.4.0"
 
 [workspace.dependencies.libloading]
-version = "0.8.6"
+version = "0.9.0"
 
 # Validating urls in config, was already a transitive dependency
 [workspace.dependencies.url]
@@ -298,7 +298,7 @@ default-features = false
 features = ["env", "toml"]
 
 [workspace.dependencies.hickory-resolver]
-version = "0.25.1"
+version = "0.25.2"
 default-features = false
 features = [
     "serde",
@@ -343,7 +343,7 @@ version = "0.1.2"
 [workspace.dependencies.ruma]
 git = "https://forgejo.ellis.link/continuwuation/ruwuma"
 #branch = "conduwuit-changes"
-rev = "3126cb5eea991ec40590e54d8c9d75637650641a"
+rev = "bb12ed288a31a23aa11b10ba0fad22b7f985eb88"
 features = [
     "compat",
     "rand",
@@ -381,6 +381,7 @@ features = [
     "unstable-pdu",
     "unstable-msc4155",
     "unstable-msc4143", # livekit well_known response
+    "unstable-msc4284"
 ]
 
 [workspace.dependencies.rust-rocksdb]
@@ -425,7 +426,7 @@ features = ["http", "grpc-tonic", "trace", "logs", "metrics"]
 
 # optional sentry metrics for crash/panic reporting
 [workspace.dependencies.sentry]
-version = "0.45.0"
+version = "0.46.0"
 default-features = false
 features = [
     "backtrace",
@@ -441,9 +442,9 @@ features = [
 ]
 
 [workspace.dependencies.sentry-tracing]
-version = "0.45.0"
+version = "0.46.0"
 [workspace.dependencies.sentry-tower]
-version = "0.45.0"
+version = "0.46.0"
 
 # jemalloc usage
 [workspace.dependencies.tikv-jemalloc-sys]
@@ -472,7 +473,7 @@ features = ["use_std"]
 version = "0.5"
 
 [workspace.dependencies.nix]
-version = "0.30.1"
+version = "0.31.0"
 default-features = false
 features = ["resource"]
 

README.md

@@ -57,10 +57,15 @@ ### What are the project's goals?
 
 ### Can I try it out?
 
-Check out the [documentation](https://continuwuity.org) for installation instructions, or join one of these vetted public homeservers running Continuwuity to get a feel for things!
+Check out the [documentation](https://continuwuity.org) for installation instructions.
 
-- https://continuwuity.rocks -- A public demo server operated by the Continuwuity Team.
-- https://federated.nexus -- Federated Nexus is a community resource hosting multiple FOSS (especially federated) services, including Matrix and Forgejo.
+If you want to try it out as a user, we have some partnered homeservers you can use:
+* You can head over to [https://federated.nexus](https://federated.nexus/) in your browser.
+  * Hit the `Apply to Join` button. Once your request has been accepted, you will receive an email with your username and password.
+  * Head over to [https://app.federated.nexus](https://app.federated.nexus/) and you can sign in there, or use any other matrix chat client you wish elsewhere.
+  * Your username for matrix will be in the form of `@username:federated.nexus`, however you can simply use the `username` part to log in. Your password is your password.
+
+* There's also [https://continuwuity.rocks/](https://continuwuity.rocks/). You can register a new account using Cinny via [this convenient link](https://app.cinny.in/register/continuwuity.rocks), or you can use Element or another matrix client *that supports registration*.
 
 ### What are we working on?
 

bin/complement

@@ -6,10 +6,10 @@ set -euo pipefail
 COMPLEMENT_SRC="${COMPLEMENT_SRC:-$1}"
 
 # A `.jsonl` file to write test logs to
-LOG_FILE="${2:-complement_test_logs.jsonl}"
+LOG_FILE="${2:-tests/test_results/complement/test_logs.jsonl}"
 
 # A `.jsonl` file to write test results to
-RESULTS_FILE="${3:-complement_test_results.jsonl}"
+RESULTS_FILE="${3:-tests/test_results/complement/test_results.jsonl}"
 
 # The base docker image to use for complement tests
 # You can build the default with `docker build -t continuwuity:complement -f ./docker/complement.Dockerfile .`

changelog.d/1421.bugfix

@@ -0,0 +1 @@
+Fixed a startup crash in the sender service if we can't detect the number of CPU cores, even if the `sender_workers' config option is set correctly. Contributed by @katie.

changelog.d/1428.feature

@@ -0,0 +1 @@
+Improved the concurrency handling of federation transactions, vastly improving performance and reliability by more accurately handling inbound transactions and reducing the amount of repeated wasted work. Contributed by @nex and @Jade.

changelog.d/1435.feature.md

@@ -0,0 +1 @@
+Added MSC3202 Device masquerading (not all of MSC3202). This should fix issues with enabling MSC4190 for some Mautrix bridges. Contributed by @Jade

changelog.d/1441.bugfix

@@ -0,0 +1 @@
+Removed the `allow_public_room_directory_without_auth` config option. Contributed by @0xnim.

changelog.d/1442.feature

@@ -0,0 +1 @@
+Implement MSC4143 MatrixRTC transport discovery endpoint. Move RTC foci configuration from `[global.well_known]` to a new `[global.matrix_rtc]` section with a `foci` field. Contributed by @0xnim

changelog.d/1445.bugfix

@@ -0,0 +1 @@
+Fixed sliding sync v5 list ranges always starting from 0, causing extra rooms to be unnecessarily processed and returned. Contributed by @0xnim

changelog.d/url-preview-fix.feature

@@ -0,0 +1 @@
+Improved URL preview fetching with a more compatible user agent for sites like YouTube Music. Added `!admin media delete-url-preview <url>` command to clear cached URL previews that were stuck and broken.

complement/complement.config.toml

@@ -9,10 +9,9 @@ address = "0.0.0.0"
 allow_device_name_federation = true
 allow_guest_registration = true
 allow_public_room_directory_over_federation = true
-allow_public_room_directory_without_auth = true
 allow_registration = true
 database_path = "/database"
-log = "trace,h2=debug,hyper=debug"
+log = "trace,h2=debug,hyper=debug,conduwuit_database=warn,conduwuit_service::manager=info,conduwuit_api::router=error,conduwuit_router=error,tower_http=error"
 port = [8008, 8448]
 trusted_servers = []
 only_query_trusted_key_servers = false
@@ -25,7 +24,7 @@ url_preview_domain_explicit_denylist = ["*"]
 media_compat_file_link = false
 media_startup_check = true
 prune_missing_media = true
-log_colors = true
+log_colors = false
 admin_room_notices = false
 allow_check_for_updates = false
 intentionally_unknown_config_option_for_testing = true
@@ -48,6 +47,7 @@ federation_idle_timeout = 300
 sender_timeout = 300
 sender_idle_timeout = 300
 sender_retry_backoff_limit = 300
+force_disable_first_run_mode = true
 
 [global.tls]
 dual_protocol = true

conduwuit-example.toml

@@ -290,6 +290,25 @@
 #
 #max_fetch_prev_events = 192
 
+# How many incoming federation transactions the server is willing to be
+# processing at any given time before it becomes overloaded and starts
+# rejecting further transactions until some slots become available.
+#
+# Setting this value too low or too high may result in unstable
+# federation, and setting it too high may cause runaway resource usage.
+#
+#max_concurrent_inbound_transactions = 150
+
+# Maximum age (in seconds) for cached federation transaction responses.
+# Entries older than this will be removed during cleanup.
+#
+#transaction_id_cache_max_age_secs = 7200 (2 hours)
+
+# Maximum number of cached federation transaction responses.
+# When the cache exceeds this limit, older entries will be removed.
+#
+#transaction_id_cache_max_entries = 8192
+
 # Default/base connection timeout (seconds). This is used only by URL
 # previews and update/news endpoint checks.
 #
@@ -527,12 +546,6 @@
 #
 #allow_public_room_directory_over_federation = false
 
-# Set this to true to allow your server's public room directory to be
-# queried without client authentication (access token) through the Client
-# APIs. Set this to false to protect against /publicRooms spiders.
-#
-#allow_public_room_directory_without_auth = false
-
 # Allow guests/unauthenticated users to access TURN credentials.
 #
 # This is the equivalent of Synapse's `turn_allow_guests` config option.
@@ -1325,7 +1338,7 @@
 # sender user's server name, inbound federation X-Matrix origin, and
 # outbound federation handler.
 #
-# You can set this to ["*"] to block all servers by default, and then
+# You can set this to [".*"] to block all servers by default, and then
 # use `allowed_remote_server_names` to allow only specific servers.
 #
 # example: ["badserver\\.tld$", "badphrase", "19dollarfortnitecards"]
@@ -1831,14 +1844,13 @@
 #
 #support_mxid =
 
-# A list of MatrixRTC foci URLs which will be served as part of the
-# MSC4143 client endpoint at /.well-known/matrix/client.  If you're
-# setting up livekit, you'd want something like:
-# rtc_focus_server_urls = [
-#     { type = "livekit", livekit_service_url = "https://livekit.example.com" },
-# ]
+# **DEPRECATED**: Use `[global.matrix_rtc].foci` instead.
 #
-# To disable, set this to be an empty vector (`[]`).
+# A list of MatrixRTC foci URLs which will be served as part of the
+# MSC4143 client endpoint at /.well-known/matrix/client.
+#
+# This option is deprecated and will be removed in a future release.
+# Please migrate to the new `[global.matrix_rtc]` config section.
 #
 #rtc_focus_server_urls = []
 
@@ -1860,6 +1872,23 @@
 #
 #blurhash_max_raw_size = 33554432
 
+[global.matrix_rtc]
+
+# A list of MatrixRTC foci (transports) which will be served via the
+# MSC4143 RTC transports endpoint at
+# `/_matrix/client/v1/rtc/transports`. If you're setting up livekit,
+# you'd want something like:
+# ```toml
+# [global.matrix_rtc]
+# foci = [
+#     { type = "livekit", livekit_service_url = "https://livekit.example.com" },
+# ]
+# ```
+#
+# To disable, set this to an empty list (`[]`).
+#
+#foci = []
+
 [global.ldap]
 
 # Whether to enable LDAP login.

docker/Dockerfile

@@ -52,7 +52,7 @@ ENV BINSTALL_VERSION=1.17.5
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree
-ENV LDDTREE_VERSION=0.4.0
+ENV LDDTREE_VERSION=0.5.0
 # renovate: datasource=crate depName=timelord-cli
 ENV TIMELORD_VERSION=3.0.1
 

docker/musl.Dockerfile

@@ -22,7 +22,7 @@ ENV BINSTALL_VERSION=1.17.5
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree
-ENV LDDTREE_VERSION=0.4.0
+ENV LDDTREE_VERSION=0.5.0
 
 # Install unpackaged tools
 RUN <<EOF

docs/calls/livekit.mdx

@@ -78,47 +78,19 @@ #### Firewall hints
 
 ### 3. Telling clients where to find LiveKit
 
-To tell clients where to find LiveKit, you need to add the address of your `lk-jwt-service` to your client .well-known file. To do so, in the config section `global.well-known`, add (or modify) the option `rtc_focus_server_urls`.
+To tell clients where to find LiveKit, you need to add the address of your `lk-jwt-service` to the `[global.matrix_rtc]` config section using the `foci` option.
 
-The variable should be a list of servers serving as MatrixRTC endpoints to serve in the well-known file to the client.
+The variable should be a list of servers serving as MatrixRTC endpoints. Clients discover these via the `/_matrix/client/v1/rtc/transports` endpoint (MSC4143).
 
 ```toml
-rtc_focus_server_urls = [
+[global.matrix_rtc]
+foci = [
     { type = "livekit", livekit_service_url = "https://livekit.example.com" },
 ]
 ```
 
 Remember to replace the URL with the address you are deploying your instance of lk-jwt-service to.
 
-#### Serving .well-known manually
-
-If you don't let Continuwuity serve your `.well-known` files, you need to add the following lines to your `.well-known/matrix/client` file, remembering to replace the URL with your own `lk-jwt-service` deployment:
-
-```json
-  "org.matrix.msc4143.rtc_foci": [
-    {
-      "type": "livekit",
-      "livekit_service_url": "https://livekit.example.com"
-    }
-  ]
-```
-
-The final file should look something like this:
-
-```json
-{
-  "m.homeserver": {
-    "base_url":"https://matrix.example.com"
-  },
-  "org.matrix.msc4143.rtc_foci": [
-    {
-      "type": "livekit",
-      "livekit_service_url": "https://livekit.example.com"
-    }
-  ]
-}
-```
-
 ### 4. Configure your Reverse Proxy
 
 Reverse proxies can be configured in many different ways - so we can't provide a step by step for this.

docs/introduction.mdx

@@ -51,7 +51,13 @@ ## Can I try it out?
 
 Check out the [documentation](https://continuwuity.org) for installation instructions.
 
-There are currently no open registration continuwuity instances available.
+If you want to try it out as a user, we have some partnered homeservers you can use:
+* You can head over to [https://federated.nexus](https://federated.nexus/) in your browser.
+  * Hit the `Apply to Join` button. Once your request has been accepted, you will receive an email with your username and password.
+  * Head over to [https://app.federated.nexus](https://app.federated.nexus/) and you can sign in there, or use any other matrix chat client you wish elsewhere.
+  * Your username for matrix will be in the form of `@username:federated.nexus`, however you can simply use the `username` part to log in. Your password is your password.
+
+* There's also [https://continuwuity.rocks/](https://continuwuity.rocks/). You can register a new account using Cinny via [this convenient link](https://app.cinny.in/register/continuwuity.rocks), or you can use Element or another matrix client *that supports registration*.
 
 ## What are we working on?
 

docs/reference/admin/media.md

@@ -36,3 +36,7 @@ ## `!admin media delete-all-from-user`
 ## `!admin media delete-all-from-server`
 
 Deletes all remote media from the specified remote server. This will always ignore errors by default
+
+## `!admin media delete-url-preview`
+
+Deletes a cached URL preview, forcing it to be re-fetched. Use --all to purge all cached URL previews

nix/packages/uwulib/build.nix

@@ -77,7 +77,12 @@ rec {
     craneLib.buildDepsOnly (
       (commonAttrs commonAttrsArgs)
       // {
-        env = uwuenv.buildDepsOnlyEnv // (makeRocksDBEnv { inherit rocksdb; });
+        env = uwuenv.buildDepsOnlyEnv
+              // (makeRocksDBEnv { inherit rocksdb; })
+              // {
+                # required since we started using unstable reqwest apparently ... otherwise the all-features build will fail
+                RUSTFLAGS = "--cfg reqwest_unstable";
+              };
         inherit (features) cargoExtraArgs;
       }
 
@@ -102,7 +107,13 @@ rec {
         '';
         cargoArtifacts = deps;
         doCheck = true;
-        env = uwuenv.buildPackageEnv // rocksdbEnv;
+        env =
+          uwuenv.buildPackageEnv
+          // rocksdbEnv
+          // {
+            # required since we started using unstable reqwest apparently ... otherwise the all-features build will fail
+            RUSTFLAGS = "--cfg reqwest_unstable";
+          };
         passthru.env = uwuenv.buildPackageEnv // rocksdbEnv;
         meta.mainProgram = crateInfo.pname;
         inherit (features) cargoExtraArgs;

src/admin/federation/commands.rs

@@ -30,12 +30,15 @@ pub(super) async fn incoming_federation(&self) -> Result {
 			.federation_handletime
 			.read();
 
-		let mut msg = format!("Handling {} incoming pdus:\n", map.len());
+		let mut msg = format!(
+			"Handling {} incoming PDUs across {} active transactions:\n",
+			map.len(),
+			self.services.transactions.txn_active_handle_count()
+		);
 		for (r, (e, i)) in map.iter() {
 			let elapsed = i.elapsed();
 			writeln!(msg, "{} {}: {}m{}s", r, e, elapsed.as_secs() / 60, elapsed.as_secs() % 60)?;
 		}
-
 		msg
 	};
 

src/admin/media/commands.rs

@@ -29,7 +29,9 @@ pub(super) async fn delete(
 			.delete(&mxc.as_str().try_into()?)
 			.await?;
 
-		return Err!("Deleted the MXC from our database and on our filesystem.",);
+		return self
+			.write_str("Deleted the MXC from our database and on our filesystem.")
+			.await;
 	}
 
 	if let Some(event_id) = event_id {
@@ -388,3 +390,19 @@ pub(super) async fn get_remote_thumbnail(
 	self.write_str(&format!("```\n{result:#?}\nreceived {len} bytes for file content.\n```"))
 		.await
 }
+
+#[admin_command]
+pub(super) async fn delete_url_preview(&self, url: Option<String>, all: bool) -> Result {
+	if all {
+		self.services.media.clear_url_previews().await;
+
+		return self.write_str("Deleted all cached URL previews.").await;
+	}
+
+	let url = url.expect("clap enforces url is required unless --all");
+
+	self.services.media.remove_url_preview(&url).await?;
+
+	self.write_str(&format!("Deleted cached URL preview for: {url}"))
+		.await
+}

src/admin/media/mod.rs

@@ -108,4 +108,16 @@ pub enum MediaCommand {
 		#[arg(long, default_value("800"))]
 		height: u32,
 	},
+
+	/// Deletes a cached URL preview, forcing it to be re-fetched.
+	/// Use --all to purge all cached URL previews.
+	DeleteUrlPreview {
+		/// The URL to clear from the saved preview data
+		#[arg(required_unless_present = "all")]
+		url: Option<String>,
+
+		/// Purge all cached URL previews
+		#[arg(long, conflicts_with = "url")]
+		all: bool,
+	},
 }

src/admin/query/raw.rs

@@ -209,7 +209,7 @@ pub(super) async fn compact(
 	let parallelism = parallelism.unwrap_or(1);
 	let results = maps
 		.into_iter()
-		.try_stream()
+		.try_stream::<conduwuit::Error>()
 		.paralleln_and_then(runtime, parallelism, move |map| {
 			map.compact_blocking(options.clone())?;
 			Ok(map.name().to_owned())

src/admin/query/resolver.rs

@@ -20,7 +20,17 @@ pub enum ResolverCommand {
 		name: Option<String>,
 	},
 
-	/// Flush a specific server from the resolver caches or everything
+	/// Flush a given server from the resolver caches or flush them completely
+	///
+	/// * Examples:
+	///   * Flush a specific server:
+	///
+	///     `!admin query resolver flush-cache matrix.example.com`
+	///
+	///   * Flush all resolver caches completely:
+	///
+	///     `!admin query resolver flush-cache --all`
+	#[command(verbatim_doc_comment)]
 	FlushCache {
 		name: Option<OwnedServerName>,
 

src/api/client/account.rs

@@ -252,6 +252,13 @@ pub(crate) async fn register_route(
 						}
 					}
 
+					// Don't allow registration with user IDs that aren't local
+					if !services.globals.user_is_local(&user_id) {
+						return Err!(Request(InvalidUsername(
+							"Username {body_username} is not local to this server"
+						)));
+					}
+
 					user_id
 				},
 				| Err(e) => {

src/api/client/account_data.rs

@@ -9,7 +9,7 @@
 	},
 	events::{
 		AnyGlobalAccountDataEventContent, AnyRoomAccountDataEventContent,
-		GlobalAccountDataEventType, RoomAccountDataEventType,
+		RoomAccountDataEventType,
 	},
 	serde::Raw,
 };
@@ -126,12 +126,6 @@ async fn set_account_data(
 		)));
 	}
 
-	if event_type_s == GlobalAccountDataEventType::PushRules.to_cow_str() {
-		return Err!(Request(BadJson(
-			"This endpoint cannot be used for setting/configuring push rules."
-		)));
-	}
-
 	let data: serde_json::Value = serde_json::from_str(data.get())
 		.map_err(|e| err!(Request(BadJson(warn!("Invalid JSON provided: {e}")))))?;
 

src/api/client/media.rs

@@ -6,6 +6,7 @@
 	Err, Result, err,
 	utils::{self, content_disposition::make_content_disposition, math::ruma_from_usize},
 };
+use conduwuit_core::error;
 use conduwuit_service::{
 	Services,
 	media::{CACHE_CONTROL_IMMUTABLE, CORP_CROSS_ORIGIN, Dim, FileMeta, MXC_LENGTH},
@@ -144,12 +145,22 @@ pub(crate) async fn get_content_route(
 		server_name: &body.server_name,
 		media_id: &body.media_id,
 	};
-
 	let FileMeta {
 		content,
 		content_type,
 		content_disposition,
-	} = fetch_file(&services, &mxc, user, body.timeout_ms, None).await?;
+	} = match fetch_file(&services, &mxc, user, body.timeout_ms, None).await {
+		| Ok(meta) => meta,
+		| Err(conduwuit::Error::Io(e)) => match e.kind() {
+			| std::io::ErrorKind::NotFound => return Err!(Request(NotFound("Media not found."))),
+			| std::io::ErrorKind::PermissionDenied => {
+				error!("Permission denied when trying to read file: {e:?}");
+				return Err!(Request(Unknown("Unknown error when fetching file.")));
+			},
+			| _ => return Err!(Request(Unknown("Unknown error when fetching file."))),
+		},
+		| Err(_) => return Err!(Request(Unknown("Unknown error when fetching file."))),
+	};
 
 	Ok(get_content::v1::Response {
 		file: content.expect("entire file contents"),
@@ -185,7 +196,18 @@ pub(crate) async fn get_content_as_filename_route(
 		content,
 		content_type,
 		content_disposition,
-	} = fetch_file(&services, &mxc, user, body.timeout_ms, Some(&body.filename)).await?;
+	} = match fetch_file(&services, &mxc, user, body.timeout_ms, None).await {
+		| Ok(meta) => meta,
+		| Err(conduwuit::Error::Io(e)) => match e.kind() {
+			| std::io::ErrorKind::NotFound => return Err!(Request(NotFound("Media not found."))),
+			| std::io::ErrorKind::PermissionDenied => {
+				error!("Permission denied when trying to read file: {e:?}");
+				return Err!(Request(Unknown("Unknown error when fetching file.")));
+			},
+			| _ => return Err!(Request(Unknown("Unknown error when fetching file."))),
+		},
+		| Err(_) => return Err!(Request(Unknown("Unknown error when fetching file."))),
+	};
 
 	Ok(get_content_as_filename::v1::Response {
 		file: content.expect("entire file contents"),

src/api/client/report.rs

@@ -4,7 +4,6 @@
 use axum_client_ip::InsecureClientIp;
 use conduwuit::{Err, Event, Result, debug_info, info, matrix::pdu::PduEvent, utils::ReadyExt};
 use conduwuit_service::Services;
-use rand::Rng;
 use ruma::{
 	EventId, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UserId,
 	api::client::{
@@ -244,7 +243,7 @@ fn build_report(report: Report) -> RoomMessageEventContent {
 /// random delay sending a response per spec suggestion regarding
 /// enumerating for potential events existing in our server.
 async fn delay_response() {
-	let time_to_wait = rand::thread_rng().gen_range(2..5);
+	let time_to_wait = rand::random_range(2..5);
 	debug_info!(
 		"Got successful /report request, waiting {time_to_wait} seconds before sending \
 		 successful response."

src/api/client/send.rs

@@ -50,8 +50,8 @@ pub(crate) async fn send_message_event_route(
 
 	// Check if this is a new transaction id
 	if let Ok(response) = services
-		.transaction_ids
-		.existing_txnid(sender_user, sender_device, &body.txn_id)
+		.transactions
+		.get_client_txn(sender_user, sender_device, &body.txn_id)
 		.await
 	{
 		// The client might have sent a txnid of the /sendToDevice endpoint
@@ -92,7 +92,7 @@ pub(crate) async fn send_message_event_route(
 		)
 		.await?;
 
-	services.transaction_ids.add_txnid(
+	services.transactions.add_client_txnid(
 		sender_user,
 		sender_device,
 		&body.txn_id,

src/api/client/sync/v5.rs

@@ -336,7 +336,9 @@ async fn handle_lists<'a, Rooms, AllRooms>(
 		let ranges = list.ranges.clone();
 
 		for mut range in ranges {
-			range.0 = uint!(0);
+			range.0 = range
+				.0
+				.min(UInt::try_from(active_rooms.len()).unwrap_or(UInt::MAX));
 			range.1 = range.1.checked_add(uint!(1)).unwrap_or(range.1);
 			range.1 = range
 				.1

src/api/client/to_device.rs

@@ -26,8 +26,8 @@ pub(crate) async fn send_event_to_device_route(
 
 	// Check if this is a new transaction id
 	if services
-		.transaction_ids
-		.existing_txnid(sender_user, sender_device, &body.txn_id)
+		.transactions
+		.get_client_txn(sender_user, sender_device, &body.txn_id)
 		.await
 		.is_ok()
 	{
@@ -104,8 +104,8 @@ pub(crate) async fn send_event_to_device_route(
 
 	// Save transaction id with empty data
 	services
-		.transaction_ids
-		.add_txnid(sender_user, sender_device, &body.txn_id, &[]);
+		.transactions
+		.add_client_txnid(sender_user, sender_device, &body.txn_id, &[]);
 
 	Ok(send_event_to_device::v3::Response {})
 }

src/api/client/well_known.rs

@@ -27,10 +27,32 @@ pub(crate) async fn well_known_client(
 		identity_server: None,
 		sliding_sync_proxy: Some(SlidingSyncProxyInfo { url: client_url }),
 		tile_server: None,
-		rtc_foci: services.config.well_known.rtc_focus_server_urls.clone(),
+		rtc_foci: services
+			.config
+			.matrix_rtc
+			.effective_foci(&services.config.well_known.rtc_focus_server_urls)
+			.to_vec(),
 	})
 }
 
+/// # `GET /_matrix/client/v1/rtc/transports`
+/// # `GET /_matrix/client/unstable/org.matrix.msc4143/rtc/transports`
+///
+/// Returns the list of MatrixRTC foci (transports) configured for this
+/// homeserver, implementing MSC4143.
+pub(crate) async fn get_rtc_transports(
+	State(services): State<crate::State>,
+	_body: Ruma<ruma::api::client::discovery::get_rtc_transports::Request>,
+) -> Result<ruma::api::client::discovery::get_rtc_transports::Response> {
+	Ok(ruma::api::client::discovery::get_rtc_transports::Response::new(
+		services
+			.config
+			.matrix_rtc
+			.effective_foci(&services.config.well_known.rtc_focus_server_urls)
+			.to_vec(),
+	))
+}
+
 /// # `GET /.well-known/matrix/support`
 ///
 /// Server support contact and support page of a homeserver's domain.

src/api/router.rs

@@ -184,6 +184,7 @@ pub fn build(router: Router<State>, server: &Server) -> Router<State> {
 		.ruma_route(&client::put_suspended_status)
 		.ruma_route(&client::well_known_support)
 		.ruma_route(&client::well_known_client)
+		.ruma_route(&client::get_rtc_transports)
 		.route("/_conduwuit/server_version", get(client::conduwuit_server_version))
 		.route("/_continuwuity/server_version", get(client::conduwuit_server_version))
 		.ruma_route(&client::room_initial_sync_route)

src/api/router/auth.rs

@@ -14,7 +14,8 @@
 	pin_mut,
 };
 use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, OwnedDeviceId, OwnedServerName, OwnedUserId, UserId,
+	CanonicalJsonObject, CanonicalJsonValue, DeviceId, OwnedDeviceId, OwnedServerName,
+	OwnedUserId, UserId,
 	api::{
 		AuthScheme, IncomingRequest, Metadata,
 		client::{
@@ -66,23 +67,17 @@ pub(super) async fn auth(
 	if metadata.authentication == AuthScheme::None {
 		match metadata {
 			| &get_public_rooms::v3::Request::METADATA => {
-				if !services
-					.server
-					.config
-					.allow_public_room_directory_without_auth
-				{
-					match token {
-						| Token::Appservice(_) | Token::User(_) => {
-							// we should have validated the token above
-							// already
-						},
-						| Token::None | Token::Invalid => {
-							return Err(Error::BadRequest(
-								ErrorKind::MissingToken,
-								"Missing or invalid access token.",
-							));
-						},
-					}
+				match token {
+					| Token::Appservice(_) | Token::User(_) => {
+						// we should have validated the token above
+						// already
+					},
+					| Token::None | Token::Invalid => {
+						return Err(Error::BadRequest(
+							ErrorKind::MissingToken,
+							"Missing or invalid access token.",
+						));
+					},
 				}
 			},
 			| &get_profile::v3::Request::METADATA
@@ -234,10 +229,33 @@ async fn auth_appservice(
 		return Err!(Request(Exclusive("User is not in namespace.")));
 	}
 
+	// MSC3202/MSC4190: Handle device_id masquerading for appservices.
+	// The device_id can be provided via `device_id` or
+	// `org.matrix.msc3202.device_id` query parameter.
+	let sender_device = if let Some(ref device_id_str) = request.query.device_id {
+		let device_id: &DeviceId = device_id_str.as_str().into();
+
+		// Verify the device exists for this user
+		if services
+			.users
+			.get_device_metadata(&user_id, device_id)
+			.await
+			.is_err()
+		{
+			return Err!(Request(Forbidden(
+				"Device does not exist for user or appservice cannot masquerade as this device."
+			)));
+		}
+
+		Some(device_id.to_owned())
+	} else {
+		None
+	};
+
 	Ok(Auth {
 		origin: None,
 		sender_user: Some(user_id),
-		sender_device: None,
+		sender_device,
 		appservice_info: Some(*info),
 	})
 }

src/api/router/request.rs

@@ -11,6 +11,10 @@
 pub(super) struct QueryParams {
 	pub(super) access_token: Option<String>,
 	pub(super) user_id: Option<String>,
+	/// Device ID for appservice device masquerading (MSC3202/MSC4190).
+	/// Can be provided as `device_id` or `org.matrix.msc3202.device_id`.
+	#[serde(alias = "org.matrix.msc3202.device_id")]
+	pub(super) device_id: Option<String>,
 }
 
 pub(super) struct Request {

src/api/server/query.rs

@@ -40,7 +40,7 @@ pub(crate) async fn get_room_information_route(
 	servers.sort_unstable();
 	servers.dedup();
 
-	servers.shuffle(&mut rand::thread_rng());
+	servers.shuffle(&mut rand::rng());
 
 	// insert our server as the very first choice if in list
 	if let Some(server_index) = servers

src/api/server/send.rs

@@ -1,27 +1,33 @@
-use std::{collections::BTreeMap, net::IpAddr, time::Instant};
+use std::{
+	collections::{BTreeMap, HashMap, HashSet},
+	net::IpAddr,
+	time::{Duration, Instant},
+};
 
 use axum::extract::State;
 use axum_client_ip::InsecureClientIp;
 use conduwuit::{
 	Err, Error, Result, debug, debug_warn, err, error,
 	result::LogErr,
+	state_res::lexicographical_topological_sort,
 	trace,
 	utils::{
 		IterStream, ReadyExt, millis_since_unix_epoch,
 		stream::{BroadbandExt, TryBroadbandExt, automatic_width},
 	},
-	warn,
 };
 use conduwuit_service::{
 	Services,
 	sending::{EDU_LIMIT, PDU_LIMIT},
 };
 use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
+use http::StatusCode;
 use itertools::Itertools;
 use ruma::{
-	CanonicalJsonObject, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, ServerName, UserId,
+	CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId,
+	RoomId, ServerName, UserId,
 	api::{
-		client::error::ErrorKind,
+		client::error::{ErrorKind, ErrorKind::LimitExceeded},
 		federation::transactions::{
 			edu::{
 				DeviceListUpdateContent, DirectDeviceContent, Edu, PresenceContent,
@@ -32,9 +38,16 @@
 		},
 	},
 	events::receipt::{ReceiptEvent, ReceiptEventContent, ReceiptType},
+	int,
 	serde::Raw,
 	to_device::DeviceIdOrAllDevices,
+	uint,
 };
+use service::transactions::{
+	FederationTxnState, TransactionError, TxnKey, WrappedTransactionResponse,
+};
+use tokio::sync::watch::{Receiver, Sender};
+use tracing::instrument;
 
 use crate::Ruma;
 
@@ -44,15 +57,6 @@
 /// # `PUT /_matrix/federation/v1/send/{txnId}`
 ///
 /// Push EDUs and PDUs to this server.
-#[tracing::instrument(
-	name = "txn",
-	level = "debug",
-	skip_all,
-	fields(
-		%client,
-		origin = body.origin().as_str()
-	),
-)]
 pub(crate) async fn send_transaction_message_route(
 	State(services): State<crate::State>,
 	InsecureClientIp(client): InsecureClientIp,
@@ -76,16 +80,73 @@ pub(crate) async fn send_transaction_message_route(
 		)));
 	}
 
-	let txn_start_time = Instant::now();
-	trace!(
-		pdus = body.pdus.len(),
-		edus = body.edus.len(),
-		elapsed = ?txn_start_time.elapsed(),
-		id = %body.transaction_id,
-		origin = %body.origin(),
-		"Starting txn",
-	);
+	let txn_key = (body.origin().to_owned(), body.transaction_id.clone());
 
+	// Atomically check cache, join active, or start new transaction
+	match services
+		.transactions
+		.get_or_start_federation_txn(txn_key.clone())?
+	{
+		| FederationTxnState::Cached(response) => {
+			// Already responded
+			Ok(response)
+		},
+		| FederationTxnState::Active(receiver) => {
+			// Another thread is processing
+			wait_for_result(receiver).await
+		},
+		| FederationTxnState::Started { receiver, sender } => {
+			// We're the first, spawn the processing task
+			services
+				.server
+				.runtime()
+				.spawn(process_inbound_transaction(services, body, client, txn_key, sender));
+			// and wait for it
+			wait_for_result(receiver).await
+		},
+	}
+}
+
+async fn wait_for_result(
+	mut recv: Receiver<WrappedTransactionResponse>,
+) -> Result<send_transaction_message::v1::Response> {
+	if tokio::time::timeout(Duration::from_secs(50), recv.changed())
+		.await
+		.is_err()
+	{
+		// Took too long, return 429 to encourage the sender to try again
+		return Err(Error::BadRequest(
+			LimitExceeded { retry_after: None },
+			"Transaction is being still being processed. Please try again later.",
+		));
+	}
+	let value = recv.borrow_and_update();
+	match value.clone() {
+		| Some(Ok(response)) => Ok(response),
+		| Some(Err(err)) => Err(transaction_error_to_response(&err)),
+		| None => Err(Error::Request(
+			ErrorKind::Unknown,
+			"Transaction processing failed unexpectedly".into(),
+			StatusCode::INTERNAL_SERVER_ERROR,
+		)),
+	}
+}
+
+#[instrument(
+	skip_all,
+	fields(
+		id = ?body.transaction_id.as_str(),
+		origin = ?body.origin()
+	)
+)]
+async fn process_inbound_transaction(
+	services: crate::State,
+	body: Ruma<send_transaction_message::v1::Request>,
+	client: IpAddr,
+	txn_key: TxnKey,
+	sender: Sender<WrappedTransactionResponse>,
+) {
+	let txn_start_time = Instant::now();
 	let pdus = body
 		.pdus
 		.iter()
@@ -102,40 +163,79 @@ 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?;
+	debug!(pdus = body.pdus.len(), edus = body.edus.len(), "Processing transaction",);
+	let results = match handle(&services, &client, body.origin(), pdus, edus).await {
+		| Ok(results) => results,
+		| Err(err) => {
+			fail_federation_txn(services, &txn_key, &sender, err);
+			return;
+		},
+	};
+
+	for (id, result) in &results {
+		if let Err(e) = result {
+			if matches!(e, Error::BadRequest(ErrorKind::NotFound, _)) {
+				debug_warn!("Incoming PDU failed {id}: {e:?}");
+			}
+		}
+	}
 
 	debug!(
 		pdus = body.pdus.len(),
 		edus = body.edus.len(),
 		elapsed = ?txn_start_time.elapsed(),
-		id = %body.transaction_id,
-		origin = %body.origin(),
-		"Finished txn",
+		"Finished processing transaction"
 	);
-	for (id, result) in &results {
-		if let Err(e) = result {
-			if matches!(e, Error::BadRequest(ErrorKind::NotFound, _)) {
-				warn!("Incoming PDU failed {id}: {e:?}");
-			}
-		}
-	}
 
-	Ok(send_transaction_message::v1::Response {
+	let response = send_transaction_message::v1::Response {
 		pdus: results
 			.into_iter()
 			.map(|(e, r)| (e, r.map_err(error::sanitized_message)))
 			.collect(),
-	})
+	};
+
+	services
+		.transactions
+		.finish_federation_txn(txn_key, sender, response);
 }
 
+/// Handles a failed federation transaction by sending the error through
+/// the channel and cleaning up the transaction state. This allows waiters to
+/// receive an appropriate error response.
+fn fail_federation_txn(
+	services: crate::State,
+	txn_key: &TxnKey,
+	sender: &Sender<WrappedTransactionResponse>,
+	err: TransactionError,
+) {
+	debug!("Transaction failed: {err}");
+
+	// Remove from active state so the transaction can be retried
+	services.transactions.remove_federation_txn(txn_key);
+
+	// Send the error to any waiters
+	if let Err(e) = sender.send(Some(Err(err))) {
+		debug_warn!("Failed to send transaction error to receivers: {e}");
+	}
+}
+
+/// Converts a TransactionError into an appropriate HTTP error response.
+fn transaction_error_to_response(err: &TransactionError) -> Error {
+	match err {
+		| TransactionError::ShuttingDown => Error::Request(
+			ErrorKind::Unknown,
+			"Server is shutting down, please retry later".into(),
+			StatusCode::SERVICE_UNAVAILABLE,
+		),
+	}
+}
 async fn handle(
 	services: &Services,
 	client: &IpAddr,
 	origin: &ServerName,
-	started: Instant,
 	pdus: impl Stream<Item = Pdu> + Send,
 	edus: impl Stream<Item = Edu> + Send,
-) -> Result<ResolvedMap> {
+) -> std::result::Result<ResolvedMap, TransactionError> {
 	// group pdus by room
 	let pdus = pdus
 		.collect()
@@ -152,7 +252,7 @@ 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())
+			handle_room(services, client, origin, room_id, pdus.into_iter())
 				.map_ok(Vec::into_iter)
 				.map_ok(IterStream::try_stream)
 		})
@@ -169,14 +269,51 @@ async fn handle(
 	Ok(results)
 }
 
+/// Attempts to build a localised directed acyclic graph out of the given PDUs,
+/// returning them in a topologically sorted order.
+///
+/// This is used to attempt to process PDUs in an order that respects their
+/// dependencies, however it is ultimately the sender's responsibility to send
+/// them in a processable order, so this is just a best effort attempt. It does
+/// not account for power levels or other tie breaks.
+async fn build_local_dag(
+	pdu_map: &HashMap<OwnedEventId, CanonicalJsonObject>,
+) -> Result<Vec<OwnedEventId>> {
+	debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs");
+	let mut dag: HashMap<OwnedEventId, HashSet<OwnedEventId>> = HashMap::new();
+
+	for (event_id, value) in pdu_map {
+		let prev_events = value
+			.get("prev_events")
+			.expect("pdu must have prev_events")
+			.as_array()
+			.expect("prev_events must be an array")
+			.iter()
+			.map(|v| {
+				OwnedEventId::parse(v.as_str().expect("prev_events values must be strings"))
+					.expect("prev_events must be valid event IDs")
+			})
+			.collect::<HashSet<OwnedEventId>>();
+
+		dag.insert(event_id.clone(), prev_events);
+	}
+	lexicographical_topological_sort(&dag, &|_| async {
+		// Note: we don't bother fetching power levels because that would massively slow
+		// this function down. This is a best-effort attempt to order events correctly
+		// for processing, however ultimately that should be the sender's job.
+		Ok((int!(0), MilliSecondsSinceUnixEpoch(uint!(0))))
+	})
+	.await
+	.map_err(|e| err!("failed to resolve local graph: {e}"))
+}
+
 async fn handle_room(
 	services: &Services,
 	_client: &IpAddr,
 	origin: &ServerName,
-	txn_start_time: Instant,
 	room_id: OwnedRoomId,
 	pdus: impl Iterator<Item = Pdu> + Send,
-) -> Result<Vec<(OwnedEventId, Result)>> {
+) -> std::result::Result<Vec<(OwnedEventId, Result)>, TransactionError> {
 	let _room_lock = services
 		.rooms
 		.event_handler
@@ -185,27 +322,40 @@ async fn handle_room(
 		.await;
 
 	let room_id = &room_id;
-	pdus.try_stream()
-		.and_then(|(_, event_id, value)| async move {
-			services.server.check_running()?;
-			let pdu_start_time = Instant::now();
-			let result = services
-				.rooms
-				.event_handler
-				.handle_incoming_pdu(origin, room_id, &event_id, value, true)
-				.await
-				.map(|_| ());
-
-			debug!(
-				pdu_elapsed = ?pdu_start_time.elapsed(),
-				txn_elapsed = ?txn_start_time.elapsed(),
-				"Finished PDU {event_id}",
-			);
-
-			Ok((event_id, result))
+	let pdu_map: HashMap<OwnedEventId, CanonicalJsonObject> = pdus
+		.into_iter()
+		.map(|(_, event_id, value)| (event_id, value))
+		.collect();
+	// Try to sort PDUs by their dependencies, but fall back to arbitrary order on
+	// failure (e.g., cycles). This is best-effort; proper ordering is the sender's
+	// responsibility.
+	let sorted_event_ids = if pdu_map.len() >= 2 {
+		build_local_dag(&pdu_map).await.unwrap_or_else(|e| {
+			debug_warn!("Failed to build local DAG for room {room_id}: {e}");
+			pdu_map.keys().cloned().collect()
 		})
-		.try_collect()
-		.await
+	} else {
+		pdu_map.keys().cloned().collect()
+	};
+	let mut results = Vec::with_capacity(sorted_event_ids.len());
+	for event_id in sorted_event_ids {
+		let value = pdu_map
+			.get(&event_id)
+			.expect("sorted event IDs must be from the original map")
+			.clone();
+		services
+			.server
+			.check_running()
+			.map_err(|_| TransactionError::ShuttingDown)?;
+		let result = services
+			.rooms
+			.event_handler
+			.handle_incoming_pdu(origin, room_id, &event_id, value, true)
+			.await
+			.map(|_| ());
+		results.push((event_id, result));
+	}
+	Ok(results)
 }
 
 async fn handle_edu(services: &Services, client: &IpAddr, origin: &ServerName, edu: Edu) {
@@ -478,8 +628,8 @@ async fn handle_edu_direct_to_device(
 
 	// Check if this is a new transaction id
 	if services
-		.transaction_ids
-		.existing_txnid(sender, None, message_id)
+		.transactions
+		.get_client_txn(sender, None, message_id)
 		.await
 		.is_ok()
 	{
@@ -498,8 +648,8 @@ async fn handle_edu_direct_to_device(
 
 	// Save transaction id with empty data
 	services
-		.transaction_ids
-		.add_txnid(sender, None, message_id, &[]);
+		.transactions
+		.add_client_txnid(sender, None, message_id, &[]);
 }
 
 async fn handle_edu_direct_to_device_user<Event: Send + Sync>(

src/core/Cargo.toml

@@ -86,6 +86,7 @@ libloading.optional = true
 log.workspace = true
 num-traits.workspace = true
 rand.workspace = true
+rand_core = { version = "0.6.4", features = ["getrandom"] }
 regex.workspace = true
 reqwest.workspace = true
 ring.workspace = true

src/core/config/mod.rs

@@ -368,6 +368,31 @@ pub struct Config {
 	#[serde(default = "default_max_fetch_prev_events")]
 	pub max_fetch_prev_events: u16,
 
+	/// How many incoming federation transactions the server is willing to be
+	/// processing at any given time before it becomes overloaded and starts
+	/// rejecting further transactions until some slots become available.
+	///
+	/// Setting this value too low or too high may result in unstable
+	/// federation, and setting it too high may cause runaway resource usage.
+	///
+	/// default: 150
+	#[serde(default = "default_max_concurrent_inbound_transactions")]
+	pub max_concurrent_inbound_transactions: usize,
+
+	/// Maximum age (in seconds) for cached federation transaction responses.
+	/// Entries older than this will be removed during cleanup.
+	///
+	/// default: 7200 (2 hours)
+	#[serde(default = "default_transaction_id_cache_max_age_secs")]
+	pub transaction_id_cache_max_age_secs: u64,
+
+	/// Maximum number of cached federation transaction responses.
+	/// When the cache exceeds this limit, older entries will be removed.
+	///
+	/// default: 8192
+	#[serde(default = "default_transaction_id_cache_max_entries")]
+	pub transaction_id_cache_max_entries: usize,
+
 	/// Default/base connection timeout (seconds). This is used only by URL
 	/// previews and update/news endpoint checks.
 	///
@@ -653,12 +678,6 @@ pub struct Config {
 	#[serde(default)]
 	pub allow_public_room_directory_over_federation: bool,
 
-	/// Set this to true to allow your server's public room directory to be
-	/// queried without client authentication (access token) through the Client
-	/// APIs. Set this to false to protect against /publicRooms spiders.
-	#[serde(default)]
-	pub allow_public_room_directory_without_auth: bool,
-
 	/// Allow guests/unauthenticated users to access TURN credentials.
 	///
 	/// This is the equivalent of Synapse's `turn_allow_guests` config option.
@@ -1525,7 +1544,7 @@ pub struct Config {
 	/// sender user's server name, inbound federation X-Matrix origin, and
 	/// outbound federation handler.
 	///
-	/// You can set this to ["*"] to block all servers by default, and then
+	/// You can set this to [".*"] to block all servers by default, and then
 	/// use `allowed_remote_server_names` to allow only specific servers.
 	///
 	/// example: ["badserver\\.tld$", "badphrase", "19dollarfortnitecards"]
@@ -2049,6 +2068,16 @@ pub struct Config {
 	pub allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure:
 		bool,
 
+	/// Forcibly disables first-run mode.
+	///
+	/// This is intended to be used for Complement testing to allow the test
+	/// suite to register users, because first-run mode interferes with open
+	/// registration.
+	///
+	/// display: hidden
+	#[serde(default)]
+	pub force_disable_first_run_mode: bool,
+
 	/// display: nested
 	#[serde(default)]
 	pub ldap: LdapConfig,
@@ -2061,6 +2090,12 @@ pub struct Config {
 	/// display: nested
 	#[serde(default)]
 	pub blurhashing: BlurhashConfig,
+
+	/// Configuration for MatrixRTC (MSC4143) transport discovery.
+	/// display: nested
+	#[serde(default)]
+	pub matrix_rtc: MatrixRtcConfig,
+
 	#[serde(flatten)]
 	#[allow(clippy::zero_sized_map_values)]
 	// this is a catchall, the map shouldn't be zero at runtime
@@ -2126,17 +2161,16 @@ pub struct WellKnownConfig {
 	/// listed.
 	pub support_mxid: Option<OwnedUserId>,
 
-	/// A list of MatrixRTC foci URLs which will be served as part of the
-	/// MSC4143 client endpoint at /.well-known/matrix/client.  If you're
-	/// setting up livekit, you'd want something like:
-	/// rtc_focus_server_urls = [
-	///     { type = "livekit", livekit_service_url = "https://livekit.example.com" },
-	/// ]
+	/// **DEPRECATED**: Use `[global.matrix_rtc].foci` instead.
 	///
-	/// To disable, set this to be an empty vector (`[]`).
+	/// A list of MatrixRTC foci URLs which will be served as part of the
+	/// MSC4143 client endpoint at /.well-known/matrix/client.
+	///
+	/// This option is deprecated and will be removed in a future release.
+	/// Please migrate to the new `[global.matrix_rtc]` config section.
 	///
 	/// default: []
-	#[serde(default = "default_rtc_focus_urls")]
+	#[serde(default)]
 	pub rtc_focus_server_urls: Vec<RtcFocusInfo>,
 }
 
@@ -2165,6 +2199,43 @@ pub struct BlurhashConfig {
 	pub blurhash_max_raw_size: u64,
 }
 
+#[derive(Clone, Debug, Deserialize, Default)]
+#[config_example_generator(filename = "conduwuit-example.toml", section = "global.matrix_rtc")]
+pub struct MatrixRtcConfig {
+	/// A list of MatrixRTC foci (transports) which will be served via the
+	/// MSC4143 RTC transports endpoint at
+	/// `/_matrix/client/v1/rtc/transports`. If you're setting up livekit,
+	/// you'd want something like:
+	/// ```toml
+	/// [global.matrix_rtc]
+	/// foci = [
+	///     { type = "livekit", livekit_service_url = "https://livekit.example.com" },
+	/// ]
+	/// ```
+	///
+	/// To disable, set this to an empty list (`[]`).
+	///
+	/// default: []
+	#[serde(default)]
+	pub foci: Vec<RtcFocusInfo>,
+}
+
+impl MatrixRtcConfig {
+	/// Returns the effective foci, falling back to the deprecated
+	/// `rtc_focus_server_urls` if the new config is empty.
+	#[must_use]
+	pub fn effective_foci<'a>(
+		&'a self,
+		deprecated_foci: &'a [RtcFocusInfo],
+	) -> &'a [RtcFocusInfo] {
+		if !self.foci.is_empty() {
+			&self.foci
+		} else {
+			deprecated_foci
+		}
+	}
+}
+
 #[derive(Clone, Debug, Default, Deserialize)]
 #[config_example_generator(filename = "conduwuit-example.toml", section = "global.ldap")]
 pub struct LdapConfig {
@@ -2358,6 +2429,7 @@ pub struct DraupnirConfig {
 	"well_known_support_email",
 	"well_known_support_mxid",
 	"registration_token_file",
+	"well_known.rtc_focus_server_urls",
 ];
 
 impl Config {
@@ -2540,6 +2612,12 @@ fn default_pusher_idle_timeout() -> u64 { 15 }
 
 fn default_max_fetch_prev_events() -> u16 { 192_u16 }
 
+fn default_max_concurrent_inbound_transactions() -> usize { 150 }
+
+fn default_transaction_id_cache_max_age_secs() -> u64 { 60 * 60 * 2 }
+
+fn default_transaction_id_cache_max_entries() -> usize { 8192 }
+
 fn default_tracing_flame_filter() -> String {
 	cfg!(debug_assertions)
 		.then_some("trace,h2=off")
@@ -2635,9 +2713,6 @@ fn default_rocksdb_stats_level() -> u8 { 1 }
 #[inline]
 pub fn default_default_room_version() -> RoomVersionId { RoomVersionId::V11 }
 
-#[must_use]
-pub fn default_rtc_focus_urls() -> Vec<RtcFocusInfo> { vec![] }
-
 fn default_ip_range_denylist() -> Vec<String> {
 	vec![
 		"127.0.0.0/8".to_owned(),

src/core/info/version.rs

@@ -14,6 +14,7 @@
 static VERSION: OnceLock<String> = OnceLock::new();
 static VERSION_UA: OnceLock<String> = OnceLock::new();
 static USER_AGENT: OnceLock<String> = OnceLock::new();
+static USER_AGENT_MEDIA: OnceLock<String> = OnceLock::new();
 
 #[inline]
 #[must_use]
@@ -21,14 +22,22 @@ pub fn name() -> &'static str { BRANDING }
 
 #[inline]
 pub fn version() -> &'static str { VERSION.get_or_init(init_version) }
+
 #[inline]
 pub fn version_ua() -> &'static str { VERSION_UA.get_or_init(init_version_ua) }
 
 #[inline]
 pub fn user_agent() -> &'static str { USER_AGENT.get_or_init(init_user_agent) }
 
+#[inline]
+pub fn user_agent_media() -> &'static str { USER_AGENT_MEDIA.get_or_init(init_user_agent_media) }
+
 fn init_user_agent() -> String { format!("{}/{} (bot; +{WEBSITE})", name(), version_ua()) }
 
+fn init_user_agent_media() -> String {
+	format!("{}/{} (embedbot; facebookexternalhit/1.1; +{WEBSITE})", name(), version_ua())
+}
+
 fn init_version_ua() -> String {
 	conduwuit_build_metadata::version_tag()
 		.map_or_else(|| SEMANTIC.to_owned(), |extra| format!("{SEMANTIC}+{extra}"))

src/core/matrix/state_res/mod.rs

@@ -1046,7 +1046,7 @@ async fn test_event_sort() {
 		// don't remove any events so we know it sorts them all correctly
 		let mut events_to_sort = events.keys().cloned().collect::<Vec<_>>();
 
-		events_to_sort.shuffle(&mut rand::thread_rng());
+		events_to_sort.shuffle(&mut rand::rng());
 
 		let power_level = resolved_power
 			.get(&(StateEventType::RoomPowerLevels, "".into()))

src/core/utils/hash/argon.rs

@@ -28,7 +28,7 @@ fn init_argon() -> Argon2<'static> {
 }
 
 pub(super) fn password(password: &str) -> Result<String> {
-	let salt = SaltString::generate(rand::thread_rng());
+	let salt = SaltString::generate(rand_core::OsRng);
 	ARGON
 		.get_or_init(init_argon)
 		.hash_password(password.as_bytes(), &salt)

src/core/utils/rand.rs

@@ -4,16 +4,16 @@
 };
 
 use arrayvec::ArrayString;
-use rand::{Rng, seq::SliceRandom, thread_rng};
+use rand::{RngExt, seq::SliceRandom};
 
 pub fn shuffle<T>(vec: &mut [T]) {
-	let mut rng = thread_rng();
+	let mut rng = rand::rng();
 	vec.shuffle(&mut rng);
 }
 
 pub fn string(length: usize) -> String {
-	thread_rng()
-		.sample_iter(&rand::distributions::Alphanumeric)
+	rand::rng()
+		.sample_iter(&rand::distr::Alphanumeric)
 		.take(length)
 		.map(char::from)
 		.collect()
@@ -22,8 +22,8 @@ pub fn string(length: usize) -> String {
 #[inline]
 pub fn string_array<const LENGTH: usize>() -> ArrayString<LENGTH> {
 	let mut ret = ArrayString::<LENGTH>::new();
-	thread_rng()
-		.sample_iter(&rand::distributions::Alphanumeric)
+	rand::rng()
+		.sample_iter(&rand::distr::Alphanumeric)
 		.take(LENGTH)
 		.map(char::from)
 		.for_each(|c| ret.push(c));
@@ -40,7 +40,4 @@ pub fn time_from_now_secs(range: Range<u64>) -> SystemTime {
 }
 
 #[must_use]
-pub fn secs(range: Range<u64>) -> Duration {
-	let mut rng = thread_rng();
-	Duration::from_secs(rng.gen_range(range))
-}
+pub fn secs(range: Range<u64>) -> Duration { Duration::from_secs(rand::random_range(range)) }

src/core/utils/stream/iter_stream.rs

@@ -3,19 +3,17 @@
 	stream::{Stream, TryStream},
 };
 
-use crate::{Error, Result};
-
 pub trait IterStream<I: IntoIterator + Send> {
 	/// Convert an Iterator into a Stream
 	fn stream(self) -> impl Stream<Item = <I as IntoIterator>::Item> + Send;
 
-	/// Convert an Iterator into a TryStream
-	fn try_stream(
+	/// Convert an Iterator into a TryStream with a generic error type
+	fn try_stream<E>(
 		self,
 	) -> impl TryStream<
 		Ok = <I as IntoIterator>::Item,
-		Error = Error,
-		Item = Result<<I as IntoIterator>::Item, Error>,
+		Error = E,
+		Item = Result<<I as IntoIterator>::Item, E>,
 	> + Send;
 }
 
@@ -28,12 +26,12 @@ impl<I> IterStream<I> for I
 	fn stream(self) -> impl Stream<Item = <I as IntoIterator>::Item> + Send { stream::iter(self) }
 
 	#[inline]
-	fn try_stream(
+	fn try_stream<E>(
 		self,
 	) -> impl TryStream<
 		Ok = <I as IntoIterator>::Item,
-		Error = Error,
-		Item = Result<<I as IntoIterator>::Item, Error>,
+		Error = E,
+		Item = Result<<I as IntoIterator>::Item, E>,
 	> + Send {
 		self.stream().map(Ok)
 	}

src/core/utils/stream/try_broadband.rs

@@ -1,9 +1,10 @@
 //! Synchronous combinator extensions to futures::TryStream
 
+use std::result::Result;
+
 use futures::{TryFuture, TryStream, TryStreamExt};
 
 use super::automatic_width;
-use crate::Result;
 
 /// Concurrency extensions to augment futures::TryStreamExt. broad_ combinators
 /// produce out-of-order

src/service/announcements/mod.rs

@@ -20,7 +20,6 @@
 use async_trait::async_trait;
 use conduwuit::{Result, Server, debug, error, warn};
 use database::{Deserialized, Map};
-use rand::Rng;
 use ruma::events::{Mentions, room::message::RoomMessageEventContent};
 use serde::Deserialize;
 use tokio::{
@@ -100,8 +99,7 @@ async fn worker(self: Arc<Self>) -> Result<()> {
 		}
 
 		let first_check_jitter = {
-			let mut rng = rand::thread_rng();
-			let jitter_percent = rng.gen_range(-50.0..=10.0);
+			let jitter_percent = rand::random_range(-50.0..=10.0);
 			self.interval.mul_f64(1.0 + jitter_percent / 100.0)
 		};
 

src/service/client/mod.rs

@@ -39,7 +39,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		let url_preview_user_agent = config
 			.url_preview_user_agent
 			.clone()
-			.unwrap_or_else(|| conduwuit::version::user_agent().to_owned());
+			.unwrap_or_else(|| conduwuit::version::user_agent_media().to_owned());
 
 		Ok(Arc::new(Self {
 			default: base(config)?

src/service/firstrun/mod.rs

@@ -67,15 +67,17 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 
 	async fn worker(self: Arc<Self>) -> Result {
-		// first run mode will be enabled if there are no local users
-		let is_first_run = self
-			.services
-			.users
-			.list_local_users()
-			.ready_filter(|user| *user != self.services.globals.server_user)
-			.next()
-			.await
-			.is_none();
+		// first run mode will be enabled if there are no local users, provided it's not
+		// forcibly disabled for Complement tests
+		let is_first_run = !self.services.config.force_disable_first_run_mode
+			&& self
+				.services
+				.users
+				.list_local_users()
+				.ready_filter(|user| *user != self.services.globals.server_user)
+				.next()
+				.await
+				.is_none();
 
 		self.first_run_marker
 			.set(if is_first_run {

src/service/media/data.rs

@@ -170,6 +170,8 @@ pub(super) fn remove_url_preview(&self, url: &str) -> Result<()> {
 		Ok(())
 	}
 
+	pub(super) async fn clear_url_previews(&self) { self.url_previews.clear().await; }
+
 	pub(super) fn set_url_preview(
 		&self,
 		url: &str,

src/service/media/preview.rs

@@ -37,6 +37,9 @@ pub async fn remove_url_preview(&self, url: &str) -> Result<()> {
 	self.db.remove_url_preview(url)
 }
 
+#[implement(Service)]
+pub async fn clear_url_previews(&self) { self.db.clear_url_previews().await; }
+
 #[implement(Service)]
 pub async fn set_url_preview(&self, url: &str, data: &UrlPreviewData) -> Result<()> {
 	let now = SystemTime::now()

src/service/mod.rs

@@ -31,7 +31,7 @@
 pub mod sending;
 pub mod server_keys;
 pub mod sync;
-pub mod transaction_ids;
+pub mod transactions;
 pub mod uiaa;
 pub mod users;
 

src/service/rooms/auth_chain/mod.rs

@@ -142,7 +142,7 @@ async fn get_auth_chain_outer(
 
 	let chunk_cache: Vec<_> = chunk
 		.into_iter()
-		.try_stream()
+		.try_stream::<conduwuit::Error>()
 		.broad_and_then(|(shortid, event_id)| async move {
 			if let Ok(cached) = self.get_cached_eventid_authchain(&[shortid]).await {
 				return Ok(cached.to_vec());

src/service/rooms/event_handler/fetch_state.rs

@@ -63,7 +63,9 @@ pub(super) async fn fetch_state<Pdu>(
 			},
 			| hash_map::Entry::Occupied(_) => {
 				return Err!(Database(
-					"State event's type and state_key combination exists multiple times.",
+					"State event's type and state_key combination exists multiple times: {}, {}",
+					pdu.kind(),
+					state_key
 				));
 			},
 		}

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -162,7 +162,9 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 			},
 			| hash_map::Entry::Occupied(_) => {
 				return Err!(Request(InvalidParam(
-					"Auth event's type and state_key combination exists multiple times.",
+					"Auth event's type and state_key combination exists multiple times: {}, {}",
+					auth_event.kind,
+					auth_event.state_key().unwrap_or("")
 				)));
 			},
 		}

src/service/sending/mod.rs

@@ -385,11 +385,13 @@ fn num_senders(args: &crate::Args<'_>) -> usize {
 	const MIN_SENDERS: usize = 1;
 	// Limit the number of senders to the number of workers threads or number of
 	// cores, conservatively.
-	let max_senders = args
-		.server
-		.metrics
-		.num_workers()
-		.min(available_parallelism());
+	let mut max_senders = args.server.metrics.num_workers();
+
+	// Work around some platforms not returning the number of cores.
+	let num_cores = available_parallelism();
+	if num_cores > 0 {
+		max_senders = max_senders.min(num_cores);
+	}
 
 	// If the user doesn't override the default 0, this is intended to then default
 	// to 1 for now as multiple senders is experimental.

src/service/services.rs

@@ -14,7 +14,7 @@
 	media, moderation, presence, pusher, registration_tokens, resolver, rooms, sending,
 	server_keys,
 	service::{self, Args, Map, Service},
-	sync, transaction_ids, uiaa, users,
+	sync, transactions, uiaa, users,
 };
 
 pub struct Services {
@@ -37,7 +37,7 @@ pub struct Services {
 	pub sending: Arc<sending::Service>,
 	pub server_keys: Arc<server_keys::Service>,
 	pub sync: Arc<sync::Service>,
-	pub transaction_ids: Arc<transaction_ids::Service>,
+	pub transactions: Arc<transactions::Service>,
 	pub uiaa: Arc<uiaa::Service>,
 	pub users: Arc<users::Service>,
 	pub moderation: Arc<moderation::Service>,
@@ -110,7 +110,7 @@ macro_rules! build {
 			sending: build!(sending::Service),
 			server_keys: build!(server_keys::Service),
 			sync: build!(sync::Service),
-			transaction_ids: build!(transaction_ids::Service),
+			transactions: build!(transactions::Service),
 			uiaa: build!(uiaa::Service),
 			users: build!(users::Service),
 			moderation: build!(moderation::Service),

src/service/transaction_ids/mod.rs

@@ -1,54 +0,0 @@
-use std::sync::Arc;
-
-use conduwuit::{Result, implement};
-use database::{Handle, Map};
-use ruma::{DeviceId, TransactionId, UserId};
-
-pub struct Service {
-	db: Data,
-}
-
-struct Data {
-	userdevicetxnid_response: Arc<Map>,
-}
-
-impl crate::Service for Service {
-	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
-		Ok(Arc::new(Self {
-			db: Data {
-				userdevicetxnid_response: args.db["userdevicetxnid_response"].clone(),
-			},
-		}))
-	}
-
-	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
-}
-
-#[implement(Service)]
-pub fn add_txnid(
-	&self,
-	user_id: &UserId,
-	device_id: Option<&DeviceId>,
-	txn_id: &TransactionId,
-	data: &[u8],
-) {
-	let mut key = user_id.as_bytes().to_vec();
-	key.push(0xFF);
-	key.extend_from_slice(device_id.map(DeviceId::as_bytes).unwrap_or_default());
-	key.push(0xFF);
-	key.extend_from_slice(txn_id.as_bytes());
-
-	self.db.userdevicetxnid_response.insert(&key, data);
-}
-
-// If there's no entry, this is a new transaction
-#[implement(Service)]
-pub async fn existing_txnid(
-	&self,
-	user_id: &UserId,
-	device_id: Option<&DeviceId>,
-	txn_id: &TransactionId,
-) -> Result<Handle<'_>> {
-	let key = (user_id, device_id, txn_id);
-	self.db.userdevicetxnid_response.qry(&key).await
-}

src/service/transactions/mod.rs

@@ -0,0 +1,326 @@
+use std::{
+	collections::HashMap,
+	fmt,
+	sync::{
+		Arc,
+		atomic::{AtomicU64, Ordering},
+	},
+	time::{Duration, SystemTime},
+};
+
+use async_trait::async_trait;
+use conduwuit::{Error, Result, SyncRwLock, debug_warn, warn};
+use database::{Handle, Map};
+use ruma::{
+	DeviceId, OwnedServerName, OwnedTransactionId, TransactionId, UserId,
+	api::{
+		client::error::ErrorKind::LimitExceeded,
+		federation::transactions::send_transaction_message,
+	},
+};
+use tokio::sync::watch::{Receiver, Sender};
+
+use crate::{Dep, config};
+
+pub type TxnKey = (OwnedServerName, OwnedTransactionId);
+pub type WrappedTransactionResponse =
+	Option<Result<send_transaction_message::v1::Response, TransactionError>>;
+
+/// Errors that can occur during federation transaction processing.
+#[derive(Debug, Clone)]
+pub enum TransactionError {
+	/// Server is shutting down - the sender should retry the entire
+	/// transaction.
+	ShuttingDown,
+}
+
+impl fmt::Display for TransactionError {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		match self {
+			| Self::ShuttingDown => write!(f, "Server is shutting down"),
+		}
+	}
+}
+
+impl std::error::Error for TransactionError {}
+
+/// Minimum interval between cache cleanup runs.
+/// Exists to prevent thrashing when the cache is full of things that can't be
+/// cleared
+const CLEANUP_INTERVAL_SECS: u64 = 30;
+
+#[derive(Clone, Debug)]
+pub struct CachedTxnResponse {
+	pub response: send_transaction_message::v1::Response,
+	pub created: SystemTime,
+}
+
+/// Internal state for a federation transaction.
+/// Either actively being processed or completed and cached.
+#[derive(Clone)]
+enum TxnState {
+	/// Transaction is currently being processed.
+	Active(Receiver<WrappedTransactionResponse>),
+
+	/// Transaction completed and response is cached.
+	Cached(CachedTxnResponse),
+}
+
+/// Result of atomically checking or starting a federation transaction.
+pub enum FederationTxnState {
+	/// Transaction already completed and cached
+	Cached(send_transaction_message::v1::Response),
+
+	/// Transaction is currently being processed by another request.
+	/// Wait on this receiver for the result.
+	Active(Receiver<WrappedTransactionResponse>),
+
+	/// This caller should process the transaction (first to request it).
+	Started {
+		receiver: Receiver<WrappedTransactionResponse>,
+		sender: Sender<WrappedTransactionResponse>,
+	},
+}
+
+pub struct Service {
+	services: Services,
+	db: Data,
+	federation_txn_state: Arc<SyncRwLock<HashMap<TxnKey, TxnState>>>,
+	last_cleanup: AtomicU64,
+}
+
+struct Services {
+	config: Dep<config::Service>,
+}
+
+struct Data {
+	userdevicetxnid_response: Arc<Map>,
+}
+
+#[async_trait]
+impl crate::Service for Service {
+	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
+		Ok(Arc::new(Self {
+			services: Services {
+				config: args.depend::<config::Service>("config"),
+			},
+			db: Data {
+				userdevicetxnid_response: args.db["userdevicetxnid_response"].clone(),
+			},
+			federation_txn_state: Arc::new(SyncRwLock::new(HashMap::new())),
+			last_cleanup: AtomicU64::new(0),
+		}))
+	}
+
+	async fn clear_cache(&self) {
+		let mut state = self.federation_txn_state.write();
+		// Only clear cached entries, preserve active transactions
+		state.retain(|_, v| matches!(v, TxnState::Active(_)));
+	}
+
+	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
+}
+
+impl Service {
+	/// Returns the count of currently active (in-progress) transactions.
+	#[must_use]
+	pub fn txn_active_handle_count(&self) -> usize {
+		let state = self.federation_txn_state.read();
+		state
+			.values()
+			.filter(|v| matches!(v, TxnState::Active(_)))
+			.count()
+	}
+
+	pub fn add_client_txnid(
+		&self,
+		user_id: &UserId,
+		device_id: Option<&DeviceId>,
+		txn_id: &TransactionId,
+		data: &[u8],
+	) {
+		let mut key = user_id.as_bytes().to_vec();
+		key.push(0xFF);
+		key.extend_from_slice(device_id.map(DeviceId::as_bytes).unwrap_or_default());
+		key.push(0xFF);
+		key.extend_from_slice(txn_id.as_bytes());
+
+		self.db.userdevicetxnid_response.insert(&key, data);
+	}
+
+	pub async fn get_client_txn(
+		&self,
+		user_id: &UserId,
+		device_id: Option<&DeviceId>,
+		txn_id: &TransactionId,
+	) -> Result<Handle<'_>> {
+		let key = (user_id, device_id, txn_id);
+		self.db.userdevicetxnid_response.qry(&key).await
+	}
+
+	/// Atomically gets a cached response, joins an active transaction, or
+	/// starts a new one.
+	pub fn get_or_start_federation_txn(&self, key: TxnKey) -> Result<FederationTxnState> {
+		// Only one upgradable lock can be held at a time, and there aren't any
+		// read-only locks, so no point being upgradable
+		let mut state = self.federation_txn_state.write();
+
+		// Check existing state for this key
+		if let Some(txn_state) = state.get(&key) {
+			return Ok(match txn_state {
+				| TxnState::Cached(cached) => FederationTxnState::Cached(cached.response.clone()),
+				| TxnState::Active(receiver) => FederationTxnState::Active(receiver.clone()),
+			});
+		}
+
+		// Check if another transaction from this origin is already running
+		let has_active_from_origin = state
+			.iter()
+			.any(|(k, v)| k.0 == key.0 && matches!(v, TxnState::Active(_)));
+
+		if has_active_from_origin {
+			debug_warn!(
+				origin = ?key.0,
+				"Got concurrent transaction request from an origin with an active transaction"
+			);
+			return Err(Error::BadRequest(
+				LimitExceeded { retry_after: None },
+				"Still processing another transaction from this origin",
+			));
+		}
+
+		let max_active_txns = self.services.config.max_concurrent_inbound_transactions;
+
+		// Check if we're at capacity
+		if state.len() >= max_active_txns
+			&& let active_count = state
+				.values()
+				.filter(|v| matches!(v, TxnState::Active(_)))
+				.count() && active_count >= max_active_txns
+		{
+			warn!(
+				active = active_count,
+				max = max_active_txns,
+				"Server is overloaded, dropping incoming transaction"
+			);
+			return Err(Error::BadRequest(
+				LimitExceeded { retry_after: None },
+				"Server is overloaded, try again later",
+			));
+		}
+
+		// Start new transaction
+		let (sender, receiver) = tokio::sync::watch::channel(None);
+		state.insert(key, TxnState::Active(receiver.clone()));
+
+		Ok(FederationTxnState::Started { receiver, sender })
+	}
+
+	/// Finishes a transaction by transitioning it from active to cached state.
+	/// Additionally may trigger cleanup of old entries.
+	pub fn finish_federation_txn(
+		&self,
+		key: TxnKey,
+		sender: Sender<WrappedTransactionResponse>,
+		response: send_transaction_message::v1::Response,
+	) {
+		// Check if cleanup might be needed before acquiring the lock
+		let should_try_cleanup = self.should_try_cleanup();
+
+		let mut state = self.federation_txn_state.write();
+
+		// Explicitly set cached first so there is no gap where receivers get a closed
+		// channel
+		state.insert(
+			key,
+			TxnState::Cached(CachedTxnResponse {
+				response: response.clone(),
+				created: SystemTime::now(),
+			}),
+		);
+
+		if let Err(e) = sender.send(Some(Ok(response))) {
+			debug_warn!("Failed to send transaction response to waiting receivers: {e}");
+		}
+
+		// Explicitly close
+		drop(sender);
+
+		// This task is dangling, we can try clean caches now
+		if should_try_cleanup {
+			self.cleanup_entries_locked(&mut state);
+		}
+	}
+
+	pub fn remove_federation_txn(&self, key: &TxnKey) {
+		let mut state = self.federation_txn_state.write();
+		state.remove(key);
+	}
+
+	/// Checks if enough time has passed since the last cleanup to consider
+	/// running another. Updates the last cleanup time if returning true.
+	fn should_try_cleanup(&self) -> bool {
+		let now = SystemTime::now()
+			.duration_since(SystemTime::UNIX_EPOCH)
+			.expect("SystemTime before UNIX_EPOCH")
+			.as_secs();
+		let last = self.last_cleanup.load(Ordering::Relaxed);
+
+		if now.saturating_sub(last) >= CLEANUP_INTERVAL_SECS {
+			// CAS: only update if no one else has updated it since we read
+			self.last_cleanup
+				.compare_exchange(last, now, Ordering::Relaxed, Ordering::Relaxed)
+				.is_ok()
+		} else {
+			false
+		}
+	}
+
+	/// Cleans up cached entries based on age and count limits.
+	///
+	/// First removes all cached entries older than the configured max age.
+	/// Then, if the cache still exceeds the max entry count, removes the oldest
+	/// cached entries until the count is within limits.
+	///
+	/// Must be called with write lock held on the state map.
+	fn cleanup_entries_locked(&self, state: &mut HashMap<TxnKey, TxnState>) {
+		let max_age_secs = self.services.config.transaction_id_cache_max_age_secs;
+		let max_entries = self.services.config.transaction_id_cache_max_entries;
+
+		// First pass: remove all cached entries older than max age
+		let cutoff = SystemTime::now()
+			.checked_sub(Duration::from_secs(max_age_secs))
+			.unwrap_or(SystemTime::UNIX_EPOCH);
+
+		state.retain(|_, v| match v {
+			| TxnState::Active(_) => true, // Never remove active transactions
+			| TxnState::Cached(cached) => cached.created > cutoff,
+		});
+
+		// Count cached entries
+		let cached_count = state
+			.values()
+			.filter(|v| matches!(v, TxnState::Cached(_)))
+			.count();
+
+		// Second pass: if still over max entries, remove oldest cached entries
+		if cached_count > max_entries {
+			let excess = cached_count.saturating_sub(max_entries);
+
+			// Collect cached entries sorted by age (oldest first)
+			let mut cached_entries: Vec<_> = state
+				.iter()
+				.filter_map(|(k, v)| match v {
+					| TxnState::Cached(cached) => Some((k.clone(), cached.created)),
+					| TxnState::Active(_) => None,
+				})
+				.collect();
+			cached_entries.sort_by(|a, b| a.1.cmp(&b.1));
+
+			// Remove the oldest cached entries to get under the limit
+			for (key, _) in cached_entries.into_iter().take(excess) {
+				state.remove(&key);
+			}
+		}
+	}
+}

src/service/users/mod.rs

@@ -184,6 +184,12 @@ pub async fn create(
 		password: Option<&str>,
 		origin: Option<&str>,
 	) -> Result<()> {
+		if !self.services.globals.user_is_local(user_id)
+			&& (password.is_some() || origin.is_some())
+		{
+			return Err!("Cannot create a nonlocal user with a set password or origin");
+		}
+
 		self.db
 			.userid_origin
 			.insert(user_id, origin.unwrap_or("password"));

tests/test_results/complement/test_results.jsonl

@@ -6,9 +6,9 @@
 {"Action":"fail","Test":"TestArchivedRoomsHistory/timeline_has_events"}
 {"Action":"fail","Test":"TestArchivedRoomsHistory/timeline_has_events/incremental_sync"}
 {"Action":"fail","Test":"TestArchivedRoomsHistory/timeline_has_events/initial_sync"}
-{"Action":"pass","Test":"TestArchivedRoomsHistory/timeline_is_empty"}
+{"Action":"fail","Test":"TestArchivedRoomsHistory/timeline_is_empty"}
 {"Action":"skip","Test":"TestArchivedRoomsHistory/timeline_is_empty/incremental_sync"}
-{"Action":"pass","Test":"TestArchivedRoomsHistory/timeline_is_empty/initial_sync"}
+{"Action":"fail","Test":"TestArchivedRoomsHistory/timeline_is_empty/initial_sync"}
 {"Action":"fail","Test":"TestAsyncUpload"}
 {"Action":"fail","Test":"TestAsyncUpload/Cannot_upload_to_a_media_ID_that_has_already_been_uploaded_to"}
 {"Action":"fail","Test":"TestAsyncUpload/Create_media"}
@@ -79,9 +79,11 @@
 {"Action":"fail","Test":"TestClientSpacesSummary/redact_link"}
 {"Action":"fail","Test":"TestClientSpacesSummary/suggested_only"}
 {"Action":"fail","Test":"TestClientSpacesSummaryJoinRules"}
+{"Action":"pass","Test":"TestComplementCanCreateValidV12Rooms"}
 {"Action":"pass","Test":"TestContent"}
 {"Action":"pass","Test":"TestContentCSAPIMediaV1"}
 {"Action":"pass","Test":"TestContentMediaV1"}
+{"Action":"fail","Test":"TestCorruptedAuthChain"}
 {"Action":"pass","Test":"TestCumulativeJoinLeaveJoinSync"}
 {"Action":"pass","Test":"TestDeactivateAccount"}
 {"Action":"pass","Test":"TestDeactivateAccount/After_deactivating_account,_can't_log_in_with_password"}
@@ -89,19 +91,18 @@
 {"Action":"pass","Test":"TestDeactivateAccount/Can_deactivate_account"}
 {"Action":"pass","Test":"TestDeactivateAccount/Password_flow_is_available"}
 {"Action":"fail","Test":"TestDelayedEvents"}
-{"Action":"fail","Test":"TestDelayedEvents/cannot_update_a_delayed_event_with_an_invalid_action"}
-{"Action":"pass","Test":"TestDelayedEvents/cannot_update_a_delayed_event_without_a_delay_ID"}
-{"Action":"fail","Test":"TestDelayedEvents/cannot_update_a_delayed_event_without_a_request_body"}
-{"Action":"fail","Test":"TestDelayedEvents/cannot_update_a_delayed_event_without_an_action"}
+{"Action":"pass","Test":"TestDelayedEvents/cannot_update_a_delayed_event_with_an_invalid_action"}
+{"Action":"pass","Test":"TestDelayedEvents/cannot_update_a_delayed_event_without_an_action"}
+{"Action":"fail","Test":"TestDelayedEvents/delayed_event_lookups_are_authenticated"}
 {"Action":"fail","Test":"TestDelayedEvents/delayed_events_are_empty_on_startup"}
 {"Action":"fail","Test":"TestDelayedEvents/delayed_message_events_are_sent_on_timeout"}
-{"Action":"fail","Test":"TestDelayedEvents/delayed_state_events_are_cancelled_by_a_more_recent_state_event_from_another_user"}
-{"Action":"fail","Test":"TestDelayedEvents/delayed_state_events_are_cancelled_by_a_more_recent_state_event_from_the_same_user"}
 {"Action":"skip","Test":"TestDelayedEvents/delayed_state_events_are_kept_on_server_restart"}
 {"Action":"fail","Test":"TestDelayedEvents/delayed_state_events_are_sent_on_timeout"}
 {"Action":"fail","Test":"TestDelayedEvents/delayed_state_events_can_be_cancelled"}
 {"Action":"fail","Test":"TestDelayedEvents/delayed_state_events_can_be_restarted"}
 {"Action":"fail","Test":"TestDelayedEvents/delayed_state_events_can_be_sent_on_request"}
+{"Action":"fail","Test":"TestDelayedEvents/delayed_state_is_cancelled_by_new_state_from_another_user"}
+{"Action":"fail","Test":"TestDelayedEvents/delayed_state_is_not_cancelled_by_new_state_from_the_same_user"}
 {"Action":"pass","Test":"TestDelayedEvents/parallel"}
 {"Action":"pass","Test":"TestDelayedEvents/parallel/cannot_cancel_a_delayed_event_without_a_matching_delay_ID"}
 {"Action":"pass","Test":"TestDelayedEvents/parallel/cannot_restart_a_delayed_event_without_a_matching_delay_ID"}
@@ -153,12 +154,14 @@
 {"Action":"fail","Test":"TestFederationKeyUploadQuery/Can_query_remote_device_keys_using_POST"}
 {"Action":"pass","Test":"TestFederationRedactSendsWithoutEvent"}
 {"Action":"pass","Test":"TestFederationRejectInvite"}
-{"Action":"pass","Test":"TestFederationRoomsInvite"}
-{"Action":"pass","Test":"TestFederationRoomsInvite/Parallel"}
+{"Action":"fail","Test":"TestFederationRoomsInvite"}
+{"Action":"fail","Test":"TestFederationRoomsInvite/Parallel"}
 {"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Invited_user_can_reject_invite_over_federation"}
 {"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Invited_user_can_reject_invite_over_federation_for_empty_room"}
 {"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Invited_user_can_reject_invite_over_federation_several_times"}
 {"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Invited_user_has_'is_direct'_flag_in_prev_content_after_joining"}
+{"Action":"fail","Test":"TestFederationRoomsInvite/Parallel/Inviter_user_can_rescind_invite_over_federation"}
+{"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Non-invitee_user_cannot_rescind_invite_over_federation"}
 {"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Remote_invited_user_can_join_the_room_when_homeserver_is_already_participating_in_the_room"}
 {"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Remote_invited_user_can_reject_invite_when_homeserver_is_already_participating_in_the_room"}
 {"Action":"pass","Test":"TestFederationRoomsInvite/Parallel/Remote_invited_user_can_see_room_metadata"}
@@ -191,6 +194,18 @@
 {"Action":"pass","Test":"TestInboundFederationProfile/Inbound_federation_can_query_profile_data"}
 {"Action":"pass","Test":"TestInboundFederationProfile/Non-numeric_ports_in_server_names_are_rejected"}
 {"Action":"fail","Test":"TestInboundFederationRejectsEventsWithRejectedAuthEvents"}
+{"Action":"pass","Test":"TestInviteFiltering"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_allow_a_user_from_a_blocked_server"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_block_a_single_user"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_block_a_user_from_an_allowed_server"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_block_a_whole_server"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_glob_serveral_servers"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_glob_serveral_users"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_ignore_a_single_user"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_ignore_a_whole_server"}
+{"Action":"pass","Test":"TestInviteFiltering/Can_invite_users_normally_without_any_rules"}
+{"Action":"pass","Test":"TestInviteFiltering/Will_allow_users_when_a_user_appears_in_multiple_fields"}
+{"Action":"pass","Test":"TestInviteFiltering/Will_ignore_null_fields"}
 {"Action":"pass","Test":"TestInviteFromIgnoredUsersDoesNotAppearInSync"}
 {"Action":"pass","Test":"TestIsDirectFlagFederation"}
 {"Action":"pass","Test":"TestIsDirectFlagLocal"}
@@ -214,18 +229,20 @@
 {"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/federation/looking_backwards,_should_be_able_to_find_event_that_was_sent_before_we_joined"}
 {"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/federation/looking_forwards,_should_be_able_to_find_event_that_was_sent_before_we_joined"}
 {"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/federation/when_looking_backwards_before_the_room_was_created,_should_be_able_to_find_event_that_was_imported"}
-{"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_find_event_after_given_timestmap"}
-{"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_find_event_before_given_timestmap"}
-{"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_find_next_event_topologically_after_given_timestmap_when_all_message_timestamps_are_the_same"}
+{"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_find_event_after_given_timestamp"}
+{"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_find_event_before_given_timestamp"}
+{"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_find_next_event_topologically_after_given_timestamp_when_all_message_timestamps_are_the_same"}
 {"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_find_next_event_topologically_before_given_timestamp_when_all_message_timestamps_are_the_same"}
-{"Action":"pass","Test":"TestJumpToDateEndpoint/parallel/should_find_nothing_after_the_latest_timestmap"}
-{"Action":"pass","Test":"TestJumpToDateEndpoint/parallel/should_find_nothing_before_the_earliest_timestmap"}
+{"Action":"pass","Test":"TestJumpToDateEndpoint/parallel/should_find_nothing_after_the_latest_timestamp"}
+{"Action":"pass","Test":"TestJumpToDateEndpoint/parallel/should_find_nothing_before_the_earliest_timestamp"}
 {"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_not_be_able_to_query_a_private_room_you_are_not_a_member_of"}
 {"Action":"fail","Test":"TestJumpToDateEndpoint/parallel/should_not_be_able_to_query_a_public_room_you_are_not_a_member_of"}
 {"Action":"fail","Test":"TestKeyChangesLocal"}
 {"Action":"fail","Test":"TestKeyChangesLocal/New_login_should_create_a_device_lists.changed_entry"}
 {"Action":"fail","Test":"TestKeyClaimOrdering"}
 {"Action":"pass","Test":"TestKeysQueryWithDeviceIDAsObjectFails"}
+{"Action":"pass","Test":"TestKnockRestrictedRoomsLocalJoinNoCreatorsUsesPowerLevelsV11"}
+{"Action":"pass","Test":"TestKnockRestrictedRoomsLocalJoinNoCreatorsUsesPowerLevelsV12"}
 {"Action":"fail","Test":"TestKnockRoomsInPublicRoomsDirectory"}
 {"Action":"fail","Test":"TestKnockRoomsInPublicRoomsDirectoryInMSC3787Room"}
 {"Action":"fail","Test":"TestKnocking"}
@@ -252,8 +269,8 @@
 {"Action":"pass","Test":"TestKnocking/Knocking_on_a_room_with_a_join_rule_other_than_'knock'_should_fail#01"}
 {"Action":"fail","Test":"TestKnocking/Knocking_on_a_room_with_join_rule_'knock'_should_succeed"}
 {"Action":"fail","Test":"TestKnocking/Knocking_on_a_room_with_join_rule_'knock'_should_succeed#01"}
-{"Action":"pass","Test":"TestKnocking/Users_in_the_room_see_a_user's_membership_update_when_they_knock"}
-{"Action":"pass","Test":"TestKnocking/Users_in_the_room_see_a_user's_membership_update_when_they_knock#01"}
+{"Action":"fail","Test":"TestKnocking/Users_in_the_room_see_a_user's_membership_update_when_they_knock"}
+{"Action":"fail","Test":"TestKnocking/Users_in_the_room_see_a_user's_membership_update_when_they_knock#01"}
 {"Action":"fail","Test":"TestKnockingInMSC3787Room"}
 {"Action":"fail","Test":"TestKnockingInMSC3787Room/A_user_can_knock_on_a_room_without_a_reason"}
 {"Action":"fail","Test":"TestKnockingInMSC3787Room/A_user_can_knock_on_a_room_without_a_reason#01"}
@@ -278,8 +295,8 @@
 {"Action":"pass","Test":"TestKnockingInMSC3787Room/Knocking_on_a_room_with_a_join_rule_other_than_'knock'_should_fail#01"}
 {"Action":"fail","Test":"TestKnockingInMSC3787Room/Knocking_on_a_room_with_join_rule_'knock'_should_succeed"}
 {"Action":"fail","Test":"TestKnockingInMSC3787Room/Knocking_on_a_room_with_join_rule_'knock'_should_succeed#01"}
-{"Action":"pass","Test":"TestKnockingInMSC3787Room/Users_in_the_room_see_a_user's_membership_update_when_they_knock"}
-{"Action":"pass","Test":"TestKnockingInMSC3787Room/Users_in_the_room_see_a_user's_membership_update_when_they_knock#01"}
+{"Action":"fail","Test":"TestKnockingInMSC3787Room/Users_in_the_room_see_a_user's_membership_update_when_they_knock"}
+{"Action":"fail","Test":"TestKnockingInMSC3787Room/Users_in_the_room_see_a_user's_membership_update_when_they_knock#01"}
 {"Action":"pass","Test":"TestLeakyTyping"}
 {"Action":"pass","Test":"TestLeaveEventInviteRejection"}
 {"Action":"fail","Test":"TestLeaveEventVisibility"}
@@ -308,10 +325,43 @@
 {"Action":"pass","Test":"TestLogout/Request_to_logout_without_an_access_token_is_rejected"}
 {"Action":"fail","Test":"TestMSC3757OwnedState"}
 {"Action":"pass","Test":"TestMSC3967"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators/PL_event_is_missing_creator_in_users_map"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators/admin_with_>PL100_cannot_kick_creator"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators/admin_with_>PL100_sorts_after_the_room_creator_for_state_resolution"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators/creator_can_kick_admin"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators/creator_can_kick_admin_above_PL100"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators/creator_can_kick_admin_at_JSON_max_value"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators/creator_cannot_set_self_in_PL_event"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators/m.room.tombstone_needs_PL150_in_the_PL_event"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators/power_level_cannot_be_set_beyond_max_canonical_JSON_int"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators/power_level_content_override_can_be_set"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators/power_level_content_override_cannot_set_the_room_creator"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_Additional"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_AdditionalCreatorsAndInvited"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_AdditionalValidation"}
+{"Action":"pass","Test":"TestMSC4289PrivilegedRoomCreators_AdditionalValidation/additional_creators_are_valid"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_AdditionalValidation/additional_creators_elements_aren't_strings"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_AdditionalValidation/additional_creators_elements_aren't_user_ID_strings"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_AdditionalValidation/additional_creators_elements_aren't_valid_user_ID_strings_(domain)"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_AdditionalValidation/additional_creators_isn't_an_array"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_InvitedAreCreators"}
+{"Action":"fail","Test":"TestMSC4289PrivilegedRoomCreators_Upgrades"}
+{"Action":"pass","Test":"TestMSC4291RoomIDAsHashOfCreateEvent"}
+{"Action":"pass","Test":"TestMSC4291RoomIDAsHashOfCreateEvent_AuthEventsOmitsCreateEvent"}
+{"Action":"pass","Test":"TestMSC4291RoomIDAsHashOfCreateEvent_CannotSendCreateEvent"}
+{"Action":"pass","Test":"TestMSC4291RoomIDAsHashOfCreateEvent_RoomIDIsOnCreateEvent"}
+{"Action":"pass","Test":"TestMSC4291RoomIDAsHashOfCreateEvent_UpgradedRooms"}
+{"Action":"fail","Test":"TestMSC4297StateResolutionV2_1_includes_conflicted_subgraph"}
+{"Action":"fail","Test":"TestMSC4297StateResolutionV2_1_starts_from_empty_set"}
+{"Action":"fail","Test":"TestMSC4308ThreadSubscriptionsSlidingSync"}
+{"Action":"fail","Test":"TestMSC4308ThreadSubscriptionsSlidingSync/Receives_thread_subscriptions_over_incremental_sliding_sync"}
+{"Action":"fail","Test":"TestMSC4308ThreadSubscriptionsSlidingSync/Receives_thread_subscriptions_over_initial_sliding_sync"}
+{"Action":"fail","Test":"TestMSC4311FullCreateEventOnStrippedState"}
 {"Action":"pass","Test":"TestMediaConfig"}
-{"Action":"pass","Test":"TestMediaFilenames"}
-{"Action":"pass","Test":"TestMediaFilenames/Parallel"}
-{"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII"}
+{"Action":"fail","Test":"TestMediaFilenames"}
+{"Action":"fail","Test":"TestMediaFilenames/Parallel"}
+{"Action":"fail","Test":"TestMediaFilenames/Parallel/ASCII"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_file_'ascii'"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_file_'ascii'_over_/_matrix/client/v1/media/download"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_file_'name;with;semicolons'"}
@@ -319,11 +369,11 @@
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_file_'name_with_spaces'"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_file_'name_with_spaces'_over_/_matrix/client/v1/media/download"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_specifying_a_different_ASCII_file_name"}
-{"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_specifying_a_different_ASCII_file_name_over__matrix/client/v1/media/download"}
+{"Action":"fail","Test":"TestMediaFilenames/Parallel/ASCII/Can_download_specifying_a_different_ASCII_file_name_over__matrix/client/v1/media/download"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/ASCII/Can_upload_with_ASCII_file_name"}
-{"Action":"pass","Test":"TestMediaFilenames/Parallel/Unicode"}
+{"Action":"fail","Test":"TestMediaFilenames/Parallel/Unicode"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/Unicode/Can_download_specifying_a_different_Unicode_file_name"}
-{"Action":"pass","Test":"TestMediaFilenames/Parallel/Unicode/Can_download_specifying_a_different_Unicode_file_name_over__matrix/client/v1/media/download"}
+{"Action":"fail","Test":"TestMediaFilenames/Parallel/Unicode/Can_download_specifying_a_different_Unicode_file_name_over__matrix/client/v1/media/download"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/Unicode/Can_download_with_Unicode_file_name_locally"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/Unicode/Can_download_with_Unicode_file_name_locally_over__matrix/client/v1/media/download"}
 {"Action":"pass","Test":"TestMediaFilenames/Parallel/Unicode/Can_download_with_Unicode_file_name_over_federation"}
@@ -352,9 +402,15 @@
 {"Action":"pass","Test":"TestMembersLocal/Parallel/Existing_members_see_new_members'_presence_(in_initial_sync)"}
 {"Action":"pass","Test":"TestMembersLocal/Parallel/New_room_members_see_their_own_join_event"}
 {"Action":"fail","Test":"TestMembershipOnEvents"}
-{"Action":"fail","Test":"TestNetworkPartitionOrdering"}
+{"Action":"fail","Test":"TestMessagesOverFederation"}
+{"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_joining_new_room_(backfill)"}
+{"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_joining_new_room_(backfill)/`messagesRequestLimit`_is_greater_than_the_number_of_messages_backfilled_(in_Synapse,_100)"}
+{"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_joining_new_room_(backfill)/`messagesRequestLimit`_is_lower_than_the_number_of_messages_backfilled_(assumed)"}
+{"Action":"fail","Test":"TestMessagesOverFederation/Visible_shared_history_after_re-joining_room_(backfill)"}
+{"Action":"fail","Test":"TestMessagesOverFederation/Visible_shared_history_after_re-joining_room_(backfill)/`messagesRequestLimit`_is_lower_than_the_number_of_messages_backfilled_(assumed)"}
+{"Action":"pass","Test":"TestNetworkPartitionOrdering"}
 {"Action":"pass","Test":"TestNotPresentUserCannotBanOthers"}
-{"Action":"pass","Test":"TestOlderLeftRoomsNotInLeaveSection"}
+{"Action":"fail","Test":"TestOlderLeftRoomsNotInLeaveSection"}
 {"Action":"fail","Test":"TestOutboundFederationEventSizeGetMissingEvents"}
 {"Action":"fail","Test":"TestOutboundFederationIgnoresMissingEventWithBadJSONForRoomVersion6"}
 {"Action":"pass","Test":"TestOutboundFederationProfile"}
@@ -379,7 +435,23 @@
 {"Action":"pass","Test":"TestProfileDisplayName"}
 {"Action":"pass","Test":"TestProfileDisplayName/GET_/profile/:user_id/displayname_publicly_accessible"}
 {"Action":"pass","Test":"TestProfileDisplayName/PUT_/profile/:user_id/displayname_sets_my_name"}
+{"Action":"pass","Test":"TestPublicRooms"}
+{"Action":"pass","Test":"TestPublicRooms/Can_search_public_room_list"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct/Creating_room_with_alias_publicroom_with_unicode_chars_name"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct/Creating_room_with_alias_publicroom_with_unicode_chars_name_topic"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct/Creating_room_with_alias_publicroom_with_unicode_chars_topic"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct/Creating_room_with_alias_publicroomalias_no_name"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct/Creating_room_with_alias_publicroomalias_with_name"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct/Creating_room_with_alias_publicroomalias_with_name_topic"}
+{"Action":"pass","Test":"TestPublicRooms/Name/topic_keys_are_correct/Creating_room_with_alias_publicroomalias_with_topic"}
 {"Action":"pass","Test":"TestPushRuleCacheHealth"}
+{"Action":"fail","Test":"TestPushRuleRoomUpgrade"}
+{"Action":"fail","Test":"TestPushRuleRoomUpgrade/parallel"}
+{"Action":"fail","Test":"TestPushRuleRoomUpgrade/parallel/joining_a_remote_manually_upgraded_room_carries_over_existing_push_rules"}
+{"Action":"fail","Test":"TestPushRuleRoomUpgrade/parallel/joining_a_remote_upgraded_room_carries_over_existing_push_rules"}
+{"Action":"fail","Test":"TestPushRuleRoomUpgrade/parallel/manually_upgrading_a_room_carries_over_existing_push_rules_for_local_users"}
+{"Action":"fail","Test":"TestPushRuleRoomUpgrade/parallel/upgrading_a_room_carries_over_existing_push_rules_for_local_users"}
 {"Action":"pass","Test":"TestPushSync"}
 {"Action":"pass","Test":"TestPushSync/Adding_a_push_rule_wakes_up_an_incremental_/sync"}
 {"Action":"pass","Test":"TestPushSync/Disabling_a_push_rule_wakes_up_an_incremental_/sync"}
@@ -440,14 +512,16 @@
 {"Action":"pass","Test":"TestRestrictedRoomsLocalJoinInMSC3787Room/Join_should_fail_with_mangled_join_rules"}
 {"Action":"pass","Test":"TestRestrictedRoomsLocalJoinInMSC3787Room/Join_should_succeed_when_invited"}
 {"Action":"fail","Test":"TestRestrictedRoomsLocalJoinInMSC3787Room/Join_should_succeed_when_joined_to_allowed_room"}
+{"Action":"pass","Test":"TestRestrictedRoomsLocalJoinNoCreatorsUsesPowerLevelsV11"}
+{"Action":"pass","Test":"TestRestrictedRoomsLocalJoinNoCreatorsUsesPowerLevelsV12"}
 {"Action":"fail","Test":"TestRestrictedRoomsRemoteJoin"}
 {"Action":"pass","Test":"TestRestrictedRoomsRemoteJoin/Join_should_fail_initially"}
 {"Action":"pass","Test":"TestRestrictedRoomsRemoteJoin/Join_should_fail_when_left_allowed_room"}
 {"Action":"pass","Test":"TestRestrictedRoomsRemoteJoin/Join_should_fail_with_mangled_join_rules"}
 {"Action":"pass","Test":"TestRestrictedRoomsRemoteJoin/Join_should_succeed_when_invited"}
 {"Action":"fail","Test":"TestRestrictedRoomsRemoteJoin/Join_should_succeed_when_joined_to_allowed_room"}
-{"Action":"fail","Test":"TestRestrictedRoomsRemoteJoinFailOver"}
-{"Action":"fail","Test":"TestRestrictedRoomsRemoteJoinFailOverInMSC3787Room"}
+{"Action":"pass","Test":"TestRestrictedRoomsRemoteJoinFailOver"}
+{"Action":"pass","Test":"TestRestrictedRoomsRemoteJoinFailOverInMSC3787Room"}
 {"Action":"fail","Test":"TestRestrictedRoomsRemoteJoinInMSC3787Room"}
 {"Action":"pass","Test":"TestRestrictedRoomsRemoteJoinInMSC3787Room/Join_should_fail_initially"}
 {"Action":"pass","Test":"TestRestrictedRoomsRemoteJoinInMSC3787Room/Join_should_fail_when_left_allowed_room"}
@@ -475,16 +549,19 @@
 {"Action":"fail","Test":"TestRoomCanonicalAlias/Parallel/m.room.canonical_alias_rejects_missing_aliases"}
 {"Action":"fail","Test":"TestRoomCanonicalAlias/Parallel/m.room.canonical_alias_rejects_missing_aliases#01"}
 {"Action":"fail","Test":"TestRoomCanonicalAlias/Parallel/m.room.canonical_alias_setting_rejects_deleted_aliases"}
-{"Action":"pass","Test":"TestRoomCreate"}
-{"Action":"pass","Test":"TestRoomCreate/Parallel"}
+{"Action":"fail","Test":"TestRoomCreate"}
+{"Action":"fail","Test":"TestRoomCreate/Parallel"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/Can_/sync_newly_created_room"}
-{"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_creates_a_room_with_the_given_version"}
+{"Action":"fail","Test":"TestRoomCreate/Parallel/POST_/createRoom_creates_a_room_with_the_given_version"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_ignores_attempts_to_set_the_room_version_via_creation_content"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_private_room"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_private_room_with_invites"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_public_room"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_room_with_a_name"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_room_with_a_topic"}
+{"Action":"fail","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_room_with_a_topic_and_writes_rich_topic_representation"}
+{"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_room_with_a_topic_via_initial_state"}
+{"Action":"fail","Test":"TestRoomCreate/Parallel/POST_/createRoom_makes_a_room_with_a_topic_via_initial_state_overwritten_by_topic"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_rejects_attempts_to_create_rooms_with_numeric_versions"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/POST_/createRoom_rejects_attempts_to_create_rooms_with_unknown_versions"}
 {"Action":"pass","Test":"TestRoomCreate/Parallel/Rooms_can_be_created_with_an_initial_invite_list_(SYN-205)"}
@@ -527,6 +604,7 @@
 {"Action":"pass","Test":"TestRoomMessagesLazyLoadingLocalUser"}
 {"Action":"pass","Test":"TestRoomReadMarkers"}
 {"Action":"pass","Test":"TestRoomReceipts"}
+{"Action":"pass","Test":"TestRoomReceipts/Receipts_DO_NOT_include_a_`room_id`_field"}
 {"Action":"pass","Test":"TestRoomSpecificUsernameAtJoin"}
 {"Action":"pass","Test":"TestRoomSpecificUsernameAtJoin/Bob_can_find_Alice_by_mxid"}
 {"Action":"pass","Test":"TestRoomSpecificUsernameAtJoin/Bob_can_find_Alice_by_profile_display_name"}
@@ -575,7 +653,7 @@
 {"Action":"pass","Test":"TestSearch/parallel/Search_results_with_recent_ordering_do_not_include_redacted_events"}
 {"Action":"pass","Test":"TestSearch/parallel/Search_works_across_an_upgraded_room_and_its_predecessor"}
 {"Action":"fail","Test":"TestSendAndFetchMessage"}
-{"Action":"skip","Test":"TestSendJoinPartialStateResponse"}
+{"Action":"fail","Test":"TestSendJoinPartialStateResponse"}
 {"Action":"pass","Test":"TestSendMessageWithTxn"}
 {"Action":"pass","Test":"TestServerCapabilities"}
 {"Action":"skip","Test":"TestServerNotices"}
@@ -589,7 +667,7 @@
 {"Action":"fail","Test":"TestSync/parallel/Newly_joined_room_has_correct_timeline_in_incremental_sync"}
 {"Action":"fail","Test":"TestSync/parallel/Newly_joined_room_includes_presence_in_incremental_sync"}
 {"Action":"pass","Test":"TestSync/parallel/Newly_joined_room_is_included_in_an_incremental_sync"}
-{"Action":"pass","Test":"TestSync/parallel/sync_should_succeed_even_if_the_sync_token_points_to_a_redaction_of_an_unknown_event"}
+{"Action":"fail","Test":"TestSync/parallel/sync_should_succeed_even_if_the_sync_token_points_to_a_redaction_of_an_unknown_event"}
 {"Action":"pass","Test":"TestSyncFilter"}
 {"Action":"pass","Test":"TestSyncFilter/Can_create_filter"}
 {"Action":"pass","Test":"TestSyncFilter/Can_download_filter"}
@@ -603,12 +681,21 @@
 {"Action":"pass","Test":"TestSyncTimelineGap/incremental"}
 {"Action":"pass","Test":"TestTentativeEventualJoiningAfterRejecting"}
 {"Action":"fail","Test":"TestThreadReceiptsInSyncMSC4102"}
+{"Action":"fail","Test":"TestThreadSubscriptions"}
+{"Action":"fail","Test":"TestThreadSubscriptions/Can_create_automatic_subscription_to_a_thread"}
+{"Action":"fail","Test":"TestThreadSubscriptions/Can_subscribe_to_and_unsubscribe_from_a_thread"}
+{"Action":"fail","Test":"TestThreadSubscriptions/Cannot_use_thread_root_as_automatic_subscription_cause_event"}
+{"Action":"fail","Test":"TestThreadSubscriptions/Error_when_using_invalid_automatic_event_ID"}
+{"Action":"fail","Test":"TestThreadSubscriptions/Manual_subscriptions_overwrite_automatic_subscriptions"}
+{"Action":"pass","Test":"TestThreadSubscriptions/Nonexistent_threads_return_404"}
+{"Action":"fail","Test":"TestThreadSubscriptions/Server-side_automatic_subscription_ordering_conflict"}
+{"Action":"fail","Test":"TestThreadSubscriptions/Unsubscribe_succeeds_even_with_no_subscription"}
 {"Action":"fail","Test":"TestThreadedReceipts"}
 {"Action":"fail","Test":"TestThreadsEndpoint"}
 {"Action":"pass","Test":"TestToDeviceMessages"}
 {"Action":"fail","Test":"TestToDeviceMessagesOverFederation"}
 {"Action":"pass","Test":"TestToDeviceMessagesOverFederation/good_connectivity"}
-{"Action":"pass","Test":"TestToDeviceMessagesOverFederation/interrupted_connectivity"}
+{"Action":"fail","Test":"TestToDeviceMessagesOverFederation/interrupted_connectivity"}
 {"Action":"fail","Test":"TestToDeviceMessagesOverFederation/stopped_server"}
 {"Action":"fail","Test":"TestTxnIdWithRefreshToken"}
 {"Action":"fail","Test":"TestTxnIdempotency"}
@@ -617,6 +704,7 @@
 {"Action":"pass","Test":"TestTxnScopeOnLocalEcho"}
 {"Action":"pass","Test":"TestTyping"}
 {"Action":"pass","Test":"TestTyping/Typing_can_be_explicitly_stopped"}
+{"Action":"pass","Test":"TestTyping/Typing_events_DO_NOT_include_a_`room_id`_field"}
 {"Action":"pass","Test":"TestTyping/Typing_notification_sent_to_local_room_members"}
 {"Action":"fail","Test":"TestUnknownEndpoints"}
 {"Action":"pass","Test":"TestUnknownEndpoints/Client-server_endpoints"}
@@ -624,7 +712,7 @@
 {"Action":"pass","Test":"TestUnknownEndpoints/Media_endpoints"}
 {"Action":"pass","Test":"TestUnknownEndpoints/Server-server_endpoints"}
 {"Action":"pass","Test":"TestUnknownEndpoints/Unknown_prefix"}
-{"Action":"fail","Test":"TestUnrejectRejectedEvents"}
+{"Action":"pass","Test":"TestUnrejectRejectedEvents"}
 {"Action":"fail","Test":"TestUploadKey"}
 {"Action":"fail","Test":"TestUploadKey/Parallel"}
 {"Action":"fail","Test":"TestUploadKey/Parallel/Can_claim_one_time_key_using_POST"}
@@ -637,7 +725,7 @@
 {"Action":"pass","Test":"TestUploadKeyIdempotency"}
 {"Action":"pass","Test":"TestUploadKeyIdempotencyOverlap"}
 {"Action":"fail","Test":"TestUrlPreview"}
-{"Action":"pass","Test":"TestUserAppearsInChangedDeviceListOnJoinOverFederation"}
+{"Action":"fail","Test":"TestUserAppearsInChangedDeviceListOnJoinOverFederation"}
 {"Action":"pass","Test":"TestVersionStructure"}
 {"Action":"pass","Test":"TestVersionStructure/Version_responds_200_OK_with_valid_structure"}
 {"Action":"pass","Test":"TestWithoutOwnedState"}