improvement: efficient /sync, mutex for federation transactions

Conduit did not send the event in /sync because of a race condition.
There is a brief moment in time where Conduit accepted the event, but
did not store it yet. So when the client /syncs it updates the since
token without sending the event. I hope I fixed it by significantly
shortening the race-condition period.

Cargo.lock

@@ -272,9 +272,9 @@ dependencies = [
 
 [[package]]
 name = "const-oid"
-version = "0.5.2"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "279bc8fc53f788a75c7804af68237d1fce02cde1e275a886a4b320604dc2aeda"
+checksum = "44c32f031ea41b4291d695026c023b95d59db2d8a2c7640800ed56bc8f510f22"
 
 [[package]]
 name = "const_fn"
@@ -394,9 +394,9 @@ dependencies = [
 
 [[package]]
 name = "der"
-version = "0.3.5"
+version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2eeb9d92785d1facb50567852ce75d0858630630e7eabea59cf7eb7474051087"
+checksum = "49f215f706081a44cb702c71c39a52c05da637822e9c1645a50b7202689e982d"
 dependencies = [
  "const-oid",
 ]
@@ -1475,9 +1475,9 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
 [[package]]
 name = "pkcs8"
-version = "0.6.1"
+version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c9c2f795bc591cb3384cb64082a578b89207ac92bb89c9d98c1ea2ace7cd8110"
+checksum = "09d156817ae0125e8aa5067710b0db24f0984830614f99875a70aa5e3b74db69"
 dependencies = [
  "der",
  "spki",
@@ -1882,8 +1882,8 @@ dependencies = [
 
 [[package]]
 name = "ruma"
-version = "0.1.2"
+version = "0.2.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "assign",
  "js_int",
@@ -1903,8 +1903,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-api"
-version = "0.17.0"
+version = "0.17.1"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "bytes",
  "http",
@@ -1919,8 +1919,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-api-macros"
-version = "0.17.0"
+version = "0.17.1"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "proc-macro-crate",
  "proc-macro2",
@@ -1930,8 +1930,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-appservice-api"
-version = "0.2.0"
+version = "0.3.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "ruma-api",
  "ruma-common",
@@ -1944,8 +1944,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-client-api"
-version = "0.10.2"
+version = "0.11.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "assign",
  "bytes",
@@ -1964,8 +1964,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-common"
-version = "0.5.3"
+version = "0.5.4"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "indexmap",
  "js_int",
@@ -1979,8 +1979,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-events"
-version = "0.22.2"
+version = "0.23.2"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "indoc",
  "js_int",
@@ -1994,8 +1994,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-events-macros"
-version = "0.22.2"
+version = "0.23.2"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "proc-macro-crate",
  "proc-macro2",
@@ -2005,8 +2005,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-federation-api"
-version = "0.1.0"
+version = "0.2.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "js_int",
  "ruma-api",
@@ -2020,8 +2020,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-identifiers"
-version = "0.19.2"
+version = "0.19.4"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "paste",
  "rand 0.8.3",
@@ -2034,8 +2034,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-identifiers-macros"
-version = "0.19.2"
+version = "0.19.4"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "quote",
  "ruma-identifiers-validation",
@@ -2045,12 +2045,12 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.4.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 
 [[package]]
 name = "ruma-identity-service-api"
-version = "0.1.0"
+version = "0.2.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "js_int",
  "ruma-api",
@@ -2062,8 +2062,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-push-gateway-api"
-version = "0.1.0"
+version = "0.2.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "js_int",
  "ruma-api",
@@ -2077,8 +2077,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-serde"
-version = "0.4.0"
+version = "0.4.1"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "bytes",
  "form_urlencoded",
@@ -2091,8 +2091,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-serde-macros"
-version = "0.4.0"
+version = "0.4.1"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "proc-macro-crate",
  "proc-macro2",
@@ -2102,8 +2102,8 @@ dependencies = [
 
 [[package]]
 name = "ruma-signatures"
-version = "0.7.2"
+version = "0.8.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "base64 0.13.0",
  "ed25519-dalek",
@@ -2115,13 +2115,12 @@ dependencies = [
  "sha2",
  "thiserror",
  "tracing",
- "untrusted",
 ]
 
 [[package]]
 name = "ruma-state-res"
-version = "0.1.0"
+version = "0.2.0"
-source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9"
+source = "git+https://github.com/DevinR528/ruma?branch=state-closure#95208b9d03876e7c85543fe8655ceb2f7dc76363"
 dependencies = [
  "itertools 0.10.0",
  "js_int",
@@ -2130,7 +2129,6 @@ dependencies = [
  "ruma-events",
  "ruma-identifiers",
  "ruma-serde",
- "ruma-signatures",
  "serde",
  "serde_json",
  "thiserror",
@@ -2444,9 +2442,9 @@ checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d"
 
 [[package]]
 name = "spki"
-version = "0.3.0"
+version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9dae7e047abc519c96350e9484a96c6bf1492348af912fd3446dd2dc323f6268"
+checksum = "987637c5ae6b3121aba9d513f869bd2bff11c4cc086c22473befd6649c0bd521"
 dependencies = [
  "der",
 ]

Cargo.toml

@@ -8,7 +8,6 @@ repository = "https://gitlab.com/famedly/conduit"
 readme = "README.md"
 version = "0.1.0"
 edition = "2018"
-rust = "1.50"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -18,7 +17,8 @@ rust = "1.50"
 rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "801e04bd5369eb39e126c75f6d11e1e9597304d8", features = ["tls"] } # Used to handle requests
 
 # Used for matrix spec type definitions and helpers
-ruma = { git = "https://github.com/ruma/ruma", rev = "5a7e2cddcf257e367465cced51442c91e8f557c9", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] }
+#ruma = { git = "https://github.com/ruma/ruma", rev = "5a7e2cddcf257e367465cced51442c91e8f557c9", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] }
+ruma = { git = "https://github.com/DevinR528/ruma", branch = "state-closure", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] }
 #ruma = { path = "../ruma/crates/ruma", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] }
 
 # Used for long polling and federation sender, should be the same as rocket::tokio

rust-toolchain

@@ -1 +1 @@
-1.50.0
+1.51.0

src/appservice_server.rs

@@ -46,7 +46,10 @@ where
     *reqwest_request.timeout_mut() = Some(Duration::from_secs(30));
 
     let url = reqwest_request.url().clone();
-    let mut response = globals.reqwest_client().execute(reqwest_request).await?;
+    let mut response = globals
+        .reqwest_client()
+        .execute(reqwest_request)
+        .await?;
 
     // reqwest::Response -> http::Response conversion
     let status = response.status();

src/client_server/membership.rs

@@ -25,7 +25,7 @@ use ruma::{
         EventType,
     },
     serde::{to_canonical_value, CanonicalJsonObject, CanonicalJsonValue, Raw},
-    state_res::{self, EventMap, RoomVersion},
+    state_res::{self, RoomVersion},
     uint, EventId, RoomId, RoomVersionId, ServerName, UserId,
 };
 use std::{
@@ -615,16 +615,9 @@ async fn join_room_by_id_helper(
         // pdu without it's state. This is okay because append_pdu can't fail.
         let statehashid = db.rooms.append_to_state(&pdu, &db.globals)?;
 
-        let count = db.globals.next_count()?;
-        let mut pdu_id = room_id.as_bytes().to_vec();
-        pdu_id.push(0xff);
-        pdu_id.extend_from_slice(&count.to_be_bytes());
-
         db.rooms.append_pdu(
             &pdu,
             utils::to_canonical_object(&pdu).expect("Pdu is valid canonical object"),
-            count,
-            &pdu_id,
             &[pdu.event_id.clone()],
             db,
         )?;
@@ -883,7 +876,6 @@ pub async fn invite_helper(
             .await?;
 
         let pub_key_map = RwLock::new(BTreeMap::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
         let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&response.event) {
@@ -906,14 +898,24 @@ pub async fn invite_helper(
         )
         .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?;
 
+        let mutex = Arc::clone(
+            db.globals
+                .roomid_mutex
+                .write()
+                .unwrap()
+                .entry(room_id.clone())
+                .or_default(),
+        );
+        let mutex_lock = mutex.lock().await;
+
         let pdu_id = server_server::handle_incoming_pdu(
             &origin,
             &event_id,
+            &room_id,
             value,
             true,
             &db,
             &pub_key_map,
-            &mut auth_cache,
         )
         .await
         .map_err(|_| {
@@ -926,6 +928,7 @@ pub async fn invite_helper(
             ErrorKind::InvalidParam,
             "Could not accept incoming PDU as timeline event.",
         ))?;
+        drop(mutex_lock);
 
         for server in db
             .rooms

src/client_server/sync.rs

@@ -88,7 +88,9 @@ pub async fn sync_events_route(
 
     let we_have_to_wait = rx.borrow().is_none();
     if we_have_to_wait {
-        let _ = rx.changed().await;
+        if let Err(e) = rx.changed().await {
+            error!("Error waiting for sync: {}", e);
+        }
     }
 
     let result = match rx
@@ -222,13 +224,16 @@ async fn sync_helper(
 
         // Database queries:
 
-        let current_shortstatehash = db.rooms.current_shortstatehash(&room_id)?;
+        let current_shortstatehash = db
+            .rooms
+            .current_shortstatehash(&room_id)?
+            .expect("All rooms have state");
 
-        // These type is Option<Option<_>>. The outer Option is None when there is no event between
+        let first_pdu_before_since = db
-        // since and the current room state, meaning there should be no updates.
+            .rooms
-        // The inner Option is None when there is an event, but there is no state hash associated
+            .pdus_until(&sender_user, &room_id, since)
-        // with it. This can happen for the RoomCreate event, so all updates should arrive.
+            .next()
-        let first_pdu_before_since = db.rooms.pdus_until(&sender_user, &room_id, since).next();
+            .transpose()?;
 
         let pdus_after_since = db
             .rooms
@@ -236,11 +241,78 @@ async fn sync_helper(
             .next()
             .is_some();
 
-        let since_shortstatehash = first_pdu_before_since.as_ref().map(|pdu| {
+        let since_shortstatehash = first_pdu_before_since
-            db.rooms
+            .as_ref()
-                .pdu_shortstatehash(&pdu.as_ref().ok()?.1.event_id)
+            .map(|pdu| {
-                .ok()?
+                db.rooms
-        });
+                    .pdu_shortstatehash(&pdu.1.event_id)
+                    .transpose()
+                    .expect("all pdus have state")
+            })
+            .transpose()?;
+
+        // Calculates joined_member_count, invited_member_count and heroes
+        let calculate_counts = || {
+            let joined_member_count = db.rooms.room_members(&room_id).count();
+            let invited_member_count = db.rooms.room_members_invited(&room_id).count();
+
+            // Recalculate heroes (first 5 members)
+            let mut heroes = Vec::new();
+
+            if joined_member_count + invited_member_count <= 5 {
+                // Go through all PDUs and for each member event, check if the user is still joined or
+                // invited until we have 5 or we reach the end
+
+                for hero in db
+                    .rooms
+                    .all_pdus(&sender_user, &room_id)
+                    .filter_map(|pdu| pdu.ok()) // Ignore all broken pdus
+                    .filter(|(_, pdu)| pdu.kind == EventType::RoomMember)
+                    .map(|(_, pdu)| {
+                        let content = serde_json::from_value::<
+                            ruma::events::room::member::MemberEventContent,
+                        >(pdu.content.clone())
+                        .map_err(|_| Error::bad_database("Invalid member event in database."))?;
+
+                        if let Some(state_key) = &pdu.state_key {
+                            let user_id = UserId::try_from(state_key.clone()).map_err(|_| {
+                                Error::bad_database("Invalid UserId in member PDU.")
+                            })?;
+
+                            // The membership was and still is invite or join
+                            if matches!(
+                                content.membership,
+                                MembershipState::Join | MembershipState::Invite
+                            ) && (db.rooms.is_joined(&user_id, &room_id)?
+                                || db.rooms.is_invited(&user_id, &room_id)?)
+                            {
+                                Ok::<_, Error>(Some(state_key.clone()))
+                            } else {
+                                Ok(None)
+                            }
+                        } else {
+                            Ok(None)
+                        }
+                    })
+                    // Filter out buggy users
+                    .filter_map(|u| u.ok())
+                    // Filter for possible heroes
+                    .flatten()
+                {
+                    if heroes.contains(&hero) || hero == sender_user.as_str() {
+                        continue;
+                    }
+
+                    heroes.push(hero);
+                }
+            }
+
+            (
+                Some(joined_member_count),
+                Some(invited_member_count),
+                heroes,
+            )
+        };
 
         let (
             heroes,
@@ -248,63 +320,107 @@ async fn sync_helper(
             invited_member_count,
             joined_since_last_sync,
             state_events,
-        ) = if pdus_after_since && Some(current_shortstatehash) != since_shortstatehash {
+        ) = if since_shortstatehash.is_none() {
-            let current_state = db.rooms.room_state_full(&room_id)?;
+            // Probably since = 0, we will do an initial sync
-            let current_members = current_state
+            let (joined_member_count, invited_member_count, heroes) = calculate_counts();
-                .iter()
-                .filter(|(key, _)| key.0 == EventType::RoomMember)
-                .map(|(key, value)| (&key.1, value)) // Only keep state key
-                .collect::<Vec<_>>();
-            let encrypted_room = current_state
-                .get(&(EventType::RoomEncryption, "".to_owned()))
-                .is_some();
-            let since_state = since_shortstatehash
-                .as_ref()
-                .map(|since_shortstatehash| {
-                    since_shortstatehash
-                        .map(|since_shortstatehash| db.rooms.state_full(since_shortstatehash))
-                        .transpose()
-                })
-                .transpose()?;
 
-            let since_encryption = since_state.as_ref().map(|state| {
+            let current_state_ids = db.rooms.state_full_ids(current_shortstatehash)?;
-                state
+            let state_events = current_state_ids
-                    .as_ref()
+                .iter()
-                    .map(|state| state.get(&(EventType::RoomEncryption, "".to_owned())))
+                .map(|id| db.rooms.get_pdu(id))
-            });
+                .filter_map(|r| r.ok().flatten())
+                .collect::<Vec<_>>();
+
+            (
+                heroes,
+                joined_member_count,
+                invited_member_count,
+                true,
+                state_events
+            )
+        } else if !pdus_after_since || since_shortstatehash == Some(current_shortstatehash) {
+            // No state changes
+            (Vec::new(), None, None, false, Vec::new())
+        } else {
+            // Incremental /sync
+            let since_shortstatehash = since_shortstatehash.unwrap();
+
+            let since_sender_member = db
+                .rooms
+                .state_get(
+                    since_shortstatehash,
+                    &EventType::RoomMember,
+                    sender_user.as_str(),
+                )?
+                .and_then(|pdu| {
+                    serde_json::from_value::<Raw<ruma::events::room::member::MemberEventContent>>(
+                        pdu.content.clone(),
+                    )
+                    .expect("Raw::from_value always works")
+                    .deserialize()
+                    .map_err(|_| Error::bad_database("Invalid PDU in database."))
+                    .ok()
+                });
+
+            let joined_since_last_sync = since_sender_member
+                .map_or(true, |member| member.membership != MembershipState::Join);
+
+            let current_state_ids = db.rooms.state_full_ids(current_shortstatehash)?;
+
+            let since_state_ids = db.rooms.state_full_ids(since_shortstatehash)?;
+
+            let state_events = if joined_since_last_sync {
+                current_state_ids
+                    .iter()
+                    .map(|id| db.rooms.get_pdu(id))
+                    .filter_map(|r| r.ok().flatten())
+                    .collect::<Vec<_>>()
+            } else {
+                current_state_ids
+                    .difference(&since_state_ids)
+                    .filter(|id| {
+                        !timeline_pdus
+                            .iter()
+                            .any(|(_, timeline_pdu)| timeline_pdu.event_id == **id)
+                    })
+                    .map(|id| db.rooms.get_pdu(id))
+                    .filter_map(|r| r.ok().flatten())
+                    .collect()
+            };
+
+            let encrypted_room = db
+                .rooms
+                .state_get(current_shortstatehash, &EventType::RoomEncryption, "")?
+                .is_some();
+
+            let since_encryption =
+                db.rooms
+                    .state_get(since_shortstatehash, &EventType::RoomEncryption, "")?;
 
             // Calculations:
-            let new_encrypted_room =
+            let new_encrypted_room = encrypted_room && since_encryption.is_none();
-                encrypted_room && since_encryption.map_or(true, |encryption| encryption.is_none());
 
-            let send_member_count = since_state.as_ref().map_or(true, |since_state| {
+            let send_member_count = state_events
-                since_state.as_ref().map_or(true, |since_state| {
+                .iter()
-                    current_members.len()
+                .any(|event| event.kind == EventType::RoomMember);
-                        != since_state
-                            .iter()
-                            .filter(|(key, _)| key.0 == EventType::RoomMember)
-                            .count()
-                })
-            });
-
-            let since_sender_member = since_state.as_ref().map(|since_state| {
-                since_state.as_ref().and_then(|state| {
-                    state
-                        .get(&(EventType::RoomMember, sender_user.as_str().to_owned()))
-                        .and_then(|pdu| {
-                            serde_json::from_value::<
-                                Raw<ruma::events::room::member::MemberEventContent>,
-                            >(pdu.content.clone())
-                            .expect("Raw::from_value always works")
-                            .deserialize()
-                            .map_err(|_| Error::bad_database("Invalid PDU in database."))
-                            .ok()
-                        })
-                })
-            });
 
             if encrypted_room {
-                for (user_id, current_member) in current_members {
+                for (user_id, current_member) in db
+                    .rooms
+                    .room_members(&room_id)
+                    .filter_map(|r| r.ok())
+                    .filter_map(|user_id| {
+                        db.rooms
+                            .state_get(
+                                current_shortstatehash,
+                                &EventType::RoomMember,
+                                user_id.as_str(),
+                            )
+                            .ok()
+                            .flatten()
+                            .map(|current_member| (user_id, current_member))
+                    })
+                {
                     let current_membership = serde_json::from_value::<
                         Raw<ruma::events::room::member::MemberEventContent>,
                     >(current_member.content.clone())
@@ -313,31 +429,23 @@ async fn sync_helper(
                     .map_err(|_| Error::bad_database("Invalid PDU in database."))?
                     .membership;
 
-                    let since_membership =
+                    let since_membership = db
-                        since_state
+                        .rooms
-                            .as_ref()
+                        .state_get(
-                            .map_or(MembershipState::Leave, |since_state| {
+                            since_shortstatehash,
-                                since_state
+                            &EventType::RoomMember,
-                                    .as_ref()
+                            user_id.as_str(),
-                                    .and_then(|since_state| {
+                        )?
-                                        since_state
+                        .and_then(|since_member| {
-                                            .get(&(EventType::RoomMember, user_id.clone()))
+                            serde_json::from_value::<
-                                            .and_then(|since_member| {
+                                Raw<ruma::events::room::member::MemberEventContent>,
-                                                serde_json::from_value::<
+                            >(since_member.content.clone())
-                                                Raw<ruma::events::room::member::MemberEventContent>,
+                            .expect("Raw::from_value always works")
-                                            >(
+                            .deserialize()
-                                                since_member.content.clone()
+                            .map_err(|_| Error::bad_database("Invalid PDU in database."))
-                                            )
+                            .ok()
-                                            .expect("Raw::from_value always works")
+                        })
-                                            .deserialize()
+                        .map_or(MembershipState::Leave, |member| member.membership);
-                                            .map_err(|_| {
-                                                Error::bad_database("Invalid PDU in database.")
-                                            })
-                                            .ok()
-                                            })
-                                    })
-                                    .map_or(MembershipState::Leave, |member| member.membership)
-                            });
 
                     let user_id = UserId::try_from(user_id.clone())
                         .map_err(|_| Error::bad_database("Invalid UserId in member PDU."))?;
@@ -359,10 +467,6 @@ async fn sync_helper(
                 }
             }
 
-            let joined_since_last_sync = since_sender_member.map_or(true, |member| {
-                member.map_or(true, |member| member.membership != MembershipState::Join)
-            });
-
             if joined_since_last_sync && encrypted_room || new_encrypted_room {
                 // If the user is in a new encrypted room, give them all joined users
                 device_list_updates.extend(
@@ -382,100 +486,11 @@ async fn sync_helper(
             }
 
             let (joined_member_count, invited_member_count, heroes) = if send_member_count {
-                let joined_member_count = db.rooms.room_members(&room_id).count();
+                calculate_counts()
-                let invited_member_count = db.rooms.room_members_invited(&room_id).count();
-
-                // Recalculate heroes (first 5 members)
-                let mut heroes = Vec::new();
-
-                if joined_member_count + invited_member_count <= 5 {
-                    // Go through all PDUs and for each member event, check if the user is still joined or
-                    // invited until we have 5 or we reach the end
-
-                    for hero in db
-                        .rooms
-                        .all_pdus(&sender_user, &room_id)
-                        .filter_map(|pdu| pdu.ok()) // Ignore all broken pdus
-                        .filter(|(_, pdu)| pdu.kind == EventType::RoomMember)
-                        .map(|(_, pdu)| {
-                            let content = serde_json::from_value::<
-                                ruma::events::room::member::MemberEventContent,
-                            >(pdu.content.clone())
-                            .map_err(|_| {
-                                Error::bad_database("Invalid member event in database.")
-                            })?;
-
-                            if let Some(state_key) = &pdu.state_key {
-                                let user_id =
-                                    UserId::try_from(state_key.clone()).map_err(|_| {
-                                        Error::bad_database("Invalid UserId in member PDU.")
-                                    })?;
-
-                                // The membership was and still is invite or join
-                                if matches!(
-                                    content.membership,
-                                    MembershipState::Join | MembershipState::Invite
-                                ) && (db.rooms.is_joined(&user_id, &room_id)?
-                                    || db.rooms.is_invited(&user_id, &room_id)?)
-                                {
-                                    Ok::<_, Error>(Some(state_key.clone()))
-                                } else {
-                                    Ok(None)
-                                }
-                            } else {
-                                Ok(None)
-                            }
-                        })
-                        // Filter out buggy users
-                        .filter_map(|u| u.ok())
-                        // Filter for possible heroes
-                        .flatten()
-                    {
-                        if heroes.contains(&hero) || hero == sender_user.as_str() {
-                            continue;
-                        }
-
-                        heroes.push(hero);
-                    }
-                }
-
-                (
-                    Some(joined_member_count),
-                    Some(invited_member_count),
-                    heroes,
-                )
             } else {
                 (None, None, Vec::new())
             };
 
-            let state_events = if joined_since_last_sync {
-                current_state
-                    .iter()
-                    .map(|(_, pdu)| pdu.to_sync_state_event())
-                    .collect()
-            } else {
-                match since_state {
-                    None => Vec::new(),
-                    Some(Some(since_state)) => current_state
-                        .iter()
-                        .filter(|(key, value)| {
-                            since_state.get(key).map(|e| &e.event_id) != Some(&value.event_id)
-                        })
-                        .filter(|(_, value)| {
-                            !timeline_pdus.iter().any(|(_, timeline_pdu)| {
-                                timeline_pdu.kind == value.kind
-                                    && timeline_pdu.state_key == value.state_key
-                            })
-                        })
-                        .map(|(_, pdu)| pdu.to_sync_state_event())
-                        .collect(),
-                    Some(None) => current_state
-                        .iter()
-                        .map(|(_, pdu)| pdu.to_sync_state_event())
-                        .collect(),
-                }
-            };
-
             (
                 heroes,
                 joined_member_count,
@@ -483,8 +498,6 @@ async fn sync_helper(
                 joined_since_last_sync,
                 state_events,
             )
-        } else {
-            (Vec::new(), None, None, false, Vec::new())
         };
 
         // Look for device list updates in this room
@@ -575,7 +588,10 @@ async fn sync_helper(
                 events: room_events,
             },
             state: sync_events::State {
-                events: state_events,
+                events: state_events
+                    .iter()
+                    .map(|pdu| pdu.to_sync_state_event())
+                    .collect(),
             },
             ephemeral: sync_events::Ephemeral { events: edus },
         };

src/client_server/to_device.rs

@@ -22,7 +22,9 @@ pub async fn send_event_to_device_route(
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
     let sender_device = body.sender_device.as_deref();
 
+    // TODO: uncomment when https://github.com/vector-im/element-android/issues/3589 is solved
     // Check if this is a new transaction id
+    /*
     if db
         .transaction_ids
         .existing_txnid(sender_user, sender_device, &body.txn_id)?
@@ -30,6 +32,7 @@ pub async fn send_event_to_device_route(
     {
         return Ok(send_event_to_device::Response.into());
     }
+    */
 
     for (target_user_id, map) in &body.messages {
         for (target_device_id_maybe, event) in map {

src/database/abstraction.rs

@@ -65,7 +65,7 @@ impl DatabaseEngine for SledEngine {
             sled::Config::default()
                 .path(&config.database_path)
                 .cache_capacity(config.cache_capacity as u64)
-                .use_compression(false)
+                .use_compression(true)
                 .open()?,
         )))
     }
@@ -111,7 +111,7 @@ impl Tree for SledEngineTree {
         backwards: bool,
     ) -> Box<dyn Iterator<Item = (Box<[u8]>, Box<[u8]>)>> {
         let iter = if backwards {
-            self.0.range(..from)
+            self.0.range(..=from)
         } else {
             self.0.range(from..)
         };

src/database/globals.rs

@@ -1,12 +1,9 @@
 use crate::{database::Config, utils, ConduitResult, Error, Result};
 use log::{error, info};
-use ruma::{
+use ruma::{DeviceId, EventId, MilliSecondsSinceUnixEpoch, RoomId, ServerName, ServerSigningKeyId, UserId, api::{
-    api::{
         client::r0::sync::sync_events,
         federation::discovery::{ServerSigningKeys, VerifyKey},
-    },
+    }};
-    DeviceId, EventId, MilliSecondsSinceUnixEpoch, ServerName, ServerSigningKeyId, UserId,
-};
 use rustls::{ServerCertVerifier, WebPKIVerifier};
 use std::{
     collections::{BTreeMap, HashMap},
@@ -15,7 +12,7 @@ use std::{
     sync::{Arc, RwLock},
     time::{Duration, Instant},
 };
-use tokio::sync::Semaphore;
+use tokio::sync::{Mutex, Semaphore};
 use trust_dns_resolver::TokioAsyncResolver;
 
 use super::abstraction::Tree;
@@ -38,6 +35,7 @@ pub struct Globals {
     pub bad_event_ratelimiter: Arc<RwLock<BTreeMap<EventId, RateLimitState>>>,
     pub bad_signature_ratelimiter: Arc<RwLock<BTreeMap<Vec<String>, RateLimitState>>>,
     pub servername_ratelimiter: Arc<RwLock<BTreeMap<Box<ServerName>, Arc<Semaphore>>>>,
+    pub roomid_mutex: RwLock<BTreeMap<RoomId, Arc<Mutex<()>>>>,
     pub sync_receivers: RwLock<
         BTreeMap<
             (UserId, Box<DeviceId>),
@@ -165,6 +163,7 @@ impl Globals {
             bad_event_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())),
             bad_signature_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())),
             servername_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())),
+            roomid_mutex: RwLock::new(BTreeMap::new()),
             sync_receivers: RwLock::new(BTreeMap::new()),
         };
 

src/database/rooms.rs

@@ -20,12 +20,7 @@ use ruma::{
     state_res::{self, Event, RoomVersion, StateMap},
     uint, EventId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
 };
-use std::{
+use std::{collections::{BTreeMap, BTreeSet, HashMap, HashSet}, convert::{TryFrom, TryInto}, mem, sync::{Arc, RwLock}};
-    collections::{BTreeMap, HashMap, HashSet},
-    convert::{TryFrom, TryInto},
-    mem,
-    sync::{Arc, RwLock},
-};
 
 use super::{abstraction::Tree, admin::AdminCommand, pusher};
 
@@ -89,7 +84,7 @@ pub struct Rooms {
 impl Rooms {
     /// Builds a StateMap by iterating over all keys that start
     /// with state_hash, this gives the full state for the given state_hash.
-    pub fn state_full_ids(&self, shortstatehash: u64) -> Result<Vec<EventId>> {
+    pub fn state_full_ids(&self, shortstatehash: u64) -> Result<BTreeSet<EventId>> {
         Ok(self
             .stateid_shorteventid
             .scan_prefix(shortstatehash.to_be_bytes().to_vec())
@@ -668,11 +663,10 @@ impl Rooms {
         &self,
         pdu: &PduEvent,
         mut pdu_json: CanonicalJsonObject,
-        count: u64,
-        pdu_id: &[u8],
         leaves: &[EventId],
         db: &Database,
-    ) -> Result<()> {
+    ) -> Result<Vec<u8>> {
+        // returns pdu id
         // Make unsigned fields correct. This is not properly documented in the spec, but state
         // events need to have previous content in the unsigned field, so clients can easily
         // interpret things like membership changes
@@ -710,20 +704,30 @@ impl Rooms {
 
         self.replace_pdu_leaves(&pdu.room_id, leaves)?;
 
+        let count1 = db.globals.next_count()?;
         // Mark as read first so the sending client doesn't get a notification even if appending
         // fails
         self.edus
-            .private_read_set(&pdu.room_id, &pdu.sender, count, &db.globals)?;
+            .private_read_set(&pdu.room_id, &pdu.sender, count1, &db.globals)?;
         self.reset_notification_counts(&pdu.sender, &pdu.room_id)?;
 
+        let count2 = db.globals.next_count()?;
+        let mut pdu_id = pdu.room_id.as_bytes().to_vec();
+        pdu_id.push(0xff);
+        pdu_id.extend_from_slice(&count2.to_be_bytes());
+
+        // There's a brief moment of time here where the count is updated but the pdu does not
+        // exist. This could theoretically lead to dropped pdus, but it's extremely rare
+
         self.pduid_pdu.insert(
-            pdu_id,
+            &pdu_id,
             &serde_json::to_vec(&pdu_json).expect("CanonicalJsonObject is always a valid"),
         )?;
 
         // This also replaces the eventid of any outliers with the correct
         // pduid, removing the place holder.
-        self.eventid_pduid.insert(pdu.event_id.as_bytes(), pdu_id)?;
+        self.eventid_pduid
+            .insert(pdu.event_id.as_bytes(), &pdu_id)?;
 
         // See if the event matches any known pushers
         for user in db
@@ -911,7 +915,7 @@ impl Rooms {
             _ => {}
         }
 
-        Ok(())
+        Ok(pdu_id)
     }
 
     pub fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
@@ -1208,6 +1212,7 @@ impl Rooms {
             state_key,
             redacts,
         } = pdu_builder;
+
         // TODO: Make sure this isn't called twice in parallel
         let prev_events = self
             .get_pdu_leaves(&room_id)?
@@ -1356,11 +1361,9 @@ impl Rooms {
         // pdu without it's state. This is okay because append_pdu can't fail.
         let statehashid = self.append_to_state(&pdu, &db.globals)?;
 
-        self.append_pdu(
+        let pdu_id = self.append_pdu(
             &pdu,
             pdu_json,
-            count,
-            &pdu_id,
             // Since this PDU references all pdu_leaves we can update the leaves
             // of the room
             &[pdu.event_id.clone()],
@@ -1497,7 +1500,7 @@ impl Rooms {
         prefix.push(0xff);
 
         let mut current = prefix.clone();
-        current.extend_from_slice(&until.to_be_bytes());
+        current.extend_from_slice(&(until.saturating_sub(1)).to_be_bytes()); // -1 because we don't want event at `until`
 
         let current: &[u8] = &current;
 

src/database/users.rs

@@ -726,10 +726,9 @@ impl Users {
         json.insert("sender".to_owned(), sender.to_string().into());
         json.insert("content".to_owned(), content);
 
-        self.todeviceid_events.insert(
+        let value = serde_json::to_vec(&json).expect("Map::to_vec always works");
-            &key,
+
-            &serde_json::to_vec(&json).expect("Map::to_vec always works"),
+        self.todeviceid_events.insert(&key, &value)?;
-        )?;
 
         Ok(())
     }
@@ -774,7 +773,7 @@ impl Users {
 
         for (key, _) in self
             .todeviceid_events
-            .iter_from(&last, true)
+            .iter_from(&last, true) // this includes last
             .take_while(move |(k, _)| k.starts_with(&prefix))
             .map(|(key, _)| {
                 Ok::<_, Error>((

src/ruma_wrapper.rs

@@ -318,6 +318,7 @@ where
             }),
             Err(e) => {
                 warn!("{:?}", e);
+                // Bad Json
                 Failure((Status::raw(583), ()))
             }
         }

src/server_server.rs

@@ -45,7 +45,7 @@ use ruma::{
     receipt::ReceiptType,
     serde::Raw,
     signatures::{CanonicalJsonObject, CanonicalJsonValue},
-    state_res::{self, Event, EventMap, RoomVersion, StateMap},
+    state_res::{self, Event, RoomVersion, StateMap},
     to_device::DeviceIdOrAllDevices,
     uint, EventId, MilliSecondsSinceUnixEpoch, RoomId, RoomVersionId, ServerName,
     ServerSigningKeyId, UserId,
@@ -612,7 +612,7 @@ pub async fn send_transaction_message_route(
     // TODO: This could potentially also be some sort of trie (suffix tree) like structure so
     // that once an auth event is known it would know (using indexes maybe) all of the auth
     // events that it references.
-    let mut auth_cache = EventMap::new();
+    // let mut auth_cache = EventMap::new();
 
     for pdu in &body.pdus {
         // We do not add the event_id field to the pdu here because of signature and hashes checks
@@ -624,21 +624,44 @@ pub async fn send_transaction_message_route(
             }
         };
 
+        // 0. Check the server is in the room
+        let room_id = match value
+            .get("room_id")
+            .and_then(|id| RoomId::try_from(id.as_str()?).ok())
+        {
+            Some(id) => id,
+            None => {
+                // Event is invalid
+                resolved_map.insert(event_id, Err("Event needs a valid RoomId.".to_string()));
+                continue;
+            }
+        };
+
+        let mutex = Arc::clone(
+            db.globals
+                .roomid_mutex
+                .write()
+                .unwrap()
+                .entry(room_id.clone())
+                .or_default(),
+        );
+        let mutex_lock = mutex.lock().await;
         let start_time = Instant::now();
         resolved_map.insert(
             event_id.clone(),
             handle_incoming_pdu(
                 &body.origin,
                 &event_id,
+                &room_id,
                 value,
                 true,
                 &db,
                 &pub_key_map,
-                &mut auth_cache,
             )
             .await
             .map(|_| ()),
         );
+        drop(mutex_lock);
 
         let elapsed = start_time.elapsed();
         if elapsed > Duration::from_secs(1) {
@@ -790,8 +813,8 @@ pub async fn send_transaction_message_route(
 type AsyncRecursiveResult<'a, T, E> = Pin<Box<dyn Future<Output = StdResult<T, E>> + 'a + Send>>;
 
 /// When receiving an event one needs to:
-/// 0. Skip the PDU if we already know about it
+/// 0. Check the server is in the room
-/// 1. Check the server is in the room
+/// 1. Skip the PDU if we already know about it
 /// 2. Check signatures, otherwise drop
 /// 3. Check content hash, redact if doesn't match
 /// 4. Fetch any missing auth events doing all checks listed here starting at 1. These are not
@@ -816,32 +839,14 @@ type AsyncRecursiveResult<'a, T, E> = Pin<Box<dyn Future<Output = StdResult<T, E
 pub fn handle_incoming_pdu<'a>(
     origin: &'a ServerName,
     event_id: &'a EventId,
+    room_id: &'a RoomId,
     value: BTreeMap<String, CanonicalJsonValue>,
     is_timeline_event: bool,
     db: &'a Database,
     pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>,
-    auth_cache: &'a mut EventMap<Arc<PduEvent>>,
 ) -> AsyncRecursiveResult<'a, Option<Vec<u8>>, String> {
     Box::pin(async move {
         // TODO: For RoomVersion6 we must check that Raw<..> is canonical do we anywhere?: https://matrix.org/docs/spec/rooms/v6#canonical-json
-
-        // 0. Skip the PDU if we already have it as a timeline event
-        if let Ok(Some(pdu_id)) = db.rooms.get_pdu_id(&event_id) {
-            return Ok(Some(pdu_id.to_vec()));
-        }
-
-        // 1. Check the server is in the room
-        let room_id = match value
-            .get("room_id")
-            .and_then(|id| RoomId::try_from(id.as_str()?).ok())
-        {
-            Some(id) => id,
-            None => {
-                // Event is invalid
-                return Err("Event needs a valid RoomId.".to_string());
-            }
-        };
-
         match db.rooms.exists(&room_id) {
             Ok(true) => {}
             _ => {
@@ -849,6 +854,11 @@ pub fn handle_incoming_pdu<'a>(
             }
         }
 
+        // 1. Skip the PDU if we already have it as a timeline event
+        if let Ok(Some(pdu_id)) = db.rooms.get_pdu_id(&event_id) {
+            return Ok(Some(pdu_id.to_vec()));
+        }
+
         // We go through all the signatures we see on the value and fetch the corresponding signing
         // keys
         fetch_required_signing_keys(&value, &pub_key_map, db)
@@ -908,15 +918,9 @@ pub fn handle_incoming_pdu<'a>(
         // 5. Reject "due to auth events" if can't get all the auth events or some of the auth events are also rejected "due to auth events"
         // EDIT: Step 5 is not applied anymore because it failed too often
         debug!("Fetching auth events for {}", incoming_pdu.event_id);
-        fetch_and_handle_events(
+        fetch_and_handle_events(db, origin, &incoming_pdu.auth_events, &room_id, pub_key_map)
-            db,
+            .await
-            origin,
+            .map_err(|e| e.to_string())?;
-            &incoming_pdu.auth_events,
-            pub_key_map,
-            auth_cache,
-        )
-        .await
-        .map_err(|e| e.to_string())?;
 
         // 6. Reject "due to auth events" if the event doesn't pass auth based on the auth events
         debug!(
@@ -927,9 +931,13 @@ pub fn handle_incoming_pdu<'a>(
         // Build map of auth events
         let mut auth_events = BTreeMap::new();
         for id in &incoming_pdu.auth_events {
-            let auth_event = auth_cache.get(id).ok_or_else(|| {
+            let auth_event = db
-                "Auth event not found, event failed recursive auth checks.".to_string()
+                .rooms
-            })?;
+                .get_pdu(id)
+                .map_err(|e| e.to_string())?
+                .ok_or_else(|| {
+                    "Auth event not found, event failed recursive auth checks.".to_string()
+                })?;
 
             match auth_events.entry((
                 auth_event.kind.clone(),
@@ -963,9 +971,9 @@ pub fn handle_incoming_pdu<'a>(
         let previous_create = if incoming_pdu.auth_events.len() == 1
             && incoming_pdu.prev_events == incoming_pdu.auth_events
         {
-            auth_cache
+            db.rooms
-                .get(&incoming_pdu.auth_events[0])
+                .get_pdu(&incoming_pdu.auth_events[0])
-                .cloned()
+                .map_err(|e| e.to_string())?
                 .filter(|maybe_create| **maybe_create == *create_event)
         } else {
             None
@@ -1008,17 +1016,16 @@ pub fn handle_incoming_pdu<'a>(
 
         debug!("Requesting state at event.");
         let mut state_at_incoming_event = None;
-        let mut incoming_auth_events = Vec::new();
 
         if incoming_pdu.prev_events.len() == 1 {
             let prev_event = &incoming_pdu.prev_events[0];
-            let state_vec = db
+            let state = db
                 .rooms
                 .pdu_shortstatehash(prev_event)
                 .map_err(|_| "Failed talking to db".to_owned())?
                 .map(|shortstatehash| db.rooms.state_full_ids(shortstatehash).ok())
                 .flatten();
-            if let Some(mut state_vec) = state_vec {
+            if let Some(mut state) = state {
                 if db
                     .rooms
                     .get_pdu(prev_event)
@@ -1028,25 +1035,31 @@ pub fn handle_incoming_pdu<'a>(
                     .state_key
                     .is_some()
                 {
-                    state_vec.push(prev_event.clone());
+                    state.insert(prev_event.clone());
                 }
                 state_at_incoming_event = Some(
-                    fetch_and_handle_events(db, origin, &state_vec, pub_key_map, auth_cache)
+                    fetch_and_handle_events(
-                        .await
+                        db,
-                        .map_err(|_| "Failed to fetch state events locally".to_owned())?
+                        origin,
-                        .into_iter()
+                        &state.into_iter().collect::<Vec<_>>(),
-                        .map(|pdu| {
+                        &room_id,
+                        pub_key_map,
+                    )
+                    .await
+                    .map_err(|_| "Failed to fetch state events locally".to_owned())?
+                    .into_iter()
+                    .map(|pdu| {
+                        (
                             (
-                                (
+                                pdu.kind.clone(),
-                                    pdu.kind.clone(),
+                                pdu.state_key
-                                    pdu.state_key
+                                    .clone()
-                                        .clone()
+                                    .expect("events from state_full_ids are state events"),
-                                        .expect("events from state_full_ids are state events"),
+                            ),
-                                ),
+                            pdu,
-                                pdu,
+                        )
-                            )
+                    })
-                        })
+                    .collect(),
-                        .collect(),
                 );
             }
             // TODO: set incoming_auth_events?
@@ -1073,8 +1086,8 @@ pub fn handle_incoming_pdu<'a>(
                         &db,
                         origin,
                         &res.pdu_ids,
+                        &room_id,
                         pub_key_map,
-                        auth_cache,
                     )
                     .await
                     {
@@ -1106,12 +1119,12 @@ pub fn handle_incoming_pdu<'a>(
                     }
 
                     debug!("Fetching auth chain events at event.");
-                    incoming_auth_events = match fetch_and_handle_events(
+                    match fetch_and_handle_events(
                         &db,
                         origin,
                         &res.auth_chain_ids,
+                        &room_id,
                         pub_key_map,
-                        auth_cache,
                     )
                     .await
                     {
@@ -1247,8 +1260,8 @@ pub fn handle_incoming_pdu<'a>(
                         &db,
                         origin,
                         &[auth_id.clone()],
+                        &room_id,
                         pub_key_map,
-                        auth_cache,
                     )
                     .await
                     {
@@ -1259,31 +1272,9 @@ pub fn handle_incoming_pdu<'a>(
                         }
                     }
                 }
-                auth_cache.extend(
-                    map.iter()
-                        .map(|pdu| (pdu.1.event_id.clone(), pdu.1.clone())),
-                );
                 auth_events.push(state_auth);
             }
 
-            // Add everything we will need to event_map
-            auth_cache.extend(
-                auth_events
-                    .iter()
-                    .map(|pdus| pdus.iter().map(|pdu| (pdu.event_id.clone(), pdu.clone())))
-                    .flatten(),
-            );
-            auth_cache.extend(
-                incoming_auth_events
-                    .into_iter()
-                    .map(|pdu| (pdu.event_id().clone(), pdu)),
-            );
-            auth_cache.extend(
-                state_after
-                    .into_iter()
-                    .map(|(_, pdu)| (pdu.event_id().clone(), pdu)),
-            );
-
             match state_res::StateResolution::resolve(
                 &room_id,
                 room_version_id,
@@ -1299,7 +1290,13 @@ pub fn handle_incoming_pdu<'a>(
                     .into_iter()
                     .map(|pdus| pdus.into_iter().map(|pdu| pdu.event_id().clone()).collect())
                     .collect(),
-                auth_cache,
+                &|id| {
+                    let res = db.rooms.get_pdu(id);
+                    if let Err(e) = &res {
+                        error!("LOOK AT ME Failed to fetch event: {}", e);
+                    }
+                    res.ok().flatten()
+                },
             ) {
                 Ok(new_state) => new_state,
                 Err(_) => {
@@ -1372,8 +1369,8 @@ pub(crate) fn fetch_and_handle_events<'a>(
     db: &'a Database,
     origin: &'a ServerName,
     events: &'a [EventId],
+    room_id: &'a RoomId,
     pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>,
-    auth_cache: &'a mut EventMap<Arc<PduEvent>>,
 ) -> AsyncRecursiveResult<'a, Vec<Arc<PduEvent>>, Error> {
     Box::pin(async move {
         let back_off = |id| match db.globals.bad_event_ratelimiter.write().unwrap().entry(id) {
@@ -1397,84 +1394,74 @@ pub(crate) fn fetch_and_handle_events<'a>(
                     continue;
                 }
             }
-            // a. Look at auth cache
+
-            let pdu = match auth_cache.get(id) {
+            // a. Look in the main timeline (pduid_pdu tree)
+            // b. Look at outlier pdu tree
+            // (get_pdu checks both)
+            let pdu = match db.rooms.get_pdu(&id)? {
                 Some(pdu) => {
-                    // We already have the auth chain for events in cache
+                    trace!("Found {} in db", id);
-                    pdu.clone()
+                    pdu
                 }
-                // b. Look in the main timeline (pduid_pdu tree)
+                None => {
-                // c. Look at outlier pdu tree
+                    // c. Ask origin server over federation
-                // (get_pdu checks both)
+                    debug!("Fetching {} over federation.", id);
-                None => match db.rooms.get_pdu(&id)? {
+                    match db
-                    Some(pdu) => {
+                        .sending
-                        trace!("Found {} in db", id);
+                        .send_federation_request(
-                        // We need to fetch the auth chain
+                            &db.globals,
-                        let _ = fetch_and_handle_events(
-                            db,
                             origin,
-                            &pdu.auth_events,
+                            get_event::v1::Request { event_id: &id },
-                            pub_key_map,
-                            auth_cache,
                         )
-                        .await?;
+                        .await
-                        pdu
+                    {
-                    }
+                        Ok(res) => {
-                    None => {
+                            debug!("Got {} over federation", id);
-                        // d. Ask origin server over federation
+                            let (event_id, mut value) =
-                        debug!("Fetching {} over federation.", id);
+                                crate::pdu::gen_event_id_canonical_json(&res.pdu)?;
-                        match db
+                            // This will also fetch the auth chain
-                            .sending
+                            match handle_incoming_pdu(
-                            .send_federation_request(
-                                &db.globals,
                                 origin,
-                                get_event::v1::Request { event_id: &id },
+                                &event_id,
+                                &room_id,
+                                value.clone(),
+                                false,
+                                db,
+                                pub_key_map,
                             )
                             .await
-                        {
+                            {
-                            Ok(res) => {
+                                Ok(_) => {
-                                debug!("Got {} over federation", id);
+                                    value.insert(
-                                let (event_id, mut value) =
+                                        "event_id".to_owned(),
-                                    crate::pdu::gen_event_id_canonical_json(&res.pdu)?;
+                                        CanonicalJsonValue::String(event_id.into()),
-                                // This will also fetch the auth chain
+                                    );
-                                match handle_incoming_pdu(
-                                    origin,
-                                    &event_id,
-                                    value.clone(),
-                                    false,
-                                    db,
-                                    pub_key_map,
-                                    auth_cache,
-                                )
-                                .await
-                                {
-                                    Ok(_) => {
-                                        value.insert(
-                                            "event_id".to_owned(),
-                                            CanonicalJsonValue::String(event_id.into()),
-                                        );
 
-                                        Arc::new(serde_json::from_value(
+                                    Arc::new(
-                                            serde_json::to_value(value).expect("canonicaljsonobject is valid value"),
+                                        serde_json::from_value(
-                                        ).expect("This is possible because handle_incoming_pdu worked"))
+                                            serde_json::to_value(value)
-                                    }
+                                                .expect("canonicaljsonobject is valid value"),
-                                    Err(e) => {
+                                        )
-                                        warn!("Authentication of event {} failed: {:?}", id, e);
+                                        .expect(
-                                        back_off(id.clone());
+                                            "This is possible because handle_incoming_pdu worked",
-                                        continue;
+                                        ),
-                                    }
+                                    )
+                                }
+                                Err(e) => {
+                                    warn!("Authentication of event {} failed: {:?}", id, e);
+                                    back_off(id.clone());
+                                    continue;
                                 }
                             }
-                            Err(_) => {
+                        }
-                                warn!("Failed to fetch event: {}", id);
+                        Err(_) => {
-                                back_off(id.clone());
+                            warn!("Failed to fetch event: {}", id);
-                                continue;
+                            back_off(id.clone());
-                            }
+                            continue;
                         }
                     }
-                },
+                }
             };
-            auth_cache.entry(id.clone()).or_insert_with(|| pdu.clone());
             pdus.push(pdu);
         }
         Ok(pdus)
@@ -1650,21 +1637,14 @@ pub(crate) fn append_incoming_pdu(
     new_room_leaves: HashSet<EventId>,
     state: &StateMap<Arc<PduEvent>>,
 ) -> Result<Vec<u8>> {
-    let count = db.globals.next_count()?;
-    let mut pdu_id = pdu.room_id.as_bytes().to_vec();
-    pdu_id.push(0xff);
-    pdu_id.extend_from_slice(&count.to_be_bytes());
-
     // We append to state before appending the pdu, so we don't have a moment in time with the
     // pdu without it's state. This is okay because append_pdu can't fail.
     db.rooms
         .set_event_state(&pdu.event_id, state, &db.globals)?;
 
-    db.rooms.append_pdu(
+    let pdu_id = db.rooms.append_pdu(
         pdu,
         pdu_json,
-        count,
-        &pdu_id,
         &new_room_leaves.into_iter().collect::<Vec<_>>(),
         &db,
     )?;
@@ -1722,7 +1702,8 @@ pub(crate) fn append_incoming_pdu(
                     .filter_map(|r| r.ok())
                     .any(|member| users.iter().any(|regex| regex.is_match(member.as_str())))
             {
-                db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
+                db.sending
+                    .send_pdu_appservice(&appservice.0, &pdu_id)?;
             }
         }
     }
@@ -1932,7 +1913,11 @@ pub fn get_room_state_ids_route(
             "Pdu state not found.",
         ))?;
 
-    let pdu_ids = db.rooms.state_full_ids(shortstatehash)?;
+    let pdu_ids = db
+        .rooms
+        .state_full_ids(shortstatehash)?
+        .into_iter()
+        .collect();
 
     let mut auth_chain_ids = BTreeSet::<EventId>::new();
     let mut todo = BTreeSet::new();
@@ -2155,7 +2140,7 @@ pub async fn create_join_event_route(
             ))?;
 
     let pub_key_map = RwLock::new(BTreeMap::new());
-    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
     let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&body.pdu) {
@@ -2178,14 +2163,23 @@ pub async fn create_join_event_route(
     )
     .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?;
 
+    let mutex = Arc::clone(
+        db.globals
+            .roomid_mutex
+            .write()
+            .unwrap()
+            .entry(body.room_id.clone())
+            .or_default(),
+    );
+    let mutex_lock = mutex.lock().await;
     let pdu_id = handle_incoming_pdu(
         &origin,
         &event_id,
+        &body.room_id,
         value,
         true,
         &db,
         &pub_key_map,
-        &mut auth_cache,
     )
     .await
     .map_err(|_| {
@@ -2198,6 +2192,7 @@ pub async fn create_join_event_route(
         ErrorKind::InvalidParam,
         "Could not accept incoming PDU as timeline event.",
     ))?;
+    drop(mutex_lock);
 
     let state_ids = db.rooms.state_full_ids(shortstatehash)?;