fix: panic on launch

Now we start the admin and sending threads at a later time.

src/api/client_server/directory.rs

@@ -29,7 +29,7 @@ use ruma::{
     },
     ServerName, UInt,
 };
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 
 /// # `POST /_matrix/client/r0/publicRooms`
 ///
@@ -279,15 +279,14 @@ pub(crate) async fn get_public_rooms_filtered_helper(
                                 JoinRule::Knock => Some(PublicRoomJoinRule::Knock),
                                 _ => None,
                             })
-                            .map_err(|_| {
-                                Error::bad_database("Invalid room join rule event in database.")
+                            .map_err(|e| {
+                                error!("Invalid room join rule event in database: {}", e);
+                                Error::BadDatabase("Invalid room join rule event in database.")
                             })
                     })
                     .transpose()?
                     .flatten()
-                    .ok_or(Error::bad_database(
-                        "Invalid room join rule event in database.",
-                    ))?,
+                    .ok_or_else(|| Error::bad_database("Missing room join rule event for room."))?,
                 room_id,
             };
             Ok(chunk)

src/database/mod.rs

@@ -1,10 +1,7 @@
 pub mod abstraction;
 pub mod key_value;
 
-use crate::{
-    services, utils, Config, Error,
-    PduEvent, Result, Services, SERVICES,
-};
+use crate::{services, utils, Config, Error, PduEvent, Result, Services, SERVICES};
 use abstraction::KeyValueDatabaseEngine;
 use abstraction::KvTree;
 use directories::ProjectDirs;
