253 implement fault tolerance mechanisms eg retries dead letter queues circuit breakers e5

crates/web-plugins/didcomm-messaging/protocols/forward/src/error.rs

@@ -10,6 +10,8 @@ pub(crate) enum ForwardError {
     UncoordinatedSender,
     #[error("Internal server error")]
     InternalServerError,
+    #[error("Service unavailable")]
+    CircuitOpen,
 }
 
 impl IntoResponse for ForwardError {
@@ -18,6 +20,7 @@ impl IntoResponse for ForwardError {
             ForwardError::MalformedBody => StatusCode::BAD_REQUEST,
             ForwardError::UncoordinatedSender => StatusCode::UNAUTHORIZED,
             ForwardError::InternalServerError => StatusCode::INTERNAL_SERVER_ERROR,
+            ForwardError::CircuitOpen => StatusCode::SERVICE_UNAVAILABLE,
         };
 
         let body = Json(serde_json::json!({

crates/web-plugins/didcomm-messaging/protocols/forward/src/handler.rs

@@ -5,9 +5,11 @@ use mongodb::bson::doc;
 use serde_json::{json, Value};
 use shared::{
     repository::entity::{Connection, RoutedMessage},
+    retry::{retry_async, RetryOptions},
     state::{AppState, AppStateRepository},
 };
 use std::sync::Arc;
+use std::time::Duration;
 
 /// Mediator receives forwarded messages, extract the next field in the message body, and the attachments in the message
 /// then stores the attachment with the next field as key for pickup
@@ -24,6 +26,11 @@ pub(crate) async fn mediator_forward_process(
         .as_ref()
         .ok_or_else(|| ForwardError::InternalServerError)?;
 
+    let circuit_breaker = state.circuit_breaker.clone();
+    if circuit_breaker.is_open() {
+        return Err(ForwardError::CircuitOpen);
+    }
+
     let next = match checks(&message, connection_repository).await.ok() {
         Some(next) => Ok(next),
         None => Err(ForwardError::InternalServerError),
@@ -36,15 +43,38 @@ pub(crate) async fn mediator_forward_process(
             AttachmentData::Base64 { value: data } => json!(data.base64),
             AttachmentData::Links { value: data } => json!(data.links),
         };
-        message_repository
-            .store(RoutedMessage {
-                id: None,
-                message: attached,
-                recipient_did: next.as_ref().unwrap().to_owned(),
-            })
-            .await
-            .map_err(|_| ForwardError::InternalServerError)?;
+
+        let result = retry_async(
+            || {
+                let attached = attached.clone();
+                let recipient_did = next.as_ref().unwrap().to_owned();
+
+                async move {
+                    message_repository
+                        .store(RoutedMessage {
+                            id: None,
+                            message: attached,
+                            recipient_did,
+                        })
+                        .await
+                }
+            },
+            RetryOptions::new()
+                .retries(5)
+                .exponential_backoff(Duration::from_millis(100))
+                .max_delay(Duration::from_secs(1)),
+        )
+        .await;
+
+        match result {
+            Ok(_) => circuit_breaker.record_success(),
+            Err(_) => {
+                circuit_breaker.record_failure();
+                return Err(ForwardError::InternalServerError);
+            }
+        };
     }
+
     Ok(None)
 }
 

crates/web-plugins/didcomm-messaging/protocols/mediator-coordination/src/handler/stateful.rs

@@ -21,9 +21,11 @@
 use shared::{
     midlw::ensure_transport_return_route_is_decorated_all,
     repository::entity::Connection,
+    retry::{retry_async, RetryOptions},
     state::{AppState, AppStateRepository},
+    CircuitBreaker::CircuitBreaker,
 };
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 use uuid::Uuid;
 
 /// Process a DIDComm mediate request
@@ -53,10 +55,24 @@
         .ok_or(MediationError::InternalServerError)?;
 
     // If there is already mediation, send mediate deny
-    if let Some(_connection) = connection_repository
-        .find_one_by(doc! { "client_did": sender_did})
-        .await
-        .map_err(|_| MediationError::InternalServerError)?
+    if let Some(_connection) = retry_async(
+        || {
+            let sender_did = sender_did.clone();
+            let connection_repository = connection_repository.clone();
+
+            async move {
+                connection_repository
+                    .find_one_by(doc! { "client_did": sender_did })
+                    .await
+            }
+        },
+        RetryOptions::new()
+            .retries(5)
+            .exponential_backoff(Duration::from_millis(100))
+            .max_delay(Duration::from_secs(1)),
+    )
+    .await
+    .map_err(|_| MediationError::InternalServerError)?
     {
         tracing::info!("Sending mediate deny.");
         return Ok(Some(
@@ -85,15 +101,24 @@
             .as_ref()
             .ok_or(MediationError::InternalServerError)?;
 
-        let diddoc = state
-            .did_resolver
-            .resolve(&routing_did)
-            .await
-            .map_err(|err| {
-                tracing::error!("Failed to resolve DID: {:?}", err);
-                MediationError::InternalServerError
-            })?
-            .ok_or(MediationError::InternalServerError)?;
+        let diddoc = retry_async(
+            || {
+                let did_resolver = state.did_resolver.clone();
+                let routing_did = routing_did.clone();
+
+                async move { did_resolver.resolve(&routing_did).await.map_err(|_| ()) }
+            },
+            RetryOptions::new()
+                .retries(5)
+                .exponential_backoff(Duration::from_millis(100))
+                .max_delay(Duration::from_secs(2)),
+        )
+        .await
+        .map_err(|err| {
+            tracing::error!("Failed to resolve DID: {:?}", err);
+            MediationError::InternalServerError
+        })?
+        .ok_or(MediationError::InternalServerError)?;
 
         let agreem_keys_jwk: Jwk = agreem_keys.try_into().unwrap();
 
@@ -229,11 +254,29 @@
 
     // Find connection for this keylist update
 
-    let connection = connection_repository
-        .find_one_by(doc! { "client_did": &sender })
-        .await
-        .unwrap()
-        .ok_or_else(|| MediationError::UncoordinatedSender)?;
+    let connection = retry_async(
+        || {
+            let connection_repository = connection_repository.clone();
+            let sender = sender.clone();
+
+            async move {
+                connection_repository
+                    .find_one_by(doc! { "client_did": &sender })
+                    .await
+                    .map_err(|_| ())
+            }
+        },
+        RetryOptions::new()
+            .retries(5)
+            .exponential_backoff(Duration::from_millis(100))
+            .max_delay(Duration::from_secs(2)),
+    )
+    .await
+    .map_err(|err| {
+        tracing::error!("Failed to find connection after retries: {:?}", err);
+        MediationError::InternalServerError
+    })?
+    .ok_or_else(|| MediationError::UncoordinatedSender)?;
 
     // Prepare handles to relevant collections
 
@@ -347,11 +390,29 @@
         .as_ref()
         .ok_or(MediationError::InternalServerError)?;
 
-    let connection = connection_repository
-        .find_one_by(doc! { "client_did": &sender })
-        .await
-        .unwrap()
-        .ok_or_else(|| MediationError::UncoordinatedSender)?;
+    let connection = retry_async(
+        || {
+            let connection_repository = connection_repository.clone();
+            let sender = sender.clone();
+
+            async move {
+                connection_repository
+                    .find_one_by(doc! { "client_did": &sender })
+                    .await
+                    .map_err(|_| ())
+            }
+        },
+        RetryOptions::new()
+            .retries(5)
+            .exponential_backoff(Duration::from_millis(100))
+            .max_delay(Duration::from_secs(2)),
+    )
+    .await
+    .map_err(|err| {
+        tracing::error!("Failed to find connection after retries: {:?}", err);
+        MediationError::InternalServerError
+    })?
+    .ok_or_else(|| MediationError::UncoordinatedSender)?;
 
     println!("keylist: {:?}", connection);
 

crates/web-plugins/didcomm-messaging/protocols/pickup/src/error.rs

@@ -14,6 +14,9 @@
 
     #[error("Malformed request. {0}")]
     MalformedRequest(String),
+
+    #[error("Service unavailable")]
+    CircuitOpen,
 }
 
 impl IntoResponse for PickupError {
@@ -22,6 +25,7 @@
             PickupError::MissingSenderDID | PickupError::MalformedRequest(_) => {
                 StatusCode::BAD_REQUEST
             }
+            PickupError::CircuitOpen => StatusCode::SERVICE_UNAVAILABLE,
             PickupError::InternalError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PickupError::MissingClientConnection => StatusCode::UNAUTHORIZED,
         };