improvement: copy watcher deduplication from sqlite

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.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c749134fda8bfc90d0de643d59bfc841dcb3ac8a1062e12b6754bd60235c48b3"
+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.16.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_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
 

src/database.rs

@@ -154,6 +154,9 @@ pub type Engine = abstraction::sqlite::Engine;
 #[cfg(feature = "heed")]
 pub type Engine = abstraction::heed::Engine;
 
+#[cfg(feature = "rocksdb")]
+pub type Engine = abstraction::rocksdb::Engine;
+
 pub struct Database {
     _db: Arc<Engine>,
     pub globals: globals::Globals,

src/database/abstraction.rs

@@ -12,6 +12,9 @@ pub mod sqlite;
 #[cfg(feature = "heed")]
 pub mod heed;
 
+#[cfg(feature = "rocksdb")]
+pub mod rocksdb;
+
 pub trait DatabaseEngine: Sized {
     fn open(config: &Config) -> Result<Arc<Self>>;
     fn open_tree(self: &Arc<Self>, name: &'static str) -> Result<Arc<dyn Tree>>;

src/database/abstraction/rocksdb.rs

@@ -0,0 +1,234 @@
+use super::super::Config;
+use crate::{utils, Result};
+use std::{future::Future, pin::Pin, sync::Arc};
+use std::{
+    collections::{hash_map, HashMap},
+    sync::RwLock};
+use super::{DatabaseEngine, Tree};
+use tokio::sync::watch;
+
+pub struct Engine {
+    rocks: rocksdb::DBWithThreadMode<rocksdb::MultiThreaded>,
+    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 Engine {
+    fn open(config: &Config) -> Result<Arc<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,
+            old_cfs: cfs,
+        }))
+    }
+
+    fn open_tree(self: &Arc<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);
+            println!("created cf");
+        }
+
+        Ok(Arc::new(RocksDbEngineTree {
+            name,
+            db: Arc::clone(self),
+            watchers: RwLock::new(HashMap::new()),
+            write_lock: RwLock::new(()),
+        }))
+    }
+
+    fn flush(self: &Arc<Self>) -> Result<()> {
+        // TODO?
+        Ok(())
+    }
+}
+
+impl RocksDbEngineTree<'_> {
+    fn cf(&self) -> 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

@@ -134,7 +134,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 +143,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 +170,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 +285,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 +304,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 +329,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/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())?;
-
+                    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::<Result<_>>()
-                .map_err(|_| "Failed to get_statekey_from_short.".to_owned())?;
+                        .collect::<StateMap<_>>()
+                })
+                .collect();
 
             let state = match state_res::resolve(
                 room_version_id,
@@ -1871,45 +1856,86 @@ 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)) => {
+            if let Ok(Some(local_pdu)) = db.rooms.get_pdu(id) {
                 trace!("Found {} in db", id);
-                    (pdu, None)
+                pdus.push((local_pdu, None));
+                continue;
             }
-                Ok(None) => {
+
             // c. Ask origin server over federation
-                    warn!("Fetching {} over federation.", id);
+            // 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;
+                }
+
+                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: id },
+                        get_event::v1::Request { event_id: &next_id },
                     )
                     .await
                 {
                     Ok(res) => {
-                            warn!("Got {} over federation", id);
+                        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((**id).to_owned());
+                                    back_off((*next_id).to_owned());
                                     continue;
                                 }
                             };
 
-                            if calculated_event_id != **id {
+                        if calculated_event_id != *next_id {
                             warn!("Server didn't return event id we requested: requested: {}, we got {}. Event: {:?}",
-                                    id, calculated_event_id, &res.pdu);
+                                next_id, calculated_event_id, &res.pdu);
                         }
 
-                            // This will also fetch the auth chain
+                        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,
-                                id,
+                    &next_id,
                     room_id,
                     value.clone(),
                     db,
@@ -1917,27 +1943,17 @@ pub(crate) fn fetch_and_handle_outliers<'a>(
                 )
                 .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;
-                        }
+                    Ok((pdu, json)) => {
+                        if next_id == id {
+                            pdus.push((pdu, Some(json)));
                         }
                     }
                     Err(e) => {
-                    warn!("Error loading {}: {}", id, e);
-                    continue;
+                        warn!("Authentication of event {} failed: {:?}", next_id, e);
+                        back_off((**next_id).to_owned());
+                    }
+                }
             }
-            };
-            pdus.push(pdu);
         }
         pdus
     })

src/utils.rs

@@ -29,6 +29,17 @@ pub fn increment(old: Option<&[u8]>) -> Option<Vec<u8>> {
     Some(number.to_be_bytes().to_vec())
 }
 
+#[cfg(feature = "rocksdb")]
+pub fn increment_rocksdb(
+    _new_key: &[u8],
+    old: Option<&[u8]>,
+    _operands: &mut rocksdb::MergeOperands,
+) -> Option<Vec<u8>> {
+    dbg!(_new_key);
+    dbg!(old);
+    increment(old)
+}
+
 pub fn generate_keypair() -> Vec<u8> {
     let mut value = random_string(8).as_bytes().to_vec();
     value.push(0xff);