editoast: reconnect on rabbitmq disconnection

Currently, when the Lapin driver loose the connection to the
RabbitMQ (like when the Kubernetes is rescheduling the RabbitMQ
server pod on another node) editoast is left in an inconsistent state.
We add a loop to try to reconnect to the RabbitMQ server when the
connection is lost.

Co-authored-by: Younes Khoudli <[email protected]>

Author: 2 people

editoast/openapi.yaml

@@ -4412,6 +4412,7 @@ components:
       - $ref: '#/components/schemas/EditoastEditionErrorSplitTrackSectionBadOffset'
       - $ref: '#/components/schemas/EditoastEditoastUrlErrorInvalidUrl'
       - $ref: '#/components/schemas/EditoastElectricalProfilesErrorNotFound'
+      - $ref: '#/components/schemas/EditoastErrorConnectionDoesNotExist'
       - $ref: '#/components/schemas/EditoastErrorLapin'
       - $ref: '#/components/schemas/EditoastErrorResponseTimeout'
       - $ref: '#/components/schemas/EditoastErrorSerialization'
@@ -4495,6 +4496,25 @@ components:
       description: Generated error type for Editoast
       discriminator:
         propertyName: type
+    EditoastErrorConnectionDoesNotExist:
+      type: object
+      required:
+      - type
+      - status
+      - message
+      properties:
+        context:
+          type: object
+        message:
+          type: string
+        status:
+          type: integer
+          enum:
+          - 500
+        type:
+          type: string
+          enum:
+          - editoast:coreclient:ConnectionDoesNotExist
     EditoastErrorLapin:
       type: object
       required:

editoast/src/core/mq_client.rs

@@ -10,11 +10,14 @@ use serde::Serialize;
 use serde_json::to_vec;
 use std::{fmt::Debug, sync::Arc};
 use thiserror::Error;
