MirrorHub/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2024-11-16 19:10:50 +01:00
Code Issues Releases Activity

refactor: use async-aware RwLocks and Mutexes where possible

Browse source
This commit is contained in:
Matthias Ahouansou 2024-03-05 14:22:54 +00:00
parent 57575b7c6f
commit becaad677f
No known key found for this signature in database
21 changed files with 1171 additions and 1006 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
Cargo.lock generated
View file

@ -80,6 +80,17 @@ version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5f093eed78becd229346bf859eec0aa4dd7ddde0757287b2b4107a1f09c80002" checksum = "5f093eed78becd229346bf859eec0aa4dd7ddde0757287b2b4107a1f09c80002"
[[package]]
name = "async-recursion"
version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5fd55a5ba1179988837d24ab4c7cc8ed6efdeff578ede0416b4225a5fca35bd0"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.48",
]
[[package]] [[package]]
name = "async-trait" name = "async-trait"
version = "0.1.77" version = "0.1.77"
@ -374,6 +385,7 @@ checksum = "3d7b894f5411737b7867f4827955924d7c254fc9f4d91a6aad6b097804b1018b"
name = "conduit" name = "conduit"
version = "0.7.0-alpha" version = "0.7.0-alpha"
dependencies = [ dependencies = [
"async-recursion",
"async-trait", "async-trait",
"axum", "axum",
"axum-server", "axum-server",

1
Cargo.toml
View file

@ -115,6 +115,7 @@ lazy_static = "1.4.0"
async-trait = "0.1.68" async-trait = "0.1.68"
sd-notify = { version = "0.4.1", optional = true } sd-notify = { version = "0.4.1", optional = true }
async-recursion = "1.0.5"
[target.'cfg(unix)'.dependencies] [target.'cfg(unix)'.dependencies]
nix = { version = "0.26.2", features = ["resource"] } nix = { version = "0.26.2", features = ["resource"] }

11
src/api/client_server/keys.rs
View file

@ -339,17 +339,19 @@ pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool>(
let mut failures = BTreeMap::new(); let mut failures = BTreeMap::new();
let back_off = |id| match services() let back_off = |id| async {
match services()
.globals .globals
.bad_query_ratelimiter .bad_query_ratelimiter
.write() .write()
.unwrap() .await
.entry(id) .entry(id)
{ {
hash_map::Entry::Vacant(e) => { hash_map::Entry::Vacant(e) => {
e.insert((Instant::now(), 1)); e.insert((Instant::now(), 1));
} }
hash_map::Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1), hash_map::Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1),
}
}; };
let mut futures: FuturesUnordered<_> = get_over_federation let mut futures: FuturesUnordered<_> = get_over_federation
@ -359,7 +361,7 @@ pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool>(
.globals .globals
.bad_query_ratelimiter .bad_query_ratelimiter
.read() .read()
.unwrap() .await
.get(server) .get(server)
{ {
// Exponential backoff // Exponential backoff
@ -428,7 +430,8 @@ pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool>(
device_keys.extend(response.device_keys); device_keys.extend(response.device_keys);
} }
_ => { _ => {
back_off(server.to_owned()); back_off(server.to_owned()).await;
failures.insert(server.to_string(), json!({})); failures.insert(server.to_string(), json!({}));
} }
} }

99
src/api/client_server/membership.rs
View file

@ -26,9 +26,10 @@ use ruma::{
use serde_json::value::{to_raw_value, RawValue as RawJsonValue}; use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
use std::{ use std::{
collections::{hash_map::Entry, BTreeMap, HashMap, HashSet}, collections::{hash_map::Entry, BTreeMap, HashMap, HashSet},
sync::{Arc, RwLock}, sync::Arc,
time::{Duration, Instant}, time::{Duration, Instant},
}; };
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn}; use tracing::{debug, error, info, warn};
use crate::{ use crate::{
@ -212,13 +213,16 @@ pub async fn kick_user_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(body.room_id.clone()) .entry(body.room_id.clone())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&event).expect("event is valid, we just created it"), content: to_raw_value(&event).expect("event is valid, we just created it"),
@ -229,7 +233,8 @@ pub async fn kick_user_route(
sender_user, sender_user,
&body.room_id, &body.room_id,
&state_lock, &state_lock,
)?; )
.await?;
drop(state_lock); drop(state_lock);
@ -276,13 +281,16 @@ pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_use
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(body.room_id.clone()) .entry(body.room_id.clone())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&event).expect("event is valid, we just created it"), content: to_raw_value(&event).expect("event is valid, we just created it"),
@ -293,7 +301,8 @@ pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_use
sender_user, sender_user,
&body.room_id, &body.room_id,
&state_lock, &state_lock,
)?; )
.await?;
drop(state_lock); drop(state_lock);
@ -334,13 +343,16 @@ pub async fn unban_user_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(body.room_id.clone()) .entry(body.room_id.clone())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&event).expect("event is valid, we just created it"), content: to_raw_value(&event).expect("event is valid, we just created it"),
@ -351,7 +363,8 @@ pub async fn unban_user_route(
sender_user, sender_user,
&body.room_id, &body.room_id,
&state_lock, &state_lock,
)?; )
.await?;
drop(state_lock); drop(state_lock);
@ -489,7 +502,7 @@ async fn join_room_by_id_helper(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
@ -680,7 +693,7 @@ async fn join_room_by_id_helper(
.iter() .iter()
.map(|pdu| validate_and_add_event_id(pdu, &room_version_id, &pub_key_map)) .map(|pdu| validate_and_add_event_id(pdu, &room_version_id, &pub_key_map))
{ {
let (event_id, value) = match result { let (event_id, value) = match result.await {
Ok(t) => t, Ok(t) => t,
Err(_) => continue, Err(_) => continue,
}; };
@ -710,7 +723,7 @@ async fn join_room_by_id_helper(
.iter() .iter()
.map(|pdu| validate_and_add_event_id(pdu, &room_version_id, &pub_key_map)) .map(|pdu| validate_and_add_event_id(pdu, &room_version_id, &pub_key_map))
{ {
let (event_id, value) = match result { let (event_id, value) = match result.await {
Ok(t) => t, Ok(t) => t,
Err(_) => continue, Err(_) => continue,
}; };
@ -784,12 +797,16 @@ async fn join_room_by_id_helper(
let statehash_after_join = services().rooms.state.append_to_state(&parsed_join_pdu)?; let statehash_after_join = services().rooms.state.append_to_state(&parsed_join_pdu)?;
info!("Appending new room join event"); info!("Appending new room join event");
services().rooms.timeline.append_pdu( services()
.rooms
.timeline
.append_pdu(
&parsed_join_pdu, &parsed_join_pdu,
join_event, join_event,
vec![(*parsed_join_pdu.event_id).to_owned()], vec![(*parsed_join_pdu.event_id).to_owned()],
&state_lock, &state_lock,
)?; )
.await?;
info!("Setting final room state for new room"); info!("Setting final room state for new room");
// We set the room state after inserting the pdu, so that we never have a moment in time // We set the room state after inserting the pdu, so that we never have a moment in time
@ -902,7 +919,10 @@ async fn join_room_by_id_helper(
}; };
// Try normal join first // Try normal join first
let error = match services().rooms.timeline.build_and_append_pdu( let error = match services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&event).expect("event is valid, we just created it"), content: to_raw_value(&event).expect("event is valid, we just created it"),
@ -913,7 +933,9 @@ async fn join_room_by_id_helper(
sender_user, sender_user,
room_id, room_id,
&state_lock, &state_lock,
) { )
.await
{
Ok(_event_id) => return Ok(join_room_by_id::v3::Response::new(room_id.to_owned())), Ok(_event_id) => return Ok(join_room_by_id::v3::Response::new(room_id.to_owned())),
Err(e) => e, Err(e) => e,
}; };
@ -1109,7 +1131,7 @@ async fn make_join_request(
make_join_response_and_server make_join_response_and_server
} }
fn validate_and_add_event_id( async fn validate_and_add_event_id(
pdu: &RawJsonValue, pdu: &RawJsonValue,
room_version: &RoomVersionId, room_version: &RoomVersionId,
pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>, pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
@ -1125,24 +1147,26 @@ fn validate_and_add_event_id(
)) ))
.expect("ruma's reference hashes are valid event ids"); .expect("ruma's reference hashes are valid event ids");
let back_off = |id| match services() let back_off = |id| async {
match services()
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.write() .write()
.unwrap() .await
.entry(id) .entry(id)
{ {
Entry::Vacant(e) => { Entry::Vacant(e) => {
e.insert((Instant::now(), 1)); e.insert((Instant::now(), 1));
} }
Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1), Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1),
}
}; };
if let Some((time, tries)) = services() if let Some((time, tries)) = services()
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.read() .read()
.unwrap() .await
.get(&event_id) .get(&event_id)
{ {
// Exponential backoff // Exponential backoff
@ -1157,15 +1181,10 @@ fn validate_and_add_event_id(
} }
} }
if let Err(e) = ruma::signatures::verify_event( if let Err(e) = ruma::signatures::verify_event(&*pub_key_map.read().await, &value, room_version)
&*pub_key_map {
.read()
.map_err(|_| Error::bad_database("RwLock is poisoned."))?,
&value,
room_version,
) {
warn!("Event {} failed verification {:?} {}", event_id, pdu, e); warn!("Event {} failed verification {:?} {}", event_id, pdu, e);
back_off(event_id); back_off(event_id).await;
return Err(Error::BadServerResponse("Event failed verification.")); return Err(Error::BadServerResponse("Event failed verification."));
} }
@ -1191,7 +1210,7 @@ pub(crate) async fn invite_helper<'a>(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
@ -1312,13 +1331,16 @@ pub(crate) async fn invite_helper<'a>(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&RoomMemberEventContent { content: to_raw_value(&RoomMemberEventContent {
@ -1339,7 +1361,8 @@ pub(crate) async fn invite_helper<'a>(
sender_user, sender_user,
room_id, room_id,
&state_lock, &state_lock,
)?; )
.await?;
drop(state_lock); drop(state_lock);
@ -1407,7 +1430,7 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId, reason: Option<Strin
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
@ -1443,7 +1466,10 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId, reason: Option<Strin
event.membership = MembershipState::Leave; event.membership = MembershipState::Leave;
event.reason = reason; event.reason = reason;
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&event).expect("event is valid, we just created it"), content: to_raw_value(&event).expect("event is valid, we just created it"),
@ -1454,7 +1480,8 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId, reason: Option<Strin
user_id, user_id,
room_id, room_id,
&state_lock, &state_lock,
)?; )
.await?;
} }
Ok(()) Ok(())

