improvement: memory usage for caches

Lazy loading

See merge request famedly/conduit!240

Cargo.lock

@@ -146,6 +146,25 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "bindgen"
+version = "0.59.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2bd2a9a458e8f4304c52c43ebb0cfbd520289f8379a52e329a38afda99bf8eb8"
+dependencies = [
+ "bitflags",
+ "cexpr",
+ "clang-sys",
+ "lazy_static",
+ "lazycell",
+ "peeking_take_while",
+ "proc-macro2",
+ "quote",
+ "regex",
+ "rustc-hash",
+ "shlex",
+]
+
 [[package]]
 name = "bitflags"
 version = "1.3.2"
@@ -205,6 +224,15 @@ dependencies = [
  "jobserver",
 ]
 
+[[package]]
+name = "cexpr"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
+dependencies = [
+ "nom",
+]
+
 [[package]]
 name = "cfg-if"
 version = "0.1.10"
@@ -230,6 +258,17 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "clang-sys"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa66045b9cb23c2e9c1520732030608b02ee07e5cfaa5a521ec15ded7fa24c90"
+dependencies = [
+ "glob",
+ "libc",
+ "libloading",
+]
+
 [[package]]
 name = "color_quant"
 version = "1.1.0"
@@ -259,6 +298,7 @@ dependencies = [
  "reqwest",
  "ring",
  "rocket",
+ "rocksdb",
  "ruma",
  "rusqlite",
  "rust-argon2",
@@ -1167,12 +1207,40 @@ version = "1.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
+[[package]]
+name = "lazycell"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
+
 [[package]]
 name = "libc"
 version = "0.2.101"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3cb00336871be5ed2c8ed44b60ae9959dc5b9f08539422ed43f09e34ecaeba21"
 
+[[package]]
+name = "libloading"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afe203d669ec979b7128619bae5a63b7b42e9203c1b29146079ee05e2f604b52"
+dependencies = [
+ "cfg-if 1.0.0",
+ "winapi",
+]
+
+[[package]]
+name = "librocksdb-sys"
+version = "6.20.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c309a9d2470844aceb9a4a098cf5286154d20596868b75a6b36357d2bb9ca25d"
+dependencies = [
+ "bindgen",
+ "cc",
+ "glob",
+ "libc",
+]
+
 [[package]]
 name = "libsqlite3-sys"
 version = "0.22.2"
@@ -1289,6 +1357,12 @@ version = "0.3.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d"
 
+[[package]]
+name = "minimal-lexical"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
+
 [[package]]
 name = "miniz_oxide"
 version = "0.3.7"
@@ -1340,6 +1414,17 @@ dependencies = [
  "version_check",
 ]
 
+[[package]]
+name = "nom"
+version = "7.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1b1d11e1ef389c76fe5b81bcaf2ea32cf88b62bc494e19f493d0b30e7a930109"
+dependencies = [
+ "memchr",
+ "minimal-lexical",
+ "version_check",
+]
+
 [[package]]
 name = "ntapi"
 version = "0.3.6"
@@ -1539,6 +1624,12 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "peeking_take_while"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
+
 [[package]]
 name = "pem"
 version = "0.8.3"
@@ -1981,6 +2072,16 @@ dependencies = [
  "uncased",
 ]
 
+[[package]]
+name = "rocksdb"
+version = "0.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a62eca5cacf2c8261128631bed9f045598d40bfbe4b29f5163f0f802f8f44a7"
+dependencies = [
+ "libc",
+ "librocksdb-sys",
+]
+
 [[package]]
 name = "ruma"
 version = "0.4.0"
@@ -2266,6 +2367,12 @@ dependencies = [
  "crossbeam-utils 0.8.5",
 ]
 
+[[package]]
+name = "rustc-hash"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2"
+
 [[package]]
 name = "rustc_version"
 version = "0.2.3"
@@ -2478,6 +2585,12 @@ dependencies = [
  "lazy_static",
 ]
 
+[[package]]
+name = "shlex"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43b2853a4d09f215c24cc5489c992ce46052d359b5109343cbafbf26bc62f8a3"
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.0"

Cargo.toml

@@ -78,16 +78,18 @@ crossbeam = { version = "0.8.1", optional = true }
 num_cpus = "1.13.0"
 threadpool = "1.8.1"
 heed = { git = "https://github.com/timokoesters/heed.git", rev = "f6f825da7fb2c758867e05ad973ef800a6fe1d5d", optional = true }
+rocksdb = { version = "0.17.0", features = ["multi-threaded-cf"], optional = true }
 thread_local = "1.1.3"
 # used for TURN server authentication
 hmac = "0.11.0"
 sha-1 = "0.9.8"
 
 [features]
-default = ["conduit_bin", "backend_sqlite"]
+default = ["conduit_bin", "backend_sqlite", "backend_rocksdb"]
 backend_sled = ["sled"]
 backend_sqlite = ["sqlite"]
 backend_heed = ["heed", "crossbeam"]
+backend_rocksdb = ["rocksdb"]
 sqlite = ["rusqlite", "parking_lot", "crossbeam", "tokio/signal"]
 conduit_bin = [] # TODO: add rocket to this when it is optional
 

conduit-example.toml

@@ -1,11 +1,15 @@
 [global]
-# The server_name is the name of this server. It is used as a suffix for user
+# The server_name is the pretty name of this server. It is used as a suffix for user
 # and room ids. Examples: matrix.org, conduit.rs
-# The Conduit server needs to be reachable at https://your.server.name/ on port
-# 443 (client-server) and 8448 (federation) OR you can create /.well-known
-# files to redirect requests. See
+
+# The Conduit server needs all /_matrix/ requests to be reachable at
+# https://your.server.name/ on port 443 (client-server) and 8448 (federation).
+
+# If that's not possible for you, you can create /.well-known files to redirect
+# requests. See
 # https://matrix.org/docs/spec/client_server/latest#get-well-known-matrix-client
-# and https://matrix.org/docs/spec/server_server/r0.1.4#get-well-known-matrix-server
+# and
+# https://matrix.org/docs/spec/server_server/r0.1.4#get-well-known-matrix-server
 # for more information
 
 # YOU NEED TO EDIT THIS
@@ -13,6 +17,7 @@
 
 # This is the only directory where Conduit will save its data
 database_path = "/var/lib/conduit/"
+database_backend = "rocksdb"
 
 # The port Conduit will be running on. You need to set up a reverse proxy in
 # your web server (e.g. apache or nginx), so all requests to /_matrix on port

src/client_server/context.rs

@@ -1,5 +1,9 @@
 use crate::{database::DatabaseGuard, ConduitResult, Error, Ruma};
-use ruma::api::client::{error::ErrorKind, r0::context::get_context};
+use ruma::{
+    api::client::{error::ErrorKind, r0::context::get_context},
+    events::EventType,
+};
+use std::collections::HashSet;
 use std::convert::TryFrom;
 
 #[cfg(feature = "conduit_bin")]
@@ -21,6 +25,7 @@ pub async fn get_context_route(
     body: Ruma<get_context::Request<'_>>,
 ) -> ConduitResult<get_context::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    let sender_device = body.sender_device.as_ref().expect("user is authenticated");
 
     if !db.rooms.is_joined(sender_user, &body.room_id)? {
         return Err(Error::BadRequest(
@@ -29,6 +34,8 @@ pub async fn get_context_route(
         ));
     }
 
+    let mut lazy_loaded = HashSet::new();
+
     let base_pdu_id = db
         .rooms
         .get_pdu_id(&body.event_id)?
@@ -45,8 +52,18 @@ pub async fn get_context_route(
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "Base event not found.",
-        ))?
-        .to_room_event();
+        ))?;
+
+    if !db.rooms.lazy_load_was_sent_before(
+        &sender_user,
+        &sender_device,
+        &body.room_id,
+        &base_event.sender,
+    )? {
+        lazy_loaded.insert(base_event.sender.clone());
+    }
+
+    let base_event = base_event.to_room_event();
 
     let events_before: Vec<_> = db
         .rooms
