Compare commits

..
Author SHA1 Message Date
Timo Kösters c1dd19265d performance tests 2021-11-25 20:59:54 +01:00
10 changed files with 70 additions and 149 deletions
+2 -2
View File
@@ -34,7 +34,7 @@ use ruma::{
EventType, EventType,
}, },
identifiers::RoomName, identifiers::RoomName,
push, RoomAliasId, RoomId, UserId, push, RoomAliasId, RoomId, RoomVersionId, UserId,
}; };
use serde_json::value::to_raw_value; use serde_json::value::to_raw_value;
use tracing::info; use tracing::info;
@@ -282,7 +282,7 @@ pub async fn register_route(
let mut content = RoomCreateEventContent::new(conduit_user.clone()); let mut content = RoomCreateEventContent::new(conduit_user.clone());
content.federate = true; content.federate = true;
content.predecessor = None; content.predecessor = None;
content.room_version = db.globals.default_room_version(); content.room_version = RoomVersionId::Version6;
// 1. The room create event // 1. The room create event
db.rooms.build_and_append_pdu( db.rooms.build_and_append_pdu(
+10 -18
View File
@@ -1,6 +1,9 @@
use crate::{database::DatabaseGuard, ConduitResult, Ruma}; use crate::{ConduitResult, Ruma};
use ruma::api::client::r0::capabilities::{ use ruma::{
get_capabilities, Capabilities, RoomVersionStability, RoomVersionsCapability, api::client::r0::capabilities::{
get_capabilities, Capabilities, RoomVersionStability, RoomVersionsCapability,
},
RoomVersionId,
}; };
use std::collections::BTreeMap; use std::collections::BTreeMap;
@@ -14,28 +17,17 @@ use rocket::get;
feature = "conduit_bin", feature = "conduit_bin",
get("/_matrix/client/r0/capabilities", data = "<_body>") get("/_matrix/client/r0/capabilities", data = "<_body>")
)] )]
#[tracing::instrument(skip(_body, db))] #[tracing::instrument(skip(_body))]
pub async fn get_capabilities_route( pub async fn get_capabilities_route(
_body: Ruma<get_capabilities::Request>, _body: Ruma<get_capabilities::Request>,
db: DatabaseGuard,
) -> ConduitResult<get_capabilities::Response> { ) -> ConduitResult<get_capabilities::Response> {
let mut available = BTreeMap::new(); let mut available = BTreeMap::new();
if db.globals.allow_unstable_room_versions() { available.insert(RoomVersionId::Version5, RoomVersionStability::Stable);
for room_version in &db.globals.unstable_room_versions { available.insert(RoomVersionId::Version6, RoomVersionStability::Stable);
available.insert(room_version.clone(), RoomVersionStability::Stable);
}
} else {
for room_version in &db.globals.unstable_room_versions {
available.insert(room_version.clone(), RoomVersionStability::Unstable);
}
}
for room_version in &db.globals.stable_room_versions {
available.insert(room_version.clone(), RoomVersionStability::Stable);
}
let mut capabilities = Capabilities::new(); let mut capabilities = Capabilities::new();
capabilities.room_versions = RoomVersionsCapability { capabilities.room_versions = RoomVersionsCapability {
default: db.globals.default_room_version(), default: RoomVersionId::Version6,
available, available,
}; };
+10 -7
View File
@@ -551,7 +551,7 @@ async fn join_room_by_id_helper(
federation::membership::create_join_event_template::v1::Request { federation::membership::create_join_event_template::v1::Request {
room_id, room_id,
user_id: sender_user, user_id: sender_user,
ver: &db.globals.supported_room_versions(), ver: &[RoomVersionId::Version5, RoomVersionId::Version6],
}, },
) )
.await; .await;
@@ -566,7 +566,12 @@ async fn join_room_by_id_helper(
let (make_join_response, remote_server) = make_join_response_and_server?; let (make_join_response, remote_server) = make_join_response_and_server?;
let room_version = match make_join_response.room_version { let room_version = match make_join_response.room_version {
Some(room_version) if db.rooms.is_supported_version(&db, &room_version) => room_version, Some(room_version)
if room_version == RoomVersionId::Version5
|| room_version == RoomVersionId::Version6 =>
{
room_version
}
_ => return Err(Error::BadServerResponse("Room version is not supported")), _ => return Err(Error::BadServerResponse("Room version is not supported")),
}; };
@@ -885,10 +890,9 @@ pub(crate) async fn invite_helper<'a>(
None None
}; };
// If there was no create event yet, assume we are creating a room with the default // If there was no create event yet, assume we are creating a version 6 room right now
// version right now
let room_version_id = create_event_content let room_version_id = create_event_content
.map_or(db.globals.default_room_version(), |create_event| { .map_or(RoomVersionId::Version6, |create_event| {
create_event.room_version create_event.room_version
}); });
let room_version = let room_version =
@@ -1035,8 +1039,7 @@ pub(crate) async fn invite_helper<'a>(
let pub_key_map = RwLock::new(BTreeMap::new()); let pub_key_map = RwLock::new(BTreeMap::new());
// We do not add the event_id field to the pdu here because of signature and hashes checks // We do not add the event_id field to the pdu here because of signature and hashes checks
let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&response.event, &db) let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&response.event) {
{
Ok(t) => t, Ok(t) => t,
Err(_) => { Err(_) => {
// Event could not be converted to canonical json // Event could not be converted to canonical json
+7 -4
View File
@@ -23,7 +23,7 @@ use ruma::{
EventType, EventType,
}, },
serde::{CanonicalJsonObject, JsonObject}, serde::{CanonicalJsonObject, JsonObject},
RoomAliasId, RoomId, RoomAliasId, RoomId, RoomVersionId,
}; };
use serde_json::{json, value::to_raw_value}; use serde_json::{json, value::to_raw_value};
use std::{ use std::{
@@ -109,7 +109,7 @@ pub async fn create_room_route(
let room_version = match body.room_version.clone() { let room_version = match body.room_version.clone() {
Some(room_version) => { Some(room_version) => {
if db.rooms.is_supported_version(&db, &room_version) { if room_version == RoomVersionId::Version5 || room_version == RoomVersionId::Version6 {
room_version room_version
} else { } else {
return Err(Error::BadRequest( return Err(Error::BadRequest(
@@ -118,7 +118,7 @@ pub async fn create_room_route(
)); ));
} }
} }
None => db.globals.default_room_version(), None => RoomVersionId::Version6,
}; };
let content = match &body.creation_content { let content = match &body.creation_content {
@@ -505,7 +505,10 @@ pub async fn upgrade_room_route(
) -> ConduitResult<upgrade_room::Response> { ) -> ConduitResult<upgrade_room::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated"); let sender_user = body.sender_user.as_ref().expect("user is authenticated");
if !db.rooms.is_supported_version(&db, &body.new_version) { if !matches!(
body.new_version,
RoomVersionId::Version5 | RoomVersionId::Version6
) {
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::UnsupportedRoomVersion, ErrorKind::UnsupportedRoomVersion,
"This server does not support that room version.", "This server does not support that room version.",
+6 -14
View File
@@ -24,7 +24,7 @@ use rocket::{
request::{FromRequest, Request}, request::{FromRequest, Request},
Shutdown, State, Shutdown, State,
}; };
use ruma::{DeviceId, EventId, RoomId, RoomVersionId, ServerName, UserId}; use ruma::{DeviceId, EventId, RoomId, ServerName, UserId};
use serde::{de::IgnoredAny, Deserialize}; use serde::{de::IgnoredAny, Deserialize};
use std::{ use std::{
collections::{BTreeMap, HashMap, HashSet}, collections::{BTreeMap, HashMap, HashSet},
@@ -63,12 +63,8 @@ pub struct Config {
allow_federation: bool, allow_federation: bool,
#[serde(default = "true_fn")] #[serde(default = "true_fn")]
allow_room_creation: bool, allow_room_creation: bool,
#[serde(default = "true_fn")]
allow_unstable_room_versions: bool,
#[serde(default = "false_fn")] #[serde(default = "false_fn")]
pub allow_jaeger: bool, pub allow_jaeger: bool,
#[serde(default = "default_room_version")]
default_room_version: RoomVersionId,
#[serde(default = "false_fn")] #[serde(default = "false_fn")]
pub tracing_flame: bool, pub tracing_flame: bool,
#[serde(default)] #[serde(default)]
@@ -141,10 +137,6 @@ fn default_max_concurrent_requests() -> u16 {
100 100
} }
fn default_room_version() -> RoomVersionId {
RoomVersionId::Version6
}
fn default_log() -> String { fn default_log() -> String {
"info,state_res=warn,rocket=off,_=off,sled=off".to_owned() "info,state_res=warn,rocket=off,_=off,sled=off".to_owned()
} }
@@ -322,11 +314,11 @@ impl Database {
.try_into() .try_into()
.expect("pdu cache capacity fits into usize"), .expect("pdu cache capacity fits into usize"),
)), )),
auth_chain_cache: Mutex::new(LruCache::new(1_000_000)), auth_chain_cache: Mutex::new(LruCache::new(10_000_000)),
shorteventid_cache: Mutex::new(LruCache::new(1_000_000)), shorteventid_cache: Mutex::new(LruCache::new(10_000_000)),
eventidshort_cache: Mutex::new(LruCache::new(1_000_000)), eventidshort_cache: Mutex::new(LruCache::new(10_000_000)),
shortstatekey_cache: Mutex::new(LruCache::new(1_000_000)), shortstatekey_cache: Mutex::new(LruCache::new(10_000_000)),
statekeyshort_cache: Mutex::new(LruCache::new(1_000_000)), statekeyshort_cache: Mutex::new(LruCache::new(10_000_000)),
our_real_users_cache: RwLock::new(HashMap::new()), our_real_users_cache: RwLock::new(HashMap::new()),
appservice_in_room_cache: RwLock::new(HashMap::new()), appservice_in_room_cache: RwLock::new(HashMap::new()),
stateinfo_cache: Mutex::new(LruCache::new(1000)), stateinfo_cache: Mutex::new(LruCache::new(1000)),
+6 -6
View File
@@ -21,7 +21,7 @@ thread_local! {
struct PreparedStatementIterator<'a> { struct PreparedStatementIterator<'a> {
pub iterator: Box<dyn Iterator<Item = TupleOfBytes> + 'a>, pub iterator: Box<dyn Iterator<Item = TupleOfBytes> + 'a>,
pub statement_ref: NonAliasingBox<rusqlite::Statement<'a>>, pub statement_ref: NonAliasingBox<rusqlite::CachedStatement<'a>>,
} }
impl Iterator for PreparedStatementIterator<'_> { impl Iterator for PreparedStatementIterator<'_> {
@@ -52,7 +52,7 @@ impl Engine {
fn prepare_conn(path: &Path, cache_size_kb: u32) -> Result<Connection> { fn prepare_conn(path: &Path, cache_size_kb: u32) -> Result<Connection> {
let conn = Connection::open(&path)?; let conn = Connection::open(&path)?;
conn.pragma_update(Some(Main), "page_size", &2048)?; conn.pragma_update(Some(Main), "page_size", &32768)?;
conn.pragma_update(Some(Main), "journal_mode", &"WAL")?; conn.pragma_update(Some(Main), "journal_mode", &"WAL")?;
conn.pragma_update(Some(Main), "synchronous", &"NORMAL")?; conn.pragma_update(Some(Main), "synchronous", &"NORMAL")?;
conn.pragma_update(Some(Main), "cache_size", &(-i64::from(cache_size_kb)))?; conn.pragma_update(Some(Main), "cache_size", &(-i64::from(cache_size_kb)))?;
@@ -136,7 +136,7 @@ impl SqliteTable {
fn get_with_guard(&self, guard: &Connection, key: &[u8]) -> Result<Option<Vec<u8>>> { fn get_with_guard(&self, guard: &Connection, key: &[u8]) -> Result<Option<Vec<u8>>> {
//dbg!(&self.name); //dbg!(&self.name);
Ok(guard Ok(guard
.prepare(format!("SELECT value FROM {} WHERE key = ?", self.name).as_str())? .prepare_cached(format!("SELECT value FROM {} WHERE key = ?", self.name).as_str())?
.query_row([key], |row| row.get(0)) .query_row([key], |row| row.get(0))
.optional()?) .optional()?)
} }
@@ -161,7 +161,7 @@ impl SqliteTable {
) -> Box<dyn Iterator<Item = TupleOfBytes> + 'a> { ) -> Box<dyn Iterator<Item = TupleOfBytes> + 'a> {
let statement = Box::leak(Box::new( let statement = Box::leak(Box::new(
guard guard
.prepare(&format!( .prepare_cached(&format!(
"SELECT key, value FROM {} ORDER BY key ASC", "SELECT key, value FROM {} ORDER BY key ASC",
&self.name &self.name
)) ))
@@ -290,7 +290,7 @@ impl Tree for SqliteTable {
if backwards { if backwards {
let statement = Box::leak(Box::new( let statement = Box::leak(Box::new(
guard guard
.prepare(&format!( .prepare_cached(&format!(
"SELECT key, value FROM {} WHERE key <= ? ORDER BY key DESC", "SELECT key, value FROM {} WHERE key <= ? ORDER BY key DESC",
&self.name &self.name
)) ))
@@ -315,7 +315,7 @@ impl Tree for SqliteTable {
} else { } else {
let statement = Box::leak(Box::new( let statement = Box::leak(Box::new(
guard guard
.prepare(&format!( .prepare_cached(&format!(
"SELECT key, value FROM {} WHERE key >= ? ORDER BY key ASC", "SELECT key, value FROM {} WHERE key >= ? ORDER BY key ASC",
&self.name &self.name
)) ))
+2 -41
View File
@@ -4,8 +4,7 @@ use ruma::{
client::r0::sync::sync_events, client::r0::sync::sync_events,
federation::discovery::{ServerSigningKeys, VerifyKey}, federation::discovery::{ServerSigningKeys, VerifyKey},
}, },
DeviceId, EventId, MilliSecondsSinceUnixEpoch, RoomId, RoomVersionId, ServerName, DeviceId, EventId, MilliSecondsSinceUnixEpoch, RoomId, ServerName, ServerSigningKeyId, UserId,
ServerSigningKeyId, UserId,
}; };
use std::{ use std::{
collections::{BTreeMap, HashMap}, collections::{BTreeMap, HashMap},
@@ -40,8 +39,6 @@ pub struct Globals {
keypair: Arc<ruma::signatures::Ed25519KeyPair>, keypair: Arc<ruma::signatures::Ed25519KeyPair>,
dns_resolver: TokioAsyncResolver, dns_resolver: TokioAsyncResolver,
jwt_decoding_key: Option<jsonwebtoken::DecodingKey<'static>>, jwt_decoding_key: Option<jsonwebtoken::DecodingKey<'static>>,
pub stable_room_versions: Vec<RoomVersionId>,
pub unstable_room_versions: Vec<RoomVersionId>,
pub(super) server_signingkeys: Arc<dyn Tree>, pub(super) server_signingkeys: Arc<dyn Tree>,
pub bad_event_ratelimiter: Arc<RwLock<HashMap<EventId, RateLimitState>>>, pub bad_event_ratelimiter: Arc<RwLock<HashMap<EventId, RateLimitState>>>,
pub bad_signature_ratelimiter: Arc<RwLock<HashMap<Vec<String>, RateLimitState>>>, pub bad_signature_ratelimiter: Arc<RwLock<HashMap<Vec<String>, RateLimitState>>>,
@@ -135,16 +132,7 @@ impl Globals {
.as_ref() .as_ref()
.map(|secret| jsonwebtoken::DecodingKey::from_secret(secret.as_bytes()).into_static()); .map(|secret| jsonwebtoken::DecodingKey::from_secret(secret.as_bytes()).into_static());
// Supported and stable room versions let s = Self {
let stable_room_versions = vec![RoomVersionId::Version6];
// Experimental, partially supported room versions
let unstable_room_versions = vec![
RoomVersionId::Version3,
RoomVersionId::Version4,
RoomVersionId::Version5,
];
let mut s = Self {
globals, globals,
config, config,
keypair: Arc::new(keypair), keypair: Arc::new(keypair),
@@ -155,8 +143,6 @@ impl Globals {
tls_name_override, tls_name_override,
server_signingkeys, server_signingkeys,
jwt_decoding_key, jwt_decoding_key,
stable_room_versions,
unstable_room_versions,
bad_event_ratelimiter: Arc::new(RwLock::new(HashMap::new())), bad_event_ratelimiter: Arc::new(RwLock::new(HashMap::new())),
bad_signature_ratelimiter: Arc::new(RwLock::new(HashMap::new())), bad_signature_ratelimiter: Arc::new(RwLock::new(HashMap::new())),
servername_ratelimiter: Arc::new(RwLock::new(HashMap::new())), servername_ratelimiter: Arc::new(RwLock::new(HashMap::new())),
@@ -169,14 +155,6 @@ impl Globals {
fs::create_dir_all(s.get_media_folder())?; fs::create_dir_all(s.get_media_folder())?;
if !s
.supported_room_versions()
.contains(&s.config.default_room_version)
{
error!("Room version in config isn't supported, falling back to Version 6");
s.config.default_room_version = RoomVersionId::Version6;
};
Ok(s) Ok(s)
} }
@@ -236,14 +214,6 @@ impl Globals {
self.config.allow_room_creation self.config.allow_room_creation
} }
pub fn allow_unstable_room_versions(&self) -> bool {
self.config.allow_unstable_room_versions
}
pub fn default_room_version(&self) -> RoomVersionId {
self.config.default_room_version.clone()
}
pub fn trusted_servers(&self) -> &[Box<ServerName>] { pub fn trusted_servers(&self) -> &[Box<ServerName>] {
&self.config.trusted_servers &self.config.trusted_servers
} }
@@ -276,15 +246,6 @@ impl Globals {
&self.config.turn_secret &self.config.turn_secret
} }
pub fn supported_room_versions(&self) -> Vec<RoomVersionId> {
let mut room_versions: Vec<RoomVersionId> = vec![];
room_versions.extend(self.stable_room_versions.clone());
if self.allow_unstable_room_versions() {
room_versions.extend(self.unstable_room_versions.clone());
};
room_versions
}
/// TODO: the key valid until timestamp is only honored in room version > 4 /// TODO: the key valid until timestamp is only honored in room version > 4
/// Remove the outdated keys and insert the new ones. /// Remove the outdated keys and insert the new ones.
/// ///
+7 -30
View File
@@ -129,12 +129,6 @@ pub struct Rooms {
} }
impl Rooms { impl Rooms {
/// Returns true if a given room version is supported
#[tracing::instrument(skip(self, db))]
pub fn is_supported_version(&self, db: &Database, room_version: &RoomVersionId) -> bool {
db.globals.supported_room_versions().contains(room_version)
}
/// Builds a StateMap by iterating over all keys that start /// Builds a StateMap by iterating over all keys that start
/// with state_hash, this gives the full state for the given state_hash. /// with state_hash, this gives the full state for the given state_hash.
#[tracing::instrument(skip(self))] #[tracing::instrument(skip(self))]
@@ -1989,10 +1983,9 @@ impl Rooms {
None None
}; };
// If there was no create event yet, assume we are creating a room with the default // If there was no create event yet, assume we are creating a version 6 room right now
// version right now
let room_version_id = create_event_content let room_version_id = create_event_content
.map_or(db.globals.default_room_version(), |create_event| { .map_or(RoomVersionId::Version6, |create_event| {
create_event.room_version create_event.room_version
}); });
let room_version = RoomVersion::new(&room_version_id).expect("room version is supported"); let room_version = RoomVersion::new(&room_version_id).expect("room version is supported");
@@ -2785,7 +2778,11 @@ impl Rooms {
let (make_leave_response, remote_server) = make_leave_response_and_server?; let (make_leave_response, remote_server) = make_leave_response_and_server?;
let room_version_id = match make_leave_response.room_version { let room_version_id = match make_leave_response.room_version {
Some(version) if self.is_supported_version(&db, &version) => version, Some(version)
if version == RoomVersionId::Version5 || version == RoomVersionId::Version6 =>
{
version
}
_ => return Err(Error::BadServerResponse("Room version is not supported")), _ => return Err(Error::BadServerResponse("Room version is not supported")),
}; };
@@ -3451,24 +3448,4 @@ impl Rooms {
Ok(()) Ok(())
} }
/// Returns the room's version.
#[tracing::instrument(skip(self))]
pub fn get_room_version(&self, room_id: &RoomId) -> Result<RoomVersionId> {
let create_event = self.room_state_get(room_id, &EventType::RoomCreate, "")?;
let create_event_content: Option<RoomCreateEventContent> = create_event
.as_ref()
.map(|create_event| {
serde_json::from_str(create_event.content.get()).map_err(|e| {
warn!("Invalid create event: {}", e);
Error::bad_database("Invalid create event in db.")
})
})
.transpose()?;
let room_version = create_event_content
.map(|create_event| create_event.room_version)
.ok_or_else(|| Error::BadDatabase("Invalid room version"))?;
Ok(room_version)
}
} }
+4 -12
View File
@@ -1,4 +1,4 @@
use crate::{Database, Error}; use crate::Error;
use ruma::{ use ruma::{
events::{ events::{
room::member::RoomMemberEventContent, AnyEphemeralRoomEvent, AnyInitialStateEvent, room::member::RoomMemberEventContent, AnyEphemeralRoomEvent, AnyInitialStateEvent,
@@ -6,7 +6,7 @@ use ruma::{
EventType, StateEvent, EventType, StateEvent,
}, },
serde::{CanonicalJsonObject, CanonicalJsonValue, Raw}, serde::{CanonicalJsonObject, CanonicalJsonValue, Raw},
state_res, EventId, MilliSecondsSinceUnixEpoch, RoomId, UInt, UserId, state_res, EventId, MilliSecondsSinceUnixEpoch, RoomId, RoomVersionId, UInt, UserId,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::{ use serde_json::{
@@ -331,24 +331,16 @@ impl Ord for PduEvent {
/// Returns a tuple of the new `EventId` and the PDU as a `BTreeMap<String, CanonicalJsonValue>`. /// Returns a tuple of the new `EventId` and the PDU as a `BTreeMap<String, CanonicalJsonValue>`.
pub(crate) fn gen_event_id_canonical_json( pub(crate) fn gen_event_id_canonical_json(
pdu: &RawJsonValue, pdu: &RawJsonValue,
db: &Database,
) -> crate::Result<(EventId, CanonicalJsonObject)> { ) -> crate::Result<(EventId, CanonicalJsonObject)> {
let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| { let value = serde_json::from_str(pdu.get()).map_err(|e| {
warn!("Error parsing incoming event {:?}: {:?}", pdu, e); warn!("Error parsing incoming event {:?}: {:?}", pdu, e);
Error::BadServerResponse("Invalid PDU in server response") Error::BadServerResponse("Invalid PDU in server response")
})?; })?;
let room_id = value
.get("room_id")
.and_then(|id| RoomId::try_from(id.as_str()?).ok())
.expect("Invalid room id in event");
let room_version_id = db.rooms.get_room_version(&room_id);
let event_id = EventId::try_from(&*format!( let event_id = EventId::try_from(&*format!(
"${}", "${}",
// Anything higher than version3 behaves the same // Anything higher than version3 behaves the same
ruma::signatures::reference_hash(&value, &room_version_id?) ruma::signatures::reference_hash(&value, &RoomVersionId::Version6)
.expect("ruma can calculate reference hashes") .expect("ruma can calculate reference hashes")
)) ))
.expect("ruma's reference hashes are valid event ids"); .expect("ruma's reference hashes are valid event ids");
+16 -15
View File
@@ -47,7 +47,6 @@ use ruma::{
}, },
AnyEphemeralRoomEvent, EventType, AnyEphemeralRoomEvent, EventType,
}, },
int,
receipt::ReceiptType, receipt::ReceiptType,
serde::JsonObject, serde::JsonObject,
signatures::{CanonicalJsonObject, CanonicalJsonValue}, signatures::{CanonicalJsonObject, CanonicalJsonValue},
@@ -725,7 +724,7 @@ pub async fn send_transaction_message_route(
for pdu in &body.pdus { for pdu in &body.pdus {
// We do not add the event_id field to the pdu here because of signature and hashes checks // We do not add the event_id field to the pdu here because of signature and hashes checks
let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(pdu, &db) { let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(pdu) {
Ok(t) => t, Ok(t) => t,
Err(_) => { Err(_) => {
// Event could not be converted to canonical json // Event could not be converted to canonical json
@@ -751,14 +750,14 @@ pub async fn send_transaction_message_route(
.roomid_mutex_federation .roomid_mutex_federation
.write() .write()
.unwrap() .unwrap()
.entry(room_id.clone()) .entry(RoomId::try_from("!asdf:asfd.asdf").unwrap())
.or_default(), .or_default(),
); );
let mutex_lock = mutex.lock().await; let mutex_lock = mutex.lock().await;
let start_time = Instant::now(); let start_time = Instant::now();
resolved_map.insert( resolved_map.insert(
event_id.clone(), event_id.clone(),
handle_incoming_pdu( dbg!(handle_incoming_pdu(
&body.origin, &body.origin,
&event_id, &event_id,
&room_id, &room_id,
@@ -767,7 +766,7 @@ pub async fn send_transaction_message_route(
&db, &db,
&pub_key_map, &pub_key_map,
) )
.await .await)
.map(|_| ()), .map(|_| ()),
); );
drop(mutex_lock); drop(mutex_lock);
@@ -956,6 +955,7 @@ pub(crate) async fn handle_incoming_pdu<'a>(
db: &'a Database, db: &'a Database,
pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>, pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>,
) -> Result<Option<Vec<u8>>, String> { ) -> Result<Option<Vec<u8>>, String> {
warn!("Handling incoming pdu: {:?}", value);
match db.rooms.exists(room_id) { match db.rooms.exists(room_id) {
Ok(true) => {} Ok(true) => {}
_ => { _ => {
@@ -1000,6 +1000,7 @@ pub(crate) async fn handle_incoming_pdu<'a>(
return Ok(None); return Ok(None);
} }
/* stop fetching prev events for now
// 9. Fetch any missing prev events doing all checks listed here starting at 1. These are timeline events // 9. Fetch any missing prev events doing all checks listed here starting at 1. These are timeline events
let mut graph = HashMap::new(); let mut graph = HashMap::new();
let mut eventid_info = HashMap::new(); let mut eventid_info = HashMap::new();
@@ -1114,6 +1115,7 @@ pub(crate) async fn handle_incoming_pdu<'a>(
); );
} }
} }
*/
upgrade_outlier_to_timeline_pdu( upgrade_outlier_to_timeline_pdu(
incoming_pdu, incoming_pdu,
@@ -1911,7 +1913,7 @@ pub(crate) fn fetch_and_handle_outliers<'a>(
Ok(res) => { Ok(res) => {
warn!("Got {} over federation", id); warn!("Got {} over federation", id);
let (calculated_event_id, value) = let (calculated_event_id, value) =
match crate::pdu::gen_event_id_canonical_json(&res.pdu, &db) { match crate::pdu::gen_event_id_canonical_json(&res.pdu) {
Ok(t) => t, Ok(t) => t,
Err(_) => { Err(_) => {
back_off((**id).clone()); back_off((**id).clone());
@@ -2670,12 +2672,10 @@ pub fn create_join_event_template_route(
None None
}; };
// If there was no create event yet, assume we are creating a room with the default version // If there was no create event yet, assume we are creating a version 6 room right now
// right now let room_version_id = create_event_content.map_or(RoomVersionId::Version6, |create_event| {
let room_version_id = create_event_content create_event.room_version
.map_or(db.globals.default_room_version(), |create_event| { });
create_event.room_version
});
let room_version = RoomVersion::new(&room_version_id).expect("room version is supported"); let room_version = RoomVersion::new(&room_version_id).expect("room version is supported");
if !body.ver.contains(&room_version_id) { if !body.ver.contains(&room_version_id) {
@@ -2815,7 +2815,7 @@ async fn create_join_event(
// let mut auth_cache = EventMap::new(); // let mut auth_cache = EventMap::new();
// We do not add the event_id field to the pdu here because of signature and hashes checks // We do not add the event_id field to the pdu here because of signature and hashes checks
let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(pdu, &db) { let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(pdu) {
Ok(t) => t, Ok(t) => t,
Err(_) => { Err(_) => {
// Event could not be converted to canonical json // Event could not be converted to canonical json
@@ -2840,7 +2840,7 @@ async fn create_join_event(
.roomid_mutex_federation .roomid_mutex_federation
.write() .write()
.unwrap() .unwrap()
.entry(room_id.clone()) .entry(RoomId::try_from("!asdf:asfd.asdf").unwrap()) // make all rooms share the same lock for now
.or_default(), .or_default(),
); );
let mutex_lock = mutex.lock().await; let mutex_lock = mutex.lock().await;
@@ -2939,7 +2939,8 @@ pub async fn create_invite_route(
return Err(Error::bad_config("Federation is disabled.")); return Err(Error::bad_config("Federation is disabled."));
} }
if !db.rooms.is_supported_version(&db, &body.room_version) { if body.room_version != RoomVersionId::Version5 && body.room_version != RoomVersionId::Version6
{
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::IncompatibleRoomVersion { ErrorKind::IncompatibleRoomVersion {
room_version: body.room_version.clone(), room_version: body.room_version.clone(),