From ae3e0852711a3d12c305a1fb4197c0d613ca9208 Mon Sep 17 00:00:00 2001 From: Kevin Wahle <62912263+KevinWahle@users.noreply.github.com> Date: Fri, 21 Aug 2026 18:54:35 -0300 Subject: [PATCH] Classify errors by severity and discard messages that cannot be read --- src/identification/errors.rs | 16 ++++++ src/lib.rs | 1 + src/node.rs | 51 ++++++++++++------ src/retry.rs | 8 ++- src/rpc/errors.rs | 101 +++++++++++++++++++++++++++++------ src/rpc/server.rs | 9 +++- src/rpc/tls_helper.rs | 13 +++-- src/storage/errors.rs | 16 +++++- src/storage/node_storage.rs | 13 ++++- 9 files changed, 185 insertions(+), 43 deletions(-) diff --git a/src/identification/errors.rs b/src/identification/errors.rs index 696922e..8d28cdb 100644 --- a/src/identification/errors.rs +++ b/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, + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 796ae5c..5abb0ed 100644 --- a/src/lib.rs +++ b/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; diff --git a/src/node.rs b/src/node.rs index b8a9785..71ab9ed 100644 --- a/src/node.rs +++ b/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 { 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(&self, key: &str, result: Result) -> Result + where + BrokerError: From, + { + 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 = 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::(&x)?; + let msg = + self.discard_row_on_err(&key, serde_json::from_str::(&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,7 +1002,6 @@ mod tests { .chain(recv1_second.into_iter()) .map(|msg| match msg { ReceivedMessage::Msg(_, data) => data, - _ => panic!("Unexpected error"), }) .collect(); let recv2_all: Vec = recv2_first @@ -990,7 +1009,6 @@ mod tests { .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; } diff --git a/src/retry.rs b/src/retry.rs index 9dfa408..3342668 100644 --- a/src/retry.rs +++ b/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 + } +} diff --git a/src/rpc/errors.rs b/src/rpc/errors.rs index 0aa0a81..dc64f48 100644 --- a/src/rpc/errors.rs +++ b/src/rpc/errors.rs @@ -4,13 +4,32 @@ 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), @@ -18,9 +37,6 @@ pub enum BrokerError { #[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 From> for BrokerError { fn from(err: PoisonError) -> 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; } diff --git a/src/rpc/server.rs b/src/rpc/server.rs index 40b2455..93cdf8a 100644 --- a/src/rpc/server.rs +++ b/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)?; diff --git a/src/rpc/tls_helper.rs b/src/rpc/tls_helper.rs index 27cb8af..3b9cf44 100644 --- a/src/rpc/tls_helper.rs +++ b/src/rpc/tls_helper.rs @@ -84,15 +84,18 @@ impl Cert { }) } pub fn from_key_file(key_path: &str) -> Result { - 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 { 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(()) } diff --git a/src/storage/errors.rs b/src/storage/errors.rs index 954d5a5..9f22270 100644 --- a/src/storage/errors.rs +++ b/src/storage/errors.rs @@ -1,4 +1,4 @@ -use crate::rpc::errors::{BrokerRpcError, FromMutexError}; +use crate::rpc::errors::{BrokerRpcError, FromMutexError, Severity}; use thiserror::Error; #[derive(Error, Debug)] @@ -16,6 +16,20 @@ pub enum BrokerStorageError { InvalidIdentifier(String), } +impl BrokerStorageError { + pub fn severity(&self) -> Severity { + match self { + BrokerStorageError::MutexPoisoned => Severity::Fatal, + // One unreadable row, or one input that never should have been keyed. + BrokerStorageError::MalformedKey(_) | BrokerStorageError::InvalidIdentifier(_) => { + Severity::NonFatal + } + // The backend reports a missing key and an unopenable database through one type. All of it is considered non-fatal. + BrokerStorageError::Backend(_) => Severity::NonFatal, + } + } +} + impl FromMutexError for BrokerStorageError { fn from_mutex_error(_context: &'static str) -> Self { BrokerStorageError::MutexPoisoned diff --git a/src/storage/node_storage.rs b/src/storage/node_storage.rs index c8ae446..444a7c9 100644 --- a/src/storage/node_storage.rs +++ b/src/storage/node_storage.rs @@ -10,6 +10,7 @@ use crate::storage::errors::BrokerStorageError; use std::net::SocketAddr; use std::rc::Rc; use storage_backend::storage::{KeyValueStore, Storage}; +use tracing::warn; pub enum QueueType { OutQueue, @@ -169,12 +170,20 @@ impl BrokerNodeStorage { pub fn store_in_msgs(&self, msgs: &[Message]) -> Result<(), BrokerStorageError> { let tx = self.storage.begin_transaction(); for msg in msgs { - let key = self.msg_key( + let key = match self.msg_key( &QueueType::InQueue, msg.uid, &msg.from.pubkey_hash, &msg.from.id.to_string(), - )?; + ) { + Ok(key) => key, + Err(e) => { + // Dropped rather than failing the batch. Failing would roll back every message + // beside it and leave the whole batch unacknowledged. + warn!("Dropping message from {}: {}", msg.from, e); + continue; + } + }; self.storage.set(&key, msg.msg.clone(), Some(tx))?; } self.storage.commit_transaction(tx)?;