Compare commits

...

8 Commits

Author SHA1 Message Date
Jacob Taylor b2122ea6e0 Feat: Merge nex/feat/global-ip-span 2026-07-12 16:58:35 -07:00
Jacob Taylor 200a2236d0 chore: Commit basic QoDana config for local QA usage 2026-07-12 14:35:57 -07:00
Jacob Taylor 7513b1622b fix: Delete silly 2026-07-12 14:35:57 -07:00
Jacob Taylor 1bcaa73f69 fix: Pre-Commit Lint Compliance Maneuver 2026-07-12 14:35:57 -07:00
Jacob Taylor ac82447a76 feat: Bump one cache a bit 2026-07-12 14:35:57 -07:00
Jacob Taylor bc2391edfe upgrade some logs to info 2026-07-12 14:35:57 -07:00
Jacob Taylor ac83e210d5 enable converged 6g at the edge in continuwuity
sender_workers scaling. this time, with feeling!
2026-07-12 14:35:57 -07:00
Jacob Taylor 67471edfe5 bump the number of allowed immutable memtables by 1, to allow for greater flood protection
this should probably not be applied if you have rocksdb_atomic_flush = false (the default)
2026-07-12 14:35:57 -07:00
34 changed files with 187 additions and 261 deletions
+1 -1
View File
@@ -51,7 +51,7 @@ repos:
hooks:
- id: cargo-clippy
name: cargo clippy
entry: cargo clippy -- -D warnings
entry: cargo clippy --
language: system
pass_filenames: false
types: [rust]
+2 -4
View File
@@ -1827,11 +1827,9 @@
#stream_amplification = 1024
# Number of sender task workers; determines sender parallelism. Default is
# '0' which means the value is determined internally, likely matching the
# number of tokio worker-threads or number of cores, etc. Override by
# setting a non-zero value.
# core count. Override by setting a different value.
#
#sender_workers = 0
#sender_workers = core count
# Enables listener sockets; can be set to false to disable listening. This
# option is intended for developer/diagnostic purposes only.
+46
View File
@@ -0,0 +1,46 @@
#-------------------------------------------------------------------------------#
# Qodana analysis is configured by qodana.yaml file #
# https://www.jetbrains.com/help/qodana/qodana-yaml.html #
#-------------------------------------------------------------------------------#
#################################################################################
# WARNING: Do not store sensitive information in this file, #
# as its contents will be included in the Qodana report. #
#################################################################################
version: "1.0"
#Specify inspection profile for code analysis
profile:
name: qodana.starter
#Enable inspections
#include:
# - name: <SomeEnabledInspectionId>
#Disable inspections
#exclude:
# - name: <SomeDisabledInspectionId>
# paths:
# - <path/where/not/run/inspection>
#Execute shell command before Qodana execution (Applied in CI/CD pipeline)
#bootstrap: sh ./prepare-qodana.sh
#Install IDE plugins before Qodana execution (Applied in CI/CD pipeline)
#plugins:
# - id: <plugin.id> #(plugin id can be found at https://plugins.jetbrains.com)
# Quality gate. Will fail the CI/CD pipeline if any condition is not met
# severityThresholds - configures maximum thresholds for different problem severities
# testCoverageThresholds - configures minimum code coverage on a whole project and newly added code
# Code Coverage is available in Ultimate and Ultimate Plus plans
#failureConditions:
# severityThresholds:
# any: 15
# critical: 5
# testCoverageThresholds:
# fresh: 70
# total: 50
#Specify Qodana linter for analysis (Applied in CI/CD pipeline)
linter: jetbrains/qodana-<linter>:2026.2
+3 -7
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Result, err, info,
pdu::PartialPdu,
@@ -43,10 +42,9 @@
///
/// Note: This will not reserve the username, so the username might become
/// invalid when trying to register
#[tracing::instrument(skip_all, fields(%client), name = "register_available", level = "info")]
#[tracing::instrument(skip_all, name = "register_available", level = "info")]
pub(crate) async fn get_register_available_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_username_availability::v3::Request>,
) -> Result<get_username_availability::v3::Response> {
let _ = services
@@ -80,10 +78,9 @@ pub(crate) async fn get_register_available_route(
/// last seen ts)
/// - Forgets to-device events
/// - Triggers device list updates
#[tracing::instrument(skip_all, fields(%client), name = "change_password", level = "info")]
#[tracing::instrument(skip_all, name = "change_password", level = "info")]
pub(crate) async fn change_password_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<change_password::v3::Request>,
) -> Result<change_password::v3::Response> {
let identity = if let Some(identity) = body.identity.as_ref() {
@@ -248,10 +245,9 @@ pub(crate) async fn whoami_route(
/// - Forgets all to-device events
/// - Triggers device list updates
/// - Removes ability to log in again
#[tracing::instrument(skip_all, fields(%client), name = "deactivate", level = "info")]
#[tracing::instrument(skip_all, name = "deactivate", level = "info")]
pub(crate) async fn deactivate_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<deactivate::v3::Request>,
) -> Result<deactivate::v3::Response> {
// Authentication for this endpoint is technically optional,
+2 -2
View File
@@ -33,10 +33,10 @@
/// /_matrix/client/v3/register/available`](fn.get_register_available_route.
/// html) to check if the user id is valid and available.
#[allow(clippy::doc_markdown)]
#[tracing::instrument(skip_all, fields(%client), name = "register", level = "info")]
#[tracing::instrument(skip_all, name = "register", level = "info")]
pub(crate) async fn register_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
ClientIp(client): ClientIp, // NOTE: Required for metadata.
body: Ruma<register::v3::Request>,
) -> Result<register::v3::Response> {
if body.kind != RegistrationKind::User {
+4 -9
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{Err, Result, at};
use futures::StreamExt;
use ruma::{
@@ -19,10 +18,9 @@
/// # `PUT /_matrix/client/../dehydrated_device`
///
/// Creates or overwrites the user's dehydrated device.
#[tracing::instrument(skip_all, fields(%client))]
#[tracing::instrument(skip_all)]
pub(crate) async fn put_dehydrated_device_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<put_dehydrated_device::Request>,
) -> Result<put_dehydrated_device::Response> {
let device_id = body.device_id.clone();
@@ -38,10 +36,9 @@ pub(crate) async fn put_dehydrated_device_route(
/// # `DELETE /_matrix/client/../dehydrated_device`
///
/// Deletes the user's dehydrated device without replacement.
#[tracing::instrument(skip_all, fields(%client))]
#[tracing::instrument(skip_all)]
pub(crate) async fn delete_dehydrated_device_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<delete_dehydrated_device::Request>,
) -> Result<delete_dehydrated_device::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -56,10 +53,9 @@ pub(crate) async fn delete_dehydrated_device_route(
/// # `GET /_matrix/client/../dehydrated_device`
///
/// Gets the user's dehydrated device
#[tracing::instrument(skip_all, fields(%client))]
#[tracing::instrument(skip_all)]
pub(crate) async fn get_dehydrated_device_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_dehydrated_device::Request>,
) -> Result<get_dehydrated_device::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -72,10 +68,9 @@ pub(crate) async fn get_dehydrated_device_route(
/// # `GET /_matrix/client/../dehydrated_device/{device_id}/events`
///
/// Paginates the events of the dehydrated device.
#[tracing::instrument(skip_all, fields(%client))]
#[tracing::instrument(skip_all)]
pub(crate) async fn get_dehydrated_events_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_events::Request>,
) -> Result<get_events::Response> {
let sender_user = body.identity.expect_sender_user()?;
+2 -2
View File
@@ -46,10 +46,10 @@ pub(crate) async fn get_device_route(
/// # `PUT /_matrix/client/r0/devices/{deviceId}`
///
/// Updates the metadata on a given device of the sender user.
#[tracing::instrument(skip_all, fields(%client), name = "update_device", level = "debug")]
#[tracing::instrument(skip_all, name = "update_device", level = "debug")]
pub(crate) async fn update_device_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
ClientIp(client): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<update_device::v3::Request>,
) -> Result<update_device::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+3 -7
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Result, err, info,
utils::{
@@ -35,10 +34,9 @@
/// Lists the public rooms on this server.
///
/// - Rooms are ordered by the number of joined members
#[tracing::instrument(skip_all, fields(%client), name = "publicrooms", level = "info")]
#[tracing::instrument(skip_all, name = "publicrooms", level = "info")]
pub(crate) async fn get_public_rooms_filtered_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_public_rooms_filtered::v3::Request>,
) -> Result<get_public_rooms_filtered::v3::Response> {
if let Some(server) = &body.server {
@@ -71,10 +69,9 @@ pub(crate) async fn get_public_rooms_filtered_route(
/// Lists the public rooms on this server.
///
/// - Rooms are ordered by the number of joined members
#[tracing::instrument(skip_all, fields(%client), name = "publicrooms", level = "info")]
#[tracing::instrument(skip_all, name = "publicrooms", level = "info")]
pub(crate) async fn get_public_rooms_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_public_rooms::v3::Request>,
) -> Result<get_public_rooms::v3::Response> {
if let Some(server) = &body.server {
@@ -106,10 +103,9 @@ pub(crate) async fn get_public_rooms_route(
/// # `PUT /_matrix/client/r0/directory/list/room/{roomId}`
///
/// Sets the visibility of a given room in the room directory.
#[tracing::instrument(skip_all, fields(%client), name = "room_directory", level = "info")]
#[tracing::instrument(skip_all, name = "room_directory", level = "info")]
pub(crate) async fn set_room_visibility_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<set_room_visibility::v3::Request>,
) -> Result<set_room_visibility::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
-36
View File
@@ -1,7 +1,6 @@
use std::time::Duration;
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Result, err,
utils::{self, content_disposition::make_content_disposition, math::ruma_from_usize},
@@ -42,15 +41,8 @@ pub(crate) async fn get_media_config_route(
///
/// - Some metadata will be saved in the database
/// - Media will be saved in the media/ directory
#[tracing::instrument(
name = "media_upload",
level = "debug",
skip_all,
fields(%client),
)]
pub(crate) async fn create_content_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<create_content::v3::Request>,
) -> Result<create_content::v3::Response> {
let user = body.identity.expect_sender_user()?;
@@ -81,15 +73,8 @@ pub(crate) async fn create_content_route(
/// # `GET /_matrix/client/v1/media/thumbnail/{serverName}/{mediaId}`
///
/// Load media thumbnail from our server or over federation.
#[tracing::instrument(
name = "media_thumbnail_get",
level = "debug",
skip_all,
fields(%client),
)]
pub(crate) async fn get_content_thumbnail_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content_thumbnail::v1::Request>,
) -> Result<get_content_thumbnail::v1::Response> {
let user = body.identity.expect_sender_user()?;
@@ -131,15 +116,8 @@ pub(crate) async fn get_content_thumbnail_route(
/// # `GET /_matrix/client/v1/media/download/{serverName}/{mediaId}`
///
/// Load media from our server or over federation.
#[tracing::instrument(
name = "media_get",
level = "debug",
skip_all,
fields(%client),
)]
pub(crate) async fn get_content_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content::v1::Request>,
) -> Result<get_content::v1::Response> {
let user = body.identity.expect_sender_user()?;
@@ -178,15 +156,8 @@ pub(crate) async fn get_content_route(
/// # `GET /_matrix/client/v1/media/download/{serverName}/{mediaId}/{fileName}`
///
/// Load media from our server or over federation as fileName.
#[tracing::instrument(
name = "media_get_af",
level = "debug",
skip_all,
fields(%client),
)]
pub(crate) async fn get_content_as_filename_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content_as_filename::v1::Request>,
) -> Result<get_content_as_filename::v1::Response> {
let user = body.identity.expect_sender_user()?;
@@ -229,15 +200,8 @@ pub(crate) async fn get_content_as_filename_route(
/// # `GET /_matrix/client/v1/media/preview_url`
///
/// Returns URL preview.
#[tracing::instrument(
name = "url_preview",
level = "debug",
skip_all,
fields(%client),
)]
pub(crate) async fn get_media_preview_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_media_preview::v1::Request>,
) -> Result<get_media_preview::v1::Response> {
let sender_user = body.identity.expect_sender_user()?;
+5 -20
View File
@@ -1,7 +1,6 @@
#![allow(deprecated)]
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Result, err,
utils::{content_disposition::make_content_disposition, math::ruma_from_usize},
@@ -50,10 +49,8 @@ pub(crate) async fn get_media_config_legacy_legacy_route(
/// # `GET /_matrix/media/v3/preview_url`
///
/// Returns URL preview.
#[tracing::instrument(skip_all, fields(%client), name = "url_preview_legacy", level = "debug")]
pub(crate) async fn get_media_preview_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_media_preview::v3::Request>,
) -> Result<get_media_preview::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -95,10 +92,9 @@ pub(crate) async fn get_media_preview_legacy_route(
/// Returns URL preview.
pub(crate) async fn get_media_preview_legacy_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_media_preview::v3::Request>,
) -> Result<RumaResponse<get_media_preview::v3::Response>> {
get_media_preview_legacy_route(State(services), ClientIp(client), body)
get_media_preview_legacy_route(State(services), body)
.await
.map(RumaResponse)
}
@@ -115,10 +111,9 @@ pub(crate) async fn get_media_preview_legacy_legacy_route(
/// - Media will be saved in the media/ directory
pub(crate) async fn create_content_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<create_content::v3::Request>,
) -> Result<RumaResponse<create_content::v3::Response>> {
create_content_route(State(services), ClientIp(client), body)
create_content_route(State(services), body)
.await
.map(RumaResponse)
}
@@ -131,10 +126,8 @@ pub(crate) async fn create_content_legacy_route(
/// - Only redirects if `allow_redirect` is true
/// - Uses client-provided `timeout_ms` if available, else defaults to 20
/// seconds
#[tracing::instrument(skip_all, fields(%client), name = "media_get_legacy", level = "debug")]
pub(crate) async fn get_content_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content::v3::Request>,
) -> Result<get_content::v3::Response> {
let mxc = Mxc {
@@ -209,13 +202,11 @@ pub(crate) async fn get_content_legacy_route(
/// - Only redirects if `allow_redirect` is true
/// - Uses client-provided `timeout_ms` if available, else defaults to 20
/// seconds
#[tracing::instrument(skip_all, fields(%client), name = "media_get_legacy", level = "debug")]
pub(crate) async fn get_content_legacy_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content::v3::Request>,
) -> Result<RumaResponse<get_content::v3::Response>> {
get_content_legacy_route(State(services), ClientIp(client), body)
get_content_legacy_route(State(services), body)
.await
.map(RumaResponse)
}
@@ -228,10 +219,8 @@ pub(crate) async fn get_content_legacy_legacy_route(
/// - Only redirects if `allow_redirect` is true
/// - Uses client-provided `timeout_ms` if available, else defaults to 20
/// seconds
#[tracing::instrument(skip_all, fields(%client), name = "media_get_legacy", level = "debug")]
pub(crate) async fn get_content_as_filename_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content_as_filename::v3::Request>,
) -> Result<get_content_as_filename::v3::Response> {
let mxc = Mxc {
@@ -307,10 +296,9 @@ pub(crate) async fn get_content_as_filename_legacy_route(
/// seconds
pub(crate) async fn get_content_as_filename_legacy_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content_as_filename::v3::Request>,
) -> Result<RumaResponse<get_content_as_filename::v3::Response>> {
get_content_as_filename_legacy_route(State(services), ClientIp(client), body)
get_content_as_filename_legacy_route(State(services), body)
.await
.map(RumaResponse)
}
@@ -323,10 +311,8 @@ pub(crate) async fn get_content_as_filename_legacy_legacy_route(
/// - Only redirects if `allow_redirect` is true
/// - Uses client-provided `timeout_ms` if available, else defaults to 20
/// seconds
#[tracing::instrument(skip_all, fields(%client), name = "media_thumbnail_get_legacy", level = "debug")]
pub(crate) async fn get_content_thumbnail_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content_thumbnail::v3::Request>,
) -> Result<get_content_thumbnail::v3::Response> {
let mxc = Mxc {
@@ -404,10 +390,9 @@ pub(crate) async fn get_content_thumbnail_legacy_route(
/// seconds
pub(crate) async fn get_content_thumbnail_legacy_legacy_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content_thumbnail::v3::Request>,
) -> Result<RumaResponse<get_content_thumbnail::v3::Response>> {
get_content_thumbnail_legacy_route(State(services), ClientIp(client), body)
get_content_thumbnail_legacy_route(State(services), body)
.await
.map(RumaResponse)
}
+3 -11
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Result, debug_error, err, info,
matrix::{event::gen_event_id_canonical_json, pdu::PartialPdu},
@@ -23,10 +22,9 @@
/// # `POST /_matrix/client/r0/rooms/{roomId}/invite`
///
/// Tries to send an invite event into the room.
#[tracing::instrument(skip_all, fields(%client), name = "invite", level = "info")]
#[tracing::instrument(skip_all, name = "invite", level = "info")]
pub(crate) async fn invite_user_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<invite_user::v3::Request>,
) -> Result<invite_user::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -42,14 +40,8 @@ pub(crate) async fn invite_user_route(
return Err!(Request(Forbidden("Invites are not allowed on this server.")));
}
banned_room_check(
&services,
sender_user,
Some(&body.room_id),
body.room_id.server_name(),
client,
)
.await?;
banned_room_check(&services, sender_user, Some(&body.room_id), body.room_id.server_name())
.await?;
match &body.recipient {
| invite_user::v3::InvitationRecipient::UserId(InviteUserId {
+7 -23
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Result, debug,
result::FlatOk,
@@ -22,10 +21,9 @@
/// rules locally
/// - If the server does not know about the room: asks other servers over
/// federation
#[tracing::instrument(skip_all, fields(%client), name = "join", level = "info")]
#[tracing::instrument(skip_all, name = "join", level = "info")]
pub(crate) async fn join_room_by_id_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<join_room_by_id::v3::Request>,
) -> Result<join_room_by_id::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -33,14 +31,8 @@ pub(crate) async fn join_room_by_id_route(
return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
}
banned_room_check(
&services,
sender_user,
Some(&body.room_id),
body.room_id.server_name(),
client,
)
.await?;
banned_room_check(&services, sender_user, Some(&body.room_id), body.room_id.server_name())
.await?;
// There is no body.server_name for /roomId/join
let mut servers: Vec<_> = services
@@ -91,10 +83,9 @@ pub(crate) async fn join_room_by_id_route(
/// - If the server does not know about the room: use the server name query
/// param if specified. if not specified, asks other servers over federation
/// via room alias server name and room ID server name
#[tracing::instrument(skip_all, fields(%client), name = "join", level = "info")]
#[tracing::instrument(skip_all, name = "join", level = "info")]
pub(crate) async fn join_room_by_id_or_alias_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<join_room_by_id_or_alias::v3::Request>,
) -> Result<join_room_by_id_or_alias::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -105,15 +96,9 @@ pub(crate) async fn join_room_by_id_or_alias_route(
let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias.clone()) {
| Ok(room_id) => {
banned_room_check(
&services,
sender_user,
Some(&room_id),
room_id.server_name(),
client,
)
.boxed()
.await?;
banned_room_check(&services, sender_user, Some(&room_id), room_id.server_name())
.boxed()
.await?;
let mut servers = body.via.clone();
if servers.is_empty() {
@@ -159,7 +144,6 @@ pub(crate) async fn join_room_by_id_or_alias_route(
sender_user,
Some(&room_id),
Some(room_alias.server_name()),
client,
)
.await?;
+3 -12
View File
@@ -1,7 +1,6 @@
use std::{borrow::Borrow, collections::HashMap, iter::once, sync::Arc};
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Result, debug, debug_info, debug_warn, err, info,
matrix::{
@@ -45,10 +44,9 @@
/// # `POST /_matrix/client/*/knock/{roomIdOrAlias}`
///
/// Tries to knock the room to ask permission to join for the sender user.
#[tracing::instrument(skip_all, fields(%client), name = "knock", level = "info")]
#[tracing::instrument(skip_all, name = "knock", level = "info")]
pub(crate) async fn knock_room_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<knock_room::v3::Request>,
) -> Result<knock_room::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -59,14 +57,8 @@ pub(crate) async fn knock_room_route(
let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias.clone()) {
| Ok(room_id) => {
banned_room_check(
&services,
sender_user,
Some(&room_id),
room_id.server_name(),
client,
)
.await?;
banned_room_check(&services, sender_user, Some(&room_id), room_id.server_name())
.await?;
let mut servers = body.via.clone();
servers.extend(
@@ -109,7 +101,6 @@ pub(crate) async fn knock_room_route(
sender_user,
Some(&room_id),
Some(room_alias.server_name()),
client,
)
.await?;
+2 -5
View File
@@ -8,8 +8,6 @@
mod members;
mod unban;
use std::net::IpAddr;
use axum::extract::State;
use conduwuit::{Err, Result, warn};
use futures::{FutureExt, StreamExt};
@@ -58,7 +56,6 @@ pub(crate) async fn banned_room_check(
user_id: &UserId,
room_id: Option<&RoomId>,
server_name: Option<&ServerName>,
client_ip: IpAddr,
) -> Result {
if services.users.is_admin(user_id).await {
return Ok(());
@@ -85,7 +82,7 @@ pub(crate) async fn banned_room_check(
.admin
.send_text(&format!(
"Automatically deactivating user {user_id} due to attempted banned \
room join from IP {client_ip}"
room join"
))
.await;
}
@@ -121,7 +118,7 @@ pub(crate) async fn banned_room_check(
.admin
.send_text(&format!(
"Automatically deactivating user {user_id} due to attempted banned \
room join from IP {client_ip}"
room join"
))
.await;
}
+1 -1
View File
@@ -72,7 +72,7 @@
/// where the user was joined, depending on `history_visibility`)
pub(crate) async fn get_message_events_route(
State(services): State<crate::State>,
ClientIp(client_ip): ClientIp,
ClientIp(client_ip): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<get_message_events::v3::Request>,
) -> Result<get_message_events::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+1 -1
View File
@@ -117,7 +117,7 @@ pub(crate) async fn set_read_marker_route(
/// Sets private read marker and public read receipt EDU.
pub(crate) async fn create_receipt_route(
State(services): State<crate::State>,
ClientIp(client_ip): ClientIp,
ClientIp(client_ip): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<create_receipt::v3::Request>,
) -> Result<create_receipt::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+1 -1
View File
@@ -14,7 +14,7 @@
/// - TODO: Handle txn id
pub(crate) async fn redact_event_route(
State(services): State<crate::State>,
ClientIp(client_ip): ClientIp,
ClientIp(client_ip): ClientIp, // NOTE: required for updating device metadata
body: Ruma<redact_event::v3::Request>,
) -> Result<redact_event::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+3 -7
View File
@@ -1,7 +1,6 @@
use std::{fmt::Write as _, time::Duration};
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{Err, Event, Result, debug_info, info, matrix::pdu::PduEvent, utils::ReadyExt};
use conduwuit_service::Services;
use ruma::{
@@ -30,10 +29,9 @@ struct Report {
/// # `POST /_matrix/client/v3/rooms/{roomId}/report`
///
/// Reports an abusive room to homeserver admins
#[tracing::instrument(skip_all, fields(%client), name = "report_room", level = "info")]
#[tracing::instrument(skip_all, name = "report_room", level = "info")]
pub(crate) async fn report_room_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<report_room::v3::Request>,
) -> Result<report_room::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -85,10 +83,9 @@ pub(crate) async fn report_room_route(
/// # `POST /_matrix/client/v3/rooms/{roomId}/report/{eventId}`
///
/// Reports an inappropriate event to homeserver admins
#[tracing::instrument(skip_all, fields(%client), name = "report_event", level = "info")]
#[tracing::instrument(skip_all, name = "report_event", level = "info")]
pub(crate) async fn report_event_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<report_content::v3::Request>,
) -> Result<report_content::v3::Response> {
// user authentication
@@ -129,10 +126,9 @@ pub(crate) async fn report_event_route(
Ok(report_content::v3::Response::new())
}
#[tracing::instrument(skip_all, fields(%client), name = "report_user", level = "info")]
#[tracing::instrument(skip_all, name = "report_user", level = "info")]
pub(crate) async fn report_user_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<report_user::v3::Request>,
) -> Result<report_user::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+1 -3
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{Err, Result};
use ruma::api::client::room::get_summary;
use service::rooms::summary::Accessibility;
@@ -9,10 +8,9 @@
/// # `GET /_matrix/client/v1/room_summary/{roomIdOrAlias}`
///
/// Returns a short description of the state of a room.
#[tracing::instrument(skip_all, fields(%client), name = "room_summary", level = "info")]
#[tracing::instrument(skip_all, name = "room_summary", level = "info")]
pub(crate) async fn get_room_summary(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_summary::v1::Request>,
) -> Result<get_summary::v1::Response> {
let (room_id, servers) = services
+1 -1
View File
@@ -19,7 +19,7 @@
/// allowed
pub(crate) async fn send_message_event_route(
State(services): State<crate::State>,
ClientIp(client_ip): ClientIp,
ClientIp(client_ip): ClientIp, // NOTE: required for updating device metadata
body: Ruma<send_message_event::v3::Request>,
) -> Result<send_message_event::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+6 -10
View File
@@ -37,10 +37,9 @@
///
/// Get the supported login types of this server. One of these should be used as
/// the `type` field when logging in.
#[tracing::instrument(skip_all, fields(%client), name = "login", level = "info")]
#[tracing::instrument(skip_all, name = "login", level = "info")]
pub(crate) async fn get_login_types_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
_body: Ruma<get_login_types::v3::Request>,
) -> Result<get_login_types::v3::Response> {
if !services.config.oauth.compatibility_mode().uiaa_available() {
@@ -114,10 +113,10 @@ pub async fn handle_login(
/// Note: You can use [`GET
/// /_matrix/client/r0/login`](fn.get_supported_versions_route.html) to see
/// supported login types.
#[tracing::instrument(skip_all, fields(%client), name = "login", level = "info")]
#[tracing::instrument(skip_all, name = "login", level = "info")]
pub(crate) async fn login_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
ClientIp(client): ClientIp, // NOTE: Required for device metadata
body: Ruma<login::v3::Request>,
) -> Result<login::v3::Response> {
if !services.config.oauth.compatibility_mode().uiaa_available() {
@@ -255,10 +254,9 @@ pub(crate) async fn login_route(
/// to log in with the m.login.token flow.
///
/// <https://spec.matrix.org/v1.13/client-server-api/#post_matrixclientv1loginget_token>
#[tracing::instrument(skip_all, fields(%client), name = "login_token", level = "info")]
#[tracing::instrument(skip_all, name = "login_token", level = "info")]
pub(crate) async fn login_token_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_login_token::v1::Request>,
) -> Result<get_login_token::v1::Response> {
if !services.config.login_via_existing_session {
@@ -291,10 +289,9 @@ pub(crate) async fn login_token_route(
/// last seen ts)
/// - Forgets to-device events
/// - Triggers device list updates
#[tracing::instrument(skip_all, fields(%client), name = "logout", level = "info")]
#[tracing::instrument(skip_all, name = "logout", level = "info")]
pub(crate) async fn logout_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<logout::v3::Request>,
) -> Result<logout::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -339,10 +336,9 @@ pub(crate) async fn logout_route(
/// Note: This is equivalent to calling [`GET
/// /_matrix/client/r0/logout`](fn.logout_route.html) from each device of this
/// user.
#[tracing::instrument(skip_all, fields(%client), name = "logout", level = "info")]
#[tracing::instrument(skip_all, name = "logout", level = "info")]
pub(crate) async fn logout_all_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<logout_all::v3::Request>,
) -> Result<logout_all::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+2 -2
View File
@@ -35,7 +35,7 @@
/// Sends a state event into the room.
pub(crate) async fn send_state_event_for_key_route(
State(services): State<crate::State>,
ClientIp(ip): ClientIp,
ClientIp(ip): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<send_state_event::v3::Request>,
) -> Result<send_state_event::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
@@ -72,7 +72,7 @@ pub(crate) async fn send_state_event_for_key_route(
/// Sends a state event into the room.
pub(crate) async fn send_state_event_for_empty_key_route(
State(services): State<crate::State>,
ClientIp(ip): ClientIp,
ClientIp(ip): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<send_state_event::v3::Request>,
) -> Result<RumaResponse<send_state_event::v3::Response>> {
send_state_event_for_key_route(State(services), ClientIp(ip), body)
+1 -1
View File
@@ -181,7 +181,7 @@ fn lazy_loading_enabled(&self) -> bool {
)]
pub(crate) async fn sync_events_route(
State(services): State<crate::State>,
ClientIp(client_ip): ClientIp,
ClientIp(client_ip): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<sync_events::v3::Request>,
) -> Result<sync_events::v3::Response> {
let sender_user = body.identity.expect_sender_user()?;
+1 -1
View File
@@ -67,7 +67,7 @@
/// [MSC4186]: https://github.com/matrix-org/matrix-spec-proposals/pull/4186
pub(crate) async fn sync_events_v5_route(
State(ref services): State<crate::State>,
ClientIp(client_ip): ClientIp,
ClientIp(client_ip): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<sync_events::v5::Request>,
) -> Result<sync_events::v5::Response> {
let sender_user = body.identity.expect_sender_user()?;
+1 -1
View File
@@ -10,7 +10,7 @@
/// Sets the typing state of the sender user.
pub(crate) async fn create_typing_event_route(
State(services): State<crate::State>,
ClientIp(ip): ClientIp,
ClientIp(ip): ClientIp, // NOTE: Required for updating device metadata
body: Ruma<create_typing_event::v3::Request>,
) -> Result<create_typing_event::v3::Response> {
use create_typing_event::v3::Typing;
+1 -3
View File
@@ -1,7 +1,6 @@
use std::collections::{HashMap, hash_map::Entry};
use axum::extract::State;
use axum_client_ip::ClientIp;
use base64::{Engine as _, engine::general_purpose};
use conduwuit::{
Err, Error, EventTypeExt, PduEvent, Result, debug, err, error,
@@ -28,10 +27,9 @@
/// # `PUT /_matrix/federation/v2/invite/{roomId}/{eventId}`
///
/// Invites a remote user to a room.
#[tracing::instrument(skip_all, fields(%client), name = "invite", level = "info")]
#[tracing::instrument(skip_all, name = "invite", level = "info")]
pub(crate) async fn create_invite_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<create_invite::v2::Request>,
) -> Result<create_invite::v2::Response> {
if !services.server.supported_room_version(&body.room_version) {
+2 -15
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{Err, Result, utils::content_disposition::make_content_disposition};
use conduwuit_service::media::{Dim, FileMeta};
use ruma::api::federation::authenticated_media::{
@@ -12,15 +11,9 @@
/// # `GET /_matrix/federation/v1/media/download/{mediaId}`
///
/// Load media from our server.
#[tracing::instrument(
name = "media_get",
level = "debug",
skip_all,
fields(%client)
)]
#[tracing::instrument(name = "media_get", level = "debug", skip_all)]
pub(crate) async fn get_content_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content::v1::Request>,
) -> Result<get_content::v1::Response> {
let mxc = Mxc {
@@ -54,15 +47,9 @@ pub(crate) async fn get_content_route(
/// # `GET /_matrix/federation/v1/media/thumbnail/{mediaId}`
///
/// Load media thumbnail from our server.
#[tracing::instrument(
name = "media_thumbnail_get",
level = "debug",
skip_all,
fields(%client)
)]
#[tracing::instrument(name = "media_thumbnail_get", level = "debug", skip_all)]
pub(crate) async fn get_content_thumbnail_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_content_thumbnail::v1::Request>,
) -> Result<get_content_thumbnail::v1::Response> {
let dim = Dim::from_ruma(body.width, body.height, body.method.clone())?;
+2 -5
View File
@@ -1,5 +1,4 @@
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{Err, Result, err};
use ruma::{
api::federation::directory::{get_public_rooms, get_public_rooms_filtered},
@@ -12,10 +11,9 @@
/// # `POST /_matrix/federation/v1/publicRooms`
///
/// Lists the public rooms on this server.
#[tracing::instrument(name = "publicrooms", level = "debug", skip_all, fields(%client))]
#[tracing::instrument(name = "publicrooms", level = "debug", skip_all)]
pub(crate) async fn get_public_rooms_filtered_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_public_rooms_filtered::v1::Request>,
) -> Result<get_public_rooms_filtered::v1::Response> {
if !services
@@ -48,10 +46,9 @@ pub(crate) async fn get_public_rooms_filtered_route(
/// # `GET /_matrix/federation/v1/publicRooms`
///
/// Lists the public rooms on this server.
#[tracing::instrument(name = "publicrooms", level = "debug", skip_all, fields(%client))]
#[tracing::instrument(name = "publicrooms", level = "debug", skip_all)]
pub(crate) async fn get_public_rooms_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<get_public_rooms::v1::Request>,
) -> Result<get_public_rooms::v1::Response> {
if !services
+13 -33
View File
@@ -1,11 +1,9 @@
use std::{
collections::{BTreeMap, HashMap},
net::IpAddr,
time::{Duration, Instant},
};
use axum::extract::State;
use axum_client_ip::ClientIp;
use conduwuit::{
Err, Error, Result, debug, debug_error, debug_warn, err, error,
result::LogErr,
@@ -57,7 +55,6 @@
/// Push EDUs and PDUs to this server.
pub(crate) async fn send_transaction_message_route(
State(services): State<crate::State>,
ClientIp(client): ClientIp,
body: Ruma<send_transaction_message::v1::Request>,
) -> Result<send_transaction_message::v1::Response> {
if body.identity != body.body.origin {
@@ -98,7 +95,7 @@ pub(crate) async fn send_transaction_message_route(
services
.server
.runtime()
.spawn(process_inbound_transaction(services, body, client, txn_key, sender));
.spawn(process_inbound_transaction(services, body, txn_key, sender));
// and wait for it
wait_for_result(receiver).await
},
@@ -141,7 +138,6 @@ async fn wait_for_result(
async fn process_inbound_transaction(
services: crate::State,
body: Ruma<send_transaction_message::v1::Request>,
client: IpAddr,
txn_key: TxnKey,
sender: Sender<WrappedTransactionResponse>,
) {
@@ -163,7 +159,7 @@ async fn process_inbound_transaction(
.stream();
debug!(pdus = body.pdus.len(), edus = body.edus.len(), "Processing transaction",);
let results = match handle(&services, &client, &body.identity, pdus, edus).await {
let results = match handle(&services, &body.identity, pdus, edus).await {
| Ok(results) => results,
| Err(err) => {
fail_federation_txn(services, &txn_key, &sender, err);
@@ -236,7 +232,6 @@ fn transaction_error_to_response(err: &TransactionError) -> Error {
}
async fn handle(
services: &Services,
client: &IpAddr,
origin: &ServerName,
pdus: impl Stream<Item = Pdu> + Send,
edus: impl Stream<Item = Edu> + Send,
@@ -257,7 +252,7 @@ async fn handle(
.into_iter()
.try_stream()
.broad_and_then(|(room_id, pdus): (_, Vec<_>)| {
handle_room(services, client, origin, room_id, pdus.into_iter())
handle_room(services, origin, room_id, pdus.into_iter())
.map_ok(Vec::into_iter)
.map_ok(IterStream::try_stream)
})
@@ -267,7 +262,7 @@ async fn handle(
.await?;
// Evaluate EDUs after PDUs in case some of the PDUs then forbid some EDUs.
edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, origin, edu))
.boxed()
.await;
@@ -276,7 +271,6 @@ async fn handle(
async fn handle_room(
services: &Services,
_client: &IpAddr,
origin: &ServerName,
room_id: OwnedRoomId,
pdus: impl Iterator<Item = Pdu> + Send,
@@ -323,25 +317,25 @@ async fn handle_room(
Ok(results)
}
async fn handle_edu(services: &Services, client: &IpAddr, origin: &ServerName, edu: Edu) {
async fn handle_edu(services: &Services, origin: &ServerName, edu: Edu) {
match edu {
| Edu::Presence(presence) if services.server.config.allow_incoming_presence =>
handle_edu_presence(services, client, origin, presence).await,
handle_edu_presence(services, origin, presence).await,
| Edu::Receipt(receipt) if services.server.config.allow_incoming_read_receipts =>
handle_edu_receipt(services, client, origin, receipt).await,
handle_edu_receipt(services, origin, receipt).await,
| Edu::Typing(typing) if services.server.config.allow_incoming_typing =>
handle_edu_typing(services, client, origin, typing).await,
handle_edu_typing(services, origin, typing).await,
| Edu::DeviceListUpdate(content) =>
handle_edu_device_list_update(services, client, origin, content).await,
handle_edu_device_list_update(services, origin, content).await,
| Edu::DirectToDevice(content) =>
handle_edu_direct_to_device(services, client, origin, content).await,
handle_edu_direct_to_device(services, origin, content).await,
| Edu::SigningKeyUpdate(content) =>
handle_edu_signing_key_update(services, client, origin, content).await,
handle_edu_signing_key_update(services, origin, content).await,
| Edu::_Custom(ref _custom) => debug_warn!(?edu, "received custom/unknown EDU"),
@@ -351,7 +345,6 @@ async fn handle_edu(services: &Services, client: &IpAddr, origin: &ServerName, e
async fn handle_edu_presence(
services: &Services,
_client: &IpAddr,
origin: &ServerName,
presence: PresenceContent,
) {
@@ -392,12 +385,7 @@ async fn handle_edu_presence_update(
.ok();
}
async fn handle_edu_receipt(
services: &Services,
_client: &IpAddr,
origin: &ServerName,
receipt: ReceiptContent,
) {
async fn handle_edu_receipt(services: &Services, origin: &ServerName, receipt: ReceiptContent) {
receipt
.receipts
.into_iter()
@@ -492,12 +480,7 @@ async fn handle_edu_receipt_room_user(
.await;
}
async fn handle_edu_typing(
services: &Services,
_client: &IpAddr,
origin: &ServerName,
typing: TypingContent,
) {
async fn handle_edu_typing(services: &Services, origin: &ServerName, typing: TypingContent) {
if typing.user_id.server_name() != origin {
debug_warn!(
%typing.user_id, %origin,
@@ -557,7 +540,6 @@ async fn handle_edu_typing(
async fn handle_edu_device_list_update(
services: &Services,
_client: &IpAddr,
origin: &ServerName,
content: DeviceListUpdateContent,
) {
@@ -576,7 +558,6 @@ async fn handle_edu_device_list_update(
async fn handle_edu_direct_to_device(
services: &Services,
_client: &IpAddr,
origin: &ServerName,
content: DirectDeviceContent,
) {
@@ -696,7 +677,6 @@ async fn handle_edu_direct_to_device_event(
async fn handle_edu_signing_key_update(
services: &Services,
_client: &IpAddr,
origin: &ServerName,
content: SigningKeyUpdateContent,
) {
+37 -24
View File
@@ -2115,12 +2115,10 @@ pub struct Config {
pub stream_amplification: usize,
/// Number of sender task workers; determines sender parallelism. Default is
/// '0' which means the value is determined internally, likely matching the
/// number of tokio worker-threads or number of cores, etc. Override by
/// setting a non-zero value.
/// core count. Override by setting a different value.
///
/// default: 0
#[serde(default)]
/// default: core count
#[serde(default = "default_sender_workers")]
pub sender_workers: usize,
/// Enables listener sockets; can be set to false to disable listening. This
@@ -2689,45 +2687,47 @@ fn default_database_backups_to_keep() -> i16 { 1 }
fn default_db_write_buffer_capacity_mb() -> f64 { 48.0 + parallelism_scaled_f64(4.0) }
fn default_db_cache_capacity_mb() -> f64 { 128.0 + parallelism_scaled_f64(64.0) }
fn default_db_cache_capacity_mb() -> f64 { 512.0 + parallelism_scaled_f64(512.0) }
fn default_pdu_cache_capacity() -> u32 { parallelism_scaled_u32(10_000).saturating_add(100_000) }
fn default_pdu_cache_capacity() -> u32 { parallelism_scaled_u32(50_000).saturating_add(100_000) }
fn default_cache_capacity_modifier() -> f64 { 1.0 }
fn default_auth_chain_cache_capacity() -> u32 {
parallelism_scaled_u32(10_000).saturating_add(100_000)
}
fn default_shorteventid_cache_capacity() -> u32 {
parallelism_scaled_u32(50_000).saturating_add(100_000)
}
fn default_shorteventid_cache_capacity() -> u32 {
parallelism_scaled_u32(100_000).saturating_add(100_000)
}
fn default_eventidshort_cache_capacity() -> u32 {
parallelism_scaled_u32(25_000).saturating_add(100_000)
parallelism_scaled_u32(50_000).saturating_add(100_000)
}
fn default_eventid_pdu_cache_capacity() -> u32 {
parallelism_scaled_u32(25_000).saturating_add(100_000)
parallelism_scaled_u32(50_000).saturating_add(100_000)
}
fn default_shortstatekey_cache_capacity() -> u32 {
parallelism_scaled_u32(10_000).saturating_add(100_000)
parallelism_scaled_u32(100_000).saturating_add(100_000)
}
fn default_statekeyshort_cache_capacity() -> u32 {
parallelism_scaled_u32(10_000).saturating_add(100_000)
parallelism_scaled_u32(50_000).saturating_add(100_000)
}
fn default_servernameevent_data_cache_capacity() -> u32 {
parallelism_scaled_u32(100_000).saturating_add(500_000)
parallelism_scaled_u32(100_000).saturating_add(100_000)
}
fn default_stateinfo_cache_capacity() -> u32 { parallelism_scaled_u32(100) }
fn default_stateinfo_cache_capacity() -> u32 { parallelism_scaled_u32(500).clamp(100, 12000) }
fn default_roomid_spacehierarchy_cache_capacity() -> u32 { parallelism_scaled_u32(1000) }
fn default_roomid_spacehierarchy_cache_capacity() -> u32 {
parallelism_scaled_u32(500).clamp(100, 12000)
}
fn default_dns_cache_entries() -> u32 { 32768 }
fn default_dns_cache_entries() -> u32 { 327_680 }
fn default_dns_min_ttl() -> u64 { 60 * 180 }
@@ -2937,15 +2937,26 @@ fn default_admin_log_capture() -> String {
fn default_admin_room_tag() -> String { "m.server_notice".to_owned() }
#[must_use]
#[allow(clippy::as_conversions, clippy::cast_precision_loss)]
fn parallelism_scaled_f64(val: f64) -> f64 { val * (sys::available_parallelism() as f64) }
pub fn parallelism_scaled_f64(val: f64) -> f64 { val * (sys::available_parallelism() as f64) }
fn parallelism_scaled_u32(val: u32) -> u32 {
let val = val.try_into().expect("failed to cast u32 to usize");
parallelism_scaled(val).try_into().unwrap_or(u32::MAX)
#[must_use]
#[allow(clippy::as_conversions, clippy::cast_possible_truncation)]
pub fn parallelism_scaled_u32(val: u32) -> u32 {
val.saturating_mul(sys::available_parallelism() as u32)
}
fn parallelism_scaled(val: usize) -> usize { val.saturating_mul(sys::available_parallelism()) }
#[must_use]
#[allow(clippy::as_conversions, clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
pub fn parallelism_scaled_i32(val: i32) -> i32 {
val.saturating_mul(sys::available_parallelism() as i32)
}
#[must_use]
pub fn parallelism_scaled(val: usize) -> usize {
val.saturating_mul(sys::available_parallelism())
}
fn default_trusted_server_batch_size() -> usize { 256 }
@@ -2965,6 +2976,8 @@ fn default_stream_width_scale() -> f32 { 1.0 }
fn default_stream_amplification() -> usize { 1024 }
fn default_sender_workers() -> usize { parallelism_scaled(1) }
fn default_client_receive_timeout() -> u64 { 75 }
fn default_client_request_timeout() -> u64 { 180 }
+3
View File
@@ -5,6 +5,7 @@
/// Sha256 hash (input gather joined by 0xFF bytes)
#[must_use]
#[tracing::instrument(skip(inputs), level = "trace")]
#[allow(clippy::unnecessary_fallible_conversions)]
pub fn delimited<'a, T, I>(mut inputs: I) -> DigestOut
where
I: Iterator<Item = T> + 'a,
@@ -25,6 +26,7 @@ pub fn delimited<'a, T, I>(mut inputs: I) -> DigestOut
/// Sha256 hash (input gather)
#[must_use]
#[tracing::instrument(skip(inputs), level = "trace")]
#[allow(clippy::unnecessary_fallible_conversions)]
pub fn concat<'a, T, I>(inputs: I) -> DigestOut
where
I: Iterator<Item = T> + 'a,
@@ -43,6 +45,7 @@ pub fn concat<'a, T, I>(inputs: I) -> DigestOut
#[inline]
#[must_use]
#[tracing::instrument(skip(input), level = "trace")]
#[allow(clippy::unnecessary_fallible_conversions)]
pub fn hash<T>(input: T) -> DigestOut
where
T: AsRef<[u8]>,
+1 -1
View File
@@ -29,7 +29,7 @@ fn descriptor_cf_options(
set_table_options(&mut opts, &desc, cache)?;
opts.set_min_write_buffer_number(1);
opts.set_max_write_buffer_number(2);
opts.set_max_write_buffer_number(3);
opts.set_write_buffer_size(desc.write_size);
opts.set_target_file_size_base(desc.file_size);
+22 -8
View File
@@ -1,10 +1,10 @@
use std::{any::Any, sync::Arc, time::Duration};
use axum::{
Router,
extract::{DefaultBodyLimit, MatchedPath},
Router, extract,
extract::{DefaultBodyLimit, FromRequestParts, MatchedPath},
};
use axum_client_ip::ClientIpSource;
use axum_client_ip::{ClientIp, ClientIpSource};
use conduwuit::{Result, Server, debug, error};
use conduwuit_service::{Services, state::Guard};
use http::{
@@ -20,7 +20,6 @@
timeout::{RequestBodyTimeoutLayer, ResponseBodyTimeoutLayer, TimeoutLayer},
trace::{DefaultOnFailure, DefaultOnRequest, DefaultOnResponse, TraceLayer},
};
use tracing::Level;
use crate::{request, router};
@@ -67,15 +66,16 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
let services_ = services.clone();
let layers = layers
.layer(SetSensitiveHeadersLayer::new([header::AUTHORIZATION]))
.layer(client_ip_layer.into_extension())
.layer(
TraceLayer::new_for_http()
.make_span_with(tracing_span::<_>)
.on_failure(DefaultOnFailure::new().level(Level::ERROR))
.on_request(DefaultOnRequest::new().level(Level::TRACE))
.on_response(DefaultOnResponse::new().level(Level::DEBUG)),
.on_failure(DefaultOnFailure::new().level(tracing::Level::ERROR))
.on_request(DefaultOnRequest::new().level(tracing::Level::TRACE))
.on_response(DefaultOnResponse::new().level(tracing::Level::DEBUG)),
)
.layer(axum::middleware::from_fn(request_ip))
.layer(axum::middleware::from_fn_with_state(Arc::clone(services), request::handle))
.layer(client_ip_layer.into_extension())
.layer(ResponseBodyTimeoutLayer::new(Duration::from_secs(
server.config.client_response_timeout,
)))
@@ -230,11 +230,25 @@ fn tracing_span<T>(request: &http::Request<T>) -> tracing::Span {
parent: None,
debug::INFO_SPAN_LEVEL,
"router",
ip=tracing::field::Empty,
method = %request.method(),
%path,
}
}
/// Annotates the tracing span with the client IP
async fn request_ip(
request: extract::Request,
next: axum::middleware::Next,
) -> axum::response::Response {
let (mut parts, body) = request.into_parts();
if let Ok(ip) = ClientIp::from_request_parts(&mut parts, &()).await {
let span = tracing::Span::current();
span.record("ip", ip.0.to_string());
}
next.run(extract::Request::from_parts(parts, body)).await
}
fn request_path_str<T>(request: &http::Request<T>) -> &str {
request
.uri()
+4 -4
View File
@@ -8,7 +8,7 @@
extract::State,
response::{IntoResponse, Response},
};
use conduwuit::{Result, debug, debug_error, debug_warn, err, error, trace};
use conduwuit::{Result, debug_warn, err, error, info, trace};
use conduwuit_service::Services;
use futures::FutureExt;
use http::{Method, StatusCode, Uri};
@@ -102,11 +102,11 @@ fn handle_result(method: &Method, uri: &Uri, result: Response) -> Result<Respons
let reason = status.canonical_reason().unwrap_or("Unknown Reason");
if status.is_server_error() {
error!(%method, %uri, "{code} {reason}");
info!(%method, %uri, "{code} {reason}");
} else if status.is_client_error() {
debug_error!(%method, %uri, "{code} {reason}");
info!(%method, %uri, "{code} {reason}");
} else if status.is_redirection() {
debug!(%method, %uri, "{code} {reason}");
trace!(%method, %uri, "{code} {reason}");
} else {
trace!(%method, %uri, "{code} {reason}");
}