studio/tuwunel
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chore: fmt

Browse Source
This commit is contained in:
Jared L
2026-01-21 07:28:12 +11:00
parent a91b01f9a2
commit 0f547115af
11 changed files with 193 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/admin/debug/commands.rs
View File

@@ -242,10 +242,11 @@ pub(super) async fn get_remote_pdu(
})
.await
{
| Err(e) =>
| Err(e) => {
return Err!(
"Remote server did not have PDU or failed sending request to remote server: {e}"
),
);
},
| Ok(response) => {
let json: CanonicalJsonObject =
serde_json::from_str(response.pdu.get()).map_err(|e| {
@@ -383,8 +384,9 @@ pub(super) async fn change_log_level(&self, filter: Option<String>, reset: bool)
.reload
.reload(&old_filter_layer, Some(handles))
{
| Err(e) =>
return Err!("Failed to modify and reload the global tracing log level: {e}"),
| Err(e) => {
return Err!("Failed to modify and reload the global tracing log level: {e}");
},
| Ok(()) => {
let value = &self.services.server.config.log;
let out = format!("Successfully changed log level back to config value {value}");
@@ -406,12 +408,14 @@ pub(super) async fn change_log_level(&self, filter: Option<String>, reset: bool)
.reload
.reload(&new_filter_layer, Some(handles))
{
| Ok(()) =>
| Ok(()) => {
return self
.write_str("Successfully changed log level")
.await,
| Err(e) =>
return Err!("Failed to modify and reload the global tracing log level: {e}"),
.await;
},
| Err(e) => {
return Err!("Failed to modify and reload the global tracing log level: {e}");
},
}
}

5
src/admin/user/commands.rs
View File

@@ -130,8 +130,9 @@ pub(super) async fn reset_password(&self, username: String, password: Option<Str
.await
{
| Err(e) => return Err!("Couldn't reset the password for user {user_id}: {e}"),
| Ok(()) =>
write!(self, "Successfully reset the password for user {user_id}: `{new_password}`"),
| Ok(()) => {
write!(self, "Successfully reset the password for user {user_id}: `{new_password}`")
},
}
.await
}

5
src/api/client/state.rs
View File

@@ -349,7 +349,7 @@ async fn allowed_to_send_state_event(
},
}
},
| StateEventType::RoomMember =>
| StateEventType::RoomMember => {
match json.deserialize_as_unchecked::<RoomMemberEventContent>() {
| Ok(membership_content) => {
let Ok(_state_key) = UserId::parse(state_key) else {
@@ -394,7 +394,8 @@ async fn allowed_to_send_state_event(
membership state: {e}"
)));
},
},
}
},
| _ => (),
}

20
src/api/router/auth.rs
View File