@@ -830,6 +827,8 @@ impl KeyValueDatabase {
         // This data is probably outdated
         db.presenceid_presence.clear()?;
 
+        services().admin.start_handler();
+
         // Set emergency access for the conduit user
         match set_emergency_access() {
             Ok(pwd_set) => {
@@ -846,6 +845,8 @@ impl KeyValueDatabase {
             }
         };
 
+        services().sending.start_handler();
+
         Self::start_cleanup_task().await;
 
         Ok(())

src/main.rs

@@ -28,7 +28,7 @@ use http::{
 };
 use opentelemetry::trace::{FutureExt, Tracer};
 use ruma::api::{client::error::ErrorKind, IncomingRequest};
-use tokio::{signal};
+use tokio::signal;
 use tower::ServiceBuilder;
 use tower_http::{
     cors::{self, CorsLayer},

src/service/account_data/mod.rs

@@ -8,7 +8,7 @@ use ruma::{
     RoomId, UserId,
 };
 
-use std::{collections::HashMap};
+use std::collections::HashMap;
 
 use crate::Result;
 

src/service/admin/mod.rs

@@ -26,7 +26,7 @@ use ruma::{
     EventId, RoomAliasId, RoomId, RoomName, RoomVersionId, ServerName, UserId,
 };
 use serde_json::value::to_raw_value;
-use tokio::sync::{mpsc, MutexGuard};
+use tokio::sync::{mpsc, Mutex, MutexGuard};
 
 use crate::{
     api::client_server::{leave_all_rooms, AUTO_GEN_PASSWORD_LENGTH},
@@ -164,25 +164,29 @@ pub enum AdminRoomEvent {
     SendMessage(RoomMessageEventContent),
 }
 
-#[derive(Clone)]
 pub struct Service {
     pub sender: mpsc::UnboundedSender<AdminRoomEvent>,
+    receiver: Mutex<mpsc::UnboundedReceiver<AdminRoomEvent>>,
 }
 
 impl Service {
     pub fn build() -> Arc<Self> {
         let (sender, receiver) = mpsc::unbounded_channel();
-        let self1 = Arc::new(Self { sender });
-        let self2 = Arc::clone(&self1);
-
-        tokio::spawn(async move {
-            self2.start_handler(receiver).await;
-        });
-
-        self1
+        Arc::new(Self {
+            sender,
+            receiver: Mutex::new(receiver),
+        })
     }
 
-    async fn start_handler(&self, mut receiver: mpsc::UnboundedReceiver<AdminRoomEvent>) {
+    pub fn start_handler(self: &Arc<Self>) {
+        let self2 = Arc::clone(&self);
+        tokio::spawn(async move {
+            self2.handler().await;
+        });
+    }
+
+    async fn handler(&self) {
+        let mut receiver = self.receiver.lock().await;
         // TODO: Use futures when we have long admin commands
         //let mut futures = FuturesUnordered::new();
 

src/service/appservice/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 
 use crate::Result;

src/service/key_backups/mod.rs

@@ -7,7 +7,7 @@ use ruma::{
     serde::Raw,
     RoomId, UserId,
 };
-use std::{collections::BTreeMap};
+use std::collections::BTreeMap;
 
 pub struct Service {
     pub db: &'static dyn Data,

src/service/rooms/alias/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 
 use crate::Result;

src/service/rooms/directory/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 use ruma::RoomId;
 

src/service/rooms/edus/presence/mod.rs

@@ -1,5 +1,5 @@
 mod data;
-use std::{collections::HashMap};
+use std::collections::HashMap;
 
 pub use data::Data;
 use ruma::{events::presence::PresenceEvent, RoomId, UserId};

src/service/rooms/edus/read_receipt/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 
 use crate::Result;

src/service/rooms/edus/typing/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 use ruma::{events::SyncEphemeralRoomEvent, RoomId, UserId};
 

src/service/rooms/lazy_loading/mod.rs

@@ -1,7 +1,7 @@
 mod data;
 use std::{
     collections::{HashMap, HashSet},
-    sync::{Mutex},
+    sync::Mutex,
 };
 
 pub use data::Data;

src/service/rooms/metadata/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 use ruma::RoomId;
 

src/service/rooms/outlier/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 use ruma::{signatures::CanonicalJsonObject, EventId};
 

src/service/rooms/search/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 
 use crate::Result;

src/service/rooms/user/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 use ruma::{RoomId, UserId};
 

src/service/sending/mod.rs

@@ -12,7 +12,7 @@ use std::{
 use crate::{
     api::{appservice_server, server_server},
     services,
-    utils::{calculate_hash},
+    utils::calculate_hash,
     Config, Error, PduEvent, Result,
 };
 use federation::transactions::send_transaction_message;
@@ -37,7 +37,7 @@ use ruma::{
 };
 use tokio::{
     select,
-    sync::{mpsc, Semaphore},
+    sync::{mpsc, Mutex, Semaphore},
 };
 use tracing::{error, warn};
 
@@ -88,6 +88,7 @@ pub struct Service {
     /// The state for a given state hash.
     pub(super) maximum_requests: Arc<Semaphore>,
     pub sender: mpsc::UnboundedSender<(OutgoingKind, SendingEventType, Vec<u8>)>,
+    receiver: Mutex<mpsc::UnboundedReceiver<(OutgoingKind, SendingEventType, Vec<u8>)>>,
 }
 
 enum TransactionStatus {
@@ -99,25 +100,24 @@ enum TransactionStatus {
 impl Service {
     pub fn build(db: &'static dyn Data, config: &Config) -> Arc<Self> {
         let (sender, receiver) = mpsc::unbounded_channel();
-
-        let self1 = Arc::new(Self {
+        Arc::new(Self {
             db,
             sender,
+            receiver: Mutex::new(receiver),
             maximum_requests: Arc::new(Semaphore::new(config.max_concurrent_requests as usize)),
-        });
-        let self2 = Arc::clone(&self1);
-
-        tokio::spawn(async move {
-            self2.start_handler(receiver).await.unwrap();
-        });
-
-        self1
+        })
     }
 
-    async fn start_handler(
-        &self,
-        mut receiver: mpsc::UnboundedReceiver<(OutgoingKind, SendingEventType, Vec<u8>)>,
-    ) -> Result<()> {
+    pub fn start_handler(self: &Arc<Self>) {
+        let self2 = Arc::clone(&self);
+        tokio::spawn(async move {
+            self2.handler().await.unwrap();
+        });
+    }
+
+    async fn handler(&self) -> Result<()> {
+        let mut receiver = self.receiver.lock().await;
+
         let mut futures = FuturesUnordered::new();
 
         let mut current_transaction_status = HashMap::<OutgoingKind, TransactionStatus>::new();

src/service/transaction_ids/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 
 use crate::Result;

src/service/uiaa/mod.rs

@@ -1,6 +1,5 @@
 mod data;
 
-
 pub use data::Data;
 
 use ruma::{