Classify errors by severity and discard messages that cannot be read

src/identification/errors.rs

@@ -1,3 +1,4 @@
+use crate::rpc::errors::Severity;
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -21,3 +22,18 @@ pub enum IdentificationError {
     #[error("Failed to parse identifier: {0}")]
     InvalidIdentifier(String),
 }
+
+impl IdentificationError {
+    pub fn severity(&self) -> Severity {
+        match self {
+            // Allow lists and routing tables are read at startup, so anything wrong with one is a configuration problem.
+            IdentificationError::IoError(_)
+            | IdentificationError::YamlParseError(_)
+            | IdentificationError::JsonParseError(_)
+            | IdentificationError::InvalidRoutingLine(_)
+            | IdentificationError::InvalidIdentifier(_) => Severity::Fatal,
+            // Editing routes while the table is in AllowAll or OnlyTo mode.
+            IdentificationError::NotInTableMode => Severity::Programming,
+        }
+    }
+}

src/lib.rs

@@ -11,4 +11,5 @@ pub use channel::remote::RemoteChannel;
 pub use node::{BrokerNode, ReceivedMessage};
 pub use rpc::client::BrokerClient;
 pub use rpc::client_async::BrokerClientAsync;
+pub use rpc::errors::{BrokerError, Severity};
 pub use rpc::server::BrokerServer;

src/node.rs