-use tokio::time::{timeout, Duration};
+use tokio::{
+    sync::RwLock,
+    time::{timeout, Duration},
+};
 
 #[derive(Debug, Clone)]
 pub struct RabbitMQClient {
-    connection: Arc<Connection>,
+    connection: Arc<RwLock<Option<Connection>>>,
     exchange: String,
     timeout: u64,
     hostname: String,
@@ -45,6 +48,9 @@ pub enum Error {
     #[error("Response timeout")]
     #[editoast_error(status = "500")]
     ResponseTimeout,
+    #[error("Connection does not exist")]
+    #[editoast_error(status = "500")]
+    ConnectionDoesNotExist,
 }
 
 pub struct MQResponse {
@@ -54,21 +60,65 @@ pub struct MQResponse {
 
 impl RabbitMQClient {
     pub async fn new(options: Options) -> Result<Self, Error> {
-        let connection = Connection::connect(&options.uri, ConnectionProperties::default())
-            .await
-            .map_err(Error::Lapin)?;
         let hostname = hostname::get()
             .map(|name| name.to_string_lossy().into_owned())
             .unwrap_or_else(|_| "unknown".to_string());
 
+        let conn = Arc::new(RwLock::new(None));
+
+        tokio::spawn(Self::connection_loop(options.uri, conn.clone()));
+
         Ok(RabbitMQClient {
-            connection: Arc::new(connection),
+            connection: conn,
             exchange: format!("{}-req-xchg", options.worker_pool_identifier),
             timeout: options.timeout,
             hostname,
         })
     }
 
+    async fn connection_ok(connection: &Arc<RwLock<Option<Connection>>>) -> bool {
+        let guard = connection.as_ref().read().await;
+        let conn = guard.as_ref();
+        let status = match conn {
+            None => return false,
+            Some(conn) => conn.status().state(),
+        };
+        match status {
+            lapin::ConnectionState::Initial => true,
+            lapin::ConnectionState::Connecting => true,
+            lapin::ConnectionState::Connected => true,
+            lapin::ConnectionState::Closing => true,
+            lapin::ConnectionState::Closed => false,
+            lapin::ConnectionState::Error => false,
+        }
+    }
+
+    async fn connection_loop(uri: String, connection: Arc<RwLock<Option<Connection>>>) {
+        loop {
+            if Self::connection_ok(&connection).await {
+                tokio::time::sleep(Duration::from_secs(2)).await;
+                continue;
+            }
+
+            tracing::info!("Reconnecting to RabbitMQ");
+
+            // Connection should be re-established
+            let new_connection = Connection::connect(&uri, ConnectionProperties::default()).await;
+
+            match new_connection {
+                Ok(new_connection) => {
+                    *connection.write().await = Some(new_connection);
+                    tracing::info!("Reconnected to RabbitMQ");
+                }
+                Err(e) => {
+                    tracing::error!("Error while reconnecting to RabbitMQ: {:?}", e);
+                }
+            }
+
+            tokio::time::sleep(Duration::from_secs(2)).await;
+        }
+    }
+
     #[allow(dead_code)]
     pub async fn call<T>(
         &self,
@@ -81,12 +131,15 @@ impl RabbitMQClient {
     where
         T: Serialize,
     {
+        // Get current connection
+        let connection = self.connection.read().await;
+        if connection.is_none() {
+            return Err(Error::ConnectionDoesNotExist);
+        }
+        let connection = connection.as_ref().unwrap();
+
         // Create a channel
-        let channel = self
-            .connection
-            .create_channel()
-            .await
-            .map_err(Error::Lapin)?;
+        let channel = connection.create_channel().await.map_err(Error::Lapin)?;
 
         let serialized_payload_vec = to_vec(published_payload).map_err(Error::Serialization)?;
         let serialized_payload = serialized_payload_vec.as_slice();
@@ -133,12 +186,15 @@ impl RabbitMQClient {
     where
         T: Serialize,
     {
+        // Get current connection
+        let connection = self.connection.read().await;
+        if connection.is_none() {
+            return Err(Error::ConnectionDoesNotExist);
+        }
+        let connection = connection.as_ref().unwrap();
+
         // Create a channel
-        let channel = self
-            .connection
-            .create_channel()
-            .await
-            .map_err(Error::Lapin)?;
+        let channel = connection.create_channel().await.map_err(Error::Lapin)?;
 
         let serialized_payload_vec = to_vec(published_payload).map_err(Error::Serialization)?;
         let serialized_payload = serialized_payload_vec.as_slice();

front/public/locales/en/errors.json

@@ -42,7 +42,8 @@
       "ResponseTimeout": "Core: request timeout",
       "Serialization": "Core: cannot serialize request",
       "StatusParsing": "Core: cannot parse status",
-      "UnparsableErrorOutput": "Core returned an error in an unknown format"
+      "UnparsableErrorOutput": "Core returned an error in an unknown format",
+      "ConnectionDoesNotExist": "Core: message queue: connection not established"
     },
     "DatabaseAccessError": "Database access fatal error",
     "document": {

front/public/locales/fr/errors.json

@@ -38,11 +38,12 @@
       "ConnectionResetByPeer": "Core: réinitialisation de la connexion. Nouvelle tentative.",
       "CoreResponseFormatError": "Core: impossible d'analyser la réponse '{{msg}}'",
       "GenericCoreError": "Core: erreur {{raw_error}}",
-      "Lapin": "Core: file d'attente de message: erreur de protocole",
+      "Lapin": "Core: file d'attente de messages: erreur de protocole",
       "ResponseTimeout": "Core: temps d'attente écoulé",
       "Serialization": "Core: impossible de sérialiser la requête",
       "StatusParsing": "Core: impossible d'obtenir le status",
-      "UnparsableErrorOutput": "Core: a renvoyé une erreur dans un format inconnu"
+      "UnparsableErrorOutput": "Core: a renvoyé une erreur dans un format inconnu",
+      "ConnectionDoesNotExist": "Core: file d'attente de messages: connexion non établie"
     },
     "document": {
       "NotFound": "Document '{{document_key}}' non trouvé"