21
src/api/client_server/message.rs
View file

@ -32,7 +32,7 @@ pub async fn send_message_event_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(body.room_id.clone()) .entry(body.room_id.clone())
.or_default(), .or_default(),
); );
@ -73,7 +73,10 @@ pub async fn send_message_event_route(
let mut unsigned = BTreeMap::new(); let mut unsigned = BTreeMap::new();
unsigned.insert("transaction_id".to_owned(), body.txn_id.to_string().into()); unsigned.insert("transaction_id".to_owned(), body.txn_id.to_string().into());
let event_id = services().rooms.timeline.build_and_append_pdu( let event_id = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: body.event_type.to_string().into(), event_type: body.event_type.to_string().into(),
content: serde_json::from_str(body.body.body.json().get()) content: serde_json::from_str(body.body.body.json().get())
@ -85,7 +88,8 @@ pub async fn send_message_event_route(
sender_user, sender_user,
&body.room_id, &body.room_id,
&state_lock, &state_lock,
)?; )
.await?;
services().transaction_ids.add_txnid( services().transaction_ids.add_txnid(
sender_user, sender_user,
@ -126,12 +130,11 @@ pub async fn get_message_events_route(
.as_ref() .as_ref()
.and_then(|t| PduCount::try_from_string(t).ok()); .and_then(|t| PduCount::try_from_string(t).ok());
services().rooms.lazy_loading.lazy_load_confirm_delivery( services()
sender_user, .rooms
sender_device, .lazy_loading
&body.room_id, .lazy_load_confirm_delivery(sender_user, sender_device, &body.room_id, from)
from, .await?;
)?;
let limit = u64::from(body.limit).min(100) as usize; let limit = u64::from(body.limit).min(100) as usize;

26
src/api/client_server/profile.rs
View file

@ -77,18 +77,17 @@ pub async fn set_displayname_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let _ = services().rooms.timeline.build_and_append_pdu( let _ = services()
pdu_builder, .rooms
sender_user, .timeline
&room_id, .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
&state_lock, .await;
);
// Presence update // Presence update
services().rooms.edus.presence.update_presence( services().rooms.edus.presence.update_presence(
@ -212,18 +211,17 @@ pub async fn set_avatar_url_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let _ = services().rooms.timeline.build_and_append_pdu( let _ = services()
pdu_builder, .rooms
sender_user, .timeline
&room_id, .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
&state_lock, .await;
);
// Presence update // Presence update
services().rooms.edus.presence.update_presence( services().rooms.edus.presence.update_presence(

10
src/api/client_server/redact.rs
View file

@ -24,13 +24,16 @@ pub async fn redact_event_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(body.room_id.clone()) .entry(body.room_id.clone())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let event_id = services().rooms.timeline.build_and_append_pdu( let event_id = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomRedaction, event_type: TimelineEventType::RoomRedaction,
content: to_raw_value(&RoomRedactionEventContent { content: to_raw_value(&RoomRedactionEventContent {
@ -45,7 +48,8 @@ pub async fn redact_event_route(
sender_user, sender_user,
&body.room_id, &body.room_id,
&state_lock, &state_lock,
)?; )
.await?;
drop(state_lock); drop(state_lock);

129
src/api/client_server/room.rs
View file

@ -61,7 +61,7 @@ pub async fn create_room_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
@ -174,7 +174,10 @@ pub async fn create_room_route(
} }
// 1. The room create event // 1. The room create event
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomCreate, event_type: TimelineEventType::RoomCreate,
content: to_raw_value(&content).expect("event is valid, we just created it"), content: to_raw_value(&content).expect("event is valid, we just created it"),
@ -185,10 +188,14 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 2. Let the room creator join // 2. Let the room creator join
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&RoomMemberEventContent { content: to_raw_value(&RoomMemberEventContent {
@ -209,7 +216,8 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 3. Power levels // 3. Power levels
@ -246,7 +254,10 @@ pub async fn create_room_route(
} }
} }
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomPowerLevels, event_type: TimelineEventType::RoomPowerLevels,
content: to_raw_value(&power_levels_content) content: to_raw_value(&power_levels_content)
@ -258,11 +269,15 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 4. Canonical room alias // 4. Canonical room alias
if let Some(room_alias_id) = &alias { if let Some(room_alias_id) = &alias {
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomCanonicalAlias, event_type: TimelineEventType::RoomCanonicalAlias,
content: to_raw_value(&RoomCanonicalAliasEventContent { content: to_raw_value(&RoomCanonicalAliasEventContent {
@ -277,13 +292,17 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
} }
// 5. Events set by preset // 5. Events set by preset
// 5.1 Join Rules // 5.1 Join Rules
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomJoinRules, event_type: TimelineEventType::RoomJoinRules,
content: to_raw_value(&RoomJoinRulesEventContent::new(match preset { content: to_raw_value(&RoomJoinRulesEventContent::new(match preset {
@ -299,10 +318,14 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 5.2 History Visibility // 5.2 History Visibility
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomHistoryVisibility, event_type: TimelineEventType::RoomHistoryVisibility,
content: to_raw_value(&RoomHistoryVisibilityEventContent::new( content: to_raw_value(&RoomHistoryVisibilityEventContent::new(
@ -316,10 +339,14 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 5.3 Guest Access // 5.3 Guest Access
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomGuestAccess, event_type: TimelineEventType::RoomGuestAccess,
content: to_raw_value(&RoomGuestAccessEventContent::new(match preset { content: to_raw_value(&RoomGuestAccessEventContent::new(match preset {
@ -334,7 +361,8 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 6. Events listed in initial_state // 6. Events listed in initial_state
for event in &body.initial_state { for event in &body.initial_state {
@ -353,17 +381,19 @@ pub async fn create_room_route(
continue; continue;
} }
services().rooms.timeline.build_and_append_pdu( services()
pdu_builder, .rooms
sender_user, .timeline
&room_id, .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
&state_lock, .await?;
)?;
} }
// 7. Events implied by name and topic // 7. Events implied by name and topic
if let Some(name) = &body.name { if let Some(name) = &body.name {
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomName, event_type: TimelineEventType::RoomName,
content: to_raw_value(&RoomNameEventContent::new(name.clone())) content: to_raw_value(&RoomNameEventContent::new(name.clone()))
@ -375,11 +405,15 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
} }
if let Some(topic) = &body.topic { if let Some(topic) = &body.topic {
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomTopic, event_type: TimelineEventType::RoomTopic,
content: to_raw_value(&RoomTopicEventContent { content: to_raw_value(&RoomTopicEventContent {
@ -393,7 +427,8 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
} }
// 8. Events implied by invite (and TODO: invite_3pid) // 8. Events implied by invite (and TODO: invite_3pid)
@ -523,7 +558,7 @@ pub async fn upgrade_room_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(body.room_id.clone()) .entry(body.room_id.clone())
.or_default(), .or_default(),
); );
@ -531,7 +566,10 @@ pub async fn upgrade_room_route(
// Send a m.room.tombstone event to the old room to indicate that it is not intended to be used any further // Send a m.room.tombstone event to the old room to indicate that it is not intended to be used any further
// Fail if the sender does not have the required permissions // Fail if the sender does not have the required permissions
let tombstone_event_id = services().rooms.timeline.build_and_append_pdu( let tombstone_event_id = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomTombstone, event_type: TimelineEventType::RoomTombstone,
content: to_raw_value(&RoomTombstoneEventContent { content: to_raw_value(&RoomTombstoneEventContent {
@ -546,7 +584,8 @@ pub async fn upgrade_room_route(
sender_user, sender_user,
&body.room_id, &body.room_id,
&state_lock, &state_lock,
)?; )
.await?;
// Change lock to replacement room // Change lock to replacement room
drop(state_lock); drop(state_lock);
@ -555,7 +594,7 @@ pub async fn upgrade_room_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(replacement_room.clone()) .entry(replacement_room.clone())
.or_default(), .or_default(),
); );
@ -613,7 +652,10 @@ pub async fn upgrade_room_route(
)); ));
} }
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomCreate, event_type: TimelineEventType::RoomCreate,
content: to_raw_value(&create_event_content) content: to_raw_value(&create_event_content)
@ -625,10 +667,14 @@ pub async fn upgrade_room_route(
sender_user, sender_user,
&replacement_room, &replacement_room,
&state_lock, &state_lock,
)?; )
.await?;
// Join the new room // Join the new room
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&RoomMemberEventContent { content: to_raw_value(&RoomMemberEventContent {
@ -649,7 +695,8 @@ pub async fn upgrade_room_route(
sender_user, sender_user,
&replacement_room, &replacement_room,
&state_lock, &state_lock,
)?; )
.await?;
// Recommended transferable state events list from the specs // Recommended transferable state events list from the specs
let transferable_state_events = vec![ let transferable_state_events = vec![
@ -676,7 +723,10 @@ pub async fn upgrade_room_route(
None => continue, // Skipping missing events. None => continue, // Skipping missing events.
}; };
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: event_type.to_string().into(), event_type: event_type.to_string().into(),
content: event_content, content: event_content,
@ -687,7 +737,8 @@ pub async fn upgrade_room_route(
sender_user, sender_user,
&replacement_room, &replacement_room,
&state_lock, &state_lock,
)?; )
.await?;
} }
// Moves any local aliases to the new room // Moves any local aliases to the new room
@ -721,7 +772,10 @@ pub async fn upgrade_room_route(
power_levels_event_content.invite = new_level; power_levels_event_content.invite = new_level;
// Modify the power levels in the old room to prevent sending of events and inviting new users // Modify the power levels in the old room to prevent sending of events and inviting new users
let _ = services().rooms.timeline.build_and_append_pdu( let _ = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomPowerLevels, event_type: TimelineEventType::RoomPowerLevels,
content: to_raw_value(&power_levels_event_content) content: to_raw_value(&power_levels_event_content)
@ -733,7 +787,8 @@ pub async fn upgrade_room_route(
sender_user, sender_user,
&body.room_id, &body.room_id,
&state_lock, &state_lock,
)?; )
.await?;
drop(state_lock); drop(state_lock);

10
src/api/client_server/state.rs
View file

@ -227,13 +227,16 @@ async fn send_state_event_for_key_helper(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let event_id = services().rooms.timeline.build_and_append_pdu( let event_id = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: event_type.to_string().into(), event_type: event_type.to_string().into(),
content: serde_json::from_str(json.json().get()).expect("content is valid json"), content: serde_json::from_str(json.json().get()).expect("content is valid json"),
@ -244,7 +247,8 @@ async fn send_state_event_for_key_helper(
sender_user, sender_user,
room_id, room_id,
&state_lock, &state_lock,
)?; )
.await?;
Ok(event_id) Ok(event_id)
} }

70
src/api/client_server/sync.rs
View file

@ -1,6 +1,7 @@
use crate::{ use crate::{
service::rooms::timeline::PduCount, services, Error, PduEvent, Result, Ruma, RumaResponse, service::rooms::timeline::PduCount, services, Error, PduEvent, Result, Ruma, RumaResponse,
}; };
use ruma::{ use ruma::{
api::client::{ api::client::{
filter::{FilterDefinition, LazyLoadOptions}, filter::{FilterDefinition, LazyLoadOptions},
@ -75,7 +76,7 @@ pub async fn sync_events_route(
.globals .globals
.sync_receivers .sync_receivers
.write() .write()
.unwrap() .await
.entry((sender_user.clone(), sender_device.clone())) .entry((sender_user.clone(), sender_device.clone()))
{ {
Entry::Vacant(v) => { Entry::Vacant(v) => {
@ -147,7 +148,7 @@ async fn sync_helper_wrapper(
.globals .globals
.sync_receivers .sync_receivers
.write() .write()
.unwrap() .await
.entry((sender_user, sender_device)) .entry((sender_user, sender_device))
{ {
Entry::Occupied(o) => { Entry::Occupied(o) => {
@ -302,11 +303,11 @@ async fn sync_helper(
.globals .globals
.roomid_mutex_insert .roomid_mutex_insert
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
drop(insert_lock); drop(insert_lock);
} }
@ -434,11 +435,11 @@ async fn sync_helper(
.globals .globals
.roomid_mutex_insert .roomid_mutex_insert
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
drop(insert_lock); drop(insert_lock);
} }
@ -577,11 +578,11 @@ async fn load_joined_room(
.globals .globals
.roomid_mutex_insert .roomid_mutex_insert
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
drop(insert_lock); drop(insert_lock);
} }
@ -599,12 +600,11 @@ async fn load_joined_room(
timeline_users.insert(event.sender.as_str().to_owned()); timeline_users.insert(event.sender.as_str().to_owned());
} }
services().rooms.lazy_loading.lazy_load_confirm_delivery( services()
sender_user, .rooms
sender_device, .lazy_loading
room_id, .lazy_load_confirm_delivery(sender_user, sender_device, room_id, sincecount)
sincecount, .await?;
)?;
// Database queries: // Database queries:
@ -797,13 +797,17 @@ async fn load_joined_room(
// The state_events above should contain all timeline_users, let's mark them as lazy // The state_events above should contain all timeline_users, let's mark them as lazy
// loaded. // loaded.
services().rooms.lazy_loading.lazy_load_mark_sent( services()
.rooms
.lazy_loading
.lazy_load_mark_sent(
sender_user, sender_user,
sender_device, sender_device,
room_id, room_id,
lazy_loaded, lazy_loaded,
next_batchcount, next_batchcount,
); )
.await;
( (
heroes, heroes,
@ -884,13 +888,17 @@ async fn load_joined_room(
} }
} }
services().rooms.lazy_loading.lazy_load_mark_sent( services()
.rooms
.lazy_loading
.lazy_load_mark_sent(
sender_user, sender_user,
sender_device, sender_device,
room_id, room_id,
lazy_loaded, lazy_loaded,
next_batchcount, next_batchcount,
); )
.await;
let encrypted_room = services() let encrypted_room = services()
.rooms .rooms
@ -1189,11 +1197,14 @@ pub async fn sync_events_v4_route(
if globalsince == 0 { if globalsince == 0 {
if let Some(conn_id) = &body.conn_id { if let Some(conn_id) = &body.conn_id {
services().users.forget_sync_request_connection( services()
.users
.forget_sync_request_connection(
sender_user.clone(), sender_user.clone(),
sender_device.clone(), sender_device.clone(),
conn_id.clone(), conn_id.clone(),
) )
.await
} }
} }
@ -1463,14 +1474,17 @@ pub async fn sync_events_v4_route(
); );
if let Some(conn_id) = &body.conn_id { if let Some(conn_id) = &body.conn_id {
services().users.update_sync_known_rooms( services()
.users
.update_sync_known_rooms(
sender_user.clone(), sender_user.clone(),
sender_device.clone(), sender_device.clone(),
conn_id.clone(), conn_id.clone(),
list_id, list_id,
new_known_rooms, new_known_rooms,
globalsince, globalsince,
); )
.await;
} }
} }
@ -1502,23 +1516,29 @@ pub async fn sync_events_v4_route(
} }
if let Some(conn_id) = &body.conn_id { if let Some(conn_id) = &body.conn_id {
services().users.update_sync_known_rooms( services()
.users
.update_sync_known_rooms(
sender_user.clone(), sender_user.clone(),
sender_device.clone(), sender_device.clone(),
conn_id.clone(), conn_id.clone(),
"subscriptions".to_owned(), "subscriptions".to_owned(),
known_subscription_rooms, known_subscription_rooms,
globalsince, globalsince,
); )
.await;
} }
if let Some(conn_id) = &body.conn_id { if let Some(conn_id) = &body.conn_id {
services().users.update_sync_subscriptions( services()
.users
.update_sync_subscriptions(
sender_user.clone(), sender_user.clone(),
sender_device.clone(), sender_device.clone(),
conn_id.clone(), conn_id.clone(),
body.room_subscriptions, body.room_subscriptions,
); )
.await;
} }
let mut rooms = BTreeMap::new(); let mut rooms = BTreeMap::new();

13
src/api/server_server.rs
View file

@ -51,9 +51,10 @@ use std::{
fmt::Debug, fmt::Debug,
mem, mem,
net::{IpAddr, SocketAddr}, net::{IpAddr, SocketAddr},
sync::{Arc, RwLock}, sync::Arc,
time::{Duration, Instant, SystemTime}, time::{Duration, Instant, SystemTime},
}; };
use tokio::sync::RwLock;
use tracing::{debug, error, warn}; use tracing::{debug, error, warn};
@ -137,7 +138,7 @@ where
.globals .globals
.actual_destination_cache .actual_destination_cache
.read() .read()
.unwrap() .await
.get(destination) .get(destination)
.cloned(); .cloned();
@ -290,7 +291,7 @@ where
.globals .globals
.actual_destination_cache .actual_destination_cache
.write() .write()
.unwrap() .await
.insert( .insert(
OwnedServerName::from(destination), OwnedServerName::from(destination),
(actual_destination, host), (actual_destination, host),
@ -740,7 +741,7 @@ pub async fn send_transaction_message_route(
.globals .globals
.roomid_mutex_federation .roomid_mutex_federation
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
@ -1409,7 +1410,7 @@ pub async fn create_join_event_template_route(
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(body.room_id.to_owned()) .entry(body.room_id.to_owned())
.or_default(), .or_default(),
); );
@ -1579,7 +1580,7 @@ async fn create_join_event(
.globals .globals
.roomid_mutex_federation .roomid_mutex_federation
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );

3
src/lib.rs
View file

@ -4,6 +4,9 @@ mod database;
mod service; mod service;
mod utils; mod utils;
// Not async due to services() being used in many closures, and async colsures are not stable as of writing
// This is the case for every other occurence of sync Mutex/RwLock, except for database related ones, where
// the current maintainer (Timo) as asked to not modify those
use std::sync::RwLock; use std::sync::RwLock;
pub use api::ruma_wrapper::{Ruma, RumaResponse}; pub use api::ruma_wrapper::{Ruma, RumaResponse};

165
src/service/admin/mod.rs
View file

@ -1,7 +1,7 @@
use std::{ use std::{
collections::BTreeMap, collections::BTreeMap,
convert::{TryFrom, TryInto}, convert::{TryFrom, TryInto},
sync::{Arc, RwLock}, sync::Arc,
time::Instant, time::Instant,
}; };
@ -26,7 +26,7 @@ use ruma::{
EventId, OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId, EventId, OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
}; };
use serde_json::value::to_raw_value; use serde_json::value::to_raw_value;
use tokio::sync::{mpsc, Mutex, MutexGuard}; use tokio::sync::{mpsc, Mutex, RwLock};
use crate::{ use crate::{
api::client_server::{leave_all_rooms, AUTO_GEN_PASSWORD_LENGTH}, api::client_server::{leave_all_rooms, AUTO_GEN_PASSWORD_LENGTH},
@ -215,27 +215,6 @@ impl Service {
.expect("@conduit:server_name is valid"); .expect("@conduit:server_name is valid");
if let Ok(Some(conduit_room)) = services().admin.get_admin_room() { if let Ok(Some(conduit_room)) = services().admin.get_admin_room() {
let send_message = |message: RoomMessageEventContent,
mutex_lock: &MutexGuard<'_, ()>| {
services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder {
event_type: TimelineEventType::RoomMessage,
content: to_raw_value(&message)
.expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
mutex_lock,
)
.unwrap();
};
loop { loop {
tokio::select! { tokio::select! {
Some(event) = receiver.recv() => { Some(event) = receiver.recv() => {
@ -248,16 +227,30 @@ impl Service {
services().globals services().globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(conduit_room.to_owned()) .entry(conduit_room.to_owned())
.or_default(), .or_default(),
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
send_message(message_content, &state_lock); services()
.rooms
drop(state_lock); .timeline
.build_and_append_pdu(
PduBuilder {
event_type: TimelineEventType::RoomMessage,
content: to_raw_value(&message_content)
.expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
&state_lock,
)
.await.unwrap();
} }
} }
} }
@ -425,11 +418,7 @@ impl Service {
Err(e) => RoomMessageEventContent::text_plain(e.to_string()), Err(e) => RoomMessageEventContent::text_plain(e.to_string()),
}, },
AdminCommand::IncomingFederation => { AdminCommand::IncomingFederation => {
let map = services() let map = services().globals.roomid_federationhandletime.read().await;
.globals
.roomid_federationhandletime
.read()
.unwrap();
let mut msg: String = format!("Handling {} incoming pdus:\n", map.len()); let mut msg: String = format!("Handling {} incoming pdus:\n", map.len());
for (r, (e, i)) in map.iter() { for (r, (e, i)) in map.iter() {
@ -543,7 +532,7 @@ impl Service {
} }
} }
AdminCommand::MemoryUsage => { AdminCommand::MemoryUsage => {
let response1 = services().memory_usage(); let response1 = services().memory_usage().await;
let response2 = services().globals.db.memory_usage(); let response2 = services().globals.db.memory_usage();
RoomMessageEventContent::text_plain(format!( RoomMessageEventContent::text_plain(format!(
@ -556,7 +545,7 @@ impl Service {
RoomMessageEventContent::text_plain("Done.") RoomMessageEventContent::text_plain("Done.")
} }
AdminCommand::ClearServiceCaches { amount } => { AdminCommand::ClearServiceCaches { amount } => {
services().clear_caches(amount); services().clear_caches(amount).await;
RoomMessageEventContent::text_plain("Done.") RoomMessageEventContent::text_plain("Done.")
} }
@ -797,7 +786,7 @@ impl Service {
.fetch_required_signing_keys(&value, &pub_key_map) .fetch_required_signing_keys(&value, &pub_key_map)
.await?; .await?;
let pub_key_map = pub_key_map.read().unwrap(); let pub_key_map = pub_key_map.read().await;
match ruma::signatures::verify_json(&pub_key_map, &value) { match ruma::signatures::verify_json(&pub_key_map, &value) {
Ok(_) => RoomMessageEventContent::text_plain("Signature correct"), Ok(_) => RoomMessageEventContent::text_plain("Signature correct"),
Err(e) => RoomMessageEventContent::text_plain(format!( Err(e) => RoomMessageEventContent::text_plain(format!(
@ -913,7 +902,7 @@ impl Service {
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
@ -932,7 +921,10 @@ impl Service {
content.room_version = services().globals.default_room_version(); content.room_version = services().globals.default_room_version();
// 1. The room create event // 1. The room create event
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomCreate, event_type: TimelineEventType::RoomCreate,
content: to_raw_value(&content).expect("event is valid, we just created it"), content: to_raw_value(&content).expect("event is valid, we just created it"),
@ -943,10 +935,14 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 2. Make conduit bot join // 2. Make conduit bot join
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&RoomMemberEventContent { content: to_raw_value(&RoomMemberEventContent {
@ -967,13 +963,17 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 3. Power levels // 3. Power levels
let mut users = BTreeMap::new(); let mut users = BTreeMap::new();
users.insert(conduit_user.clone(), 100.into()); users.insert(conduit_user.clone(), 100.into());
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomPowerLevels, event_type: TimelineEventType::RoomPowerLevels,
content: to_raw_value(&RoomPowerLevelsEventContent { content: to_raw_value(&RoomPowerLevelsEventContent {
@ -988,10 +988,14 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 4.1 Join Rules // 4.1 Join Rules
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomJoinRules, event_type: TimelineEventType::RoomJoinRules,
content: to_raw_value(&RoomJoinRulesEventContent::new(JoinRule::Invite)) content: to_raw_value(&RoomJoinRulesEventContent::new(JoinRule::Invite))
@ -1003,10 +1007,14 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 4.2 History Visibility // 4.2 History Visibility
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomHistoryVisibility, event_type: TimelineEventType::RoomHistoryVisibility,
content: to_raw_value(&RoomHistoryVisibilityEventContent::new( content: to_raw_value(&RoomHistoryVisibilityEventContent::new(
@ -1020,13 +1028,19 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 4.3 Guest Access // 4.3 Guest Access
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomGuestAccess, event_type: TimelineEventType::RoomGuestAccess,
content: to_raw_value(&RoomGuestAccessEventContent::new(GuestAccess::Forbidden)) content: to_raw_value(&RoomGuestAccessEventContent::new(
GuestAccess::Forbidden,
))
.expect("event is valid, we just created it"), .expect("event is valid, we just created it"),
unsigned: None, unsigned: None,
state_key: Some("".to_owned()), state_key: Some("".to_owned()),
@ -1035,11 +1049,15 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 5. Events implied by name and topic // 5. Events implied by name and topic
let room_name = format!("{} Admin Room", services().globals.server_name()); let room_name = format!("{} Admin Room", services().globals.server_name());
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomName, event_type: TimelineEventType::RoomName,
content: to_raw_value(&RoomNameEventContent::new(room_name)) content: to_raw_value(&RoomNameEventContent::new(room_name))
@ -1051,9 +1069,13 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomTopic, event_type: TimelineEventType::RoomTopic,
content: to_raw_value(&RoomTopicEventContent { content: to_raw_value(&RoomTopicEventContent {
@ -1067,14 +1089,18 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 6. Room alias // 6. Room alias
let alias: OwnedRoomAliasId = format!("#admins:{}", services().globals.server_name()) let alias: OwnedRoomAliasId = format!("#admins:{}", services().globals.server_name())
.try_into() .try_into()
.expect("#admins:server_name is a valid alias name"); .expect("#admins:server_name is a valid alias name");
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomCanonicalAlias, event_type: TimelineEventType::RoomCanonicalAlias,
content: to_raw_value(&RoomCanonicalAliasEventContent { content: to_raw_value(&RoomCanonicalAliasEventContent {
@ -1089,7 +1115,8 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
services().rooms.alias.set_alias(&alias, &room_id)?; services().rooms.alias.set_alias(&alias, &room_id)?;
@ -1125,7 +1152,7 @@ impl Service {
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
@ -1137,7 +1164,10 @@ impl Service {
.expect("@conduit:server_name is valid"); .expect("@conduit:server_name is valid");
// Invite and join the real user // Invite and join the real user
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&RoomMemberEventContent { content: to_raw_value(&RoomMemberEventContent {
@ -1158,8 +1188,12 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
services().rooms.timeline.build_and_append_pdu( .await?;
services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomMember, event_type: TimelineEventType::RoomMember,
content: to_raw_value(&RoomMemberEventContent { content: to_raw_value(&RoomMemberEventContent {
@ -1180,14 +1214,18 @@ impl Service {
user_id, user_id,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// Set power level // Set power level
let mut users = BTreeMap::new(); let mut users = BTreeMap::new();
users.insert(conduit_user.to_owned(), 100.into()); users.insert(conduit_user.to_owned(), 100.into());
users.insert(user_id.to_owned(), 100.into()); users.insert(user_id.to_owned(), 100.into());
services().rooms.timeline.build_and_append_pdu( services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: TimelineEventType::RoomPowerLevels, event_type: TimelineEventType::RoomPowerLevels,
content: to_raw_value(&RoomPowerLevelsEventContent { content: to_raw_value(&RoomPowerLevelsEventContent {
@ -1202,7 +1240,8 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// Send welcome message // Send welcome message
services().rooms.timeline.build_and_append_pdu( services().rooms.timeline.build_and_append_pdu(
@ -1220,7 +1259,7 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; ).await?;
} }
Ok(()) Ok(())
} }

18
src/service/globals/mod.rs
View file

@ -31,11 +31,11 @@ use std::{
path::PathBuf, path::PathBuf,
sync::{ sync::{
atomic::{self, AtomicBool}, atomic::{self, AtomicBool},
Arc, Mutex, RwLock, Arc, RwLock as SyncRwLock,
}, },
time::{Duration, Instant}, time::{Duration, Instant},
}; };
use tokio::sync::{broadcast, watch::Receiver, Mutex as TokioMutex, Semaphore}; use tokio::sync::{broadcast, watch::Receiver, Mutex, RwLock, Semaphore};
use tracing::{error, info}; use tracing::{error, info};
use trust_dns_resolver::TokioAsyncResolver; use trust_dns_resolver::TokioAsyncResolver;
@ -53,7 +53,7 @@ pub struct Service {
pub db: &'static dyn Data, pub db: &'static dyn Data,
pub actual_destination_cache: Arc<RwLock<WellKnownMap>>, // actual_destination, host pub actual_destination_cache: Arc<RwLock<WellKnownMap>>, // actual_destination, host
pub tls_name_override: Arc<RwLock<TlsNameMap>>, pub tls_name_override: Arc<SyncRwLock<TlsNameMap>>,
pub config: Config, pub config: Config,
keypair: Arc<ruma::signatures::Ed25519KeyPair>, keypair: Arc<ruma::signatures::Ed25519KeyPair>,
dns_resolver: TokioAsyncResolver, dns_resolver: TokioAsyncResolver,
@ -68,8 +68,8 @@ pub struct Service {
pub servername_ratelimiter: Arc<RwLock<HashMap<OwnedServerName, Arc<Semaphore>>>>, pub servername_ratelimiter: Arc<RwLock<HashMap<OwnedServerName, Arc<Semaphore>>>>,
pub sync_receivers: RwLock<HashMap<(OwnedUserId, OwnedDeviceId), SyncHandle>>, pub sync_receivers: RwLock<HashMap<(OwnedUserId, OwnedDeviceId), SyncHandle>>,
pub roomid_mutex_insert: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>, pub roomid_mutex_insert: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>,
pub roomid_mutex_state: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>, pub roomid_mutex_state: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>,
pub roomid_mutex_federation: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>, // this lock will be held longer pub roomid_mutex_federation: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>, // this lock will be held longer
pub roomid_federationhandletime: RwLock<HashMap<OwnedRoomId, (OwnedEventId, Instant)>>, pub roomid_federationhandletime: RwLock<HashMap<OwnedRoomId, (OwnedEventId, Instant)>>,
pub stateres_mutex: Arc<Mutex<()>>, pub stateres_mutex: Arc<Mutex<()>>,
pub rotate: RotationHandler, pub rotate: RotationHandler,
@ -109,11 +109,11 @@ impl Default for RotationHandler {
pub struct Resolver { pub struct Resolver {
inner: GaiResolver, inner: GaiResolver,
overrides: Arc<RwLock<TlsNameMap>>, overrides: Arc<SyncRwLock<TlsNameMap>>,
} }
impl Resolver { impl Resolver {
pub fn new(overrides: Arc<RwLock<TlsNameMap>>) -> Self { pub fn new(overrides: Arc<SyncRwLock<TlsNameMap>>) -> Self {
Resolver { Resolver {
inner: GaiResolver::new(), inner: GaiResolver::new(),
overrides, overrides,
@ -125,7 +125,7 @@ impl Resolve for Resolver {
fn resolve(&self, name: Name) -> Resolving { fn resolve(&self, name: Name) -> Resolving {
self.overrides self.overrides
.read() .read()
.expect("lock should not be poisoned") .unwrap()
.get(name.as_str()) .get(name.as_str())
.and_then(|(override_name, port)| { .and_then(|(override_name, port)| {
override_name.first().map(|first_name| { override_name.first().map(|first_name| {
@ -159,7 +159,7 @@ impl Service {
} }
}; };
let tls_name_override = Arc::new(RwLock::new(TlsNameMap::new())); let tls_name_override = Arc::new(SyncRwLock::new(TlsNameMap::new()));
let jwt_decoding_key = config let jwt_decoding_key = config
.jwt_secret .jwt_secret

39
src/service/mod.rs
View file

@ -1,9 +1,10 @@
use std::{ use std::{
collections::{BTreeMap, HashMap}, collections::{BTreeMap, HashMap},
sync::{Arc, Mutex}, sync::{Arc, Mutex as SyncMutex},
}; };
use lru_cache::LruCache; use lru_cache::LruCache;
use tokio::sync::Mutex;
use crate::{Config, Result}; use crate::{Config, Result};
@ -79,17 +80,17 @@ impl Services {
state: rooms::state::Service { db }, state: rooms::state::Service { db },
state_accessor: rooms::state_accessor::Service { state_accessor: rooms::state_accessor::Service {
db, db,
server_visibility_cache: Mutex::new(LruCache::new( server_visibility_cache: SyncMutex::new(LruCache::new(
(100.0 * config.conduit_cache_capacity_modifier) as usize, (100.0 * config.conduit_cache_capacity_modifier) as usize,
)), )),
user_visibility_cache: Mutex::new(LruCache::new( user_visibility_cache: SyncMutex::new(LruCache::new(
(100.0 * config.conduit_cache_capacity_modifier) as usize, (100.0 * config.conduit_cache_capacity_modifier) as usize,
)), )),
}, },
state_cache: rooms::state_cache::Service { db }, state_cache: rooms::state_cache::Service { db },
state_compressor: rooms::state_compressor::Service { state_compressor: rooms::state_compressor::Service {
db, db,
stateinfo_cache: Mutex::new(LruCache::new( stateinfo_cache: SyncMutex::new(LruCache::new(
(100.0 * config.conduit_cache_capacity_modifier) as usize, (100.0 * config.conduit_cache_capacity_modifier) as usize,
)), )),
}, },
@ -107,7 +108,7 @@ impl Services {
uiaa: uiaa::Service { db }, uiaa: uiaa::Service { db },
users: users::Service { users: users::Service {
db, db,
connections: Mutex::new(BTreeMap::new()), connections: SyncMutex::new(BTreeMap::new()),
}, },
account_data: account_data::Service { db }, account_data: account_data::Service { db },
admin: admin::Service::build(), admin: admin::Service::build(),
@ -118,14 +119,8 @@ impl Services {
globals: globals::Service::load(db, config)?, globals: globals::Service::load(db, config)?,
}) })
} }
fn memory_usage(&self) -> String { async fn memory_usage(&self) -> String {
let lazy_load_waiting = self let lazy_load_waiting = self.rooms.lazy_loading.lazy_load_waiting.lock().await.len();
.rooms
.lazy_loading
.lazy_load_waiting
.lock()
.unwrap()
.len();
let server_visibility_cache = self let server_visibility_cache = self
.rooms .rooms
.state_accessor .state_accessor
@ -152,15 +147,9 @@ impl Services {
.timeline .timeline
.lasttimelinecount_cache .lasttimelinecount_cache
.lock() .lock()
.unwrap() .await
.len();
let roomid_spacechunk_cache = self
.rooms
.spaces
.roomid_spacechunk_cache
.lock()
.unwrap()
.len(); .len();
let roomid_spacechunk_cache = self.rooms.spaces.roomid_spacechunk_cache.lock().await.len();
format!( format!(
"\ "\
@ -173,13 +162,13 @@ roomid_spacechunk_cache: {roomid_spacechunk_cache}\
" "
) )
} }
fn clear_caches(&self, amount: u32) { async fn clear_caches(&self, amount: u32) {
if amount > 0 { if amount > 0 {
self.rooms self.rooms
.lazy_loading .lazy_loading
.lazy_load_waiting .lazy_load_waiting
.lock() .lock()
.unwrap() .await
.clear(); .clear();
} }
if amount > 1 { if amount > 1 {
@ -211,7 +200,7 @@ roomid_spacechunk_cache: {roomid_spacechunk_cache}\
.timeline .timeline
.lasttimelinecount_cache .lasttimelinecount_cache
.lock() .lock()
.unwrap() .await
.clear(); .clear();
} }
if amount > 5 { if amount > 5 {
@ -219,7 +208,7 @@ roomid_spacechunk_cache: {roomid_spacechunk_cache}\
.spaces .spaces
.roomid_spacechunk_cache .roomid_spacechunk_cache
.lock() .lock()
.unwrap() .await
.clear(); .clear();
} }
} }

135
src/service/rooms/event_handler/mod.rs
View file

@ -1,25 +1,24 @@
/// An async function that can recursively call itself. /// An async function that can recursively call itself.
type AsyncRecursiveType<'a, T> = Pin<Box<dyn Future<Output = T> + 'a + Send>>; type AsyncRecursiveType<'a, T> = Pin<Box<dyn Future<Output = T> + 'a + Send>>;
use ruma::{
api::federation::discovery::{get_remote_server_keys, get_server_keys},
CanonicalJsonObject, CanonicalJsonValue, OwnedServerName, OwnedServerSigningKeyId,
RoomVersionId,
};
use std::{ use std::{
collections::{hash_map, BTreeMap, HashMap, HashSet}, collections::{hash_map, BTreeMap, HashMap, HashSet},
pin::Pin, pin::Pin,
sync::{Arc, RwLock, RwLockWriteGuard}, sync::Arc,
time::{Duration, Instant, SystemTime}, time::{Duration, Instant, SystemTime},
}; };
use tokio::sync::Semaphore;
use async_recursion::async_recursion;
use futures_util::{stream::FuturesUnordered, Future, StreamExt}; use futures_util::{stream::FuturesUnordered, Future, StreamExt};
use ruma::{ use ruma::{
api::{ api::{
client::error::ErrorKind, client::error::ErrorKind,
federation::{ federation::{
discovery::get_remote_server_keys_batch::{self, v2::QueryCriteria}, discovery::{
get_remote_server_keys,
get_remote_server_keys_batch::{self, v2::QueryCriteria},
get_server_keys,
},
event::{get_event, get_room_state_ids}, event::{get_event, get_room_state_ids},
membership::create_join_event, membership::create_join_event,
}, },
@ -31,9 +30,11 @@ use ruma::{
int, int,
serde::Base64, serde::Base64,
state_res::{self, RoomVersion, StateMap}, state_res::{self, RoomVersion, StateMap},
uint, EventId, MilliSecondsSinceUnixEpoch, RoomId, ServerName, uint, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
OwnedServerName, OwnedServerSigningKeyId, RoomId, RoomVersionId, ServerName,
}; };
use serde_json::value::RawValue as RawJsonValue; use serde_json::value::RawValue as RawJsonValue;
use tokio::sync::{RwLock, RwLockWriteGuard, Semaphore};
use tracing::{debug, error, info, trace, warn}; use tracing::{debug, error, info, trace, warn};
use crate::{service::*, services, Error, PduEvent, Result}; use crate::{service::*, services, Error, PduEvent, Result};
@ -168,7 +169,7 @@ impl Service {
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.read() .read()
.unwrap() .await
.get(&*prev_id) .get(&*prev_id)
{ {
// Exponential backoff // Exponential backoff
@ -189,7 +190,7 @@ impl Service {
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.write() .write()
.unwrap() .await
.entry((*prev_id).to_owned()) .entry((*prev_id).to_owned())
{ {
hash_map::Entry::Vacant(e) => { hash_map::Entry::Vacant(e) => {
@ -213,7 +214,7 @@ impl Service {
.globals .globals
.roomid_federationhandletime .roomid_federationhandletime
.write() .write()
.unwrap() .await
.insert(room_id.to_owned(), ((*prev_id).to_owned(), start_time)); .insert(room_id.to_owned(), ((*prev_id).to_owned(), start_time));
if let Err(e) = self if let Err(e) = self
@ -233,7 +234,7 @@ impl Service {
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.write() .write()
.unwrap() .await
.entry((*prev_id).to_owned()) .entry((*prev_id).to_owned())
{ {
hash_map::Entry::Vacant(e) => { hash_map::Entry::Vacant(e) => {
@ -249,7 +250,7 @@ impl Service {
.globals .globals
.roomid_federationhandletime .roomid_federationhandletime
.write() .write()
.unwrap() .await
.remove(&room_id.to_owned()); .remove(&room_id.to_owned());
debug!( debug!(
"Handling prev event {} took {}m{}s", "Handling prev event {} took {}m{}s",
@ -267,7 +268,7 @@ impl Service {
.globals .globals
.roomid_federationhandletime .roomid_federationhandletime
.write() .write()
.unwrap() .await
.insert(room_id.to_owned(), (event_id.to_owned(), start_time)); .insert(room_id.to_owned(), (event_id.to_owned(), start_time));
let r = services() let r = services()
.rooms .rooms
@ -285,7 +286,7 @@ impl Service {
.globals .globals
.roomid_federationhandletime .roomid_federationhandletime
.write() .write()
.unwrap() .await
.remove(&room_id.to_owned()); .remove(&room_id.to_owned());
r r
@ -326,11 +327,8 @@ impl Service {
let room_version = let room_version =
RoomVersion::new(room_version_id).expect("room version is supported"); RoomVersion::new(room_version_id).expect("room version is supported");
let mut val = match ruma::signatures::verify_event( let guard = pub_key_map.read().await;
&pub_key_map.read().expect("RwLock is poisoned."), let mut val = match ruma::signatures::verify_event(&guard, &value, room_version_id) {
&value,
room_version_id,
) {
Err(e) => { Err(e) => {
// Drop // Drop
warn!("Dropping bad event {}: {}", event_id, e,); warn!("Dropping bad event {}: {}", event_id, e,);
@ -365,6 +363,8 @@ impl Service {
Ok(ruma::signatures::Verified::All) => value, Ok(ruma::signatures::Verified::All) => value,
}; };
drop(guard);
// Now that we have checked the signature and hashes we can add the eventID and convert // Now that we have checked the signature and hashes we can add the eventID and convert
// to our PduEvent type // to our PduEvent type
val.insert( val.insert(
@ -692,13 +692,15 @@ impl Service {
{ {
Ok(res) => { Ok(res) => {
debug!("Fetching state events at event."); debug!("Fetching state events at event.");
let collect = res
.pdu_ids
.iter()
.map(|x| Arc::from(&**x))
.collect::<Vec<_>>();
let state_vec = self let state_vec = self
.fetch_and_handle_outliers( .fetch_and_handle_outliers(
origin, origin,
&res.pdu_ids &collect,
.iter()
.map(|x| Arc::from(&**x))
.collect::<Vec<_>>(),
create_event, create_event,
room_id, room_id,
room_version_id, room_version_id,
@ -805,7 +807,7 @@ impl Service {
.globals .globals
.roomid_mutex_state .roomid_mutex_state
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
@ -884,14 +886,18 @@ impl Service {
debug!("Starting soft fail auth check"); debug!("Starting soft fail auth check");
if soft_fail { if soft_fail {
services().rooms.timeline.append_incoming_pdu( services()
.rooms
.timeline
.append_incoming_pdu(
&incoming_pdu, &incoming_pdu,
val, val,
extremities.iter().map(|e| (**e).to_owned()).collect(), extremities.iter().map(|e| (**e).to_owned()).collect(),
state_ids_compressed, state_ids_compressed,
soft_fail, soft_fail,
&state_lock, &state_lock,
)?; )
.await?;
// Soft fail, we keep the event as an outlier but don't add it to the timeline // Soft fail, we keep the event as an outlier but don't add it to the timeline
warn!("Event was soft failed: {:?}", incoming_pdu); warn!("Event was soft failed: {:?}", incoming_pdu);
@ -912,14 +918,18 @@ impl Service {
// We use the `state_at_event` instead of `state_after` so we accurately // We use the `state_at_event` instead of `state_after` so we accurately
// represent the state for this event. // represent the state for this event.
let pdu_id = services().rooms.timeline.append_incoming_pdu( let pdu_id = services()
.rooms
.timeline
.append_incoming_pdu(
&incoming_pdu, &incoming_pdu,
val, val,
extremities.iter().map(|e| (**e).to_owned()).collect(), extremities.iter().map(|e| (**e).to_owned()).collect(),
state_ids_compressed, state_ids_compressed,
soft_fail, soft_fail,
&state_lock, &state_lock,
)?; )
.await?;
debug!("Appended incoming pdu"); debug!("Appended incoming pdu");
@ -1034,7 +1044,8 @@ impl Service {
/// d. TODO: Ask other servers over federation? /// d. TODO: Ask other servers over federation?
#[allow(clippy::type_complexity)] #[allow(clippy::type_complexity)]
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
pub(crate) fn fetch_and_handle_outliers<'a>( #[async_recursion]
pub(crate) async fn fetch_and_handle_outliers<'a>(
&'a self, &'a self,
origin: &'a ServerName, origin: &'a ServerName,
events: &'a [Arc<EventId>], events: &'a [Arc<EventId>],
@ -1042,20 +1053,20 @@ impl Service {
room_id: &'a RoomId, room_id: &'a RoomId,
room_version_id: &'a RoomVersionId, room_version_id: &'a RoomVersionId,
pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, Base64>>>, pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
) -> AsyncRecursiveType<'a, Vec<(Arc<PduEvent>, Option<BTreeMap<String, CanonicalJsonValue>>)>> ) -> Vec<(Arc<PduEvent>, Option<BTreeMap<String, CanonicalJsonValue>>)> {
{ let back_off = |id| async move {
Box::pin(async move { match services()
let back_off = |id| match services()
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.write() .write()
.unwrap() .await
.entry(id) .entry(id)
{ {
hash_map::Entry::Vacant(e) => { hash_map::Entry::Vacant(e) => {
e.insert((Instant::now(), 1)); e.insert((Instant::now(), 1));
} }
hash_map::Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1), hash_map::Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1),
}
}; };
let mut pdus = vec![]; let mut pdus = vec![];
@ -1081,7 +1092,7 @@ impl Service {
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.read() .read()
.unwrap() .await
.get(&*next_id) .get(&*next_id)
{ {
// Exponential backoff // Exponential backoff
@ -1128,7 +1139,7 @@ impl Service {
match pdu::gen_event_id_canonical_json(&res.pdu, room_version_id) { match pdu::gen_event_id_canonical_json(&res.pdu, room_version_id) {
Ok(t) => t, Ok(t) => t,
Err(_) => { Err(_) => {
back_off((*next_id).to_owned()); back_off((*next_id).to_owned()).await;
continue; continue;
} }
}; };
@ -1160,7 +1171,7 @@ impl Service {
} }
Err(_) => { Err(_) => {
warn!("Failed to fetch event: {}", next_id); warn!("Failed to fetch event: {}", next_id);
back_off((*next_id).to_owned()); back_off((*next_id).to_owned()).await;
} }
} }
} }
@ -1170,7 +1181,7 @@ impl Service {
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.read() .read()
.unwrap() .await
.get(&**next_id) .get(&**next_id)
{ {
// Exponential backoff // Exponential backoff
@ -1205,13 +1216,12 @@ impl Service {
} }
Err(e) => { Err(e) => {
warn!("Authentication of event {} failed: {:?}", next_id, e); warn!("Authentication of event {} failed: {:?}", next_id, e);
back_off((**next_id).to_owned()); back_off((**next_id).to_owned()).await;
} }
} }
} }
} }
pdus pdus
})
} }
async fn fetch_unknown_prev_events( async fn fetch_unknown_prev_events(
@ -1360,7 +1370,7 @@ impl Service {
pub_key_map pub_key_map
.write() .write()
.map_err(|_| Error::bad_database("RwLock is poisoned."))? .await
.insert(signature_server.clone(), keys); .insert(signature_server.clone(), keys);
} }
@ -1369,7 +1379,7 @@ impl Service {
// Gets a list of servers for which we don't have the signing key yet. We go over // Gets a list of servers for which we don't have the signing key yet. We go over
// the PDUs and either cache the key or add it to the list that needs to be retrieved. // the PDUs and either cache the key or add it to the list that needs to be retrieved.
fn get_server_keys_from_cache( async fn get_server_keys_from_cache(
&self, &self,
pdu: &RawJsonValue, pdu: &RawJsonValue,
servers: &mut BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, QueryCriteria>>, servers: &mut BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, QueryCriteria>>,
@ -1393,7 +1403,7 @@ impl Service {
.globals .globals
.bad_event_ratelimiter .bad_event_ratelimiter
.read() .read()
.unwrap() .await
.get(event_id) .get(event_id)
{ {
// Exponential backoff // Exponential backoff
@ -1469,17 +1479,19 @@ impl Service {
> = BTreeMap::new(); > = BTreeMap::new();
{ {
let mut pkm = pub_key_map let mut pkm = pub_key_map.write().await;
.write()
.map_err(|_| Error::bad_database("RwLock is poisoned."))?;
// Try to fetch keys, failure is okay // Try to fetch keys, failure is okay
// Servers we couldn't find in the cache will be added to `servers` // Servers we couldn't find in the cache will be added to `servers`
for pdu in &event.room_state.state { for pdu in &event.room_state.state {
let _ = self.get_server_keys_from_cache(pdu, &mut servers, room_version, &mut pkm); let _ = self
.get_server_keys_from_cache(pdu, &mut servers, room_version, &mut pkm)
.await;
} }
for pdu in &event.room_state.auth_chain { for pdu in &event.room_state.auth_chain {
let _ = self.get_server_keys_from_cache(pdu, &mut servers, room_version, &mut pkm); let _ = self
.get_server_keys_from_cache(pdu, &mut servers, room_version, &mut pkm)
.await;
} }
drop(pkm); drop(pkm);
@ -1503,9 +1515,7 @@ impl Service {
.await .await
{ {
trace!("Got signing keys: {:?}", keys); trace!("Got signing keys: {:?}", keys);
let mut pkm = pub_key_map let mut pkm = pub_key_map.write().await;
.write()
.map_err(|_| Error::bad_database("RwLock is poisoned."))?;
for k in keys.server_keys { for k in keys.server_keys {
let k = match k.deserialize() { let k = match k.deserialize() {
Ok(key) => key, Ok(key) => key,
@ -1564,10 +1574,7 @@ impl Service {
.into_iter() .into_iter()
.map(|(k, v)| (k.to_string(), v.key)) .map(|(k, v)| (k.to_string(), v.key))
.collect(); .collect();
pub_key_map pub_key_map.write().await.insert(origin.to_string(), result);
.write()
.map_err(|_| Error::bad_database("RwLock is poisoned."))?
.insert(origin.to_string(), result);
} }
} }
info!("Done handling result"); info!("Done handling result");
@ -1632,14 +1639,14 @@ impl Service {
.globals .globals
.servername_ratelimiter .servername_ratelimiter
.read() .read()
.unwrap() .await
.get(origin) .get(origin)
.map(|s| Arc::clone(s).acquire_owned()); .map(|s| Arc::clone(s).acquire_owned());
let permit = match permit { let permit = match permit {
Some(p) => p, Some(p) => p,
None => { None => {
let mut write = services().globals.servername_ratelimiter.write().unwrap(); let mut write = services().globals.servername_ratelimiter.write().await;
let s = Arc::clone( let s = Arc::clone(
write write
.entry(origin.to_owned()) .entry(origin.to_owned())
@ -1651,24 +1658,26 @@ impl Service {
} }
.await; .await;
let back_off = |id| match services() let back_off = |id| async {
match services()
.globals .globals
.bad_signature_ratelimiter .bad_signature_ratelimiter
.write() .write()
.unwrap() .await
.entry(id) .entry(id)
{ {
hash_map::Entry::Vacant(e) => { hash_map::Entry::Vacant(e) => {
e.insert((Instant::now(), 1)); e.insert((Instant::now(), 1));
} }
hash_map::Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1), hash_map::Entry::Occupied(mut e) => *e.get_mut() = (Instant::now(), e.get().1 + 1),
}
}; };
if let Some((time, tries)) = services() if let Some((time, tries)) = services()
.globals .globals
.bad_signature_ratelimiter .bad_signature_ratelimiter
.read() .read()
.unwrap() .await
.get(&signature_ids) .get(&signature_ids)
{ {
// Exponential backoff // Exponential backoff
@ -1775,7 +1784,7 @@ impl Service {
drop(permit); drop(permit);
back_off(signature_ids); back_off(signature_ids).await;
warn!("Failed to find public key for server: {}", origin); warn!("Failed to find public key for server: {}", origin);
Err(Error::BadServerResponse( Err(Error::BadServerResponse(

14
src/service/rooms/lazy_loading/mod.rs
View file

@ -1,11 +1,9 @@
mod data; mod data;
use std::{ use std::collections::{HashMap, HashSet};
collections::{HashMap, HashSet},
sync::Mutex,
};
pub use data::Data; pub use data::Data;
use ruma::{DeviceId, OwnedDeviceId, OwnedRoomId, OwnedUserId, RoomId, UserId}; use ruma::{DeviceId, OwnedDeviceId, OwnedRoomId, OwnedUserId, RoomId, UserId};
use tokio::sync::Mutex;
use crate::Result; use crate::Result;
@ -33,7 +31,7 @@ impl Service {
} }
#[tracing::instrument(skip(self))] #[tracing::instrument(skip(self))]
pub fn lazy_load_mark_sent( pub async fn lazy_load_mark_sent(
&self, &self,
user_id: &UserId, user_id: &UserId,
device_id: &DeviceId, device_id: &DeviceId,
@ -41,7 +39,7 @@ impl Service {
lazy_load: HashSet<OwnedUserId>, lazy_load: HashSet<OwnedUserId>,
count: PduCount, count: PduCount,
) { ) {
self.lazy_load_waiting.lock().unwrap().insert( self.lazy_load_waiting.lock().await.insert(
( (
user_id.to_owned(), user_id.to_owned(),
device_id.to_owned(), device_id.to_owned(),
@ -53,14 +51,14 @@ impl Service {
} }
#[tracing::instrument(skip(self))] #[tracing::instrument(skip(self))]
pub fn lazy_load_confirm_delivery( pub async fn lazy_load_confirm_delivery(
&self, &self,
user_id: &UserId, user_id: &UserId,
device_id: &DeviceId, device_id: &DeviceId,
room_id: &RoomId, room_id: &RoomId,
since: PduCount, since: PduCount,
) -> Result<()> { ) -> Result<()> {
if let Some(user_ids) = self.lazy_load_waiting.lock().unwrap().remove(&( if let Some(user_ids) = self.lazy_load_waiting.lock().await.remove(&(
user_id.to_owned(), user_id.to_owned(),
device_id.to_owned(), device_id.to_owned(),
room_id.to_owned(), room_id.to_owned(),

11
src/service/rooms/spaces/mod.rs
View file

@ -1,4 +1,4 @@
use std::sync::{Arc, Mutex}; use std::sync::Arc;
use lru_cache::LruCache; use lru_cache::LruCache;
use ruma::{ use ruma::{
@ -25,6 +25,7 @@ use ruma::{
space::SpaceRoomJoinRule, space::SpaceRoomJoinRule,
OwnedRoomId, RoomId, UserId, OwnedRoomId, RoomId, UserId,
}; };
use tokio::sync::Mutex;
use tracing::{debug, error, warn}; use tracing::{debug, error, warn};
@ -79,7 +80,7 @@ impl Service {
if let Some(cached) = self if let Some(cached) = self
.roomid_spacechunk_cache .roomid_spacechunk_cache
.lock() .lock()
.unwrap() .await
.get_mut(&current_room.to_owned()) .get_mut(&current_room.to_owned())
.as_ref() .as_ref()
{ {
@ -171,7 +172,7 @@ impl Service {
.transpose()? .transpose()?
.unwrap_or(JoinRule::Invite); .unwrap_or(JoinRule::Invite);
self.roomid_spacechunk_cache.lock().unwrap().insert( self.roomid_spacechunk_cache.lock().await.insert(
current_room.clone(), current_room.clone(),
Some(CachedSpaceChunk { Some(CachedSpaceChunk {
chunk, chunk,
@ -265,7 +266,7 @@ impl Service {
} }
} }
self.roomid_spacechunk_cache.lock().unwrap().insert( self.roomid_spacechunk_cache.lock().await.insert(
current_room.clone(), current_room.clone(),
Some(CachedSpaceChunk { Some(CachedSpaceChunk {
chunk, chunk,
@ -289,7 +290,7 @@ impl Service {
} else { } else {
self.roomid_spacechunk_cache self.roomid_spacechunk_cache
.lock() .lock()
.unwrap() .await
.insert(current_room.clone(), None); .insert(current_room.clone(), None);
} }
} }

2
src/service/rooms/state/mod.rs
View file

@ -95,7 +95,7 @@ impl Service {
.spaces .spaces
.roomid_spacechunk_cache .roomid_spacechunk_cache
.lock() .lock()
.unwrap() .await
.remove(&pdu.room_id); .remove(&pdu.room_id);
} }
_ => continue, _ => continue,

40
src/service/rooms/timeline/mod.rs
View file

@ -2,12 +2,8 @@ mod data;
use std::{ use std::{
cmp::Ordering, cmp::Ordering,
collections::{BTreeMap, HashMap}, collections::{BTreeMap, HashMap, HashSet},
}; sync::Arc,
use std::{
collections::HashSet,
sync::{Arc, Mutex, RwLock},
}; };
pub use data::Data; pub use data::Data;
@ -32,7 +28,7 @@ use ruma::{
}; };
use serde::Deserialize; use serde::Deserialize;
use serde_json::value::{to_raw_value, RawValue as RawJsonValue}; use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
use tokio::sync::MutexGuard; use tokio::sync::{Mutex, MutexGuard, RwLock};
use tracing::{error, info, warn}; use tracing::{error, info, warn};
use crate::{ use crate::{
@ -201,7 +197,7 @@ impl Service {
/// ///
/// Returns pdu id /// Returns pdu id
#[tracing::instrument(skip(self, pdu, pdu_json, leaves))] #[tracing::instrument(skip(self, pdu, pdu_json, leaves))]
pub fn append_pdu<'a>( pub async fn append_pdu<'a>(
&self, &self,
pdu: &PduEvent, pdu: &PduEvent,
mut pdu_json: CanonicalJsonObject, mut pdu_json: CanonicalJsonObject,
@ -263,11 +259,11 @@ impl Service {
.globals .globals
.roomid_mutex_insert .roomid_mutex_insert
.write() .write()
.unwrap() .await
.entry(pdu.room_id.clone()) .entry(pdu.room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
let count1 = services().globals.next_count()?; let count1 = services().globals.next_count()?;
// Mark as read first so the sending client doesn't get a notification even if appending // Mark as read first so the sending client doesn't get a notification even if appending
@ -395,7 +391,7 @@ impl Service {
.spaces .spaces
.roomid_spacechunk_cache .roomid_spacechunk_cache
.lock() .lock()
.unwrap() .await
.remove(&pdu.room_id); .remove(&pdu.room_id);
} }
} }
@ -806,7 +802,7 @@ impl Service {
/// Creates a new persisted data unit and adds it to a room. This function takes a /// Creates a new persisted data unit and adds it to a room. This function takes a
/// roomid_mutex_state, meaning that only this function is able to mutate the room state. /// roomid_mutex_state, meaning that only this function is able to mutate the room state.
#[tracing::instrument(skip(self, state_lock))] #[tracing::instrument(skip(self, state_lock))]
pub fn build_and_append_pdu( pub async fn build_and_append_pdu(
&self, &self,
pdu_builder: PduBuilder, pdu_builder: PduBuilder,
sender: &UserId, sender: &UserId,
@ -902,14 +898,16 @@ impl Service {
// pdu without it's state. This is okay because append_pdu can't fail. // pdu without it's state. This is okay because append_pdu can't fail.
let statehashid = services().rooms.state.append_to_state(&pdu)?; let statehashid = services().rooms.state.append_to_state(&pdu)?;
let pdu_id = self.append_pdu( let pdu_id = self
.append_pdu(
&pdu, &pdu,
pdu_json, pdu_json,
// Since this PDU references all pdu_leaves we can update the leaves // Since this PDU references all pdu_leaves we can update the leaves
// of the room // of the room
vec![(*pdu.event_id).to_owned()], vec![(*pdu.event_id).to_owned()],
state_lock, state_lock,
)?; )
.await?;
// We set the room state after inserting the pdu, so that we never have a moment in time // We set the room state after inserting the pdu, so that we never have a moment in time
// where events in the current room state do not exist // where events in the current room state do not exist
@ -947,7 +945,7 @@ impl Service {
/// Append the incoming event setting the state snapshot to the state from the /// Append the incoming event setting the state snapshot to the state from the
/// server that sent the event. /// server that sent the event.
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
pub fn append_incoming_pdu<'a>( pub async fn append_incoming_pdu<'a>(
&self, &self,
pdu: &PduEvent, pdu: &PduEvent,
pdu_json: CanonicalJsonObject, pdu_json: CanonicalJsonObject,
@ -977,11 +975,11 @@ impl Service {
return Ok(None); return Ok(None);
} }
let pdu_id = let pdu_id = services()
services()
.rooms .rooms
.timeline .timeline
.append_pdu(pdu, pdu_json, new_room_leaves, state_lock)?; .append_pdu(pdu, pdu_json, new_room_leaves, state_lock)
.await?;
Ok(Some(pdu_id)) Ok(Some(pdu_id))
} }
@ -1118,7 +1116,7 @@ impl Service {
.globals .globals
.roomid_mutex_federation .roomid_mutex_federation
.write() .write()
.unwrap() .await
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default(), .or_default(),
); );
@ -1150,11 +1148,11 @@ impl Service {
.globals .globals
.roomid_mutex_insert .roomid_mutex_insert
.write() .write()
.unwrap() .await
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
let count = services().globals.next_count()?; let count = services().globals.next_count()?;
let mut pdu_id = shortroomid.to_be_bytes().to_vec(); let mut pdu_id = shortroomid.to_be_bytes().to_vec();

6
src/service/users/mod.rs
View file

@ -45,7 +45,7 @@ impl Service {
self.db.exists(user_id) self.db.exists(user_id)
} }
pub fn forget_sync_request_connection( pub async fn forget_sync_request_connection(
&self, &self,
user_id: OwnedUserId, user_id: OwnedUserId,
device_id: OwnedDeviceId, device_id: OwnedDeviceId,
@ -186,7 +186,7 @@ impl Service {
cached.known_rooms.clone() cached.known_rooms.clone()
} }
pub fn update_sync_subscriptions( pub async fn update_sync_subscriptions(
&self, &self,
user_id: OwnedUserId, user_id: OwnedUserId,
device_id: OwnedDeviceId, device_id: OwnedDeviceId,
@ -212,7 +212,7 @@ impl Service {
cached.subscriptions = subscriptions; cached.subscriptions = subscriptions;
} }
pub fn update_sync_known_rooms( pub async fn update_sync_known_rooms(
&self, &self,
user_id: OwnedUserId, user_id: OwnedUserId,
device_id: OwnedDeviceId, device_id: OwnedDeviceId,