@@ -34,7 +34,6 @@ use crate::{
 #[derive(Debug)]
 pub enum ReceivedMessage {
     Msg(Identifier, String), //Id, Msg
-    Error(BrokerError),
 }
 
 // What the node was built for. Peers talk to other brokers over the network, services hand messages
@@ -254,9 +253,7 @@ impl BrokerNode {
     pub fn create_local_channel(&self, id: Identifier) -> Result<LocalChannel, BrokerError> {
         self.require_mode(NodeMode::Services)?;
         if id == self.local_id {
-            return Err(BrokerError::Other(
-                "Cannot create a local channel to the node's own local_id".to_string(),
-            ));
+            return Err(BrokerError::LocalChannelForOwnId);
         }
         Ok(self.server.create_local_channel(id))
     }
@@ -301,6 +298,23 @@ impl BrokerNode {
         self.local_channel.send(dest, data)
     }
 
+    /// Removes a row that could not be read, before the error travels.
+    /// A row left in place is read again on the next tick and fails the same way, so the queue never
+    /// drains past it.
+    fn discard_row_on_err<T, E>(&self, key: &str, result: Result<T, E>) -> Result<T, BrokerError>
+    where
+        BrokerError: From<E>,
+    {
+        match result {
+            Ok(value) => Ok(value),
+            Err(e) => {
+                warn!("Discarding unreadable key {}", key);
+                self.storage.remove(key)?;
+                Err(e.into())
+            }
+        }
+    }
+
     fn process_out_queue(&self) -> Result<(), BrokerError> {
         // send up to 50% of max capacity messages per tick
         let mut sent_per_dest: HashMap<String, usize> = HashMap::new();
@@ -313,14 +327,16 @@ impl BrokerNode {
 
         for key in self.storage.sorted_keys(&QueueType::OutQueue, None)? {
             if let Some(raw) = self.storage.get(&key)? {
-                let (pubk_hash, address) = BrokerNodeStorage::dest_from_key(&key)?;
+                let (pubk_hash, address) =
+                    self.discard_row_on_err(&key, BrokerNodeStorage::dest_from_key(&key))?;
 
                 // check if destination has not exceeded max messages per tick by destination pubk_hash
                 let sent = sent_per_dest.entry(pubk_hash.clone()).or_insert(0);
                 if *sent >= max_per_dest {
                     continue; // destination exhausted for this tick
                 }
-                let mut msg: OutgoingMsg = serde_json::from_str(&raw)?;
+                let mut msg: OutgoingMsg =
+                    self.discard_row_on_err(&key, serde_json::from_str(&raw))?;
                 if msg.retry.is_ready(now) == false {
                     continue;
                 }
@@ -339,11 +355,14 @@ impl BrokerNode {
                     self.storage.remove(&key)?;
                     *sent += 1;
                 } else {
+                    // Ok(false) is the broker refusing the route.
+                    let reason = match &attempt_to_send {
+                        Ok(_) => "routing denied".to_string(),
+                        Err(e) => e.to_string(),
+                    };
                     warn!(
                         "Failed to send queued message to {} at {}: {}",
-                        pubk_hash,
-                        address,
-                        attempt_to_send.as_ref().err().unwrap()
+                        pubk_hash, address, reason
                     );
 
                     msg.retry.record_attempt(&self.retry_policy, now);
@@ -444,14 +463,17 @@ impl BrokerNode {
             if let Some(x) = self.storage.get(&key)? {
                 let (identifier, data, ctx) = match queue_type {
                     QueueType::InQueue => {
-                        let identifier = BrokerNodeStorage::sender_from_key(&key)?;
+                        let identifier = self
+                            .discard_row_on_err(&key, BrokerNodeStorage::sender_from_key(&key))?;
                         (identifier, x, None)
                     }
                     QueueType::DeadLetterQueue => {
                         // No receiver id in deadletter, use COMMS_ID as default
-                        let (pubk_hash, _) = BrokerNodeStorage::dest_from_key(&key)?;
+                        let (pubk_hash, _) =
+                            self.discard_row_on_err(&key, BrokerNodeStorage::dest_from_key(&key))?;
                         let identifier = Identifier::new(pubk_hash, COMMS_ID);
-                        let msg = serde_json::from_str::<OutgoingMsg>(&x)?;
+                        let msg =
+                            self.discard_row_on_err(&key, serde_json::from_str::<OutgoingMsg>(&x))?;
                         (identifier, msg.payload, Some(msg.ctx))
                     }
                     _ => continue,
@@ -707,7 +729,6 @@ mod tests {
                     assert_eq!(data, &expected_msgs[i]);
                     assert_eq!(identifier.pubkey_hash, expected_pubk_hashes[i]);
                 }
-                _ => panic!("Expected message"),
             }
         }
     }
@@ -901,7 +922,6 @@ mod tests {
                 ReceivedMessage::Msg(_, data) => {
                     assert_eq!(data, &sent_msgs[i], "Message order violated at index {}", i);
                 }
-                _ => panic!("Expected Msg"),
             }
         }
 
@@ -982,15 +1002,13 @@ mod tests {
             .chain(recv1_second.into_iter())
             .map(|msg| match msg {
                 ReceivedMessage::Msg(_, data) => data,
-                _ => panic!("Unexpected error"),
             })
             .collect();
         let recv2_all: Vec<String> = recv2_first
             .into_iter()
             .chain(recv2_second.into_iter())
             .map(|msg| match msg {
                 ReceivedMessage::Msg(_, data) => data,
-                _ => panic!("Unexpected error"),
             })
             .collect();
         assert_eq!(recv1_all, sent_msgs_r1);
@@ -1043,7 +1061,6 @@ mod tests {
                         assert_eq!(data, &msg);
                         assert_eq!(ctx, CTX);
                     }
-                    _ => panic!("Expected dead letter message"),
                 }
                 break;
             }

src/retry.rs

@@ -3,7 +3,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use thiserror::Error;
 use tracing::info;
 
-use crate::rpc::config::BrokerNodeConfig;
+use crate::rpc::{config::BrokerNodeConfig, errors::Severity};
 
 #[derive(Debug, Clone)]
 pub struct RetryPolicy {
@@ -111,3 +111,9 @@ pub enum RetryPolicyError {
     )]
     DelayRangeTooSmall { min: u64, max: u64, attempts: u8 },
 }
+
+impl RetryPolicyError {
+    pub fn severity(&self) -> Severity {
+        Severity::Fatal
+    }
+}

src/rpc/errors.rs

@@ -4,23 +4,39 @@ use serde::{Deserialize, Serialize};
 use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
 use thiserror::Error;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Severity {
+    /// Storage is gone, a lock is poisoned, or the configuration is invalid. Nothing can proceed.
+    Fatal,
+    /// One message or one peer was refused or failed. Keep serving.
+    NonFatal,
+    /// The caller misused the API.
+    Programming,
+}
+
+impl Severity {
+    pub fn is_fatal(&self) -> bool {
+        !matches!(self, Severity::NonFatal)
+    }
+}
+
 #[derive(Error, Debug)]
 pub enum BrokerError {
     #[error("Rpc error")]
     RpcError(#[from] tarpc::client::RpcError),
 
-    #[error("IO error")]
-    IoError(#[from] std::io::Error),
+    #[error("Failed to reach a peer: {0}")]
+    ConnectError(#[from] std::io::Error),
+
+    #[error("Failed to bind the listener: {0}")]
+    BindError(std::io::Error),
 
     #[error("Serialization error {0}")]
     SerdeSerializationError(#[from] serde_json::Error),
 
     #[error("Identification error: {0}")]
     IdentificationError(#[from] identification::errors::IdentificationError),
 
-    #[error("Error parsing int")]
-    ParseIntError(#[from] std::num::ParseIntError),
-
     #[error("Broker client is disconnected")]
     Disconnected,
 
@@ -36,15 +52,9 @@ pub enum BrokerError {
     #[error("Broker storage error: {0}")]
     BrokerStorageError(#[from] storage::BrokerStorageError),
 
-    #[error("Failed to get address: {0}")]
+    #[error("Failed to get the peer address: {0}")]
     AddressError(#[from] std::net::AddrParseError),
 
-    #[error("Invalid identifier: {0}")]
-    InvalidIdentifier(String),
-
-    #[error("Closed channel")]
-    ClosedChannel,
-
     #[error("Mutex error: {0}")]
     MutexError(String),
 
@@ -78,19 +88,63 @@ pub enum BrokerError {
     #[error("Operation not available on a node built in {0} mode")]
     WrongNodeMode(String),
 
+    #[error("Cannot create a local channel to the node's own local_id")]
+    LocalChannelForOwnId,
+
     #[error("Time error: {0}")]
     TimeError(#[from] std::time::SystemTimeError),
 
     #[error("Retry policy error: {0}")]
     RetryPolicyError(#[from] RetryPolicyError),
 
-    #[error("Other error: {0}")]
-    Other(String),
-
     #[error("Setting file error: {0}")]
     Settings(#[from] ConfigError),
 }
 
+impl BrokerError {
+    pub fn severity(&self) -> Severity {
+        match self {
+            // Only a constructor reaches these.
+            BrokerError::BindError(_)
+            | BrokerError::RcgenError(_)
+            | BrokerError::RsaError(_)
+            | BrokerError::InvalidPrivateKey(_)
+            | BrokerError::X509ParseError(_)
+            | BrokerError::MutexError(_)
+            | BrokerError::Settings(_) => Severity::Fatal,
+
+            BrokerError::WrongNodeMode(_) | BrokerError::LocalChannelForOwnId => {
+                Severity::Programming
+            }
+
+            // Delegated.
+            BrokerError::BrokerRpcError(e) => e.severity(),
+            BrokerError::BrokerStorageError(e) => e.severity(),
+            BrokerError::IdentificationError(e) => e.severity(),
+            BrokerError::RetryPolicyError(e) => e.severity(),
+
+            // One message, one row, one peer, or on every dial.
+            BrokerError::AboutCertsAllow(_) // Considering certificate rotation.
+            | BrokerError::TlsError(_)
+            | BrokerError::RustlsError(_)
+            | BrokerError::PemParseError(_)
+            | BrokerError::RpcError(_)
+            | BrokerError::ConnectError(_)
+            | BrokerError::AddressError(_)
+            | BrokerError::Disconnected
+            | BrokerError::UnauthorizedFingerprint(_)
+            | BrokerError::SerdeSerializationError(_)
+            | BrokerError::MessageTooLarge(_, _)
+            | BrokerError::InvalidMessageContext { .. }
+            | BrokerError::TimeError(_) => Severity::NonFatal,
+        }
+    }
+
+    pub fn is_fatal(&self) -> bool {
+        self.severity().is_fatal()
+    }
+}
+
 impl<T> From<PoisonError<T>> for BrokerError {
     fn from(err: PoisonError<T>) -> Self {
         BrokerError::MutexError(err.to_string())
@@ -115,6 +169,23 @@ pub enum BrokerRpcError {
     QueueFull(String, u64),
 }
 
+impl BrokerRpcError {
+    pub fn severity(&self) -> Severity {
+        match self {
+            BrokerRpcError::MutexError(_) => Severity::Fatal,
+
+            BrokerRpcError::ParseError(_)
+            | BrokerRpcError::MessageTooLarge(_, _)
+            | BrokerRpcError::RateLimitExceeded
+            | BrokerRpcError::QueueFull(_, _) => Severity::NonFatal,
+        }
+    }
+
+    pub fn is_fatal(&self) -> bool {
+        self.severity().is_fatal()
+    }
+}
+
 pub trait FromMutexError {
     fn from_mutex_error(context: &'static str) -> Self;
 }

src/rpc/server.rs

@@ -56,10 +56,15 @@ impl BrokerServer {
         let rt = Runtime::new()?;
         let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
 
-        let listener = rt.block_on(TcpListener::bind(config.bind_addr()))?;
+        let listener = rt
+            .block_on(TcpListener::bind(config.bind_addr()))
+            .map_err(BrokerError::BindError)?;
         info!(
             "Listening with TLS on port {}",
-            listener.local_addr()?.port()
+            listener
+                .local_addr()
+                .map_err(BrokerError::BindError)?
+                .port()
         );
 
         let handler = ConnectionHandler::new(config, cert, storage.clone(), allow_list, routing)?;

src/rpc/tls_helper.rs

@@ -84,15 +84,18 @@ impl Cert {
         })
     }
     pub fn from_key_file(key_path: &str) -> Result<Self, BrokerError> {
-        let key_pem = std::fs::read_to_string(key_path)?;
+        let key_pem =
+            std::fs::read_to_string(key_path).map_err(|e| BrokerError::AboutCertsAllow(e.into()))?;
         Self::new_with_privk(&key_pem)
     }
 
     pub fn from_file(path: &str, name: &str) -> Result<Self, BrokerError> {
         let cert_path = format!("{path}/{name}.pem");
         let key_path = format!("{path}/{name}.key");
-        let cert_pem = std::fs::read_to_string(cert_path)?;
-        let key_pem = std::fs::read_to_string(key_path)?;
+        let cert_pem =
+            std::fs::read_to_string(cert_path).map_err(|e| BrokerError::AboutCertsAllow(e.into()))?;
+        let key_pem =
+            std::fs::read_to_string(key_path).map_err(|e| BrokerError::AboutCertsAllow(e.into()))?;
 
         let cert_blocks = pem::parse_many(&cert_pem)?;
         let first_cert_block = cert_blocks
@@ -173,9 +176,9 @@ impl Cert {
         bits: usize,
     ) -> Result<(), BrokerError> {
         let key = Self::generate_private_key(rng, bits)?;
-        std::fs::create_dir_all(path)?;
+        std::fs::create_dir_all(path).map_err(|e| BrokerError::AboutCertsAllow(e.into()))?;
         let key_path = format!("{path}/{name}.key");
-        std::fs::write(key_path, key)?;
+        std::fs::write(key_path, key).map_err(|e| BrokerError::AboutCertsAllow(e.into()))?;
         info!("Private key saved to {path}/{name}.key");
         Ok(())
     }