@@ -60,6 +77,17 @@ pub async fn get_context_route(
         .filter_map(|r| r.ok()) // Remove buggy events
         .collect();
 
+    for (_, event) in &events_before {
+        if !db.rooms.lazy_load_was_sent_before(
+            &sender_user,
+            &sender_device,
+            &body.room_id,
+            &event.sender,
+        )? {
+            lazy_loaded.insert(event.sender.clone());
+        }
+    }
+
     let start_token = events_before
         .last()
         .and_then(|(pdu_id, _)| db.rooms.pdu_count(pdu_id).ok())
@@ -82,6 +110,17 @@ pub async fn get_context_route(
         .filter_map(|r| r.ok()) // Remove buggy events
         .collect();
 
+    for (_, event) in &events_after {
+        if !db.rooms.lazy_load_was_sent_before(
+            &sender_user,
+            &sender_device,
+            &body.room_id,
+            &event.sender,
+        )? {
+            lazy_loaded.insert(event.sender.clone());
+        }
+    }
+
     let end_token = events_after
         .last()
         .and_then(|(pdu_id, _)| db.rooms.pdu_count(pdu_id).ok())
@@ -98,12 +137,15 @@ pub async fn get_context_route(
     resp.events_before = events_before;
     resp.event = Some(base_event);
     resp.events_after = events_after;
-    resp.state = db // TODO: State at event
-        .rooms
-        .room_state_full(&body.room_id)?
-        .values()
-        .map(|pdu| pdu.to_state_event())
-        .collect();
+    resp.state = Vec::new();
+    for ll_id in &lazy_loaded {
+        if let Some(member_event) =
+            db.rooms
+                .room_state_get(&body.room_id, &EventType::RoomMember, ll_id.as_str())?
+        {
+            resp.state.push(member_event.to_state_event());
+        }
+    }
 
     Ok(resp.into())
 }

src/client_server/filter.rs

@@ -1,32 +1,47 @@
-use crate::{utils, ConduitResult};
-use ruma::api::client::r0::filter::{self, create_filter, get_filter};
+use crate::{database::DatabaseGuard, ConduitResult, Error, Ruma};
+use ruma::api::client::{
+    error::ErrorKind,
+    r0::filter::{create_filter, get_filter},
+};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, post};
 
 /// # `GET /_matrix/client/r0/user/{userId}/filter/{filterId}`
 ///
-/// TODO: Loads a filter that was previously created.
-#[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/user/<_>/filter/<_>"))]
-#[tracing::instrument]
-pub async fn get_filter_route() -> ConduitResult<get_filter::Response> {
-    // TODO
-    Ok(get_filter::Response::new(filter::IncomingFilterDefinition {
-        event_fields: None,
-        event_format: filter::EventFormat::default(),
-        account_data: filter::IncomingFilter::default(),
-        room: filter::IncomingRoomFilter::default(),
-        presence: filter::IncomingFilter::default(),
-    })
-    .into())
+/// Loads a filter that was previously created.
+///
+/// - A user can only access their own filters
+#[cfg_attr(
+    feature = "conduit_bin",
+    get("/_matrix/client/r0/user/<_>/filter/<_>", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn get_filter_route(
+    db: DatabaseGuard,
+    body: Ruma<get_filter::Request<'_>>,
+) -> ConduitResult<get_filter::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    let filter = match db.users.get_filter(sender_user, &body.filter_id)? {
+        Some(filter) => filter,
+        None => return Err(Error::BadRequest(ErrorKind::NotFound, "Filter not found.")),
+    };
+
+    Ok(get_filter::Response::new(filter).into())
 }
 
 /// # `PUT /_matrix/client/r0/user/{userId}/filter`
 ///
-/// TODO: Creates a new filter to be used by other endpoints.
-#[cfg_attr(feature = "conduit_bin", post("/_matrix/client/r0/user/<_>/filter"))]
-#[tracing::instrument]
-pub async fn create_filter_route() -> ConduitResult<create_filter::Response> {
-    // TODO
-    Ok(create_filter::Response::new(utils::random_string(10)).into())
+/// Creates a new filter to be used by other endpoints.
+#[cfg_attr(
+    feature = "conduit_bin",
+    post("/_matrix/client/r0/user/<_>/filter", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn create_filter_route(
+    db: DatabaseGuard,
+    body: Ruma<create_filter::Request<'_>>,
+) -> ConduitResult<create_filter::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    Ok(create_filter::Response::new(db.users.create_filter(sender_user, &body.filter)?).into())
 }

src/client_server/message.rs

@@ -6,7 +6,11 @@ use ruma::{
     },
     events::EventType,
 };
-use std::{collections::BTreeMap, convert::TryInto, sync::Arc};
+use std::{
+    collections::{BTreeMap, HashSet},
+    convert::TryInto,
+    sync::Arc,
+};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, put};
@@ -117,6 +121,7 @@ pub async fn get_message_events_route(
     body: Ruma<get_message_events::Request<'_>>,
 ) -> ConduitResult<get_message_events::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    let sender_device = body.sender_device.as_ref().expect("user is authenticated");
 
     if !db.rooms.is_joined(sender_user, &body.room_id)? {
         return Err(Error::BadRequest(
@@ -133,9 +138,18 @@ pub async fn get_message_events_route(
 
     let to = body.to.as_ref().map(|t| t.parse());
 
+    db.rooms
+        .lazy_load_confirm_delivery(&sender_user, &sender_device, &body.room_id, from)?;
+
     // Use limit or else 10
     let limit = body.limit.try_into().map_or(10_usize, |l: u32| l as usize);
 
+    let next_token;
+
+    let mut resp = get_message_events::Response::new();
+
+    let mut lazy_loaded = HashSet::new();
+
     match body.dir {
         get_message_events::Direction::Forward => {
             let events_after: Vec<_> = db
@@ -152,7 +166,18 @@ pub async fn get_message_events_route(
                 .take_while(|&(k, _)| Some(Ok(k)) != to) // Stop at `to`
                 .collect();
 
-            let end_token = events_after.last().map(|(count, _)| count.to_string());
+            for (_, event) in &events_after {
+                if !db.rooms.lazy_load_was_sent_before(
+                    &sender_user,
+                    &sender_device,
+                    &body.room_id,
+                    &event.sender,
+                )? {
+                    lazy_loaded.insert(event.sender.clone());
+                }
+            }
+
+            next_token = events_after.last().map(|(count, _)| count).copied();
 
             let events_after: Vec<_> = events_after
                 .into_iter()
@@ -161,11 +186,8 @@ pub async fn get_message_events_route(
 
             let mut resp = get_message_events::Response::new();
             resp.start = Some(body.from.to_owned());
-            resp.end = end_token;
+            resp.end = next_token.map(|count| count.to_string());
             resp.chunk = events_after;
-            resp.state = Vec::new();
-
-            Ok(resp.into())
         }
         get_message_events::Direction::Backward => {
             let events_before: Vec<_> = db
@@ -182,20 +204,49 @@ pub async fn get_message_events_route(
                 .take_while(|&(k, _)| Some(Ok(k)) != to) // Stop at `to`
                 .collect();
 
-            let start_token = events_before.last().map(|(count, _)| count.to_string());
+            for (_, event) in &events_before {
+                if !db.rooms.lazy_load_was_sent_before(
+                    &sender_user,
+                    &sender_device,
+                    &body.room_id,
+                    &event.sender,
+                )? {
+                    lazy_loaded.insert(event.sender.clone());
+                }
+            }
+
+            next_token = events_before.last().map(|(count, _)| count).copied();
 
             let events_before: Vec<_> = events_before
                 .into_iter()
                 .map(|(_, pdu)| pdu.to_room_event())
                 .collect();
 
-            let mut resp = get_message_events::Response::new();
             resp.start = Some(body.from.to_owned());
-            resp.end = start_token;
+            resp.end = next_token.map(|count| count.to_string());
             resp.chunk = events_before;
-            resp.state = Vec::new();
-
-            Ok(resp.into())
         }
     }
+
+    resp.state = Vec::new();
+    for ll_id in &lazy_loaded {
+        if let Some(member_event) =
+            db.rooms
+                .room_state_get(&body.room_id, &EventType::RoomMember, ll_id.as_str())?
+        {
+            resp.state.push(member_event.to_state_event());
+        }
+    }
+
+    if let Some(next_token) = next_token {
+        db.rooms.lazy_load_mark_sent(
+            &sender_user,
+            &sender_device,
+            &body.room_id,
+            lazy_loaded,
+            next_token,
+        );
+    }
+
+    Ok(resp.into())
 }

src/client_server/sync.rs

@@ -1,6 +1,10 @@
 use crate::{database::DatabaseGuard, ConduitResult, Database, Error, Result, Ruma, RumaResponse};
 use ruma::{
-    api::client::r0::{sync::sync_events, uiaa::UiaaResponse},
+    api::client::r0::{
+        filter::{IncomingFilterDefinition, LazyLoadOptions},
+        sync::sync_events,
+        uiaa::UiaaResponse,
+    },
     events::{
         room::member::{MembershipState, RoomMemberEventContent},
         AnySyncEphemeralRoomEvent, EventType,
@@ -36,13 +40,15 @@ use rocket::{get, tokio};
 /// Calling this endpoint with a `since` parameter from a previous `next_batch` returns:
 /// For joined rooms:
 /// - Some of the most recent events of each timeline that happened after since
-/// - If user joined the room after since: All state events and device list updates in that room
+/// - If user joined the room after since: All state events (unless lazy loading is activated) and
+/// all device list updates in that room
 /// - If the user was already in the room: A list of all events that are in the state now, but were
 /// not in the state at `since`
 /// - If the state we send contains a member event: Joined and invited member counts, heroes
 /// - Device list updates that happened after `since`
 /// - If there are events in the timeline we send or the user send updated his read mark: Notification counts
 /// - EDUs that are active now (read receipts, typing updates, presence)
+/// - TODO: Allow multiple sync streams to support Pantalaimon
 ///
 /// For invited rooms:
 /// - If the user was invited after `since`: A subset of the state of the room at the point of the invite
@@ -77,34 +83,32 @@ pub async fn sync_events_route(
         Entry::Vacant(v) => {
             let (tx, rx) = tokio::sync::watch::channel(None);
 
+            v.insert((body.since.clone(), rx.clone()));
+
             tokio::spawn(sync_helper_wrapper(
                 Arc::clone(&arc_db),
                 sender_user.clone(),
                 sender_device.clone(),
-                body.since.clone(),
-                body.full_state,
-                body.timeout,
+                body,
                 tx,
             ));
 
-            v.insert((body.since.clone(), rx)).1.clone()
+            rx
         }
         Entry::Occupied(mut o) => {
             if o.get().0 != body.since {
                 let (tx, rx) = tokio::sync::watch::channel(None);
 
+                o.insert((body.since.clone(), rx.clone()));
+
                 tokio::spawn(sync_helper_wrapper(
                     Arc::clone(&arc_db),
                     sender_user.clone(),
                     sender_device.clone(),
-                    body.since.clone(),
-                    body.full_state,
-                    body.timeout,
+                    body,
                     tx,
                 ));
 
-                o.insert((body.since.clone(), rx.clone()));
-
                 rx
             } else {
                 o.get().1.clone()
@@ -135,18 +139,16 @@ async fn sync_helper_wrapper(
     db: Arc<DatabaseGuard>,
     sender_user: Box<UserId>,
     sender_device: Box<DeviceId>,
-    since: Option<String>,
-    full_state: bool,
-    timeout: Option<Duration>,
+    body: sync_events::IncomingRequest,
     tx: Sender<Option<ConduitResult<sync_events::Response>>>,
 ) {
+    let since = body.since.clone();
+
     let r = sync_helper(
         Arc::clone(&db),
         sender_user.clone(),
         sender_device.clone(),
-        since.clone(),
-        full_state,
-        timeout,
+        body,
     )
     .await;
 
@@ -179,9 +181,7 @@ async fn sync_helper(
     db: Arc<DatabaseGuard>,
     sender_user: Box<UserId>,
     sender_device: Box<DeviceId>,
-    since: Option<String>,
-    full_state: bool,
-    timeout: Option<Duration>,
+    body: sync_events::IncomingRequest,
     // bool = caching allowed
 ) -> Result<(sync_events::Response, bool), Error> {
     // TODO: match body.set_presence {
@@ -193,8 +193,26 @@ async fn sync_helper(
     let next_batch = db.globals.current_count()?;
     let next_batch_string = next_batch.to_string();
 
+    // Load filter
+    let filter = match body.filter {
+        None => IncomingFilterDefinition::default(),
+        Some(sync_events::IncomingFilter::FilterDefinition(filter)) => filter,
+        Some(sync_events::IncomingFilter::FilterId(filter_id)) => db
+            .users
+            .get_filter(&sender_user, &filter_id)?
+            .unwrap_or_default(),
+    };
+
+    let (lazy_load_enabled, lazy_load_send_redundant) = match filter.room.state.lazy_load_options {
+        LazyLoadOptions::Enabled {
+            include_redundant_members: redundant,
+        } => (true, redundant),
+        _ => (false, false),
+    };
+
     let mut joined_rooms = BTreeMap::new();
-    let since = since
+    let since = body
+        .since
         .clone()
         .and_then(|string| string.parse().ok())
         .unwrap_or(0);
@@ -264,6 +282,14 @@ async fn sync_helper(
         // limited unless there are events in non_timeline_pdus
         let limited = non_timeline_pdus.next().is_some();
 
+        let mut timeline_users = HashSet::new();
+        for (_, event) in &timeline_pdus {
+            timeline_users.insert(event.sender.as_str().to_owned());
+        }
+
+        db.rooms
+            .lazy_load_confirm_delivery(&sender_user, &sender_device, &room_id, since)?;
+
         // Database queries:
 
         let current_shortstatehash = db
@@ -344,14 +370,58 @@ 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 current_state_ids = db.rooms.state_full_ids(current_shortstatehash)?;
-            let state_events: Vec<_> = current_state_ids
-                .iter()
-                .map(|(_, id)| db.rooms.get_pdu(id))
-                .filter_map(|r| r.ok().flatten())
-                .collect();
+
+            let mut state_events = Vec::new();
+            let mut lazy_loaded = HashSet::new();
+
+            for (shortstatekey, id) in current_state_ids {
+                let (event_type, state_key) = db.rooms.get_statekey_from_short(shortstatekey)?;
+
+                if event_type != EventType::RoomMember {
+                    let pdu = match db.rooms.get_pdu(&id)? {
+                        Some(pdu) => pdu,
+                        None => {
+                            error!("Pdu in state not found: {}", id);
+                            continue;
+                        }
+                    };
+                    state_events.push(pdu);
+                } else if !lazy_load_enabled
+                    || body.full_state
+                    || timeline_users.contains(&state_key)
+                {
+                    let pdu = match db.rooms.get_pdu(&id)? {
+                        Some(pdu) => pdu,
+                        None => {
+                            error!("Pdu in state not found: {}", id);
+                            continue;
+                        }
+                    };
+                    lazy_loaded.insert(
+                        UserId::parse(state_key.as_ref())
+                            .expect("they are in timeline_users, so they should be correct"),
+                    );
+                    state_events.push(pdu);
+                }
+            }
+
+            // Reset lazy loading because this is an initial sync
+            db.rooms
+                .lazy_load_reset(&sender_user, &sender_device, &room_id)?;
+
+            // The state_events above should contain all timeline_users, let's mark them as lazy
+            // loaded.
+            db.rooms.lazy_load_mark_sent(
+                &sender_user,
+                &sender_device,
+                &room_id,
+                lazy_loaded,
+                next_batch,
+            );
 
             (
                 heroes,
@@ -387,20 +457,67 @@ async fn sync_helper(
 
             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
-                    .iter()
-                    .filter(|(key, id)| since_state_ids.get(key) != Some(id))
-                    .map(|(_, id)| db.rooms.get_pdu(id))
-                    .filter_map(|r| r.ok().flatten())
-                    .collect()
-            };
+            let mut state_events = Vec::new();
+            let mut lazy_loaded = HashSet::new();
+
+            for (key, id) in current_state_ids {
+                if body.full_state || since_state_ids.get(&key) != Some(&id) {
+                    let pdu = match db.rooms.get_pdu(&id)? {
+                        Some(pdu) => pdu,
+                        None => {
+                            error!("Pdu in state not found: {}", id);
+                            continue;
+                        }
+                    };
+
+                    if pdu.kind == EventType::RoomMember {
+                        match UserId::parse(
+                            pdu.state_key
+                                .as_ref()
+                                .expect("State event has state key")
+                                .clone(),
+                        ) {
+                            Ok(state_key_userid) => {
+                                lazy_loaded.insert(state_key_userid);
+                            }
+                            Err(e) => error!("Invalid state key for member event: {}", e),
+                        }
+                    }
+
+                    state_events.push(pdu);
+                }
+            }
+
+            for (_, event) in &timeline_pdus {
+                if lazy_loaded.contains(&event.sender) {
+                    continue;
+                }
+
+                if !db.rooms.lazy_load_was_sent_before(
+                    &sender_user,
+                    &sender_device,
+                    &room_id,
+                    &event.sender,
+                )? || lazy_load_send_redundant
+                {
+                    if let Some(member_event) = db.rooms.room_state_get(
+                        &room_id,
+                        &EventType::RoomMember,
+                        event.sender.as_str(),
+                    )? {
+                        lazy_loaded.insert(event.sender.clone());
+                        state_events.push(member_event);
+                    }
+                }
+            }
+
+            db.rooms.lazy_load_mark_sent(
+                &sender_user,
+                &sender_device,
+                &room_id,
+                lazy_loaded,
+                next_batch,
+            );
 
             let encrypted_room = db
                 .rooms
@@ -765,7 +882,7 @@ async fn sync_helper(
     };
 
     // TODO: Retry the endpoint instead of returning (waiting for #118)
-    if !full_state
+    if !body.full_state
         && response.rooms.is_empty()
         && response.presence.is_empty()
         && response.account_data.is_empty()
@@ -774,7 +891,7 @@ async fn sync_helper(
     {
         // Hang a few seconds so requests are not spammed
         // Stop hanging if new info arrives
-        let mut duration = timeout.unwrap_or_default();
+        let mut duration = body.timeout.unwrap_or_default();
         if duration.as_secs() > 30 {
             duration = Duration::from_secs(30);
         }

src/database.rs

@@ -44,13 +44,15 @@ use self::proxy::ProxyConfig;
 #[derive(Clone, Debug, Deserialize)]
 pub struct Config {
     server_name: Box<ServerName>,
+    #[serde(default = "default_database_backend")]
+    database_backend: String,
     database_path: String,
     #[serde(default = "default_db_cache_capacity_mb")]
     db_cache_capacity_mb: f64,
     #[serde(default = "default_pdu_cache_capacity")]
     pdu_cache_capacity: u32,
-    #[serde(default = "default_sqlite_wal_clean_second_interval")]
-    sqlite_wal_clean_second_interval: u32,
+    #[serde(default = "default_cleanup_second_interval")]
+    cleanup_second_interval: u32,
     #[serde(default = "default_max_request_size")]
     max_request_size: u32,
     #[serde(default = "default_max_concurrent_requests")]
@@ -117,6 +119,10 @@ fn true_fn() -> bool {
     true
 }
 
+fn default_database_backend() -> String {
+    "sqlite".to_owned()
+}
+
 fn default_db_cache_capacity_mb() -> f64 {
     200.0
 }
@@ -125,7 +131,7 @@ fn default_pdu_cache_capacity() -> u32 {
     100_000
 }
 
-fn default_sqlite_wal_clean_second_interval() -> u32 {
+fn default_cleanup_second_interval() -> u32 {
     1 * 60 // every minute
 }
 
@@ -145,17 +151,8 @@ fn default_turn_ttl() -> u64 {
     60 * 60 * 24
 }
 
-#[cfg(feature = "sled")]
-pub type Engine = abstraction::sled::Engine;
-
-#[cfg(feature = "sqlite")]
-pub type Engine = abstraction::sqlite::Engine;
-
-#[cfg(feature = "heed")]
-pub type Engine = abstraction::heed::Engine;
-
 pub struct Database {
-    _db: Arc<Engine>,
+    _db: Arc<dyn DatabaseEngine>,
     pub globals: globals::Globals,
     pub users: users::Users,
     pub uiaa: uiaa::Uiaa,
@@ -183,27 +180,53 @@ impl Database {
         Ok(())
     }
 
-    fn check_sled_or_sqlite_db(config: &Config) -> Result<()> {
-        #[cfg(feature = "backend_sqlite")]
-        {
-            let path = Path::new(&config.database_path);
+    fn check_db_setup(config: &Config) -> Result<()> {
+        let path = Path::new(&config.database_path);
 
-            let sled_exists = path.join("db").exists();
-            let sqlite_exists = path.join("conduit.db").exists();
-            if sled_exists {
-                if sqlite_exists {
-                    // most likely an in-place directory, only warn
-                    warn!("Both sled and sqlite databases are detected in database directory");
-                    warn!("Currently running from the sqlite database, but consider removing sled database files to free up space")
-                } else {
-                    error!(
-                        "Sled database detected, conduit now uses sqlite for database operations"
-                    );
-                    error!("This database must be converted to sqlite, go to https://github.com/ShadowJonathan/conduit_toolbox#conduit_sled_to_sqlite");
-                    return Err(Error::bad_config(
-                        "sled database detected, migrate to sqlite",
-                    ));
-                }
+        let sled_exists = path.join("db").exists();
+        let sqlite_exists = path.join("conduit.db").exists();
+        let rocksdb_exists = path.join("IDENTITY").exists();
+
+        let mut count = 0;
+
+        if sled_exists {
+            count += 1;
+        }
+
+        if sqlite_exists {
+            count += 1;
+        }
+
+        if rocksdb_exists {
+            count += 1;
+        }
+
+        if count > 1 {
+            warn!("Multiple databases at database_path detected");
+            return Ok(());
+        }
+
+        if sled_exists {
+            if config.database_backend != "sled" {
+                return Err(Error::bad_config(
+                    "Found sled at database_path, but is not specified in config.",
+                ));
+            }
+        }
+
+        if sqlite_exists {
+            if config.database_backend != "sqlite" {
+                return Err(Error::bad_config(
+                    "Found sqlite at database_path, but is not specified in config.",
+                ));
+            }
+        }
+
+        if rocksdb_exists {
+            if config.database_backend != "rocksdb" {
+                return Err(Error::bad_config(
+                    "Found rocksdb at database_path, but is not specified in config.",
+                ));
             }
         }
 
@@ -212,14 +235,30 @@ impl Database {
 
     /// Load an existing database or create a new one.
     pub async fn load_or_create(config: &Config) -> Result<Arc<TokioRwLock<Self>>> {
-        Self::check_sled_or_sqlite_db(config)?;
+        Self::check_db_setup(config)?;
 
         if !Path::new(&config.database_path).exists() {
             std::fs::create_dir_all(&config.database_path)
                 .map_err(|_| Error::BadConfig("Database folder doesn't exists and couldn't be created (e.g. due to missing permissions). Please create the database folder yourself."))?;
         }
 
-        let builder = Engine::open(config)?;
+        let builder: Arc<dyn DatabaseEngine> = match &*config.database_backend {
+            "sqlite" => {
+                #[cfg(not(feature = "sqlite"))]
+                return Err(Error::BadConfig("Database backend not found."));
+                #[cfg(feature = "sqlite")]
+                Arc::new(Arc::<abstraction::sqlite::Engine>::open(config)?)
+            }
+            "rocksdb" => {
+                #[cfg(not(feature = "rocksdb"))]
+                return Err(Error::BadConfig("Database backend not found."));
+                #[cfg(feature = "rocksdb")]
+                Arc::new(Arc::<abstraction::rocksdb::Engine>::open(config)?)
+            }
+            _ => {
+                return Err(Error::BadConfig("Database backend not found."));
+            }
+        };
 
         if config.max_request_size < 1024 {
             eprintln!("ERROR: Max request size is less than 1KB. Please increase it.");
@@ -246,6 +285,7 @@ impl Database {
                 userid_masterkeyid: builder.open_tree("userid_masterkeyid")?,
                 userid_selfsigningkeyid: builder.open_tree("userid_selfsigningkeyid")?,
                 userid_usersigningkeyid: builder.open_tree("userid_usersigningkeyid")?,
+                userfilterid_filter: builder.open_tree("userfilterid_filter")?,
                 todeviceid_events: builder.open_tree("todeviceid_events")?,
             },
             uiaa: uiaa::Uiaa {
@@ -286,6 +326,8 @@ impl Database {
                 userroomid_leftstate: builder.open_tree("userroomid_leftstate")?,
                 roomuserid_leftcount: builder.open_tree("roomuserid_leftcount")?,
 
+                lazyloadedids: builder.open_tree("lazyloadedids")?,
+
                 userroomid_notificationcount: builder.open_tree("userroomid_notificationcount")?,
                 userroomid_highlightcount: builder.open_tree("userroomid_highlightcount")?,
 
@@ -321,6 +363,7 @@ impl Database {
                 statekeyshort_cache: Mutex::new(LruCache::new(1_000_000)),
                 our_real_users_cache: RwLock::new(HashMap::new()),
                 appservice_in_room_cache: RwLock::new(HashMap::new()),
+                lazy_load_waiting: Mutex::new(HashMap::new()),
                 stateinfo_cache: Mutex::new(LruCache::new(1000)),
             },
             account_data: account_data::AccountData {
@@ -769,10 +812,7 @@ impl Database {
 
         drop(guard);
 
-        #[cfg(feature = "sqlite")]
-        {
-            Self::start_wal_clean_task(Arc::clone(&db), config).await;
-        }
+        Self::start_cleanup_task(Arc::clone(&db), config).await;
 
         Ok(db)
     }
@@ -910,15 +950,8 @@ impl Database {
         res
     }
 
-    #[cfg(feature = "sqlite")]
-    #[tracing::instrument(skip(self))]
-    pub fn flush_wal(&self) -> Result<()> {
-        self._db.flush_wal()
-    }
-
-    #[cfg(feature = "sqlite")]
     #[tracing::instrument(skip(db, config))]
-    pub async fn start_wal_clean_task(db: Arc<TokioRwLock<Self>>, config: &Config) {
+    pub async fn start_cleanup_task(db: Arc<TokioRwLock<Self>>, config: &Config) {
         use tokio::time::interval;
 
         #[cfg(unix)]
@@ -927,7 +960,7 @@ impl Database {
 
         use std::time::{Duration, Instant};
 
-        let timer_interval = Duration::from_secs(config.sqlite_wal_clean_second_interval as u64);
+        let timer_interval = Duration::from_secs(config.cleanup_second_interval as u64);
 
         tokio::spawn(async move {
             let mut i = interval(timer_interval);
@@ -938,23 +971,23 @@ impl Database {
                 #[cfg(unix)]
                 tokio::select! {
                     _ = i.tick() => {
-                        info!("wal-trunc: Timer ticked");
+                        info!("cleanup: Timer ticked");
                     }
                     _ = s.recv() => {
-                        info!("wal-trunc: Received SIGHUP");
+                        info!("cleanup: Received SIGHUP");
                     }
                 };
                 #[cfg(not(unix))]
                 {
                     i.tick().await;
-                    info!("wal-trunc: Timer ticked")
+                    info!("cleanup: Timer ticked")
                 }
 
                 let start = Instant::now();
-                if let Err(e) = db.read().await.flush_wal() {
-                    error!("wal-trunc: Errored: {}", e);
+                if let Err(e) = db.read().await._db.cleanup() {
+                    error!("cleanup: Errored: {}", e);
                 } else {
-                    info!("wal-trunc: Flushed in {:?}", start.elapsed());
+                    info!("cleanup: Finished in {:?}", start.elapsed());
                 }
             }
         });

src/database/abstraction.rs

@@ -12,10 +12,21 @@ pub mod sqlite;
 #[cfg(feature = "heed")]
 pub mod heed;
 
-pub trait DatabaseEngine: Sized {
-    fn open(config: &Config) -> Result<Arc<Self>>;
-    fn open_tree(self: &Arc<Self>, name: &'static str) -> Result<Arc<dyn Tree>>;
-    fn flush(self: &Arc<Self>) -> Result<()>;
+#[cfg(feature = "rocksdb")]
+pub mod rocksdb;
+
+pub trait DatabaseEngine: Send + Sync {
+    fn open(config: &Config) -> Result<Self>
+    where
+        Self: Sized;
+    fn open_tree(&self, name: &'static str) -> Result<Arc<dyn Tree>>;
+    fn flush(self: &Self) -> Result<()>;
+    fn cleanup(self: &Self) -> Result<()> {
+        Ok(())
+    }
+    fn memory_usage(self: &Self) -> Result<String> {
+        Ok("Current database engine does not support memory usage reporting.".to_string())
+    }
 }
 
 pub trait Tree: Send + Sync {

src/database/abstraction/rocksdb.rs

@@ -0,0 +1,251 @@
+use super::super::Config;
+use super::{DatabaseEngine, Tree};
+use crate::{utils, Result};
+use std::{
+    collections::{hash_map, HashMap},
+    sync::RwLock,
+};
+use std::{future::Future, pin::Pin, sync::Arc};
+use tokio::sync::watch;
+
+pub struct Engine {
+    rocks: rocksdb::DBWithThreadMode<rocksdb::MultiThreaded>,
+    cache: rocksdb::Cache,
+    old_cfs: Vec<String>,
+}
+
+pub struct RocksDbEngineTree<'a> {
+    db: Arc<Engine>,
+    name: &'a str,
+    watchers: RwLock<HashMap<Vec<u8>, (watch::Sender<()>, watch::Receiver<()>)>>,
+    write_lock: RwLock<()>,
+}
+
+impl DatabaseEngine for Arc<Engine> {
+    fn open(config: &Config) -> Result<Self> {
+        let mut db_opts = rocksdb::Options::default();
+        db_opts.create_if_missing(true);
+        db_opts.set_max_open_files(512);
+        db_opts.set_compaction_style(rocksdb::DBCompactionStyle::Level);
+        db_opts.set_compression_type(rocksdb::DBCompressionType::Zstd);
+        db_opts.set_target_file_size_base(2 << 22);
+        db_opts.set_max_bytes_for_level_base(2 << 24);
+        db_opts.set_max_bytes_for_level_multiplier(2.0);
+        db_opts.set_num_levels(8);
+        db_opts.set_write_buffer_size(2 << 27);
+
+        let rocksdb_cache =
+            rocksdb::Cache::new_lru_cache((config.db_cache_capacity_mb * 1024.0 * 1024.0) as usize)
+                .unwrap();
+
+        let mut block_based_options = rocksdb::BlockBasedOptions::default();
+        block_based_options.set_block_size(2 << 19);
+        block_based_options.set_block_cache(&rocksdb_cache);
+        db_opts.set_block_based_table_factory(&block_based_options);
+
+        let cfs = rocksdb::DBWithThreadMode::<rocksdb::MultiThreaded>::list_cf(
+            &db_opts,
+            &config.database_path,
+        )
+        .unwrap_or_default();
+
+        let db = rocksdb::DBWithThreadMode::<rocksdb::MultiThreaded>::open_cf_descriptors(
+            &db_opts,
+            &config.database_path,
+            cfs.iter().map(|name| {
+                let mut options = rocksdb::Options::default();
+                let prefix_extractor = rocksdb::SliceTransform::create_fixed_prefix(1);
+                options.set_prefix_extractor(prefix_extractor);
+
+                rocksdb::ColumnFamilyDescriptor::new(name, options)
+            }),
+        )?;
+
+        Ok(Arc::new(Engine {
+            rocks: db,
+            cache: rocksdb_cache,
+            old_cfs: cfs,
+        }))
+    }
+
+    fn open_tree(&self, name: &'static str) -> Result<Arc<dyn Tree>> {
+        if !self.old_cfs.contains(&name.to_owned()) {
+            // Create if it didn't exist
+            let mut options = rocksdb::Options::default();
+            let prefix_extractor = rocksdb::SliceTransform::create_fixed_prefix(1);
+            options.set_prefix_extractor(prefix_extractor);
+
+            let _ = self.rocks.create_cf(name, &options);
+        }
+
+        Ok(Arc::new(RocksDbEngineTree {
+            name,
+            db: Arc::clone(self),
+            watchers: RwLock::new(HashMap::new()),
+            write_lock: RwLock::new(()),
+        }))
+    }
+
+    fn flush(&self) -> Result<()> {
+        // TODO?
+        Ok(())
+    }
+
+    fn memory_usage(&self) -> Result<String> {
+        let stats =
+            rocksdb::perf::get_memory_usage_stats(Some(&[&self.rocks]), Some(&[&self.cache]))?;
+        Ok(format!(
+            "Approximate memory usage of all the mem-tables: {:.3} MB\n\
+                   Approximate memory usage of un-flushed mem-tables: {:.3} MB\n\
+                   Approximate memory usage of all the table readers: {:.3} MB\n\
+                   Approximate memory usage by cache: {:.3} MB",
+            stats.mem_table_total as f64 / 1024.0 / 1024.0,
+            stats.mem_table_unflushed as f64 / 1024.0 / 1024.0,
+            stats.mem_table_readers_total as f64 / 1024.0 / 1024.0,
+            stats.cache_total as f64 / 1024.0 / 1024.0
+        ))
+    }
+}
+
+impl RocksDbEngineTree<'_> {
+    fn cf(&self) -> Arc<rocksdb::BoundColumnFamily<'_>> {
+        self.db.rocks.cf_handle(self.name).unwrap()
+    }
+}
+
+impl Tree for RocksDbEngineTree<'_> {
+    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
+        Ok(self.db.rocks.get_cf(&self.cf(), key)?)
+    }
+
+    fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> {
+        let lock = self.write_lock.read().unwrap();
+
+        let result = self.db.rocks.put_cf(&self.cf(), key, value)?;
+
+        drop(lock);
+
+        let watchers = self.watchers.read().unwrap();
+        let mut triggered = Vec::new();
+
+        for length in 0..=key.len() {
+            if watchers.contains_key(&key[..length]) {
+                triggered.push(&key[..length]);
+            }
+        }
+
+        drop(watchers);
+
+        if !triggered.is_empty() {
+            let mut watchers = self.watchers.write().unwrap();
+            for prefix in triggered {
+                if let Some(tx) = watchers.remove(prefix) {
+                    let _ = tx.0.send(());
+                }
+            }
+        };
+
+        Ok(result)
+    }
+
+    fn insert_batch<'a>(&self, iter: &mut dyn Iterator<Item = (Vec<u8>, Vec<u8>)>) -> Result<()> {
+        for (key, value) in iter {
+            self.db.rocks.put_cf(&self.cf(), key, value)?;
+        }
+
+        Ok(())
+    }
+
+    fn remove(&self, key: &[u8]) -> Result<()> {
+        Ok(self.db.rocks.delete_cf(&self.cf(), key)?)
+    }
+
+    fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> {
+        Box::new(
+            self.db
+                .rocks
+                .iterator_cf(&self.cf(), rocksdb::IteratorMode::Start)
+                .map(|(k, v)| (Vec::from(k), Vec::from(v))),
+        )
+    }
+
+    fn iter_from<'a>(
+        &'a self,
+        from: &[u8],
+        backwards: bool,
+    ) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> {
+        Box::new(
+            self.db
+                .rocks
+                .iterator_cf(
+                    &self.cf(),
+                    rocksdb::IteratorMode::From(
+                        from,
+                        if backwards {
+                            rocksdb::Direction::Reverse
+                        } else {
+                            rocksdb::Direction::Forward
+                        },
+                    ),
+                )
+                .map(|(k, v)| (Vec::from(k), Vec::from(v))),
+        )
+    }
+
+    fn increment(&self, key: &[u8]) -> Result<Vec<u8>> {
+        let lock = self.write_lock.write().unwrap();
+
+        let old = self.db.rocks.get_cf(&self.cf(), &key)?;
+        let new = utils::increment(old.as_deref()).unwrap();
+        self.db.rocks.put_cf(&self.cf(), key, &new)?;
+
+        drop(lock);
+        Ok(new)
+    }
+
+    fn increment_batch<'a>(&self, iter: &mut dyn Iterator<Item = Vec<u8>>) -> Result<()> {
+        let lock = self.write_lock.write().unwrap();
+
+        for key in iter {
+            let old = self.db.rocks.get_cf(&self.cf(), &key)?;
+            let new = utils::increment(old.as_deref()).unwrap();
+            self.db.rocks.put_cf(&self.cf(), key, new)?;
+        }
+
+        drop(lock);
+
+        Ok(())
+    }
+
+    fn scan_prefix<'a>(
+        &'a self,
+        prefix: Vec<u8>,
+    ) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> {
+        Box::new(
+            self.db
+                .rocks
+                .iterator_cf(
+                    &self.cf(),
+                    rocksdb::IteratorMode::From(&prefix, rocksdb::Direction::Forward),
+                )
+                .map(|(k, v)| (Vec::from(k), Vec::from(v)))
+                .take_while(move |(k, _)| k.starts_with(&prefix)),
+        )
+    }
+
+    fn watch_prefix<'a>(&'a self, prefix: &[u8]) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
+        let mut rx = match self.watchers.write().unwrap().entry(prefix.to_vec()) {
+            hash_map::Entry::Occupied(o) => o.get().1.clone(),
+            hash_map::Entry::Vacant(v) => {
+                let (tx, rx) = tokio::sync::watch::channel(());
+                v.insert((tx, rx.clone()));
+                rx
+            }
+        };
+
+        Box::pin(async move {
+            // Tx is never destroyed
+            rx.changed().await.unwrap();
+        })
+    }
+}

src/database/abstraction/sqlite.rs

@@ -82,8 +82,8 @@ impl Engine {
     }
 }
 
-impl DatabaseEngine for Engine {
-    fn open(config: &Config) -> Result<Arc<Self>> {
+impl DatabaseEngine for Arc<Engine> {
+    fn open(config: &Config) -> Result<Self> {
         let path = Path::new(&config.database_path).join("conduit.db");
 
         // calculates cache-size per permanent connection
@@ -94,7 +94,7 @@ impl DatabaseEngine for Engine {
             / ((num_cpus::get().max(1) * 2) + 1) as f64)
             as u32;
 
-        let writer = Mutex::new(Self::prepare_conn(&path, cache_size_per_thread)?);
+        let writer = Mutex::new(Engine::prepare_conn(&path, cache_size_per_thread)?);
 
         let arc = Arc::new(Engine {
             writer,
@@ -107,7 +107,7 @@ impl DatabaseEngine for Engine {
         Ok(arc)
     }
 
-    fn open_tree(self: &Arc<Self>, name: &str) -> Result<Arc<dyn Tree>> {
+    fn open_tree(&self, name: &str) -> Result<Arc<dyn Tree>> {
         self.write_lock().execute(&format!("CREATE TABLE IF NOT EXISTS {} ( \"key\" BLOB PRIMARY KEY, \"value\" BLOB NOT NULL )", name), [])?;
 
         Ok(Arc::new(SqliteTable {
@@ -117,10 +117,14 @@ impl DatabaseEngine for Engine {
         }))
     }
 
-    fn flush(self: &Arc<Self>) -> Result<()> {
+    fn flush(&self) -> Result<()> {
         // we enabled PRAGMA synchronous=normal, so this should not be necessary
         Ok(())
     }
+
+    fn cleanup(&self) -> Result<()> {
+        self.flush_wal()
+    }
 }
 
 pub struct SqliteTable {
@@ -134,7 +138,7 @@ type TupleOfBytes = (Vec<u8>, Vec<u8>);
 impl SqliteTable {
     #[tracing::instrument(skip(self, guard, key))]
     fn get_with_guard(&self, guard: &Connection, key: &[u8]) -> Result<Option<Vec<u8>>> {
-        //dbg!(&self.name);
+        dbg!(&self.name);
         Ok(guard
             .prepare(format!("SELECT value FROM {} WHERE key = ?", self.name).as_str())?
             .query_row([key], |row| row.get(0))
@@ -143,7 +147,7 @@ impl SqliteTable {
 
     #[tracing::instrument(skip(self, guard, key, value))]
     fn insert_with_guard(&self, guard: &Connection, key: &[u8], value: &[u8]) -> Result<()> {
-        //dbg!(&self.name);
+        dbg!(&self.name);
         guard.execute(
             format!(
                 "INSERT OR REPLACE INTO {} (key, value) VALUES (?, ?)",
@@ -170,14 +174,14 @@ impl SqliteTable {
 
         let statement_ref = NonAliasingBox(statement);
 
-        //let name = self.name.clone();
+        let name = self.name.clone();
 
         let iterator = Box::new(
             statement
                 .query_map([], |row| Ok((row.get_unwrap(0), row.get_unwrap(1))))
                 .unwrap()
                 .map(move |r| {
-                    //dbg!(&name);
+                    dbg!(&name);
                     r.unwrap()
                 }),
         );
@@ -285,7 +289,7 @@ impl Tree for SqliteTable {
         let guard = self.engine.read_lock_iterator();
         let from = from.to_vec(); // TODO change interface?
 
-        //let name = self.name.clone();
+        let name = self.name.clone();
 
         if backwards {
             let statement = Box::leak(Box::new(
@@ -304,7 +308,7 @@ impl Tree for SqliteTable {
                     .query_map([from], |row| Ok((row.get_unwrap(0), row.get_unwrap(1))))
                     .unwrap()
                     .map(move |r| {
-                        //dbg!(&name);
+                        dbg!(&name);
                         r.unwrap()
                     }),
             );
@@ -329,7 +333,7 @@ impl Tree for SqliteTable {
                     .query_map([from], |row| Ok((row.get_unwrap(0), row.get_unwrap(1))))
                     .unwrap()
                     .map(move |r| {
-                        //dbg!(&name);
+                        dbg!(&name);
                         r.unwrap()
                     }),
             );

src/database/admin.rs

@@ -13,6 +13,7 @@ use tracing::warn;
 pub enum AdminCommand {
     RegisterAppservice(serde_yaml::Value),
     ListAppservices,
+    ShowMemoryUsage,
     SendMessage(RoomMessageEventContent),
 }
 
@@ -109,6 +110,13 @@ impl Admin {
                                     send_message(RoomMessageEventContent::text_plain("Failed to get appservices."), guard, &state_lock);
                                 }
                             }
+                            AdminCommand::ShowMemoryUsage => {
+                                if let Ok(response) = guard._db.memory_usage() {
+                                    send_message(RoomMessageEventContent::text_plain(response), guard, &state_lock);
+                                } else {
+                                    send_message(RoomMessageEventContent::text_plain("Failed to get database memory usage.".to_string()), guard, &state_lock);
+                                }
+                            }
                             AdminCommand::SendMessage(message) => {
                                 send_message(message, guard, &state_lock);
                             }

src/database/rooms.rs

@@ -28,7 +28,7 @@ use ruma::{
     push::{Action, Ruleset, Tweak},
     serde::{CanonicalJsonObject, CanonicalJsonValue, Raw},
     state_res::{self, RoomVersion, StateMap},
-    uint, EventId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
+    uint, DeviceId, EventId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
 };
 use serde::Deserialize;
 use serde_json::value::to_raw_value;
@@ -79,6 +79,8 @@ pub struct Rooms {
     pub(super) userroomid_leftstate: Arc<dyn Tree>,
     pub(super) roomuserid_leftcount: Arc<dyn Tree>,
 
+    pub(super) lazyloadedids: Arc<dyn Tree>, // LazyLoadedIds = UserId + DeviceId + RoomId + LazyLoadedUserId
+
     pub(super) userroomid_notificationcount: Arc<dyn Tree>, // NotifyCount = u64
     pub(super) userroomid_highlightcount: Arc<dyn Tree>,    // HightlightCount = u64
 
@@ -117,6 +119,8 @@ pub struct Rooms {
     pub(super) shortstatekey_cache: Mutex<LruCache<u64, (EventType, String)>>,
     pub(super) our_real_users_cache: RwLock<HashMap<Box<RoomId>, Arc<HashSet<Box<UserId>>>>>,
     pub(super) appservice_in_room_cache: RwLock<HashMap<Box<RoomId>, HashMap<String, bool>>>,
+    pub(super) lazy_load_waiting:
+        Mutex<HashMap<(Box<UserId>, Box<DeviceId>, Box<RoomId>, u64), HashSet<Box<UserId>>>>,
     pub(super) stateinfo_cache: Mutex<
         LruCache<
             u64,
@@ -1676,6 +1680,9 @@ impl Rooms {
                                         ));
                                     }
                                 }
+                                "database_memory_usage" => {
+                                    db.admin.send(AdminCommand::ShowMemoryUsage);
+                                }
                                 _ => {
                                     db.admin.send(AdminCommand::SendMessage(
                                         RoomMessageEventContent::text_plain(format!(
@@ -3453,4 +3460,94 @@ impl Rooms {
 
         Ok(())
     }
+
+    #[tracing::instrument(skip(self))]
+    pub fn lazy_load_was_sent_before(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+        ll_user: &UserId,
+    ) -> Result<bool> {
+        let mut key = user_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(&device_id.as_bytes());
+        key.push(0xff);
+        key.extend_from_slice(&room_id.as_bytes());
+        key.push(0xff);
+        key.extend_from_slice(&ll_user.as_bytes());
+        Ok(self.lazyloadedids.get(&key)?.is_some())
+    }
+
+    #[tracing::instrument(skip(self))]
+    pub fn lazy_load_mark_sent(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+        lazy_load: HashSet<Box<UserId>>,
+        count: u64,
+    ) {
+        self.lazy_load_waiting.lock().unwrap().insert(
+            (
+                user_id.to_owned(),
+                device_id.to_owned(),
+                room_id.to_owned(),
+                count,
+            ),
+            lazy_load,
+        );
+    }
+
+    #[tracing::instrument(skip(self))]
+    pub fn lazy_load_confirm_delivery(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+        since: u64,
+    ) -> Result<()> {
+        if let Some(user_ids) = self.lazy_load_waiting.lock().unwrap().remove(&(
+            user_id.to_owned(),
+            device_id.to_owned(),
+            room_id.to_owned(),
+            since,
+        )) {
+            let mut prefix = user_id.as_bytes().to_vec();
+            prefix.push(0xff);
+            prefix.extend_from_slice(&device_id.as_bytes());
+            prefix.push(0xff);
+            prefix.extend_from_slice(&room_id.as_bytes());
+            prefix.push(0xff);
+
+            for ll_id in user_ids {
+                let mut key = prefix.clone();
+                key.extend_from_slice(&ll_id.as_bytes());
+                self.lazyloadedids.insert(&key, &[])?;
+            }
+        }
+
+        Ok(())
+    }
+
+    #[tracing::instrument(skip(self))]
+    pub fn lazy_load_reset(
+        &self,
+        user_id: &Box<UserId>,
+        device_id: &Box<DeviceId>,
+        room_id: &Box<RoomId>,
+    ) -> Result<()> {
+        let mut prefix = user_id.as_bytes().to_vec();
+        prefix.push(0xff);
+        prefix.extend_from_slice(&device_id.as_bytes());
+        prefix.push(0xff);
+        prefix.extend_from_slice(&room_id.as_bytes());
+        prefix.push(0xff);
+
+        for (key, _) in self.lazyloadedids.scan_prefix(prefix) {
+            self.lazyloadedids.remove(&key)?;
+        }
+
+        Ok(())
+    }
 }

src/database/users.rs

@@ -1,6 +1,9 @@
 use crate::{utils, Error, Result};
 use ruma::{
-    api::client::{error::ErrorKind, r0::device::Device},
+    api::client::{
+        error::ErrorKind,
+        r0::{device::Device, filter::IncomingFilterDefinition},
+    },
     encryption::{CrossSigningKey, DeviceKeys, OneTimeKey},
     events::{AnyToDeviceEvent, EventType},
     identifiers::MxcUri,
@@ -31,6 +34,8 @@ pub struct Users {
     pub(super) userid_selfsigningkeyid: Arc<dyn Tree>,
     pub(super) userid_usersigningkeyid: Arc<dyn Tree>,
 
+    pub(super) userfilterid_filter: Arc<dyn Tree>, // UserFilterId = UserId + FilterId
+
     pub(super) todeviceid_events: Arc<dyn Tree>, // ToDeviceId = UserId + DeviceId + Count
 }
 
@@ -990,4 +995,45 @@ impl Users {
         // TODO: Unhook 3PID
         Ok(())
     }
+
+    /// Creates a new sync filter. Returns the filter id.
+    #[tracing::instrument(skip(self))]
+    pub fn create_filter(
+        &self,
+        user_id: &UserId,
+        filter: &IncomingFilterDefinition,
+    ) -> Result<String> {
+        let filter_id = utils::random_string(4);
+
+        let mut key = user_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(filter_id.as_bytes());
+
+        self.userfilterid_filter.insert(
+            &key,
+            &serde_json::to_vec(&filter).expect("filter is valid json"),
+        )?;
+
+        Ok(filter_id)
+    }
+
+    #[tracing::instrument(skip(self))]
+    pub fn get_filter(
+        &self,
+        user_id: &UserId,
+        filter_id: &str,
+    ) -> Result<Option<IncomingFilterDefinition>> {
+        let mut key = user_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(filter_id.as_bytes());
+
+        let raw = self.userfilterid_filter.get(&key)?;
+
+        if let Some(raw) = raw {
+            serde_json::from_slice(&raw)
+                .map_err(|_| Error::bad_database("Invalid filter event in db."))
+        } else {
+            Ok(None)
+        }
+    }
 }

src/error.rs

@@ -39,6 +39,12 @@ pub enum Error {
     #[cfg(feature = "heed")]
     #[error("There was a problem with the connection to the heed database: {error}")]
     HeedError { error: String },
+    #[cfg(feature = "rocksdb")]
+    #[error("There was a problem with the connection to the rocksdb database: {source}")]
+    RocksDbError {
+        #[from]
+        source: rocksdb::Error,
+    },
     #[error("Could not generate an image.")]
     ImageError {
         #[from]

src/server_server.rs

@@ -1392,12 +1392,11 @@ async fn upgrade_outlier_to_timeline_pdu(
                 let mut starting_events = Vec::with_capacity(leaf_state.len());
 
                 for (k, id) in leaf_state {
-                    let k = db
-                        .rooms
-                        .get_statekey_from_short(k)
-                        .map_err(|_| "Failed to get_statekey_from_short.".to_owned())?;
-
-                    state.insert(k, id.clone());
+                    if let Ok(k) = db.rooms.get_statekey_from_short(k) {
+                        state.insert(k, id.clone());
+                    } else {
+                        warn!("Failed to get_statekey_from_short.");
+                    }
                     starting_events.push(id);
                 }
 
@@ -1687,25 +1686,6 @@ async fn upgrade_outlier_to_timeline_pdu(
         // We do this by adding the current state to the list of fork states
         extremity_sstatehashes.remove(&current_sstatehash);
         fork_states.push(current_state_ids);
-        dbg!(&extremity_sstatehashes);
-
-        for (sstatehash, leaf_pdu) in extremity_sstatehashes {
-            let mut leaf_state = db
-                .rooms
-                .state_full_ids(sstatehash)
-                .map_err(|_| "Failed to ask db for room state.".to_owned())?;
-
-            if let Some(state_key) = &leaf_pdu.state_key {
-                let shortstatekey = db
-                    .rooms
-                    .get_or_create_shortstatekey(&leaf_pdu.kind, state_key, &db.globals)
-                    .map_err(|_| "Failed to create shortstatekey.".to_owned())?;
-                leaf_state.insert(shortstatekey, Arc::from(&*leaf_pdu.event_id));
-                // Now it's the state after the pdu
-            }
-
-            fork_states.push(leaf_state);
-        }
 
         // We also add state after incoming event to the fork states
         let mut state_after = state_at_incoming_event.clone();
@@ -1755,11 +1735,16 @@ async fn upgrade_outlier_to_timeline_pdu(
                 .into_iter()
                 .map(|map| {
                     map.into_iter()
-                        .map(|(k, id)| db.rooms.get_statekey_from_short(k).map(|k| (k, id)))
-                        .collect::<Result<StateMap<_>>>()
+                        .filter_map(|(k, id)| {
+                            db.rooms
+                                .get_statekey_from_short(k)
+                                .map(|k| (k, id))
+                                .map_err(|e| warn!("Failed to get_statekey_from_short: {}", e))
+                                .ok()
+                        })
+                        .collect::<StateMap<_>>()
                 })
-                .collect::<Result<_>>()
-                .map_err(|_| "Failed to get_statekey_from_short.".to_owned())?;
+                .collect();
 
             let state = match state_res::resolve(
                 room_version_id,
@@ -1871,73 +1856,104 @@ pub(crate) fn fetch_and_handle_outliers<'a>(
             // a. Look in the main timeline (pduid_pdu tree)
             // b. Look at outlier pdu tree
             // (get_pdu_json checks both)
-            let local_pdu = db.rooms.get_pdu(id);
-            let pdu = match local_pdu {
-                Ok(Some(pdu)) => {
-                    trace!("Found {} in db", id);
-                    (pdu, None)
-                }
-                Ok(None) => {
-                    // c. Ask origin server over federation
-                    warn!("Fetching {} over federation.", id);
-                    match db
-                        .sending
-                        .send_federation_request(
-                            &db.globals,
-                            origin,
-                            get_event::v1::Request { event_id: id },
-                        )
-                        .await
-                    {
-                        Ok(res) => {
-                            warn!("Got {} over federation", id);
-                            let (calculated_event_id, value) =
-                                match crate::pdu::gen_event_id_canonical_json(&res.pdu) {
-                                    Ok(t) => t,
-                                    Err(_) => {
-                                        back_off((**id).to_owned());
-                                        continue;
-                                    }
-                                };
+            if let Ok(Some(local_pdu)) = db.rooms.get_pdu(id) {
+                trace!("Found {} in db", id);
+                pdus.push((local_pdu, None));
+                continue;
+            }
 
-                            if calculated_event_id != **id {
-                                warn!("Server didn't return event id we requested: requested: {}, we got {}. Event: {:?}",
-                                    id, calculated_event_id, &res.pdu);
-                            }
-
-                            // This will also fetch the auth chain
-                            match handle_outlier_pdu(
-                                origin,
-                                create_event,
-                                id,
-                                room_id,
-                                value.clone(),
-                                db,
-                                pub_key_map,
-                            )
-                            .await
-                            {
-                                Ok((pdu, json)) => (pdu, Some(json)),
-                                Err(e) => {
-                                    warn!("Authentication of event {} failed: {:?}", id, e);
-                                    back_off((**id).to_owned());
-                                    continue;
-                                }
-                            }
-                        }
-                        Err(_) => {
-                            warn!("Failed to fetch event: {}", id);
-                            back_off((**id).to_owned());
-                            continue;
-                        }
-                    }
-                }
-                Err(e) => {
-                    warn!("Error loading {}: {}", id, e);
+            // c. Ask origin server over federation
+            // We also handle its auth chain here so we don't get a stack overflow in
+            // handle_outlier_pdu.
+            let mut todo_auth_events = vec![Arc::clone(id)];
+            let mut events_in_reverse_order = Vec::new();
+            let mut events_all = HashSet::new();
+            while let Some(next_id) = todo_auth_events.pop() {
+                if events_all.contains(&next_id) {
                     continue;
                 }
-            };
-            pdus.push(pdu);
+
+                if let Ok(Some(_)) = db.rooms.get_pdu(&next_id) {
+                    trace!("Found {} in db", id);
+                    continue;
+                }
+
+                warn!("Fetching {} over federation.", next_id);
+                match db
+                    .sending
+                    .send_federation_request(
+                        &db.globals,
+                        origin,
+                        get_event::v1::Request { event_id: &next_id },
+                    )
+                    .await
+                {
+                    Ok(res) => {
+                        warn!("Got {} over federation", next_id);
+                        let (calculated_event_id, value) =
+                            match crate::pdu::gen_event_id_canonical_json(&res.pdu) {
+                                Ok(t) => t,
+                                Err(_) => {
+                                    back_off((*next_id).to_owned());
+                                    continue;
+                                }
+                            };
+
+                        if calculated_event_id != *next_id {
+                            warn!("Server didn't return event id we requested: requested: {}, we got {}. Event: {:?}",
+                                next_id, calculated_event_id, &res.pdu);
+                        }
+
+                        if let Some(auth_events) =
+                            value.get("auth_events").and_then(|c| c.as_array())
+                        {
+                            for auth_event in auth_events {
+                                if let Ok(auth_event) =
+                                    serde_json::from_value(auth_event.clone().into())
+                                {
+                                    let a: Arc<EventId> = auth_event;
+                                    todo_auth_events.push(a);
+                                } else {
+                                    warn!("Auth event id is not valid");
+                                }
+                            }
+                        } else {
+                            warn!("Auth event list invalid");
+                        }
+
+                        events_in_reverse_order.push((next_id.clone(), value));
+                        events_all.insert(next_id);
+                    }
+                    Err(_) => {
+                        warn!("Failed to fetch event: {}", next_id);
+                        back_off((*next_id).to_owned());
+                    }
+                }
+            }
+
+            for (next_id, value) in events_in_reverse_order.iter().rev() {
+                match handle_outlier_pdu(
+                    origin,
+                    create_event,
+                    &next_id,
+                    room_id,
+                    value.clone(),
+                    db,
+                    pub_key_map,
+                )
+                .await
+                {
+                    Ok((pdu, json)) => {
+                        if next_id == id {
+                            pdus.push((pdu, Some(json)));
+                        }
+                    }
+                    Err(e) => {
+                        warn!("Authentication of event {} failed: {:?}", next_id, e);
+                        back_off((**next_id).to_owned());
+                    }
+                }
+            }
         }
         pdus
     })