feat: more admin commands, better logging

src/database/admin.rs

@@ -231,9 +231,15 @@ enum AdminCommand {
     /// List all the currently registered appservices
     ListAppservices,
 
+    /// List all rooms the server knows about
+    ListRooms,
+
     /// List users in the database
     ListLocalUsers,
 
+    /// List all rooms we are currently handling an incoming pdu from
+    IncomingFederation,
+
     /// Get the auth_chain of a PDU
     GetAuthChain {
         /// An event ID (the $ character followed by the base64 reference hash)
@@ -269,6 +275,7 @@ enum AdminCommand {
         /// Username of the user for whom the password should be reset
         username: String,
     },
+
     /// Create a new user
     CreateUser {
         /// Username of the new user
@@ -276,6 +283,11 @@ enum AdminCommand {
         /// Password of the new user, if unspecified one is generated
         password: Option<String>,
     },
+
+    /// Disables incoming federation handling for a room.
+    DisableRoom { room_id: Box<RoomId> },
+    /// Enables incoming federation handling for a room again.
+    EnableRoom { room_id: Box<RoomId> },
 }
 
 fn process_admin_command(
@@ -336,6 +348,26 @@ fn process_admin_command(
                 RoomMessageEventContent::text_plain("Failed to get appservices.")
             }
         }
+        AdminCommand::ListRooms => {
+            let room_ids = db.rooms.iter_ids();
+            let output = format!(
+                "Rooms:\n{}",
+                room_ids
+                    .filter_map(|r| r.ok())
+                    .map(|id| id.to_string()
+                        + "\tMembers: "
+                        + &db
+                            .rooms
+                            .room_joined_count(&id)
+                            .ok()
+                            .flatten()
+                            .unwrap_or(0)
+                            .to_string())
+                    .collect::<Vec<_>>()
+                    .join("\n")
+            );
+            RoomMessageEventContent::text_plain(output)
+        }
         AdminCommand::ListLocalUsers => match db.users.list_local_users() {
             Ok(users) => {
                 let mut msg: String = format!("Found {} local user account(s):\n", users.len());
@@ -344,6 +376,22 @@ fn process_admin_command(
             }
             Err(e) => RoomMessageEventContent::text_plain(e.to_string()),
         },
+        AdminCommand::IncomingFederation => {
+            let map = db.globals.roomid_federationhandletime.read().unwrap();
+            let mut msg: String = format!("Handling {} incoming pdus:\n", map.len());
+
+            for (r, (e, i)) in map.iter() {
+                let elapsed = i.elapsed();
+                msg += &format!(
+                    "{} {}: {}m{}s\n",
+                    r,
+                    e,
+                    elapsed.as_secs() / 60,
+                    elapsed.as_secs() % 60
+                );
+            }
+            RoomMessageEventContent::text_plain(&msg)
+        }
         AdminCommand::GetAuthChain { event_id } => {
             let event_id = Arc::<EventId>::from(event_id);
             if let Some(event) = db.rooms.get_pdu_json(&event_id)? {
@@ -545,6 +593,14 @@ fn process_admin_command(
                 "Created user with user_id: {user_id} and password: {password}"
             ))
         }
+        AdminCommand::DisableRoom { room_id } => {
+            db.rooms.disabledroomids.insert(room_id.as_bytes(), &[])?;
+            RoomMessageEventContent::text_plain("Room disabled.")
+        }
+        AdminCommand::EnableRoom { room_id } => {
+            db.rooms.disabledroomids.remove(room_id.as_bytes())?;
+            RoomMessageEventContent::text_plain("Room enabled.")
+        }
     };
 
     Ok(reply_message_content)

src/database/globals.rs

@@ -52,6 +52,8 @@ pub struct Globals {
     pub roomid_mutex_insert: RwLock<HashMap<Box<RoomId>, Arc<Mutex<()>>>>,
     pub roomid_mutex_state: RwLock<HashMap<Box<RoomId>, Arc<TokioMutex<()>>>>,
     pub roomid_mutex_federation: RwLock<HashMap<Box<RoomId>, Arc<TokioMutex<()>>>>, // this lock will be held longer
+    pub roomid_federationhandletime: RwLock<HashMap<Box<RoomId>, (Box<EventId>, Instant)>>,
+    pub stateres_mutex: Arc<Mutex<()>>,
     pub rotate: RotationHandler,
 }
 
@@ -183,6 +185,8 @@ impl Globals {
             roomid_mutex_state: RwLock::new(HashMap::new()),
             roomid_mutex_insert: RwLock::new(HashMap::new()),
             roomid_mutex_federation: RwLock::new(HashMap::new()),
+            roomid_federationhandletime: RwLock::new(HashMap::new()),
+            stateres_mutex: Arc::new(Mutex::new(())),
             sync_receivers: RwLock::new(HashMap::new()),
             rotate: RotationHandler::new(),
         };

src/database/rooms.rs

@@ -76,6 +76,8 @@ pub struct Rooms {
     pub(super) userroomid_leftstate: Arc<dyn Tree>,
     pub(super) roomuserid_leftcount: Arc<dyn Tree>,
 
+    pub(super) disabledroomids: Arc<dyn Tree>, // Rooms where incoming federation handling is disabled
+
     pub(super) lazyloadedids: Arc<dyn Tree>, // LazyLoadedIds = UserId + DeviceId + RoomId + LazyLoadedUserId
 
     pub(super) userroomid_notificationcount: Arc<dyn Tree>, // NotifyCount = u64
@@ -2858,6 +2860,18 @@ impl Rooms {
         Ok(self.publicroomids.get(room_id.as_bytes())?.is_some())
     }
 
+    #[tracing::instrument(skip(self))]
+    pub fn iter_ids(&self) -> impl Iterator<Item = Result<Box<RoomId>>> + '_ {
+        self.roomid_shortroomid.iter().map(|(bytes, _)| {
+            RoomId::parse(
+                utils::string_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Room ID in publicroomids is invalid unicode.")
+                })?,
+            )
+            .map_err(|_| Error::bad_database("Room ID in roomid_shortroomid is invalid."))
+        })
+    }
+
     #[tracing::instrument(skip(self))]
     pub fn public_rooms(&self) -> impl Iterator<Item = Result<Box<RoomId>>> + '_ {
         self.publicroomids.iter().map(|(bytes, _)| {
@@ -3140,6 +3154,10 @@ impl Rooms {
             .transpose()
     }
 
+    pub fn is_disabled(&self, room_id: &RoomId) -> Result<bool> {
+        Ok(self.disabledroomids.get(room_id.as_bytes())?.is_some())
+    }
+
     /// Returns an iterator over all rooms this user joined.
     #[tracing::instrument(skip(self))]
     pub fn rooms_joined<'a>(