feat: better error handling

Cargo.lock

@@ -6511,6 +6511,7 @@ dependencies = [
  "revolt_rocket_okapi",
  "rocket",
  "schemars",
+ "sentry",
  "serde",
  "serde_json",
  "utoipa",

crates/bonfire/src/config.rs

@@ -1,7 +1,7 @@
 use async_tungstenite::tungstenite::{handshake, Message};
 use futures::channel::oneshot::Sender;
 use revolt_database::events::client::ReadyPayloadFields;
-use revolt_result::{create_error, Result};
+use revolt_result::{create_error, Result, ToRevoltError};
 use serde::{Deserialize, Serialize};
 
 /// Enumeration of supported protocol formats
@@ -38,14 +38,14 @@ impl ProtocolConfiguration {
         match self.format {
             ProtocolFormat::Json => {
                 if let Message::Text(text) = msg {
-                    serde_json::from_str(text).map_err(|_| create_error!(InternalError))
+                    serde_json::from_str(text).to_internal_error()
                 } else {
                     Err(create_error!(InternalError))
                 }
             }
             ProtocolFormat::Msgpack => {
                 if let Message::Binary(buf) = msg {
-                    rmp_serde::from_slice(buf).map_err(|_| create_error!(InternalError))
+                    rmp_serde::from_slice(buf).to_internal_error()
                 } else {
                     Err(create_error!(InternalError))
                 }

crates/core/database/src/models/users/model.rs

@@ -705,6 +705,8 @@ impl User {
         duration_days: Option<usize>,
         reason: Option<Vec<String>>,
     ) -> Result<()> {
+        // TODO: authifier Error should implement Error
+
         let authifier = db.clone().to_authifier().await;
         let mut account = authifier
             .database

crates/core/files/Cargo.toml

@@ -14,7 +14,7 @@ imagesize = "0.13.0"
 tempfile = "3.12.0"
 
 base64 = "0.22.1"
-aes-gcm = "0.10.3"
+aes-gcm = { version = "0.10.3", features = ["std"] }
 typenum = "1.17.0"
 
 aws-config = "1.5.5"

crates/core/files/src/lib.rs

@@ -6,7 +6,7 @@ use aes_gcm::{
 };
 use image::{DynamicImage, ImageBuffer};
 use revolt_config::{config, report_internal_error, FilesS3};
-use revolt_result::{create_error, Result};
+use revolt_result::{create_error, Result, ToRevoltError};
 
 use aws_sdk_s3::{
     config::{Credentials, Region},
@@ -78,7 +78,7 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
     // Decrypt the file
     create_cipher(&config.files.encryption_key)
         .decrypt_in_place(nonce, b"", &mut buf)
-        .map_err(|_| create_error!(InternalError))?;
+        .to_internal_error()?;
 
     Ok(buf)
 }
@@ -97,7 +97,7 @@ pub async fn upload_to_s3(bucket_id: &str, path: &str, buf: &[u8]) -> Result<Str
     // Encrypt the file in place
     create_cipher(&config.files.encryption_key)
         .encrypt_in_place(&nonce, b"", &mut buf)
-        .map_err(|_| create_error!(InternalError))?;
+        .to_internal_error()?;
 
     // Upload the file to remote
     report_internal_error!(

crates/core/result/Cargo.toml

@@ -34,3 +34,6 @@ revolt_okapi = { version = "0.9.1", optional = true }
 
 # Axum
 axum = { version = "0.7.5", optional = true }
+
+# Sentry
+sentry = "0.31.5"

crates/core/result/src/axum.rs

@@ -2,10 +2,9 @@ use axum::{http::StatusCode, response::IntoResponse, Json};
 
 use crate::{Error, ErrorType};
 
-/// HTTP response builder for Error enum
+impl Error {
-impl IntoResponse for Error {
+    pub fn axum_status(&self) -> StatusCode {
-    fn into_response(self) -> axum::response::Response {
+        match self.error_type {
-        let status = match self.error_type {
             ErrorType::LabelMe => StatusCode::INTERNAL_SERVER_ERROR,
 
             ErrorType::AlreadyOnboarded => StatusCode::FORBIDDEN,
@@ -74,7 +73,10 @@ impl IntoResponse for Error {
             ErrorType::VosoUnavailable => StatusCode::BAD_REQUEST,
             ErrorType::NotFound => StatusCode::NOT_FOUND,
             ErrorType::NoEffect => StatusCode::OK,
-            ErrorType::FailedValidation { .. } => StatusCode::BAD_REQUEST,
+            ErrorType::IOError => StatusCode::BAD_REQUEST,
+            ErrorType::UnprocessableEntity => StatusCode::UNPROCESSABLE_ENTITY,
+            ErrorType::DeserializationError { .. } => StatusCode::UNPROCESSABLE_ENTITY,
+            ErrorType::FailedValidation { .. } => StatusCode::UNPROCESSABLE_ENTITY,
             ErrorType::InvalidFlagValue => StatusCode::BAD_REQUEST,
             ErrorType::FeatureDisabled { .. } => StatusCode::BAD_REQUEST,
 
@@ -84,8 +86,13 @@ impl IntoResponse for Error {
             ErrorType::FileTypeNotAllowed => StatusCode::BAD_REQUEST,
             ErrorType::ImageProcessingFailed => StatusCode::INTERNAL_SERVER_ERROR,
             ErrorType::NoEmbedData => StatusCode::BAD_REQUEST,
-        };
+        }
+    }
+}
 
-        (status, Json(&self)).into_response()
+/// HTTP response builder for Error enum
+impl IntoResponse for Error {
+    fn into_response(self) -> axum::response::Response {
+        (self.axum_status(), Json(&self)).into_response()
     }
 }

crates/core/result/src/lib.rs

@@ -1,3 +1,4 @@
+use std::panic::Location;
 use std::fmt::Display;
 
 #[cfg(feature = "serde")]
@@ -154,6 +155,11 @@ pub enum ErrorType {
     DuplicateNonce,
     NotFound,
     NoEffect,
+    IOError,
+    UnprocessableEntity,
+    DeserializationError {
+        error: String,
+    },
     FailedValidation {
         error: String,
     },
@@ -197,6 +203,53 @@ macro_rules! create_database_error {
     };
 }
 
+pub trait ToRevoltError<T>: Sized {
+    fn capture_error(self) -> Self;
+
+    #[track_caller]
+    fn to_internal_error(self) -> Result<T, Error>;
+}
+
+impl<T, E: std::error::Error> ToRevoltError<T> for Result<T, E> {
+    fn capture_error(self) -> Self {
+        self.inspect_err(|e| { sentry::capture_error(e); })
+    }
+
+    #[track_caller]
+    fn to_internal_error(self) -> Result<T, Error> {
+        let loc = Location::caller();
+
+        self
+            .capture_error()
+            .map_err(|_| {
+                Error {
+                    error_type: ErrorType::InternalError,
+                    location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column())
+                }
+            })
+    }
+}
+
+impl<T: std::error::Error> ToRevoltError<T> for Option<T> {
+    fn capture_error(self) -> Self {
+        self.inspect(|e| { sentry::capture_error(e); })
+    }
+
+    #[track_caller]
+    fn to_internal_error(self) -> Result<T, Error> {
+        let loc = Location::caller();
+
+        self
+        .capture_error()
+        .ok_or_else(|| {
+            Error {
+                error_type: ErrorType::InternalError,
+                location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column())
+            }
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::ErrorType;

crates/core/result/src/rocket.rs

@@ -8,10 +8,9 @@ use rocket::{
 
 use crate::{Error, ErrorType};
 
-/// HTTP response builder for Error enum
+impl Error {
-impl<'r> Responder<'r, 'static> for Error {
+    pub fn rocket_status(&self) -> Status {
-    fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
+        match self.error_type {
-        let status = match self.error_type {
             ErrorType::LabelMe => Status::InternalServerError,
 
             ErrorType::AlreadyOnboarded => Status::Forbidden,
@@ -81,7 +80,10 @@ impl<'r> Responder<'r, 'static> for Error {
             ErrorType::VosoUnavailable => Status::BadRequest,
             ErrorType::NotFound => Status::NotFound,
             ErrorType::NoEffect => Status::Ok,
-            ErrorType::FailedValidation { .. } => Status::BadRequest,
+            ErrorType::IOError => Status::BadRequest,
+            ErrorType::UnprocessableEntity => Status::UnprocessableEntity,
+            ErrorType::DeserializationError { .. } => Status::UnprocessableEntity,
+            ErrorType::FailedValidation { .. } => Status::UnprocessableEntity,
             ErrorType::FeatureDisabled { .. } => Status::BadRequest,
 
             ErrorType::ProxyError => Status::BadRequest,
@@ -90,8 +92,13 @@ impl<'r> Responder<'r, 'static> for Error {
             ErrorType::FileTypeNotAllowed => Status::BadRequest,
             ErrorType::ImageProcessingFailed => Status::InternalServerError,
             ErrorType::NoEmbedData => Status::BadRequest,
-        };
+        }
+    }
+}
 
+/// HTTP response builder for Error enum
+impl<'r> Responder<'r, 'static> for Error {
+    fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
         // Serialize the error data structure into JSON.
         let string = serde_json::to_string(&self).unwrap();
 
@@ -99,7 +106,7 @@ impl<'r> Responder<'r, 'static> for Error {
         Response::build()
             .sized_body(string.len(), Cursor::new(string))
             .header(ContentType::new("application", "json"))
-            .status(status)
+            .status(self.rocket_status())
             .ok()
     }
 }

crates/daemons/crond/src/main.rs

@@ -1,19 +1,32 @@
-use revolt_config::configure;
+use std::{future::Future, time::Duration};
-use revolt_database::DatabaseInfo;
+
+use revolt_config::{configure, capture_error};
+use revolt_database::{Database, DatabaseInfo};
 use revolt_result::Result;
 use tasks::{file_deletion, prune_dangling_files};
-use tokio::try_join;
+use tokio::{join, time::sleep};
 
 pub mod tasks;
 
+pub async fn cron_task_wrapper<Fut: Future<Output = Result<()>>>(func: fn(Database) -> Fut, db: Database) {
+    loop {
+        if let Err(error) = func(db.clone()).await {
+            log::error!("cron task failed unexpectidly: {error:?}\nRetrying after 60s");
+            capture_error(&error);
+        }
+
+        sleep(Duration::from_secs(60)).await;
+    }
+}
+
 #[tokio::main]
-async fn main() -> Result<()> {
+async fn main() {
     configure!(crond);
 
     let db = DatabaseInfo::Auto.connect().await.expect("database");
-    try_join!(
+
-        file_deletion::task(db.clone()),
+    join!(
-        prune_dangling_files::task(db)
+        cron_task_wrapper(file_deletion::task, db.clone()),
-    )
+        cron_task_wrapper(prune_dangling_files::task, db.clone()),
-    .map(|_| ())
+    );
 }

crates/daemons/pushd/src/consumers/inbound/ack.rs

@@ -123,9 +123,9 @@ impl AsyncConsumer for AckConsumer {
                     token: session.subscription.as_ref().unwrap().auth.clone(),
                     extras: Default::default(),
                 };
-                let raw_service_payload = serde_json::to_string(&service_payload);
 
-                if let Ok(p) = raw_service_payload {
+                match serde_json::to_string(&service_payload) {
+                    Ok(p) => {
                         let args = BasicPublishArguments::new(
                             config.pushd.exchange.as_str(),
                             config.pushd.apn.queue.as_str(),
@@ -138,9 +138,11 @@ impl AsyncConsumer for AckConsumer {
                         );
 
                         publish_message(self, p.into(), args).await;
-                } else {
+                    },
+                    Err(e) => {
                         log::warn!("Failed to serialize ack badge update payload!");
-                    revolt_config::capture_error(&raw_service_payload.unwrap_err());
+                        revolt_config::capture_error(&e);
+                    }
                 }
             }
         }

crates/delta/src/main.rs

@@ -135,6 +135,7 @@ pub async fn web() -> Rocket<Build> {
         .manage(cors.clone())
         .attach(util::ratelimiter::RatelimitFairing)
         .attach(cors)
+        .register("/", util::catchers::all_catchers())
         .configure(rocket::Config {
             limits: rocket::data::Limits::default().limit("string", 5.megabytes()),
             address: Ipv4Addr::new(0, 0, 0, 0).into(),

crates/delta/src/routes/channels/message_send.rs

@@ -8,9 +8,9 @@ use revolt_models::v0;
 use revolt_permissions::PermissionQuery;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::serde::json::Json;
+use crate::util::json::{Json, Validate};
 use rocket::State;
-use validator::Validate;
+// use validator::Validate;
 
 /// # Send Message
 ///
@@ -22,15 +22,10 @@ pub async fn message_send(
     amqp: &State<AMQP>,
     user: User,
     target: Reference<'_>,
-    data: Json<v0::DataMessageSend>,
+    data: Validate<Json<v0::DataMessageSend>>,
     idempotency: IdempotencyKey,
 ) -> Result<Json<v0::Message>> {
     let data = data.into_inner();
-    data.validate().map_err(|error| {
-        create_error!(FailedValidation {
-            error: error.to_string()
-        })
-    })?;
 
     // Ensure we have permissions to send a message
     let channel = target.as_channel(db).await?;

crates/delta/src/routes/webhooks/webhook_execute_github.rs

@@ -701,7 +701,7 @@ fn safe_from_str<T: for<'de> Deserialize<'de>>(data: &str) -> Result<T> {
     match serde_json::from_str(data) {
         Ok(output) => Ok(output),
         Err(err) => {
-            log::error!("{err:?}");
+            revolt_config::capture_error(&err);
             Err(create_error!(InvalidOperation))
         }
     }

crates/delta/src/util/catchers.rs

@@ -0,0 +1,19 @@
+use rocket::{catch, Catcher, Request};
+use revolt_result::{create_error, Error, Result};
+
+#[catch(404)]
+pub fn not_found() -> Result<()> {
+    Err(create_error!(NotFound))
+}
+
+#[catch(422)]
+pub fn unprocessable_entity(req: &Request) -> Result<()> {
+    match req.local_cache(|| None::<Error>) {
+        Some(e) => Err(e.clone()),
+        None => Err(create_error!(UnprocessableEntity))
+    }
+}
+
+pub fn all_catchers() -> Vec<Catcher> {
+    catchers![not_found, unprocessable_entity]
+}

crates/delta/src/util/json.rs

@@ -0,0 +1,176 @@
+use std::fmt::Debug;
+
+use revolt_rocket_okapi::{
+    r#gen::OpenApiGenerator,
+    response::OpenApiResponderInner,
+    revolt_okapi::openapi3::Responses,
+    util::add_schema_response,
+    request::OpenApiFromData,
+    revolt_okapi::{
+        openapi3::{MediaType, RequestBody},
+        Map,
+    },
+};
+use rocket::data::{Data, FromData, Limits, Outcome};
+use rocket::response::{self, Responder, content};
+use rocket::request::{local_cache, Request};
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use revolt_result::{create_error, Error, ToRevoltError};
+
+// A lot of this code is modified versions of rocket::serde::json so we can store
+// the error so it can be passed to the error catcher.
+
+macro_rules! fn_request_body {
+    ($gen:ident, $ty:path, $mime_type:expr) => {{
+        let schema = $gen.json_schema::<$ty>();
+        Ok(RequestBody {
+            content: {
+                let mut map = Map::new();
+                map.insert(
+                    $mime_type.to_owned(),
+                    MediaType {
+                        schema: Some(schema),
+                        ..MediaType::default()
+                    },
+                );
+                map
+            },
+            required: true,
+            ..revolt_rocket_okapi::revolt_okapi::openapi3::RequestBody::default()
+        })
+    }};
+}
+
+#[derive(Debug, Clone)]
+pub struct Json<T>(pub T);
+
+impl<'r, T: Deserialize<'r>> Json<T> {
+    #[inline]
+    pub fn into_inner(self) -> T {
+        self.0
+    }
+
+    fn from_str(s: &'r str) -> Result<Self, Error> {
+        serde_json::from_str(s)
+            .map(Json)
+            .map_err(|e| create_error!(DeserializationError { error: e.to_string() }))
+    }
+
+    async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Result<Self, Error> {
+        let limit = req.limits().get("json").unwrap_or(Limits::JSON);
+        let string = match data.open(limit).into_string().await {
+            Ok(s) if s.is_complete() => s.into_inner(),
+            Ok(_) => {
+                return Err(create_error!(PayloadTooLarge));
+            },
+            Err(_) => return Err(create_error!(IOError)),
+        };
+
+        Self::from_str(local_cache!(req, string))
+    }
+}
+
+impl<T> std::ops::Deref for Json<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+#[async_trait]
+impl<'r, T: Deserialize<'r> + std::fmt::Debug> FromData<'r> for Json<T> {
+    type Error = Error;
+
+    async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Outcome<'r, Self> {
+        let r = Self::from_data(req, data).await;
+
+        match r {
+            Ok(value) => Outcome::Success(value),
+            Err(e) => {
+                req.local_cache(|| Some(e.clone()));
+                rocket::outcome::Outcome::Error((e.rocket_status(), e))
+            }
+        }
+    }
+}
+
+impl<'r, T: Serialize> Responder<'r, 'static> for Json<T> {
+    fn respond_to(self, req: &'r Request<'_>) -> response::Result<'static> {
+        match serde_json::to_string(&self.0).capture_error() {
+            Ok(string) => content::RawJson(string).respond_to(req),
+            Err(_) => create_error!(InternalError).respond_to(req)
+        }
+    }
+}
+
+impl<'r, T: JsonSchema + Deserialize<'r> + Debug> OpenApiFromData<'r> for Json<T> {
+    fn request_body(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result<RequestBody> {
+        fn_request_body!(gen, T, "application/json")
+    }
+}
+
+impl<T: Serialize + JsonSchema + Send> OpenApiResponderInner for Json<T> {
+    fn responses(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result<Responses> {
+        let mut responses = Responses::default();
+        let schema = gen.json_schema::<T>();
+        add_schema_response(&mut responses, 200, "application/json", schema)?;
+        Ok(responses)
+    }
+}
+
+impl<T: validator::Validate> validator::Validate for Json<T> {
+    fn validate(&self) -> Result<(), validator::ValidationErrors> {
+        self.0.validate()
+    }
+}
+
+pub struct Validate<T>(pub T);
+
+impl<T> Validate<Json<T>> {
+    pub fn into_inner(self) -> T {
+        self.0.0
+    }
+}
+
+#[async_trait]
+impl<'r, T: validator::Validate + Deserialize<'r> + Debug> FromData<'r> for Validate<Json<T>> {
+    type Error = Error;
+
+    async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Outcome<'r, Self> {
+        <Json::<T> as FromData<'r>>::from_data(req, data).await.and_then(|inner| {
+            if let Err(e) = inner.validate() {
+                let error = create_error!(FailedValidation { error: e.to_string() });
+                req.local_cache(|| Some(error.clone()));
+
+                return Outcome::Error((error.rocket_status(), error))
+            };
+
+            Outcome::Success(Self(inner))
+        })
+    }
+}
+
+impl<'r, T: JsonSchema + Deserialize<'r> + validator::Validate + Debug> OpenApiFromData<'r> for Validate<Json<T>> {
+    fn request_body(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result<RequestBody> {
+        fn_request_body!(gen, T, "application/json")
+    }
+}
+
+impl<T: Serialize + JsonSchema + Send> OpenApiResponderInner for Validate<Json<T>> {
+    fn responses(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result<Responses> {
+        let mut responses = Responses::default();
+        let schema = gen.json_schema::<T>();
+        add_schema_response(&mut responses, 200, "application/json", schema)?;
+        Ok(responses)
+    }
+}
+
+impl<T> std::ops::Deref for Validate<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}

crates/delta/src/util/mod.rs

@@ -1,2 +1,4 @@
 pub mod ratelimiter;
 pub mod test;
+pub mod catchers;
+pub mod json;