Also calculate sum of cache sizes

Signed-off-by: Jonas Zohren <git-pbkyr@jzohren.de>
Add test to measure memory usage of caches
2021-08-25 00:51:46 +02:00 · 2021-08-25 00:43:01 +02:00 · 2021-08-24 20:12:10 +00:00 · 2021-08-24 21:55:18 +02:00 · 2021-08-24 18:16:56 +00:00 · 2021-08-24 17:54:17 +00:00
20 changed files with 2387 additions and 1114 deletions
@@ -27,6 +27,9 @@ variables:
      - cargohome
      - target/
    key: "build_cache-$TARGET-release"
+  variables:
+    CARGO_PROFILE_RELEASE_LTO=true
+    CARGO_PROFILE_RELEASE_CODEGEN_UNITS=1
  before_script:
    - 'echo "Building for target $TARGET"'
    - 'mkdir -p cargohome && CARGOHOME="cargohome"'
@@ -248,7 +248,7 @@ dependencies = [
 "jsonwebtoken",
 "lru-cache",
 "num_cpus",
- "opentelemetry",
+ "opentelemetry 0.16.0",
 "opentelemetry-jaeger",
 "parking_lot",
 "pretty_env_logger",
@@ -1466,16 +1466,46 @@ dependencies = [
 ]

 [[package]]
-name = "opentelemetry-jaeger"
-version = "0.14.0"
+name = "opentelemetry"
+version = "0.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09a9fc8192722e7daa0c56e59e2336b797122fb8598383dcb11c8852733b435c"
+checksum = "e1cf9b1c4e9a6c4de793c632496fa490bdc0e1eea73f0c91394f7b6990935d22"
+dependencies = [
+ "async-trait",
+ "crossbeam-channel",
+ "futures",
+ "js-sys",
+ "lazy_static",
+ "percent-encoding",
+ "pin-project",
+ "rand 0.8.4",
+ "thiserror",
+ "tokio",
+ "tokio-stream",
+]
+
+[[package]]
+name = "opentelemetry-jaeger"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "db22f492873ea037bc267b35a0e8e4fb846340058cb7c864efe3d0bf23684593"
 dependencies = [
 "async-trait",
 "lazy_static",
- "opentelemetry",
+ "opentelemetry 0.16.0",
+ "opentelemetry-semantic-conventions",
 "thiserror",
 "thrift",
+ "tokio",
+]
+
+[[package]]
+name = "opentelemetry-semantic-conventions"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ffeac823339e8b0f27b961f4385057bf9f97f2863bc745bd015fd6091f2270e9"
+dependencies = [
+ "opentelemetry 0.16.0",
 ]

 [[package]]
@@ -2014,8 +2044,8 @@ dependencies = [

 [[package]]
 name = "ruma"
-version = "0.2.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.3.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "assign",
 "js_int",
@@ -2035,8 +2065,8 @@ dependencies = [

 [[package]]
 name = "ruma-api"
-version = "0.17.1"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.18.3"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "bytes",
 "http",
@@ -2051,8 +2081,8 @@ dependencies = [

 [[package]]
 name = "ruma-api-macros"
-version = "0.17.1"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.18.3"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "proc-macro-crate",
 "proc-macro2",
@@ -2062,8 +2092,8 @@ dependencies = [

 [[package]]
 name = "ruma-appservice-api"
-version = "0.3.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.4.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "ruma-api",
 "ruma-common",
@@ -2076,8 +2106,8 @@ dependencies = [

 [[package]]
 name = "ruma-client-api"
-version = "0.11.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.12.2"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "assign",
 "bytes",
@@ -2096,8 +2126,8 @@ dependencies = [

 [[package]]
 name = "ruma-common"
-version = "0.5.4"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.6.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "indexmap",
 "js_int",
@@ -2111,8 +2141,8 @@ dependencies = [

 [[package]]
 name = "ruma-events"
-version = "0.23.2"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.24.4"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "indoc",
 "js_int",
@@ -2127,8 +2157,8 @@ dependencies = [

 [[package]]
 name = "ruma-events-macros"
-version = "0.23.2"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.24.4"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "proc-macro-crate",
 "proc-macro2",
@@ -2138,8 +2168,8 @@ dependencies = [

 [[package]]
 name = "ruma-federation-api"
-version = "0.2.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.3.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "js_int",
 "ruma-api",
@@ -2153,8 +2183,8 @@ dependencies = [

 [[package]]
 name = "ruma-identifiers"
-version = "0.19.4"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.20.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "paste",
 "rand 0.8.4",
@@ -2167,8 +2197,8 @@ dependencies = [

 [[package]]
 name = "ruma-identifiers-macros"
-version = "0.19.4"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.20.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "quote",
 "ruma-identifiers-validation",
@@ -2177,13 +2207,16 @@ dependencies = [

 [[package]]
 name = "ruma-identifiers-validation"
-version = "0.4.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.5.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
+dependencies = [
+ "thiserror",
+]

 [[package]]
 name = "ruma-identity-service-api"
-version = "0.2.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.3.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "js_int",
 "ruma-api",
@@ -2195,8 +2228,8 @@ dependencies = [

 [[package]]
 name = "ruma-push-gateway-api"
-version = "0.2.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.3.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "js_int",
 "ruma-api",
@@ -2210,8 +2243,8 @@ dependencies = [

 [[package]]
 name = "ruma-serde"
-version = "0.4.1"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.5.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "bytes",
 "form_urlencoded",
@@ -2224,8 +2257,8 @@ dependencies = [

 [[package]]
 name = "ruma-serde-macros"
-version = "0.4.1"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.5.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "proc-macro-crate",
 "proc-macro2",
@@ -2235,8 +2268,8 @@ dependencies = [

 [[package]]
 name = "ruma-signatures"
-version = "0.8.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.9.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "base64 0.13.0",
 "ed25519-dalek",
@@ -2252,8 +2285,8 @@ dependencies = [

 [[package]]
 name = "ruma-state-res"
-version = "0.2.0"
-source = "git+https://github.com/timokoesters/ruma?rev=a2d93500e1dbc87e7032a3c74f3b2479a7f84e93#a2d93500e1dbc87e7032a3c74f3b2479a7f84e93"
+version = "0.3.0"
+source = "git+https://github.com/timokoesters/ruma?rev=2215049b60a1c3358f5a52215adf1e7bb88619a1#2215049b60a1c3358f5a52215adf1e7bb88619a1"
 dependencies = [
 "itertools 0.10.1",
 "js_int",
@@ -3022,7 +3055,7 @@ version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c47440f2979c4cd3138922840eec122e3c0ba2148bc290f756bd7fd60fc97fff"
 dependencies = [
- "opentelemetry",
+ "opentelemetry 0.15.0",
 "tracing",
 "tracing-core",
 "tracing-log",
@@ -18,8 +18,8 @@ edition = "2018"
 rocket = { version = "0.5.0-rc.1", features = ["tls"] } # Used to handle requests

 # Used for matrix spec type definitions and helpers
-#ruma = { git = "https://github.com/ruma/ruma", rev = "eb19b0e08a901b87d11b3be0890ec788cc760492", 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/timokoesters/ruma", rev = "a2d93500e1dbc87e7032a3c74f3b2479a7f84e93", 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 = "f5ab038e22421ed338396ece977b6b2844772ced", 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/timokoesters/ruma", rev = "2215049b60a1c3358f5a52215adf1e7bb88619a1", 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
@@ -66,11 +66,11 @@ regex = "1.5.4"
 jsonwebtoken = "7.2.0"
 # Performance measurements
 tracing = { version = "0.1.26", features = ["release_max_level_warn"] }
-opentelemetry = "0.15.0"
 tracing-subscriber = "0.2.19"
 tracing-opentelemetry = "0.14.0"
 tracing-flame = "0.1.0"
-opentelemetry-jaeger = "0.14.0"
+opentelemetry = { version = "0.16.0", features = ["rt-tokio"] }
+opentelemetry-jaeger = { version = "0.15.0", features = ["rt-tokio"] }
 pretty_env_logger = "0.4.0"
 lru-cache = "0.1.2"
 rusqlite = { version = "0.25.3", optional = true, features = ["bundled"] }
@@ -124,16 +124,15 @@ lto = 'thin'
 incremental = true

 [profile.release]
-lto = true
+lto = 'thin'
 incremental = true
-codegen-units = 1
+
+codegen-units=32
 # If you want to make flamegraphs, enable debug info:
 # debug = true

 # For releases also try to max optimizations for dependencies:
 [profile.release.build-override]
 opt-level = 3
-codegen-units = 1
 [profile.release.package."*"]
 opt-level = 3
-codegen-units = 1
@@ -249,6 +249,8 @@ pub async fn register_route(

        let room_id = RoomId::new(db.globals.server_name());

+        db.rooms.get_or_create_shortroomid(&room_id, &db.globals)?;
+
        let mutex_state = Arc::clone(
            db.globals
                .roomid_mutex_state
@@ -290,6 +292,7 @@ pub async fn register_route(
                    is_direct: None,
                    third_party_invite: None,
                    blurhash: None,
+                    reason: None,
                })
                .expect("event is valid, we just created it"),
                unsigned: None,
@@ -455,6 +458,7 @@ pub async fn register_route(
                    is_direct: None,
                    third_party_invite: None,
                    blurhash: None,
+                    reason: None,
                })
                .expect("event is valid, we just created it"),
                unsigned: None,
@@ -476,6 +480,7 @@ pub async fn register_route(
                    is_direct: None,
                    third_party_invite: None,
                    blurhash: None,
+                    reason: None,
                })
                .expect("event is valid, we just created it"),
                unsigned: None,
@@ -681,6 +686,7 @@ pub async fn deactivate_route(
            is_direct: None,
            third_party_invite: None,
            blurhash: None,
+            reason: None,
        };

        let mutex_state = Arc::clone(
@@ -731,7 +737,7 @@ pub async fn deactivate_route(
 pub async fn third_party_route(
    body: Ruma<get_contacts::Request>,
 ) -> ConduitResult<get_contacts::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");

    Ok(get_contacts::Response::new(Vec::new()).into())
 }
@@ -44,7 +44,7 @@ pub async fn get_context_route(

    let events_before = db
        .rooms
-        .pdus_until(&sender_user, &body.room_id, base_token)
+        .pdus_until(&sender_user, &body.room_id, base_token)?
        .take(
            u32::try_from(body.limit).map_err(|_| {
                Error::BadRequest(ErrorKind::InvalidParam, "Limit value is invalid.")
@@ -66,7 +66,7 @@ pub async fn get_context_route(

    let events_after = db
        .rooms
-        .pdus_after(&sender_user, &body.room_id, base_token)
+        .pdus_after(&sender_user, &body.room_id, base_token)?
        .take(
            u32::try_from(body.limit).map_err(|_| {
                Error::BadRequest(ErrorKind::InvalidParam, "Limit value is invalid.")
@@ -262,6 +262,7 @@ pub async fn ban_user_route(
                is_direct: None,
                third_party_invite: None,
                blurhash: db.users.blurhash(&body.user_id)?,
+                reason: None,
            }),
            |event| {
                let mut event = serde_json::from_value::<Raw<member::MemberEventContent>>(
@@ -563,6 +564,7 @@ async fn join_room_by_id_helper(
                is_direct: None,
                third_party_invite: None,
                blurhash: db.users.blurhash(&sender_user)?,
+                reason: None,
            })
            .expect("event is valid, we just created it"),
        );
@@ -609,6 +611,8 @@ async fn join_room_by_id_helper(
            )
            .await?;

+        db.rooms.get_or_create_shortroomid(&room_id, &db.globals)?;
+
        let pdu = PduEvent::from_id_val(&event_id, join_event.clone())
            .map_err(|_| Error::BadServerResponse("Invalid join event PDU."))?;

@@ -636,23 +640,40 @@ async fn join_room_by_id_helper(

            db.rooms.add_pdu_outlier(&event_id, &value)?;
            if let Some(state_key) = &pdu.state_key {
-                state.insert((pdu.kind.clone(), state_key.clone()), pdu.event_id.clone());
+                let shortstatekey =
+                    db.rooms
+                        .get_or_create_shortstatekey(&pdu.kind, state_key, &db.globals)?;
+                state.insert(shortstatekey, pdu.event_id.clone());
            }
        }

-        state.insert(
-            (
-                pdu.kind.clone(),
-                pdu.state_key.clone().expect("join event has state key"),
-            ),
-            pdu.event_id.clone(),
-        );
+        let incoming_shortstatekey = db.rooms.get_or_create_shortstatekey(
+            &pdu.kind,
+            pdu.state_key
+                .as_ref()
+                .expect("Pdu is a membership state event"),
+            &db.globals,
+        )?;

-        if state.get(&(EventType::RoomCreate, "".to_owned())).is_none() {
+        state.insert(incoming_shortstatekey, pdu.event_id.clone());
+
+        let create_shortstatekey = db
+            .rooms
+            .get_shortstatekey(&EventType::RoomCreate, "")?
+            .expect("Room exists");
+
+        if state.get(&create_shortstatekey).is_none() {
            return Err(Error::BadServerResponse("State contained no create event."));
        }

-        db.rooms.force_state(room_id, state, &db)?;
+        db.rooms.force_state(
+            room_id,
+            state
+                .into_iter()
+                .map(|(k, id)| db.rooms.compress_state_event(k, &id, &db.globals))
+                .collect::<Result<HashSet<_>>>()?,
+            &db,
+        )?;

        for result in futures::future::join_all(
            send_join_response
@@ -693,6 +714,7 @@ async fn join_room_by_id_helper(
            is_direct: None,
            third_party_invite: None,
            blurhash: db.users.blurhash(&sender_user)?,
+            reason: None,
        };

        db.rooms.build_and_append_pdu(
@@ -817,7 +839,10 @@ pub async fn invite_helper<'a>(
                    serde_json::from_value::<Raw<CreateEventContent>>(create_event.content.clone())
                        .expect("Raw::from_value always works.")
                        .deserialize()
-                        .map_err(|_| Error::bad_database("Invalid PowerLevels event in db."))
+                        .map_err(|e| {
+                            warn!("Invalid create event: {}", e);
+                            Error::bad_database("Invalid create event in db.")
+                        })
                })
                .transpose()?;

@@ -844,6 +869,7 @@ pub async fn invite_helper<'a>(
                membership: MembershipState::Invite,
                third_party_invite: None,
                blurhash: None,
+                reason: None,
            })
            .expect("member event is valid value");

@@ -904,8 +930,8 @@ pub async fn invite_helper<'a>(
                &room_version,
                &Arc::new(pdu.clone()),
                create_prev_event,
-                &auth_events,
                None, // TODO: third_party_invite
+                |k, s| auth_events.get(&(k.clone(), s.to_owned())).map(Arc::clone),
            )
            .map_err(|e| {
                error!("{:?}", e);
@@ -1038,6 +1064,7 @@ pub async fn invite_helper<'a>(
                is_direct: Some(is_direct),
                third_party_invite: None,
                blurhash: db.users.blurhash(&user_id)?,
+                reason: None,
            })
            .expect("event is valid, we just created it"),
            unsigned: None,
@@ -128,7 +128,7 @@ pub async fn get_message_events_route(
        get_message_events::Direction::Forward => {
            let events_after = db
                .rooms
-                .pdus_after(&sender_user, &body.room_id, from)
+                .pdus_after(&sender_user, &body.room_id, from)?
                .take(limit)
                .filter_map(|r| r.ok()) // Filter out buggy events
                .filter_map(|(pdu_id, pdu)| {
@@ -158,7 +158,7 @@ pub async fn get_message_events_route(
        get_message_events::Direction::Backward => {
            let events_before = db
                .rooms
-                .pdus_until(&sender_user, &body.room_id, from)
+                .pdus_until(&sender_user, &body.room_id, from)?
                .take(limit)
                .filter_map(|r| r.ok()) // Filter out buggy events
                .filter_map(|(pdu_id, pdu)| {
@@ -33,6 +33,8 @@ pub async fn create_room_route(

    let room_id = RoomId::new(db.globals.server_name());

+    db.rooms.get_or_create_shortroomid(&room_id, &db.globals)?;
+
    let mutex_state = Arc::clone(
        db.globals
            .roomid_mutex_state
@@ -105,6 +107,7 @@ pub async fn create_room_route(
                is_direct: Some(body.is_direct),
                third_party_invite: None,
                blurhash: db.users.blurhash(&sender_user)?,
+                reason: None,
            })
            .expect("event is valid, we just created it"),
            unsigned: None,
@@ -173,7 +176,6 @@ pub async fn create_room_route(
    )?;

    // 4. Canonical room alias
-
    if let Some(room_alias_id) = &alias {
        db.rooms.build_and_append_pdu(
            PduBuilder {
@@ -193,7 +195,7 @@ pub async fn create_room_route(
            &room_id,
            &db,
            &state_lock,
-        );
+        )?;
    }

    // 5. Events set by preset
@@ -421,6 +423,8 @@ pub async fn upgrade_room_route(

    // Create a replacement room
    let replacement_room = RoomId::new(db.globals.server_name());
+    db.rooms
+        .get_or_create_shortroomid(&replacement_room, &db.globals)?;

    let mutex_state = Arc::clone(
        db.globals
@@ -516,6 +520,7 @@ pub async fn upgrade_room_route(
                is_direct: None,
                third_party_invite: None,
                blurhash: db.users.blurhash(&sender_user)?,
+                reason: None,
            })
            .expect("event is valid, we just created it"),
            unsigned: None,
@@ -3,7 +3,10 @@ use crate::{database::DatabaseGuard, utils, ConduitResult, Error, Ruma};
 use ruma::{
    api::client::{
        error::ErrorKind,
-        r0::session::{get_login_types, login, logout, logout_all},
+        r0::{
+            session::{get_login_types, login, logout, logout_all},
+            uiaa::IncomingUserIdentifier,
+        },
    },
    UserId,
 };
@@ -60,7 +63,7 @@ pub async fn login_route(
            identifier,
            password,
        } => {
-            let username = if let login::IncomingUserIdentifier::MatrixId(matrix_id) = identifier {
+            let username = if let IncomingUserIdentifier::MatrixId(matrix_id) = identifier {
                matrix_id
            } else {
                return Err(Error::BadRequest(ErrorKind::Forbidden, "Bad login type."));
@@ -205,7 +205,7 @@ async fn sync_helper(

        let mut non_timeline_pdus = db
            .rooms
-            .pdus_until(&sender_user, &room_id, u64::MAX)
+            .pdus_until(&sender_user, &room_id, u64::MAX)?
            .filter_map(|r| {
                // Filter out buggy events
                if r.is_err() {
@@ -248,13 +248,13 @@ async fn sync_helper(

        let first_pdu_before_since = db
            .rooms
-            .pdus_until(&sender_user, &room_id, since)
+            .pdus_until(&sender_user, &room_id, since)?
            .next()
            .transpose()?;

        let pdus_after_since = db
            .rooms
-            .pdus_after(&sender_user, &room_id, since)
+            .pdus_after(&sender_user, &room_id, since)?
            .next()
            .is_some();

@@ -286,7 +286,7 @@ async fn sync_helper(

                for hero in db
                    .rooms
-                    .all_pdus(&sender_user, &room_id)
+                    .all_pdus(&sender_user, &room_id)?
                    .filter_map(|pdu| pdu.ok()) // Ignore all broken pdus
                    .filter(|(_, pdu)| pdu.kind == EventType::RoomMember)
                    .map(|(_, pdu)| {
@@ -328,11 +328,11 @@ async fn sync_helper(
                }
            }

-            (
+            Ok::<_, Error>((
                Some(joined_member_count),
                Some(invited_member_count),
                heroes,
-            )
+            ))
        };

        let (
@@ -343,12 +343,12 @@ async fn sync_helper(
            state_events,
        ) = if since_shortstatehash.is_none() {
            // Probably since = 0, we will do an initial sync
-            let (joined_member_count, invited_member_count, heroes) = calculate_counts();
+            let (joined_member_count, invited_member_count, heroes) = calculate_counts()?;

            let current_state_ids = db.rooms.state_full_ids(current_shortstatehash)?;
            let state_events = current_state_ids
                .iter()
-                .map(|id| db.rooms.get_pdu(id))
+                .map(|(_, id)| db.rooms.get_pdu(id))
                .filter_map(|r| r.ok().flatten())
                .collect::<Vec<_>>();

@@ -393,18 +393,19 @@ async fn sync_helper(
            let state_events = if joined_since_last_sync {
                current_state_ids
                    .iter()
-                    .map(|id| db.rooms.get_pdu(id))
+                    .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| {
+                    .iter()
+                    .filter(|(key, id)| since_state_ids.get(key) != Some(id))
+                    .filter(|(_, id)| {
                        !timeline_pdus
                            .iter()
                            .any(|(_, timeline_pdu)| timeline_pdu.event_id == **id)
                    })
-                    .map(|id| db.rooms.get_pdu(id))
+                    .map(|(_, id)| db.rooms.get_pdu(id))
                    .filter_map(|r| r.ok().flatten())
                    .collect()
            };
@@ -510,7 +511,7 @@ async fn sync_helper(
            }

            let (joined_member_count, invited_member_count, heroes) = if send_member_count {
-                calculate_counts()
+                calculate_counts()?
            } else {
                (None, None, Vec::new())
            };
@@ -24,13 +24,14 @@ use rocket::{
    request::{FromRequest, Request},
    Shutdown, State,
 };
-use ruma::{DeviceId, RoomId, ServerName, UserId};
+use ruma::{DeviceId, EventId, RoomId, ServerName, UserId};
 use serde::{de::IgnoredAny, Deserialize};
 use std::{
-    collections::{BTreeMap, HashMap},
-    convert::TryFrom,
+    collections::{BTreeMap, HashMap, HashSet},
+    convert::{TryFrom, TryInto},
    fs::{self, remove_dir_all},
    io::Write,
+    mem::size_of,
    ops::Deref,
    path::Path,
    sync::{Arc, Mutex, RwLock},
@@ -107,7 +108,7 @@ fn default_db_cache_capacity_mb() -> f64 {
 }

 fn default_sqlite_wal_clean_second_interval() -> u32 {
-    15 * 60 // every 15 minutes
+    1 * 60 // every minute
 }

 fn default_max_request_size() -> u32 {
@@ -261,7 +262,11 @@ impl Database {
                userroomid_highlightcount: builder.open_tree("userroomid_highlightcount")?,

                statekey_shortstatekey: builder.open_tree("statekey_shortstatekey")?,
-                stateid_shorteventid: builder.open_tree("stateid_shorteventid")?,
+                shortstatekey_statekey: builder.open_tree("shortstatekey_statekey")?,
+
+                roomid_shortroomid: builder.open_tree("roomid_shortroomid")?,
+
+                shortstatehash_statediff: builder.open_tree("shortstatehash_statediff")?,
                eventid_shorteventid: builder.open_tree("eventid_shorteventid")?,
                shorteventid_eventid: builder.open_tree("shorteventid_eventid")?,
                shorteventid_shortstatehash: builder.open_tree("shorteventid_shortstatehash")?,
@@ -270,8 +275,13 @@ impl Database {

                eventid_outlierpdu: builder.open_tree("eventid_outlierpdu")?,
                referencedevents: builder.open_tree("referencedevents")?,
-                pdu_cache: Mutex::new(LruCache::new(1_000_000)),
-                auth_chain_cache: Mutex::new(LruCache::new(1_000_000)),
+                pdu_cache: Mutex::new(LruCache::new(100_000)),
+                auth_chain_cache: Mutex::new(LruCache::new(100_000)),
+                shorteventid_cache: Mutex::new(LruCache::new(1_000_000)),
+                eventidshort_cache: Mutex::new(LruCache::new(1_000_000)),
+                shortstatekey_cache: Mutex::new(LruCache::new(1_000_000)),
+                statekeyshort_cache: Mutex::new(LruCache::new(1_000_000)),
+                stateinfo_cache: Mutex::new(LruCache::new(1000)),
            },
            account_data: account_data::AccountData {
                roomuserdataid_accountdata: builder.open_tree("roomuserdataid_accountdata")?,
@@ -424,7 +434,6 @@ impl Database {
            }

            if db.globals.database_version()? < 6 {
-                // TODO update to 6
                // Set room member count
                for (roomid, _) in db.rooms.roomid_shortstatehash.iter() {
                    let room_id =
@@ -437,6 +446,273 @@ impl Database {

                println!("Migration: 5 -> 6 finished");
            }
+
+            if db.globals.database_version()? < 7 {
+                // Upgrade state store
+                let mut last_roomstates: HashMap<RoomId, u64> = HashMap::new();
+                let mut current_sstatehash: Option<u64> = None;
+                let mut current_room = None;
+                let mut current_state = HashSet::new();
+                let mut counter = 0;
+
+                let mut handle_state =
+                    |current_sstatehash: u64,
+                     current_room: &RoomId,
+                     current_state: HashSet<_>,
+                     last_roomstates: &mut HashMap<_, _>| {
+                        counter += 1;
+                        println!("counter: {}", counter);
+                        let last_roomsstatehash = last_roomstates.get(current_room);
+
+                        let states_parents = last_roomsstatehash.map_or_else(
+                            || Ok(Vec::new()),
+                            |&last_roomsstatehash| {
+                                db.rooms.load_shortstatehash_info(dbg!(last_roomsstatehash))
+                            },
+                        )?;
+
+                        let (statediffnew, statediffremoved) =
+                            if let Some(parent_stateinfo) = states_parents.last() {
+                                let statediffnew = current_state
+                                    .difference(&parent_stateinfo.1)
+                                    .cloned()
+                                    .collect::<HashSet<_>>();
+
+                                let statediffremoved = parent_stateinfo
+                                    .1
+                                    .difference(&current_state)
+                                    .cloned()
+                                    .collect::<HashSet<_>>();
+
+                                (statediffnew, statediffremoved)
+                            } else {
+                                (current_state, HashSet::new())
+                            };
+
+                        db.rooms.save_state_from_diff(
+                            dbg!(current_sstatehash),
+                            statediffnew,
+                            statediffremoved,
+                            2, // every state change is 2 event changes on average
+                            states_parents,
+                        )?;
+
+                        /*
+                        let mut tmp = db.rooms.load_shortstatehash_info(&current_sstatehash, &db)?;
+                        let state = tmp.pop().unwrap();
+                        println!(
+                            "{}\t{}{:?}: {:?} + {:?} - {:?}",
+                            current_room,
+                            "  ".repeat(tmp.len()),
+                            utils::u64_from_bytes(&current_sstatehash).unwrap(),
+                            tmp.last().map(|b| utils::u64_from_bytes(&b.0).unwrap()),
+                            state
+                                .2
+                                .iter()
+                                .map(|b| utils::u64_from_bytes(&b[size_of::<u64>()..]).unwrap())
+                                .collect::<Vec<_>>(),
+                            state
+                                .3
+                                .iter()
+                                .map(|b| utils::u64_from_bytes(&b[size_of::<u64>()..]).unwrap())
+                                .collect::<Vec<_>>()
+                        );
+                        */
+
+                        Ok::<_, Error>(())
+                    };
+
+                for (k, seventid) in db._db.open_tree("stateid_shorteventid")?.iter() {
+                    let sstatehash = utils::u64_from_bytes(&k[0..size_of::<u64>()])
+                        .expect("number of bytes is correct");
+                    let sstatekey = k[size_of::<u64>()..].to_vec();
+                    if Some(sstatehash) != current_sstatehash {
+                        if let Some(current_sstatehash) = current_sstatehash {
+                            handle_state(
+                                current_sstatehash,
+                                current_room.as_ref().unwrap(),
+                                current_state,
+                                &mut last_roomstates,
+                            )?;
+                            last_roomstates
+                                .insert(current_room.clone().unwrap(), current_sstatehash);
+                        }
+                        current_state = HashSet::new();
+                        current_sstatehash = Some(sstatehash);
+
+                        let event_id = db
+                            .rooms
+                            .shorteventid_eventid
+                            .get(&seventid)
+                            .unwrap()
+                            .unwrap();
+                        let event_id =
+                            EventId::try_from(utils::string_from_bytes(&event_id).unwrap())
+                                .unwrap();
+                        let pdu = db.rooms.get_pdu(&event_id).unwrap().unwrap();
+
+                        if Some(&pdu.room_id) != current_room.as_ref() {
+                            current_room = Some(pdu.room_id.clone());
+                        }
+                    }
+
+                    let mut val = sstatekey;
+                    val.extend_from_slice(&seventid);
+                    current_state.insert(val.try_into().expect("size is correct"));
+                }
+
+                if let Some(current_sstatehash) = current_sstatehash {
+                    handle_state(
+                        current_sstatehash,
+                        current_room.as_ref().unwrap(),
+                        current_state,
+                        &mut last_roomstates,
+                    )?;
+                }
+
+                db.globals.bump_database_version(7)?;
+
+                println!("Migration: 6 -> 7 finished");
+            }
+
+            if db.globals.database_version()? < 8 {
+                // Generate short room ids for all rooms
+                for (room_id, _) in db.rooms.roomid_shortstatehash.iter() {
+                    let shortroomid = db.globals.next_count()?.to_be_bytes();
+                    db.rooms.roomid_shortroomid.insert(&room_id, &shortroomid)?;
+                    println!("Migration: 8");
+                }
+                // Update pduids db layout
+                let mut batch = db.rooms.pduid_pdu.iter().filter_map(|(key, v)| {
+                    if !key.starts_with(b"!") {
+                        return None;
+                    }
+                    let mut parts = key.splitn(2, |&b| b == 0xff);
+                    let room_id = parts.next().unwrap();
+                    let count = parts.next().unwrap();
+
+                    let short_room_id = db
+                        .rooms
+                        .roomid_shortroomid
+                        .get(&room_id)
+                        .unwrap()
+                        .expect("shortroomid should exist");
+
+                    let mut new_key = short_room_id;
+                    new_key.extend_from_slice(count);
+
+                    Some((new_key, v))
+                });
+
+                db.rooms.pduid_pdu.insert_batch(&mut batch)?;
+
+                let mut batch2 = db.rooms.eventid_pduid.iter().filter_map(|(k, value)| {
+                    if !value.starts_with(b"!") {
+                        return None;
+                    }
+                    let mut parts = value.splitn(2, |&b| b == 0xff);
+                    let room_id = parts.next().unwrap();
+                    let count = parts.next().unwrap();
+
+                    let short_room_id = db
+                        .rooms
+                        .roomid_shortroomid
+                        .get(&room_id)
+                        .unwrap()
+                        .expect("shortroomid should exist");
+
+                    let mut new_value = short_room_id;
+                    new_value.extend_from_slice(count);
+
+                    Some((k, new_value))
+                });
+
+                db.rooms.eventid_pduid.insert_batch(&mut batch2)?;
+
+                db.globals.bump_database_version(8)?;
+
+                println!("Migration: 7 -> 8 finished");
+            }
+
+            if db.globals.database_version()? < 9 {
+                // Update tokenids db layout
+                let batch = db
+                    .rooms
+                    .tokenids
+                    .iter()
+                    .filter_map(|(key, _)| {
+                        if !key.starts_with(b"!") {
+                            return None;
+                        }
+                        let mut parts = key.splitn(4, |&b| b == 0xff);
+                        let room_id = parts.next().unwrap();
+                        let word = parts.next().unwrap();
+                        let _pdu_id_room = parts.next().unwrap();
+                        let pdu_id_count = parts.next().unwrap();
+
+                        let short_room_id = db
+                            .rooms
+                            .roomid_shortroomid
+                            .get(&room_id)
+                            .unwrap()
+                            .expect("shortroomid should exist");
+                        let mut new_key = short_room_id;
+                        new_key.extend_from_slice(word);
+                        new_key.push(0xff);
+                        new_key.extend_from_slice(pdu_id_count);
+                        println!("old {:?}", key);
+                        println!("new {:?}", new_key);
+                        Some((new_key, Vec::new()))
+                    })
+                    .collect::<Vec<_>>();
+
+                let mut iter = batch.into_iter().peekable();
+
+                while iter.peek().is_some() {
+                    db.rooms
+                        .tokenids
+                        .insert_batch(&mut iter.by_ref().take(1000))?;
+                    println!("smaller batch done");
+                }
+
+                println!("Deleting starts");
+
+                let batch2 = db
+                    .rooms
+                    .tokenids
+                    .iter()
+                    .filter_map(|(key, _)| {
+                        if key.starts_with(b"!") {
+                            println!("del {:?}", key);
+                            Some(key)
+                        } else {
+                            None
+                        }
+                    })
+                    .collect::<Vec<_>>();
+
+                for key in batch2 {
+                    println!("del");
+                    db.rooms.tokenids.remove(&key)?;
+                }
+
+                db.globals.bump_database_version(9)?;
+
+                println!("Migration: 8 -> 9 finished");
+            }
+
+            if db.globals.database_version()? < 10 {
+                // Add other direction for shortstatekeys
+                for (statekey, shortstatekey) in db.rooms.statekey_shortstatekey.iter() {
+                    db.rooms
+                        .shortstatekey_statekey
+                        .insert(&shortstatekey, &statekey)?;
+                }
+
+                db.globals.bump_database_version(10)?;
+
+                println!("Migration: 9 -> 10 finished");
+            }
        }

        let guard = db.read().await;
@@ -35,6 +35,7 @@ pub trait Tree: Send + Sync {
    ) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a>;

    fn increment(&self, key: &[u8]) -> Result<Vec<u8>>;
+    fn increment_batch<'a>(&self, iter: &mut dyn Iterator<Item = Vec<u8>>) -> Result<()>;

    fn scan_prefix<'a>(
        &'a self,
@@ -9,15 +9,13 @@ use std::{
    path::{Path, PathBuf},
    pin::Pin,
    sync::Arc,
-    time::{Duration, Instant},
 };
 use tokio::sync::oneshot::Sender;
-use tracing::{debug, warn};
-
-pub const MILLI: Duration = Duration::from_millis(1);
+use tracing::debug;

 thread_local! {
    static READ_CONNECTION: RefCell<Option<&'static Connection>> = RefCell::new(None);
+    static READ_CONNECTION_ITERATOR: RefCell<Option<&'static Connection>> = RefCell::new(None);
 }

 struct PreparedStatementIterator<'a> {
@@ -51,11 +49,11 @@ impl Engine {
    fn prepare_conn(path: &Path, cache_size_kb: u32) -> Result<Connection> {
        let conn = Connection::open(&path)?;

-        conn.pragma_update(Some(Main), "page_size", &32768)?;
+        conn.pragma_update(Some(Main), "page_size", &2048)?;
        conn.pragma_update(Some(Main), "journal_mode", &"WAL")?;
        conn.pragma_update(Some(Main), "synchronous", &"NORMAL")?;
        conn.pragma_update(Some(Main), "cache_size", &(-i64::from(cache_size_kb)))?;
-        conn.pragma_update(Some(Main), "wal_autocheckpoint", &0)?;
+        conn.pragma_update(Some(Main), "wal_autocheckpoint", &2000)?;

        Ok(conn)
    }
@@ -79,9 +77,25 @@ impl Engine {
        })
    }

+    fn read_lock_iterator(&self) -> &'static Connection {
+        READ_CONNECTION_ITERATOR.with(|cell| {
+            let connection = &mut cell.borrow_mut();
+
+            if (*connection).is_none() {
+                let c = Box::leak(Box::new(
+                    Self::prepare_conn(&self.path, self.cache_size_per_thread).unwrap(),
+                ));
+                **connection = Some(c);
+            }
+
+            connection.unwrap()
+        })
+    }
+
    pub fn flush_wal(self: &Arc<Self>) -> Result<()> {
-        self.write_lock()
-            .pragma_update(Some(Main), "wal_checkpoint", &"TRUNCATE")?;
+        // We use autocheckpoints
+        //self.write_lock()
+        //.pragma_update(Some(Main), "wal_checkpoint", &"TRUNCATE")?;
        Ok(())
    }
 }
@@ -153,6 +167,34 @@ impl SqliteTable {
        )?;
        Ok(())
    }
+
+    pub fn iter_with_guard<'a>(
+        &'a self,
+        guard: &'a Connection,
+    ) -> Box<dyn Iterator<Item = TupleOfBytes> + 'a> {
+        let statement = Box::leak(Box::new(
+            guard
+                .prepare(&format!(
+                    "SELECT key, value FROM {} ORDER BY key ASC",
+                    &self.name
+                ))
+                .unwrap(),
+        ));
+
+        let statement_ref = NonAliasingBox(statement);
+
+        let iterator = Box::new(
+            statement
+                .query_map([], |row| Ok((row.get_unwrap(0), row.get_unwrap(1))))
+                .unwrap()
+                .map(|r| r.unwrap()),
+        );
+
+        Box::new(PreparedStatementIterator {
+            iterator,
+            statement_ref,
+        })
+    }
 }

 impl Tree for SqliteTable {
@@ -164,16 +206,7 @@ impl Tree for SqliteTable {
    #[tracing::instrument(skip(self, key, value))]
    fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> {
        let guard = self.engine.write_lock();
-
-        let start = Instant::now();
-
        self.insert_with_guard(&guard, key, value)?;
-
-        let elapsed = start.elapsed();
-        if elapsed > MILLI {
-            warn!("insert took {:?} : {}", elapsed, &self.name);
-        }
-
        drop(guard);

        let watchers = self.watchers.read();
@@ -216,53 +249,41 @@ impl Tree for SqliteTable {
        Ok(())
    }

+    #[tracing::instrument(skip(self, iter))]
+    fn increment_batch<'a>(&self, iter: &mut dyn Iterator<Item = Vec<u8>>) -> Result<()> {
+        let guard = self.engine.write_lock();
+
+        guard.execute("BEGIN", [])?;
+        for key in iter {
+            let old = self.get_with_guard(&guard, &key)?;
+            let new = crate::utils::increment(old.as_deref())
+                .expect("utils::increment always returns Some");
+            self.insert_with_guard(&guard, &key, &new)?;
+        }
+        guard.execute("COMMIT", [])?;
+
+        drop(guard);
+
+        Ok(())
+    }
+
    #[tracing::instrument(skip(self, key))]
    fn remove(&self, key: &[u8]) -> Result<()> {
        let guard = self.engine.write_lock();

-        let start = Instant::now();
-
        guard.execute(
            format!("DELETE FROM {} WHERE key = ?", self.name).as_str(),
            [key],
        )?;

-        let elapsed = start.elapsed();
-
-        if elapsed > MILLI {
-            debug!("remove:    took {:012?} : {}", elapsed, &self.name);
-        }
-        // debug!("remove key: {:?}", &key);
-
        Ok(())
    }

    #[tracing::instrument(skip(self))]
    fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = TupleOfBytes> + 'a> {
-        let guard = self.engine.read_lock();
+        let guard = self.engine.read_lock_iterator();

-        let statement = Box::leak(Box::new(
-            guard
-                .prepare(&format!(
-                    "SELECT key, value FROM {} ORDER BY key ASC",
-                    &self.name
-                ))
-                .unwrap(),
-        ));
-
-        let statement_ref = NonAliasingBox(statement);
-
-        let iterator = Box::new(
-            statement
-                .query_map([], |row| Ok((row.get_unwrap(0), row.get_unwrap(1))))
-                .unwrap()
-                .map(|r| r.unwrap()),
-        );
-
-        Box::new(PreparedStatementIterator {
-            iterator,
-            statement_ref,
-        })
+        self.iter_with_guard(&guard)
    }

    #[tracing::instrument(skip(self, from, backwards))]
@@ -271,7 +292,7 @@ impl Tree for SqliteTable {
        from: &[u8],
        backwards: bool,
    ) -> Box<dyn Iterator<Item = TupleOfBytes> + 'a> {
-        let guard = self.engine.read_lock();
+        let guard = self.engine.read_lock_iterator();
        let from = from.to_vec(); // TODO change interface?

        if backwards {
@@ -326,8 +347,6 @@ impl Tree for SqliteTable {
    fn increment(&self, key: &[u8]) -> Result<Vec<u8>> {
        let guard = self.engine.write_lock();

-        let start = Instant::now();
-
        let old = self.get_with_guard(&guard, key)?;

        let new =
@@ -335,26 +354,11 @@ impl Tree for SqliteTable {

        self.insert_with_guard(&guard, key, &new)?;

-        let elapsed = start.elapsed();
-
-        if elapsed > MILLI {
-            debug!("increment: took {:012?} : {}", elapsed, &self.name);
-        }
-        // debug!("increment key: {:?}", &key);
-
        Ok(new)
    }

    #[tracing::instrument(skip(self, prefix))]
    fn scan_prefix<'a>(&'a self, prefix: Vec<u8>) -> Box<dyn Iterator<Item = TupleOfBytes> + 'a> {
-        // let name = self.name.clone();
-        // self.iter_from_thread(
-        //     format!(
-        //         "SELECT key, value FROM {} WHERE key BETWEEN ?1 AND ?1 || X'FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF' ORDER BY key ASC",
-        //         name
-        //     )
-        //     [prefix]
-        // )
        Box::new(
            self.iter_from(&prefix, false)
                .take_while(move |(key, _)| key.starts_with(&prefix)),
@@ -1,5 +1,5 @@
 use std::{
-    collections::{BTreeMap, HashMap},
+    collections::{BTreeMap, HashMap, HashSet},
    convert::{TryFrom, TryInto},
    fmt::Debug,
    sync::Arc,
@@ -20,14 +20,17 @@ use ruma::{
        appservice,
        federation::{
            self,
-            transactions::edu::{Edu, ReceiptContent, ReceiptData, ReceiptMap},
+            transactions::edu::{
+                DeviceListUpdateContent, Edu, ReceiptContent, ReceiptData, ReceiptMap,
+            },
        },
        OutgoingRequest,
    },
+    device_id,
    events::{push_rules, AnySyncEphemeralRoomEvent, EventType},
    push,
    receipt::ReceiptType,
-    MilliSecondsSinceUnixEpoch, ServerName, UInt, UserId,
+    uint, MilliSecondsSinceUnixEpoch, ServerName, UInt, UserId,
 };
 use tokio::{
    select,
@@ -317,8 +320,19 @@ impl Sending {
            })?;
        let mut events = Vec::new();
        let mut max_edu_count = since;
+        let mut device_list_changes = HashSet::new();
+
        'outer: for room_id in db.rooms.server_rooms(server) {
            let room_id = room_id?;
+            // Look for device list updates in this room
+            device_list_changes.extend(
+                db.users
+                    .keys_changed(&room_id.to_string(), since, None)
+                    .filter_map(|r| r.ok())
+                    .filter(|user_id| user_id.server_name() == db.globals.server_name()),
+            );
+
+            // Look for read receipts in this room
            for r in db.rooms.edus.readreceipts_since(&room_id, since) {
                let (user_id, count, read_receipt) = r?;

@@ -378,6 +392,22 @@ impl Sending {
            }
        }

+        for user_id in device_list_changes {
+            // Empty prev id forces synapse to resync: https://github.com/matrix-org/synapse/blob/98aec1cc9da2bd6b8e34ffb282c85abf9b8b42ca/synapse/handlers/device.py#L767
+            // Because synapse resyncs, we can just insert dummy data
+            let edu = Edu::DeviceListUpdate(DeviceListUpdateContent {
+                user_id,
+                device_id: device_id!("dummy"),
+                device_display_name: "Dummy".to_owned(),
+                stream_id: uint!(1),
+                prev_id: Vec::new(),
+                deleted: None,
+                keys: None,
+            });
+
+            events.push(serde_json::to_vec(&edu).expect("json can be serialized"));
+        }
+
        Ok((events, max_edu_count))
    }

@@ -4,11 +4,14 @@ use crate::{client_server::SESSION_ID_LENGTH, utils, Error, Result};
 use ruma::{
    api::client::{
        error::ErrorKind,
-        r0::uiaa::{IncomingAuthData, UiaaInfo},
+        r0::uiaa::{
+            IncomingAuthData, IncomingPassword, IncomingUserIdentifier::MatrixId, UiaaInfo,
+        },
    },
    signatures::CanonicalJsonValue,
    DeviceId, UserId,
 };
+use tracing::error;

 use super::abstraction::Tree;

@@ -49,14 +52,8 @@ impl Uiaa {
        users: &super::users::Users,
        globals: &super::globals::Globals,
    ) -> Result<(bool, UiaaInfo)> {
-        if let IncomingAuthData::DirectRequest {
-            kind,
-            session,
-            auth_parameters,
-        } = &auth
-        {
-            let mut uiaainfo = session
-                .as_ref()
+        let mut uiaainfo = auth
+            .session()
            .map(|session| self.get_uiaa_session(&user_id, &device_id, session))
            .unwrap_or_else(|| Ok(uiaainfo.clone()))?;

@@ -64,55 +61,29 @@ impl Uiaa {
            uiaainfo.session = Some(utils::random_string(SESSION_ID_LENGTH));
        }

+        match auth {
            // Find out what the user completed
-            match &**kind {
-                "m.login.password" => {
-                    let identifier = auth_parameters.get("identifier").ok_or(Error::BadRequest(
-                        ErrorKind::MissingParam,
-                        "m.login.password needs identifier.",
-                    ))?;
-
-                    let identifier_type = identifier.get("type").ok_or(Error::BadRequest(
-                        ErrorKind::MissingParam,
-                        "Identifier needs a type.",
-                    ))?;
-
-                    if identifier_type != "m.id.user" {
+            IncomingAuthData::Password(IncomingPassword {
+                identifier,
+                password,
+                ..
+            }) => {
+                let username = match identifier {
+                    MatrixId(username) => username,
+                    _ => {
                        return Err(Error::BadRequest(
                            ErrorKind::Unrecognized,
                            "Identifier type not recognized.",
-                        ));
+                        ))
                    }
+                };

-                    let username = identifier
-                        .get("user")
-                        .ok_or(Error::BadRequest(
-                            ErrorKind::MissingParam,
-                            "Identifier needs user field.",
-                        ))?
-                        .as_str()
-                        .ok_or(Error::BadRequest(
-                            ErrorKind::BadJson,
-                            "User is not a string.",
-                        ))?;
-
-                    let user_id = UserId::parse_with_server_name(username, globals.server_name())
+                let user_id =
+                    UserId::parse_with_server_name(username.clone(), globals.server_name())
                        .map_err(|_| {
                            Error::BadRequest(ErrorKind::InvalidParam, "User ID is invalid.")
                        })?;

-                    let password = auth_parameters
-                        .get("password")
-                        .ok_or(Error::BadRequest(
-                            ErrorKind::MissingParam,
-                            "Password is missing.",
-                        ))?
-                        .as_str()
-                        .ok_or(Error::BadRequest(
-                            ErrorKind::BadJson,
-                            "Password is not a string.",
-                        ))?;
-
                // Check if password is correct
                if let Some(hash) = users.password_hash(&user_id)? {
                    let hash_matches =
@@ -130,10 +101,10 @@ impl Uiaa {
                // Password was correct! Let's add it to `completed`
                uiaainfo.completed.push("m.login.password".to_owned());
            }
-                "m.login.dummy" => {
+            IncomingAuthData::Dummy(_) => {
                uiaainfo.completed.push("m.login.dummy".to_owned());
            }
-                k => panic!("type not supported: {}", k),
+            k => error!("type not supported: {:?}", k),
        }

        // Check if a flow now succeeds
@@ -166,9 +137,6 @@ impl Uiaa {
            None,
        )?;
        Ok((true, uiaainfo))
-        } else {
-            panic!("FallbackAcknowledgement is not supported yet");
-        }
    }

    fn set_uiaa_request(
@@ -673,7 +673,7 @@ impl Users {
    }

    #[tracing::instrument(skip(self, user_id, rooms, globals))]
-    fn mark_device_key_update(
+    pub fn mark_device_key_update(
        &self,
        user_id: &UserId,
        rooms: &super::rooms::Rooms,
@@ -17,7 +17,7 @@ use std::sync::Arc;
 use database::Config;
 pub use database::Database;
 pub use error::{Error, Result};
-use opentelemetry::trace::Tracer;
+use opentelemetry::trace::{FutureExt, Tracer};
 pub use pdu::PduEvent;
 pub use rocket::State;
 use ruma::api::client::error::ErrorKind;
@@ -160,7 +160,8 @@ fn setup_rocket(config: Figment, data: Arc<RwLock<Database>>) -> rocket::Rocket<
                server_server::get_room_state_route,
                server_server::get_room_state_ids_route,
                server_server::create_join_event_template_route,
-                server_server::create_join_event_route,
+                server_server::create_join_event_v1_route,
+                server_server::create_join_event_v2_route,
                server_server::create_invite_route,
                server_server::get_devices_route,
                server_server::get_room_information_route,
@@ -220,14 +221,17 @@ async fn main() {
    };

    if config.allow_jaeger {
+        opentelemetry::global::set_text_map_propagator(opentelemetry_jaeger::Propagator::new());
        let tracer = opentelemetry_jaeger::new_pipeline()
-            .with_service_name("conduit")
-            .install_simple()
+            .install_batch(opentelemetry::runtime::Tokio)
            .unwrap();

        let span = tracer.start("conduit");
-        start.await;
+        start.with_current_context().await;
        drop(span);
+
+        println!("exporting");
+        opentelemetry::global::shutdown_tracer_provider();
    } else {
        std::env::set_var("RUST_LOG", &config.log);

@@ -12,7 +12,7 @@ use ruma::{
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use std::{cmp::Ordering, collections::BTreeMap, convert::TryFrom};
-use tracing::error;
+use tracing::warn;

 #[derive(Clone, Deserialize, Serialize, Debug)]
 pub struct PduEvent {
@@ -322,7 +322,7 @@ pub(crate) fn gen_event_id_canonical_json(
    pdu: &Raw<ruma::events::pdu::Pdu>,
 ) -> crate::Result<(EventId, CanonicalJsonObject)> {
    let value = serde_json::from_str(pdu.json().get()).map_err(|e| {
-        error!("{:?}: {:?}", pdu, e);
+        warn!("Error parsing incoming event {:?}: {:?}", pdu, e);
        Error::BadServerResponse("Invalid PDU in server response")
    })?;
Author	SHA1	Message	Date
Jonas Zohren	84d54e0eaf	Also calculate sum of cache sizes Signed-off-by: Jonas Zohren <git-pbkyr@jzohren.de>	2021-08-25 00:51:46 +02:00
Jonas Zohren	927a23814b	Add test to measure memory usage of caches Signed-off-by: Jonas Zohren <git-pbkyr@jzohren.de>	2021-08-25 00:43:01 +02:00
Timo Kösters	5aa56b92ee	Merge branch 'speed' into 'master' improvement: more efficient auth chain cache See merge request famedly/conduit!164	2021-08-24 20:12:10 +00:00
Timo Kösters	dd87066546	improvement: more efficient auth chain cache	2021-08-24 21:55:18 +02:00
Timo Kösters	30b309b708	Merge branch 'speed' into 'master' Better E2EE and faster incoming event handling See merge request famedly/conduit!163	2021-08-24 18:16:56 +00:00
Timo Kösters	d5695fb9fe	Merge branch 'only-set-max-optimzations-in-ci' into 'master' Only apply max. optimizations in CI builds. See merge request famedly/conduit!162	2021-08-24 17:54:17 +00:00
Timo Kösters	81e056417c	improvement: better e2ee over fed, faster incoming event handling	2021-08-24 19:52:05 +02:00
Timo Kösters	72dd95f500	Merge branch 'createwarnings' into 'master' fix: correct create event warnings See merge request famedly/conduit!161	2021-08-24 17:51:53 +00:00
Jonas Zohren	b6e755f67e	Only apply max. optimizations in CI builds. The average german man has a life expectancy of 78.7 years, or 689884.2 hours. Assuming that Timo is 20 years old, he has rougly 514564.2 hours left on planet earth. Also assuming that cross release builds took him 25 minutes before, but 2-2.5x of that with the current release compilation config he wasted roughly an hour waiting for it to complete. If he continued to work on Conduit for 20 more years (or 175320 hours), and makes a release compilation about once per day, this means 7305 hours or 304 days wasted waiting for the rust compiler. By cutting that back down to the original settings, he get's 182 days of his life back. That's about 0.63% of his remaining life. 182 joyful days he can spend with family and loved ones.	2021-08-22 21:05:32 +00:00
Timo Kösters	4f8cf3bed4	Merge branch 'send_join_v1' into 'master' Implement federation/v1/send_join See merge request famedly/conduit!141	2021-08-22 12:06:26 +00:00
Timo Kösters	b3ea6adc9c	Merge branch 'get-pdu-json' into 'master' Consistent and escaped response in get_pdu See merge request famedly/conduit!160	2021-08-22 11:08:11 +00:00
Timo Kösters	6d83954c40	fix: room upgrades	2021-08-22 13:06:50 +02:00
Timo Kösters	667ffb4239	fix: correct create event warnings	2021-08-22 13:01:13 +02:00
Kurt Roeckx	48494c9464	Implement federation/v1/send_join	2021-08-22 10:07:38 +02:00
Tom Smeding	2ef23b213a	Consistent and escaped response in get_pdu 1. The fallback text of the get_pdu admin room command response message now contains the same text as the formatted_body content (namely, the json instead of Debug-formatting of a serde type). 2. The formatted_body content of the get_pdu response is now html-escaped.	2021-08-21 15:31:26 +02:00
Timo Kösters	33481ec062	Merge branch 'statediffs' into 'master' Statediffs See merge request famedly/conduit!145	2021-08-21 13:05:38 +00:00
Timo Kösters	3b78e43a18	fmt	2021-08-21 14:24:10 +02:00
Timo Kösters	4956fb9fba	improvement: limit prev event fetching	2021-08-21 14:22:21 +02:00
Timo Kösters	b09499c2df	fix: don't save empty tokens	2021-08-19 14:05:23 +02:00
Timo Kösters	46d8a46e1f	improvement: faster incoming transaction handling	2021-08-19 11:01:43 +02:00
Timo Kösters	bf7e019a68	improvement: better prev event fetching, perf improvements	2021-08-17 16:06:09 +02:00
Timo Kösters	75ba8bb565	fix: faster room joins	2021-08-17 00:22:52 +02:00
Timo Kösters	0823506d05	fix: don't load endless prev events and fix room join bug	2021-08-16 23:24:52 +02:00
Timo Kösters	2c3bee34a0	improvement: better sqlite	2021-08-15 13:27:03 +02:00
Timo Kösters	a4310f840e	improvement: state info cache	2021-08-15 06:48:25 +02:00
Timo Kösters	5bd5b41c70	fix: fetch event multiple times	2021-08-15 06:48:25 +02:00
Timo Kösters	f9a2edc0dd	fix: also fetch prev events that are outliers already	2021-08-15 06:48:25 +02:00
Timo Kösters	ecd1e45a44	fix: fetch more than one prev event	2021-08-15 06:48:24 +02:00
Timo Kösters	1d46569929	fix: don't use recursion for prev events	2021-08-15 06:48:24 +02:00
Timo Kösters	1e3a8ca35d	fmt	2021-08-15 06:48:24 +02:00
Timo Kösters	0cb22996be	remove prev event fetch limit	2021-08-15 06:48:14 +02:00
Timo Kösters	38effda799	fix: delta calculation	2021-08-14 19:11:10 +02:00
Timo Kösters	3cf0145bc5	fix: room exists panic	2021-08-14 08:26:45 +02:00
Timo Kösters	3eabaa2a95	finish implementing better state store	2021-08-13 19:49:16 +02:00
Timo Kösters	31f60ad6fd	improvement: migrations, batch inserts	2021-08-12 17:56:35 +02:00
Timo Kösters	41dd620d74	WIP improvement: much better state storage	2021-08-12 17:56:35 +02:00
Timo Kösters	9410d3ef9c	fix: long prev event fetch times for huge rooms	2021-08-12 17:56:14 +02:00
Timo Kösters	665aee11c0	less warnings	2021-08-12 17:56:14 +02:00
Timo Kösters	5173d0deb5	improvement: cache for short event ids	2021-08-12 17:56:14 +02:00
Timo Kösters	c2c6a8673e	improvement: use u64s in auth chain cache	2021-08-12 17:56:14 +02:00
Timo Kösters	096e0971f1	improvement: smaller cache, better prev event fetching	2021-08-12 17:56:13 +02:00