@@ -122,11 +122,13 @@ pub(super) async fn auth(
Err(BadRequest(UnknownToken { soft_logout: true }, "Expired access token."))
},
| (AppserviceToken, User(_)) =>
Err!(Request(Unauthorized("Appservice tokens must be used on this endpoint."))),
| (AppserviceToken, User(_)) => {
Err!(Request(Unauthorized("Appservice tokens must be used on this endpoint.")))
},
| (ServerSignatures, Appservice(_) | User(_)) =>
Err!(Request(Unauthorized("Server signatures must be used on this endpoint."))),
| (ServerSignatures, Appservice(_) | User(_)) => {
Err!(Request(Unauthorized("Server signatures must be used on this endpoint.")))
},
| (ServerSignatures, Token::None) => Ok(auth_server(services, request, json_body).await?),
@@ -182,8 +184,9 @@ fn check_auth_still_required(services: &Services, metadata: &Metadata, token: &T
.require_auth_for_profile_requests =>
match token {
| Token::Appservice(_) | Token::User(_) => Ok(()),
| Token::None | Token::Expired(_) | Token::Invalid =>
Err!(Request(MissingToken("Missing or invalid access token."))),
| Token::None | Token::Expired(_) | Token::Invalid => {
Err!(Request(MissingToken("Missing or invalid access token.")))
},
},
| &get_public_rooms::v3::Request::METADATA
if !services
@@ -192,8 +195,9 @@ fn check_auth_still_required(services: &Services, metadata: &Metadata, token: &T
.allow_public_room_directory_without_auth =>
match token {
| Token::Appservice(_) | Token::User(_) => Ok(()),
| Token::None | Token::Expired(_) | Token::Invalid =>
Err!(Request(MissingToken("Missing or invalid access token."))),
| Token::None | Token::Expired(_) | Token::Invalid => {
Err!(Request(MissingToken("Missing or invalid access token.")))
},
},
| _ => Ok(()),
}

7
src/service/admin/grant.rs
View File

@@ -144,7 +144,7 @@ pub async fn make_user_admin(&self, user_id: &UserId) -> Result {
if self.services.server.config.admin_room_notices {
let welcome_message = String::from(
"## Thank you for trying out tuwunel!\n\nTuwunel is a continuation of conduwuit which was technically a hard fork of Conduit.\n\nHelpful links:\n> GitHub Repo: https://github.com/matrix-construct/tuwunel\n> Documentation: https://matrix-construct.github.io/tuwunel\n> Report issues: https://github.com/matrix-construct/tuwunel/issues\n\nFor a list of available commands, send the following message in this room: `!admin --help`"
"## Thank you for trying out tuwunel!\n\nTuwunel is a continuation of conduwuit which was technically a hard fork of Conduit.\n\nHelpful links:\n> GitHub Repo: https://github.com/matrix-construct/tuwunel\n> Documentation: https://matrix-construct.github.io/tuwunel\n> Report issues: https://github.com/matrix-construct/tuwunel/issues\n\nFor a list of available commands, send the following message in this room: `!admin --help`",
);
// Send welcome message
@@ -183,8 +183,9 @@ pub async fn revoke_admin(&self, user_id: &UserId) -> Result {
| Err(e) => return Err!(error!(?e, "Failure occurred while attempting revoke.")),
| Ok(event) if !matches!(event.membership, Invite | Knock | Join) =>
return Err!("Cannot revoke {user_id} in membership state {:?}.", event.membership),
| Ok(event) if !matches!(event.membership, Invite | Knock | Join) => {
return Err!("Cannot revoke {user_id} in membership state {:?}.", event.membership);
},
| Ok(event) => {
assert!(

48
src/service/presence/aggregate.rs
View File

@@ -7,8 +7,8 @@
use std::collections::HashMap;
use ruma::{OwnedDeviceId, OwnedUserId, UInt, UserId, presence::PresenceState};
use tuwunel_core::debug;
use tokio::sync::RwLock;
use tuwunel_core::debug;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) enum DeviceKey {
@@ -45,9 +45,7 @@ impl PresenceAggregator {
pub(crate) fn new() -> Self { Self::default() }
/// Clear all tracked device state.
pub(crate) async fn clear(&self) {
self.inner.write().await.clear();
}
pub(crate) async fn clear(&self) { self.inner.write().await.clear(); }
/// Update presence state for a single device.
#[allow(clippy::too_many_arguments)]
@@ -69,13 +67,15 @@ impl PresenceAggregator {
| Some(ago) => now_ms.saturating_sub(ago.into()),
};
let entry = devices.entry(device_key).or_insert_with(|| DevicePresence {
state: state.clone(),
currently_active: currently_active.unwrap_or(false),
last_active_ts,
last_update_ts: now_ms,
status_msg: status_msg.clone(),
});
let entry = devices
.entry(device_key)
.or_insert_with(|| DevicePresence {
state: state.clone(),
currently_active: currently_active.unwrap_or(false),
last_active_ts,
last_update_ts: now_ms,
status_msg: status_msg.clone(),
});
entry.state = state.clone();
entry.currently_active = currently_active.unwrap_or(false);
@@ -131,14 +131,19 @@ impl PresenceAggregator {
best_state = effective_state.clone();
}
if (effective_state == PresenceState::Online || effective_state == PresenceState::Busy)
if (effective_state == PresenceState::Online
|| effective_state == PresenceState::Busy)
&& device.currently_active
&& last_active_age < idle_timeout_ms
{
any_currently_active = true;
}
if let Some(msg) = device.status_msg.as_ref().filter(|msg| !msg.is_empty()) {
if let Some(msg) = device
.status_msg
.as_ref()
.filter(|msg| !msg.is_empty())
{
match latest_status {
| None => {
latest_status = Some((device.last_update_ts, msg.clone()));
@@ -199,20 +204,18 @@ fn effective_device_state(
offline_timeout_ms: u64,
) -> PresenceState {
match state {
| PresenceState::Busy | PresenceState::Online => {
| PresenceState::Busy | PresenceState::Online =>
if last_active_age >= idle_timeout_ms {
PresenceState::Unavailable
} else {
state.clone()
}
},
| PresenceState::Unavailable => {
},
| PresenceState::Unavailable =>
if last_active_age >= offline_timeout_ms {
PresenceState::Offline
} else {
PresenceState::Unavailable
}
},
},
| PresenceState::Offline => PresenceState::Offline,
| _ => state.clone(),
}
@@ -229,9 +232,10 @@ fn state_rank(state: &PresenceState) -> u8 {
#[cfg(test)]
mod tests {
use super::*;
use ruma::{device_id, uint, user_id};
use super::*;
#[tokio::test]
async fn aggregates_rank_and_status_msg() {
let aggregator = PresenceAggregator::new();
@@ -314,9 +318,7 @@ mod tests {
)
.await;
let aggregated = aggregator
.aggregate(user, 1_000, 100, 100)
.await;
let aggregated = aggregator.aggregate(user, 1_000, 100, 100).await;
assert_eq!(aggregated.device_count, 0);
assert_eq!(aggregated.state, PresenceState::Offline);

16
src/service/presence/mod.rs
View File

@@ -7,17 +7,15 @@ mod presence;
use std::{collections::HashMap, sync::Arc, time::Duration};
use async_trait::async_trait;
use futures::{Stream, StreamExt, TryFutureExt, future::{AbortHandle, Abortable}, stream::FuturesUnordered};
use futures::{
Stream, StreamExt, TryFutureExt,
future::{AbortHandle, Abortable},
stream::FuturesUnordered,
};
use loole::{Receiver, Sender};
use ruma::{
OwnedUserId, UserId, events::presence::PresenceEvent, presence::PresenceState,
};
use ruma::{OwnedUserId, UserId, events::presence::PresenceEvent, presence::PresenceState};
use tokio::sync::RwLock;
use tuwunel_core::{
Result, checked, debug, debug_warn,
result::LogErr,
trace,
};
use tuwunel_core::{Result, checked, debug, debug_warn, result::LogErr, trace};
use self::{aggregate::PresenceAggregator, data::Data, presence::Presence};

104
src/service/presence/pipeline.rs
View File

@@ -8,18 +8,17 @@ use std::time::Duration;
use futures::TryFutureExt;
use ruma::{
DeviceId, OwnedUserId, UInt, UserId,
events::presence::PresenceEvent,
presence::PresenceState,
DeviceId, OwnedUserId, UInt, UserId, events::presence::PresenceEvent, presence::PresenceState,
};
use tokio::time::sleep;
use tuwunel_core::{
Error, Result, debug, error, trace,
Error, Result, debug, error,
result::LogErr,
trace,
utils::{future::OptionFutureExt, option::OptionExt},
};
use super::{TimerFired, aggregate, Service};
use super::{Service, TimerFired, aggregate};
impl Service {
fn device_key(device_id: Option<&DeviceId>, is_remote: bool) -> aggregate::DeviceKey {
@@ -46,8 +45,16 @@ impl Service {
}
let timeout = match presence_state {
| PresenceState::Online => self.services.server.config.presence_idle_timeout_s,
| _ => self.services.server.config.presence_offline_timeout_s,
| PresenceState::Online =>
self.services
.server
.config
.presence_idle_timeout_s,
| _ =>
self.services
.server
.config
.presence_offline_timeout_s,
};
self.timer_channel
@@ -136,27 +143,25 @@ impl Service {
};
if !state_changed
&& let Some((count, last_last_active_ago)) = Self::refresh_skip_decision(
refresh_window_ms,
last_event.as_ref(),
last_count,
) {
let presence = last_event
.as_ref()
.map(|event| &event.content.presence)
.unwrap_or(state);
&& let Some((count, last_last_active_ago)) =
Self::refresh_skip_decision(refresh_window_ms, last_event.as_ref(), last_count)
{
let presence = last_event
.as_ref()
.map(|event| &event.content.presence)
.unwrap_or(state);
self.schedule_presence_timer(user_id, presence, count)
.log_err()
.ok();
debug!(
?user_id,
?state,
last_last_active_ago,
"Skipping presence update: refresh window (timer rescheduled)"
);
return Ok(());
}
self.schedule_presence_timer(user_id, presence, count)
.log_err()
.ok();
debug!(
?user_id,
?state,
last_last_active_ago,
"Skipping presence update: refresh window (timer rescheduled)"
);
return Ok(());
}
let fallback_status = last_event
.and_then(|event| event.content.status_msg)
@@ -213,12 +218,7 @@ impl Service {
Some(REFRESH_TIMEOUT),
);
debug!(
?user_id,
?new_state,
currently_active,
"Presence ping accepted"
);
debug!(?user_id, ?new_state, currently_active, "Presence ping accepted");
futures::future::try_join(set_presence, update_device_seen.unwrap_or(Ok(())))
.map_ok(|_| ())
@@ -309,12 +309,7 @@ impl Service {
};
if Self::timer_is_stale(expected_count, current_count) {
trace!(
?user_id,
expected_count,
current_count,
"Skipping stale presence timer"
);
trace!(?user_id, expected_count, current_count, "Skipping stale presence timer");
return Ok(());
}
@@ -339,24 +334,21 @@ impl Service {
};
debug!(
"Processed presence timer for user '{user_id}': Old state = {presence_state}, New \
state = {new_state:?}"
"Processed presence timer for user '{user_id}': Old state = {presence_state}, \
New state = {new_state:?}"
);
if let Some(new_state) = new_state {
if matches!(new_state, PresenceState::Unavailable | PresenceState::Offline) {
self.services
.sending
.schedule_flush_suppressed_for_user(user_id.to_owned(), "presence->inactive");
.schedule_flush_suppressed_for_user(
user_id.to_owned(),
"presence->inactive",
);
}
self.set_presence(
user_id,
&new_state,
Some(false),
last_active_ago,
status_msg,
)
.await?;
self.set_presence(user_id, &new_state, Some(false), last_active_ago, status_msg)
.await?;
}
return Ok(());
@@ -369,16 +361,15 @@ impl Service {
return Ok(());
}
if matches!(
aggregated.state,
PresenceState::Unavailable | PresenceState::Offline
) {
if matches!(aggregated.state, PresenceState::Unavailable | PresenceState::Offline) {
self.services
.sending
.schedule_flush_suppressed_for_user(user_id.to_owned(), "presence->inactive");
}
let status_msg = aggregated.status_msg.or_else(|| presence.status_msg());
let status_msg = aggregated
.status_msg
.or_else(|| presence.status_msg());
let last_active_ago =
Some(UInt::new_saturating(now.saturating_sub(aggregated.last_active_ts)));
@@ -407,9 +398,10 @@ pub(super) async fn presence_timer(
#[cfg(test)]
mod tests {
use super::*;
use ruma::{presence::PresenceState, uint, user_id};
use super::*;
#[test]
fn refresh_window_skip_decision() {
let user_id = user_id!("@alice:example.com");

31
src/service/pusher/suppressed.rs
View File

@@ -3,8 +3,10 @@
//! Stores suppressed push events in memory until they can be flushed. This is
//! intentionally in-memory only: suppressed events are discarded on restart.
use std::collections::{HashMap, VecDeque};
use std::sync::Mutex;
use std::{
collections::{HashMap, VecDeque},
sync::Mutex,
};
use ruma::{OwnedRoomId, OwnedUserId, RoomId, UserId};
use tuwunel_core::{debug, implement, trace, utils};
@@ -45,7 +47,10 @@ impl SuppressedQueue {
}
fn drain_room(queue: VecDeque<SuppressedEvent>) -> Vec<RawPduId> {
queue.into_iter().map(|event| event.pdu_id).collect()
queue
.into_iter()
.map(|event| event.pdu_id)
.collect()
}
fn drop_one_front(queue: &mut VecDeque<SuppressedEvent>, total_events: &mut usize) -> bool {
@@ -69,9 +74,7 @@ pub fn queue_suppressed_push(
) -> bool {
let mut inner = self.suppressed.lock();
let user_entry = inner.entry(user_id.to_owned()).or_default();
let push_entry = user_entry
.entry(pushkey.to_owned())
.or_default();
let push_entry = user_entry.entry(pushkey.to_owned()).or_default();
if !push_entry.rooms.contains_key(room_id)
&& push_entry.rooms.len() >= SUPPRESSED_MAX_ROOMS_PER_PUSHKEY
@@ -95,12 +98,7 @@ pub fn queue_suppressed_push(
.back()
.is_some_and(|event| event.pdu_id == pdu_id)
{
trace!(
?user_id,
?room_id,
pushkey,
"Suppressed push event is duplicate; skipping"
);
trace!(?user_id, ?room_id, pushkey, "Suppressed push event is duplicate; skipping");
return false;
}
@@ -161,10 +159,7 @@ pub fn take_suppressed_for_pushkey(
/// Take and remove all suppressed PDUs for a given user across all pushkeys.
#[implement(super::Service)]
pub fn take_suppressed_for_user(
&self,
user_id: &UserId,
) -> SuppressedPushes {
pub fn take_suppressed_for_user(&self, user_id: &UserId) -> SuppressedPushes {
let mut inner = self.suppressed.lock();
let Some(user_entry) = inner.remove(user_id) else {
return Vec::new();
@@ -195,7 +190,9 @@ pub fn clear_suppressed_room(&self, user_id: &UserId, room_id: &RoomId) -> usize
user_entry.retain(|_, push_entry| {
if let Some(queue) = push_entry.rooms.remove(room_id) {
removed = removed.saturating_add(queue.len());
push_entry.total_events = push_entry.total_events.saturating_sub(queue.len());
push_entry.total_events = push_entry
.total_events
.saturating_sub(queue.len());
}
!push_entry.rooms.is_empty()

113
src/service/sending/sender.rs
View File

@@ -50,9 +50,8 @@ use tuwunel_core::{
warn,
};
use crate::rooms::timeline::RawPduId;
use super::{Destination, EduBuf, EduVec, Msg, SendingEvent, Service, data::QueueItem};
use crate::rooms::timeline::RawPduId;
#[derive(Debug)]
enum TransactionStatus {
@@ -736,13 +735,14 @@ impl Service {
pdu_jsons.push(pdu.to_format());
}
},
| SendingEvent::Edu(edu) =>
| SendingEvent::Edu(edu) => {
if appservice.receive_ephemeral
&& let Ok(edu) =
serde_json::from_slice(edu).and_then(|edu| Raw::new(&edu))
{
edu_jsons.push(edu);
},
}
},
| SendingEvent::Flush => {}, // flush only; no new content
}
}
@@ -875,15 +875,13 @@ impl Service {
});
}
pub fn schedule_flush_suppressed_for_user(
&self,
user_id: OwnedUserId,
reason: &'static str,
) {
pub fn schedule_flush_suppressed_for_user(&self, user_id: OwnedUserId, reason: &'static str) {
let sending = self.services.sending.clone();
let runtime = self.server.runtime();
runtime.spawn(async move {
sending.flush_suppressed_for_user(user_id, reason).await;
sending
.flush_suppressed_for_user(user_id, reason)
.await;
});
}
@@ -899,7 +897,12 @@ impl Service {
continue;
};
let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await else {
let Ok(pdu) = self
.services
.timeline
.get_pdu_from_id(pdu_id)
.await
else {
debug!(?user_id, ?pdu_id, "Suppressing push but PDU is missing");
continue;
};
@@ -909,11 +912,12 @@ impl Service {
continue;
}
if self
.services
.pusher
.queue_suppressed_push(user_id, pushkey, pdu.room_id(), *pdu_id)
{
if self.services.pusher.queue_suppressed_push(
user_id,
pushkey,
pdu.room_id(),
*pdu_id,
) {
queued = queued.saturating_add(1);
}
}
@@ -934,12 +938,7 @@ impl Service {
return;
}
debug!(
?user_id,
pushkey,
rooms = rooms.len(),
"Flushing suppressed pushes ({reason})"
);
debug!(?user_id, pushkey, rooms = rooms.len(), "Flushing suppressed pushes ({reason})");
for (room_id, pdu_ids) in rooms {
let unread = self
@@ -948,16 +947,17 @@ impl Service {
.notification_count(user_id, &room_id)
.await;
if unread == 0 {
trace!(
?user_id,
?room_id,
"Skipping suppressed push flush: no unread"
);
trace!(?user_id, ?room_id, "Skipping suppressed push flush: no unread");
continue;
}
for pdu_id in pdu_ids {
let Ok(pdu) = self.services.timeline.get_pdu_from_id(&pdu_id).await else {
let Ok(pdu) = self
.services
.timeline
.get_pdu_from_id(&pdu_id)
.await
else {
debug!(?user_id, ?pdu_id, "Suppressed PDU missing during flush");
continue;
};
@@ -973,12 +973,10 @@ impl Service {
.send_push_notice(user_id, pusher, rules_for_user, &pdu)
.await
{
let requeued = self.services.pusher.queue_suppressed_push(
user_id,
pushkey,
&room_id,
pdu_id,
);
let requeued = self
.services
.pusher
.queue_suppressed_push(user_id, pushkey, &room_id, pdu_id);
warn!(
?user_id,
?room_id,
@@ -1005,7 +1003,12 @@ impl Service {
return;
}
let pusher = match self.services.pusher.get_pusher(&user_id, &pushkey).await {
let pusher = match self
.services
.pusher
.get_pusher(&user_id, &pushkey)
.await
{
| Ok(pusher) => pusher,
| Err(error) => {
warn!(?user_id, pushkey, ?error, "Missing pusher for suppressed flush");
@@ -1034,12 +1037,11 @@ impl Service {
.await;
}
pub async fn flush_suppressed_for_user(
&self,
user_id: OwnedUserId,
reason: &'static str,
) {
let suppressed = self.services.pusher.take_suppressed_for_user(&user_id);
pub async fn flush_suppressed_for_user(&self, user_id: OwnedUserId, reason: &'static str) {
let suppressed = self
.services
.pusher
.take_suppressed_for_user(&user_id);
if suppressed.is_empty() {
return;
}
@@ -1055,7 +1057,12 @@ impl Service {
};
for (pushkey, rooms) in suppressed {
let pusher = match self.services.pusher.get_pusher(&user_id, &pushkey).await {
let pusher = match self
.services
.pusher
.get_pusher(&user_id, &pushkey)
.await
{
| Ok(pusher) => pusher,
| Err(error) => {
warn!(?user_id, pushkey, ?error, "Missing pusher for suppressed flush");
@@ -1104,11 +1111,7 @@ impl Service {
.unwrap_or(u64::MAX);
if presence_age_ms >= 65_000 {
debug!(
?user_id,
presence_age_ms,
"push not suppressed: presence too old"
);
debug!(?user_id, presence_age_ms, "push not suppressed: presence too old");
return false;
}
@@ -1121,10 +1124,18 @@ impl Service {
let considered_active = sync_gap_ms.is_some_and(|gap| gap < 32_000);
match sync_gap_ms {
| Some(gap) if gap < 32_000 =>
debug!(?user_id, presence_age_ms, sync_gap_ms = gap, "suppressing push: active heuristic"),
| Some(gap) =>
debug!(?user_id, presence_age_ms, sync_gap_ms = gap, "push not suppressed: sync gap too large"),
| Some(gap) if gap < 32_000 => debug!(
?user_id,
presence_age_ms,
sync_gap_ms = gap,
"suppressing push: active heuristic"
),
| Some(gap) => debug!(
?user_id,
presence_age_ms,
sync_gap_ms = gap,
"push not suppressed: sync gap too large"
),
| None => debug!(?user_id, presence_age_ms, "push not suppressed: no recent sync"),
}

5
src/service/users/mod.rs
View File

@@ -264,10 +264,11 @@ impl Service {
self.db.userid_password.insert(user_id, hash);
self.db.userid_origin.insert(user_id, "password");
},
| Some(Err(e)) =>
| Some(Err(e)) => {
return Err!(Request(InvalidParam(
"Password does not meet the requirements: {e}"
))),
)));
},
}
Ok(())