misc cleanup

src/client_server/account.rs

@@ -1,7 +1,7 @@
-use std::{collections::BTreeMap, convert::TryInto, sync::Arc};
+use std::{collections::BTreeMap, convert::TryInto};
 
-use super::{State, DEVICE_ID_LENGTH, SESSION_ID_LENGTH, TOKEN_LENGTH};
+use super::{DEVICE_ID_LENGTH, SESSION_ID_LENGTH, TOKEN_LENGTH};
-use crate::{database::ReadGuard, pdu::PduBuilder, utils, ConduitResult, Database, Error, Ruma};
+use crate::{database::ReadGuard, pdu::PduBuilder, utils, ConduitResult, Error, Ruma};
 use log::info;
 use ruma::{
     api::client::{

src/client_server/alias.rs

@@ -1,6 +1,3 @@
-use std::sync::Arc;
-
-use super::State;
 use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use regex::Regex;
 use ruma::{

src/client_server/backup.rs

@@ -1,7 +1,4 @@
-use std::sync::Arc;
+use crate::{database::ReadGuard, ConduitResult, Error, Ruma};
-
-use super::State;
-use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use ruma::api::client::{
     error::ErrorKind,
     r0::backup::{

src/client_server/config.rs

@@ -1,7 +1,4 @@
-use std::sync::Arc;
+use crate::{database::ReadGuard, ConduitResult, Error, Ruma};
-
-use super::State;
-use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::{
         error::ErrorKind,

src/client_server/context.rs

@@ -1,7 +1,6 @@
-use super::State;
+use crate::{database::ReadGuard, ConduitResult, Error, Ruma};
-use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use ruma::api::client::{error::ErrorKind, r0::context::get_context};
-use std::{convert::TryFrom, sync::Arc};
+use std::convert::TryFrom;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::get;

src/client_server/device.rs

@@ -1,7 +1,4 @@
-use std::sync::Arc;
+use crate::{database::ReadGuard, utils, ConduitResult, Error, Ruma};
-
-use super::State;
-use crate::{database::ReadGuard, utils, ConduitResult, Database, Error, Ruma};
 use ruma::api::client::{
     error::ErrorKind,
     r0::{

src/client_server/directory.rs

@@ -1,6 +1,3 @@
-use std::sync::Arc;
-
-use super::State;
 use crate::{database::ReadGuard, ConduitResult, Database, Error, Result, Ruma};
 use log::info;
 use ruma::{

src/client_server/keys.rs

@@ -1,4 +1,4 @@
-use super::{State, SESSION_ID_LENGTH};
+use super::SESSION_ID_LENGTH;
 use crate::{database::ReadGuard, utils, ConduitResult, Database, Error, Result, Ruma};
 use ruma::{
     api::client::{
@@ -14,10 +14,7 @@ use ruma::{
     encryption::UnsignedDeviceInfo,
     DeviceId, DeviceKeyAlgorithm, UserId,
 };
-use std::{
+use std::collections::{BTreeMap, HashSet};
-    collections::{BTreeMap, HashSet},
-    sync::Arc,
-};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, post};

src/client_server/media.rs

@@ -1,15 +1,12 @@
-use super::State;
+use crate::{database::media::FileMeta, database::ReadGuard, utils, ConduitResult, Error, Ruma};
-use crate::{
-    database::media::FileMeta, database::ReadGuard, utils, ConduitResult, Database, Error, Ruma,
-};
 use ruma::api::client::{
     error::ErrorKind,
     r0::media::{create_content, get_content, get_content_thumbnail, get_media_config},
 };
+use std::convert::TryInto;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, post};
-use std::{convert::TryInto, sync::Arc};
 
 const MXC_LENGTH: usize = 32;
 

src/client_server/membership.rs

@@ -1,4 +1,3 @@
-use super::State;
 use crate::{
     client_server,
     database::ReadGuard,

src/client_server/message.rs

@@ -1,5 +1,4 @@
-use super::State;
+use crate::{database::ReadGuard, pdu::PduBuilder, utils, ConduitResult, Error, Ruma};
-use crate::{database::ReadGuard, pdu::PduBuilder, utils, ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::{
         error::ErrorKind,
@@ -11,7 +10,6 @@ use ruma::{
 use std::{
     collections::BTreeMap,
     convert::{TryFrom, TryInto},
-    sync::Arc,
 };
 
 #[cfg(feature = "conduit_bin")]

src/client_server/mod.rs

@@ -64,9 +64,7 @@ pub use voip::*;
 use super::State;
 #[cfg(feature = "conduit_bin")]
 use {
-    crate::ConduitResult,
+    crate::ConduitResult, rocket::options, ruma::api::client::r0::to_device::send_event_to_device,
-    rocket::{options, State},
-    ruma::api::client::r0::to_device::send_event_to_device,
 };
 
 pub const DEVICE_ID_LENGTH: usize = 10;

src/client_server/presence.rs

@@ -1,7 +1,6 @@
-use super::State;
+use crate::{database::ReadGuard, utils, ConduitResult, Ruma};
-use crate::{database::ReadGuard, utils, ConduitResult, Database, Ruma};
 use ruma::api::client::r0::presence::{get_presence, set_presence};
-use std::{convert::TryInto, sync::Arc, time::Duration};
+use std::{convert::TryInto, time::Duration};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, put};

src/client_server/profile.rs

@@ -1,5 +1,4 @@
-use super::State;
+use crate::{database::ReadGuard, pdu::PduBuilder, utils, ConduitResult, Error, Ruma};
-use crate::{database::ReadGuard, pdu::PduBuilder, utils, ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::{
         error::ErrorKind,
@@ -10,10 +9,10 @@ use ruma::{
     events::EventType,
     serde::Raw,
 };
+use std::convert::TryInto;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, put};
-use std::{convert::TryInto, sync::Arc};
 
 #[cfg_attr(
     feature = "conduit_bin",

src/client_server/push.rs

@@ -1,7 +1,4 @@
-use std::sync::Arc;
+use crate::{database::ReadGuard, ConduitResult, Error, Ruma};
-
-use super::State;
-use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::{
         error::ErrorKind,

src/client_server/read_marker.rs

@@ -1,5 +1,4 @@
-use super::State;
+use crate::{database::ReadGuard, ConduitResult, Error, Ruma};
-use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::{
         error::ErrorKind,
@@ -9,10 +8,10 @@ use ruma::{
     receipt::ReceiptType,
     MilliSecondsSinceUnixEpoch,
 };
+use std::collections::BTreeMap;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::post;
-use std::{collections::BTreeMap, sync::Arc};
 
 #[cfg_attr(
     feature = "conduit_bin",

src/client_server/redact.rs

@@ -1,10 +1,8 @@
-use super::State;
+use crate::{database::ReadGuard, pdu::PduBuilder, ConduitResult, Ruma};
-use crate::{database::ReadGuard, pdu::PduBuilder, ConduitResult, Database, Ruma};
 use ruma::{
     api::client::r0::redact::redact_event,
     events::{room::redaction, EventType},
 };
-use std::sync::Arc;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::put;

src/client_server/room.rs

@@ -1,7 +1,5 @@
-use super::State;
 use crate::{
-    client_server::invite_helper, database::ReadGuard, pdu::PduBuilder, ConduitResult, Database,
+    client_server::invite_helper, database::ReadGuard, pdu::PduBuilder, ConduitResult, Error, Ruma,
-    Error, Ruma,
 };
 use log::info;
 use ruma::{
@@ -16,7 +14,7 @@ use ruma::{
     serde::Raw,
     RoomAliasId, RoomId, RoomVersionId,
 };
-use std::{cmp::max, collections::BTreeMap, convert::TryFrom, sync::Arc};
+use std::{cmp::max, collections::BTreeMap, convert::TryFrom};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, post};

src/client_server/search.rs

@@ -1,7 +1,5 @@
-use super::State;
+use crate::{database::ReadGuard, ConduitResult, Error, Ruma};
-use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use ruma::api::client::{error::ErrorKind, r0::search::search_events};
-use std::sync::Arc;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::post;

src/client_server/session.rs

@@ -1,7 +1,5 @@
-use std::sync::Arc;
+use super::{DEVICE_ID_LENGTH, TOKEN_LENGTH};
-
+use crate::{database::ReadGuard, utils, ConduitResult, Error, Ruma};
-use super::{State, DEVICE_ID_LENGTH, TOKEN_LENGTH};
-use crate::{database::ReadGuard, utils, ConduitResult, Database, Error, Ruma};
 use log::info;
 use ruma::{
     api::client::{

src/client_server/state.rs

@@ -1,6 +1,3 @@
-use std::sync::Arc;
-
-use super::State;
 use crate::{database::ReadGuard, pdu::PduBuilder, ConduitResult, Database, Error, Result, Ruma};
 use ruma::{
     api::client::{

src/client_server/sync.rs

@@ -1,4 +1,3 @@
-use super::State;
 use crate::{database::ReadGuard, ConduitResult, Database, Error, Result, Ruma, RumaResponse};
 use log::error;
 use ruma::{

src/client_server/tag.rs

@@ -1,10 +1,9 @@
-use super::State;
+use crate::{database::ReadGuard, ConduitResult, Ruma};
-use crate::{database::ReadGuard, ConduitResult, Database, Ruma};
 use ruma::{
     api::client::r0::tag::{create_tag, delete_tag, get_tags},
     events::EventType,
 };
-use std::{collections::BTreeMap, sync::Arc};
+use std::collections::BTreeMap;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{delete, get, put};

src/client_server/to_device.rs

@@ -1,7 +1,4 @@
-use std::sync::Arc;
+use crate::{database::ReadGuard, ConduitResult, Error, Ruma};
-
-use super::State;
-use crate::{database::ReadGuard, ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::{error::ErrorKind, r0::to_device::send_event_to_device},
     to_device::DeviceIdOrAllDevices,

src/client_server/typing.rs

@@ -1,7 +1,4 @@
-use std::sync::Arc;
+use crate::{database::ReadGuard, utils, ConduitResult, Ruma};
-
-use super::State;
-use crate::{database::ReadGuard, utils, ConduitResult, Database, Ruma};
 use create_typing_event::Typing;
 use ruma::api::client::r0::typing::create_typing_event;
 

src/client_server/user_directory.rs

@@ -1,7 +1,4 @@
-use std::sync::Arc;
+use crate::{database::ReadGuard, ConduitResult, Ruma};
-
-use super::State;
-use crate::{database::ReadGuard, ConduitResult, Database, Ruma};
 use ruma::api::client::r0::user_directory::search_users;
 
 #[cfg(feature = "conduit_bin")]

src/database.rs

@@ -34,7 +34,7 @@ use std::{
     ops::Deref,
     sync::{Arc, RwLock},
 };
-use tokio::sync::{OwnedRwLockReadGuard, RwLock as TokioRwLock, RwLockReadGuard, Semaphore};
+use tokio::sync::{OwnedRwLockReadGuard, RwLock as TokioRwLock, Semaphore};
 
 use self::proxy::ProxyConfig;
 
@@ -462,7 +462,6 @@ impl Deref for ReadGuard {
     }
 }
 
-#[cfg(feature = "conduit_bin")]
 #[rocket::async_trait]
 impl<'r> FromRequest<'r> for ReadGuard {
     type Error = ();
@@ -473,3 +472,9 @@ impl<'r> FromRequest<'r> for ReadGuard {
         Ok(ReadGuard(Arc::clone(&db).read_owned().await)).or_forward(())
     }
 }
+
+impl Into<ReadGuard> for OwnedRwLockReadGuard<Database> {
+    fn into(self) -> ReadGuard {
+        ReadGuard(self)
+    }
+}

src/database/admin.rs

@@ -10,7 +10,7 @@ use ruma::{
     events::{room::message, EventType},
     UserId,
 };
-use tokio::sync::RwLock;
+use tokio::sync::{RwLock, RwLockReadGuard};
 
 pub enum AdminCommand {
     RegisterAppservice(serde_yaml::Value),
@@ -54,47 +54,52 @@ impl Admin {
 
             drop(guard);
 
-            let send_message = |message: message::MessageEventContent| {};
+            let send_message =
+                |message: message::MessageEventContent, guard: RwLockReadGuard<'_, Database>| {
+                    if let Some(conduit_room) = &conduit_room {
+                        guard
+                            .rooms
+                            .build_and_append_pdu(
+                                PduBuilder {
+                                    event_type: EventType::RoomMessage,
+                                    content: serde_json::to_value(message)
+                                        .expect("event is valid, we just created it"),
+                                    unsigned: None,
+                                    state_key: None,
+                                    redacts: None,
+                                },
+                                &conduit_user,
+                                &conduit_room,
+                                &guard,
+                            )
+                            .unwrap();
+                    }
+                };
 
             loop {
                 tokio::select! {
                     Some(event) = receiver.next() => {
+                        let guard = db.read().await;
+
                         match event {
                             AdminCommand::RegisterAppservice(yaml) => {
-                                db.read().await.appservice.register_appservice(yaml).unwrap(); // TODO handle error
+                                guard.appservice.register_appservice(yaml).unwrap(); // TODO handle error
                             }
                             AdminCommand::ListAppservices => {
-                                if let Ok(appservices) = db.read().await.appservice.iter_ids().map(|ids| ids.collect::<Vec<_>>()) {
+                                if let Ok(appservices) = guard.appservice.iter_ids().map(|ids| ids.collect::<Vec<_>>()) {
                                     let count = appservices.len();
                                     let output = format!(
                                         "Appservices ({}): {}",
                                         count,
                                         appservices.into_iter().filter_map(|r| r.ok()).collect::<Vec<_>>().join(", ")
                                     );
-                                    send_message(message::MessageEventContent::text_plain(output));
+                                    send_message(message::MessageEventContent::text_plain(output), guard);
                                 } else {
-                                    send_message(message::MessageEventContent::text_plain("Failed to get appservices."));
+                                    send_message(message::MessageEventContent::text_plain("Failed to get appservices."), guard);
                                 }
                             }
                             AdminCommand::SendMessage(message) => {
-                                if let Some(conduit_room) = &conduit_room {
+                                send_message(message, guard)
-                                    let guard = db.read().await;
-                                    guard.rooms
-                                        .build_and_append_pdu(
-                                            PduBuilder {
-                                                event_type: EventType::RoomMessage,
-                                                content: serde_json::to_value(message)
-                                                    .expect("event is valid, we just created it"),
-                                                unsigned: None,
-                                                state_key: None,
-                                                redacts: None,
-                                            },
-                                            &conduit_user,
-                                            &conduit_room,
-                                            &guard,
-                                        )
-                                        .unwrap();
-                                }
                             }
                         }
                     }

src/lib.rs

@@ -15,8 +15,7 @@ pub use error::{Error, Result};
 pub use pdu::PduEvent;
 pub use rocket::Config;
 pub use ruma_wrapper::{ConduitResult, Ruma, RumaResponse};
-use std::{ops::Deref, sync::Arc};
+use std::ops::Deref;
-use tokio::sync::RwLock;
 
 pub struct State<'r, T: Send + Sync + 'static>(pub &'r T);
 

src/main.rs

@@ -231,8 +231,11 @@ async fn main() {
                     };
                     log::warn!("wal-trunc: locked, flushing...");
                     let start = Instant::now();
-                    guard.flush_wal();
+                    if let Err(e) = guard.flush_wal() {
-                    log::warn!("wal-trunc: flushed in {:?}", start.elapsed());
+                        log::warn!("wal-trunc: errored: {}", e);
+                    } else {
+                        log::warn!("wal-trunc: flushed in {:?}", start.elapsed());
+                    }
                 } else {
                     break;
                 }

src/ruma_wrapper.rs

@@ -9,7 +9,7 @@ use std::ops::Deref;
 
 #[cfg(feature = "conduit_bin")]
 use {
-    crate::{server_server, Database},
+    crate::server_server,
     log::{debug, warn},
     rocket::{
         data::{self, ByteUnit, Data, FromData},
@@ -17,13 +17,12 @@ use {
         outcome::Outcome::*,
         response::{self, Responder},
         tokio::io::AsyncReadExt,
-        Request, State,
+        Request,
     },
     ruma::api::{AuthScheme, IncomingRequest},
     std::collections::BTreeMap,
     std::convert::TryFrom,
     std::io::Cursor,
-    std::sync::Arc,
 };
 
 /// This struct converts rocket requests into ruma structs by converting them into http requests

src/server_server.rs

@@ -7,7 +7,7 @@ use get_profile_information::v1::ProfileField;
 use http::header::{HeaderValue, AUTHORIZATION, HOST};
 use log::{debug, error, info, trace, warn};
 use regex::Regex;
-use rocket::{response::content::Json, State};
+use rocket::response::content::Json;
 use ruma::{
     api::{
         client::error::{Error as RumaError, ErrorKind},