feat: error handling improvements

Author: evansims

crates/raft/src/auto_recovery.rs

@@ -28,7 +28,7 @@ use crate::error::{
     ApplyOperationsSnafu, BlockArchiveNotConfiguredSnafu, BlockReadSnafu, IndexLookupSnafu,
     RecoveryError, StateRootComputationSnafu,
 };
-use crate::log_storage::{AppliedStateAccessor, VaultHealthStatus, MAX_RECOVERY_ATTEMPTS};
+use crate::log_storage::{AppliedStateAccessor, MAX_RECOVERY_ATTEMPTS, VaultHealthStatus};
 use crate::types::{LedgerNodeId, LedgerRequest, LedgerResponse, LedgerTypeConfig};
 
 /// Default interval between recovery scans.
@@ -144,7 +144,10 @@ impl AutoRecoveryJob {
     /// Calculate retry delay with exponential backoff.
     fn retry_delay(&self, attempt: u8) -> Duration {
         let multiplier = 2u64.saturating_pow(attempt.saturating_sub(1) as u32);
-        let delay = self.config.base_retry_delay.saturating_mul(multiplier as u32);
+        let delay = self
+            .config
+            .base_retry_delay
+            .saturating_mul(multiplier as u32);
         std::cmp::min(delay, self.config.max_retry_delay)
     }
 
@@ -171,7 +174,10 @@ impl AutoRecoveryJob {
                 VaultHealthStatus::Diverged { .. } => {
                     needs_recovery.push((namespace_id, vault_id, health));
                 }
-                VaultHealthStatus::Recovering { started_at, attempt } => {
+                VaultHealthStatus::Recovering {
+                    started_at,
+                    attempt,
+                } => {
                     if *attempt < MAX_RECOVERY_ATTEMPTS
                         && self.is_ready_for_retry(*started_at, *attempt)
                     {
@@ -213,7 +219,16 @@ impl AutoRecoveryJob {
         // First, transition to Recovering state
         let now = chrono::Utc::now().timestamp();
         if let Err(e) = self
-            .propose_health_update(namespace_id, vault_id, false, None, None, None, Some(attempt), Some(now))
+            .propose_health_update(
+                namespace_id,
+                vault_id,
+                false,
+                None,
+                None,
+                None,
+                Some(attempt),
+                Some(now),
+            )
             .await
         {
             warn!(
@@ -228,9 +243,7 @@ impl AutoRecoveryJob {
 
         info!(
             namespace_id,
-            vault_id,
-            attempt,
-            "Starting vault recovery attempt"
+            vault_id, attempt, "Starting vault recovery attempt"
         );
 
         // Get the expected state root from the divergence info
@@ -258,7 +271,16 @@ impl AutoRecoveryJob {
                 if computed_root == expected_root || expected_root == ledger_types::ZERO_HASH {
                     // Recovery successful
                     if let Err(e) = self
-                        .propose_health_update(namespace_id, vault_id, true, None, None, None, None, None)
+                        .propose_health_update(
+                            namespace_id,
+                            vault_id,
+                            true,
+                            None,
+                            None,
+                            None,
+                            None,
+                            None,
+                        )
                         .await
                     {
                         warn!(
@@ -320,10 +342,7 @@ impl AutoRecoveryJob {
 
         debug!(
             namespace_id,
-            vault_id,
-            start_height,
-            tip_height,
-            "Replaying vault state for recovery"
+            vault_id, start_height, tip_height, "Replaying vault state for recovery"
         );
 
         // Replay blocks from start_height to tip
@@ -385,10 +404,7 @@ impl AutoRecoveryJob {
                             .iter()
                             .find(|v| v.vault_id == vault_id)
                         {
-                            return Ok((
-                                vault_state.vault_height + 1,
-                                vault_state.state_root,
-                            ));
+                            return Ok((vault_state.vault_height + 1, vault_state.state_root));
                         }
                     }
                 }
@@ -423,14 +439,14 @@ impl AutoRecoveryJob {
             recovery_started_at,
         };
 
-        let result = self
-            .raft
-            .client_write(request)
-            .await
-            .map_err(|e| RecoveryError::RaftConsensus {
-                message: format!("{:?}", e),
-                backtrace: snafu::Backtrace::generate(),
-            })?;
+        let result =
+            self.raft
+                .client_write(request)
+                .await
+                .map_err(|e| RecoveryError::RaftConsensus {
+                    message: format!("{:?}", e),
+                    backtrace: snafu::Backtrace::generate(),
+                })?;
 
         match result.data {
             LedgerResponse::VaultHealthUpdated { success: true } => Ok(()),

crates/raft/src/batching.rs

@@ -581,7 +581,7 @@ mod tests {
     #[tokio::test]
     async fn test_eager_commit_flushes_quickly() {
         let config = BatchConfig {
-            max_batch_size: 100, // High limit
+            max_batch_size: 100,                    // High limit
             batch_timeout: Duration::from_secs(10), // Very long timeout
             tick_interval: Duration::from_millis(5),
             eager_commit: true, // Enable eager commit

crates/raft/src/error.rs

@@ -56,7 +56,10 @@ pub enum RecoveryError {
 
     /// Raft consensus write failed.
     #[snafu(display("Raft consensus write failed: {message}"))]
-    RaftConsensus { message: String, backtrace: Backtrace },
+    RaftConsensus {
+        message: String,
+        backtrace: Backtrace,
+    },
 
     /// Health update was rejected by the state machine.
     #[snafu(display("Health update rejected: {reason}"))]
@@ -88,7 +91,10 @@ pub enum SagaError {
 
     /// Raft consensus write failed.
     #[snafu(display("Raft write failed for saga operation: {message}"))]
-    SagaRaftWrite { message: String, backtrace: Backtrace },
+    SagaRaftWrite {
+        message: String,
+        backtrace: Backtrace,
+    },
 
     /// Entity read from state failed.
     #[snafu(display("Failed to read {entity_type} from state: {source}"))]
@@ -106,7 +112,10 @@ pub enum SagaError {
 
     /// Sequence allocation failed.
     #[snafu(display("Sequence allocation failed: {message}"))]
-    SequenceAllocation { message: String, backtrace: Backtrace },
+    SequenceAllocation {
+        message: String,
+        backtrace: Backtrace,
+    },
 
     /// Unexpected response from state machine.
     #[snafu(display("Unexpected saga response: {description}"))]
@@ -123,7 +132,10 @@ pub enum SagaError {
 pub enum OrphanCleanupError {
     /// Raft consensus write failed.
     #[snafu(display("Raft write failed during orphan cleanup: {message}"))]
-    OrphanRaftWrite { message: String, backtrace: Backtrace },
+    OrphanRaftWrite {
+        message: String,
+        backtrace: Backtrace,
+    },
 
     /// Failed to delete orphaned entity.
     #[snafu(display("Failed to delete orphan entity {key}: {reason}"))]
@@ -142,7 +154,10 @@ pub enum OrphanCleanupError {
 pub enum ServiceError {
     /// Raft consensus operation failed.
     #[snafu(display("Raft operation failed: {message}"))]
-    Raft { message: String, backtrace: Backtrace },
+    Raft {
+        message: String,
+        backtrace: Backtrace,
+    },
 
     /// Storage operation failed.
     #[snafu(display("Storage operation failed: {source}"))]

crates/raft/src/file_lock.rs

@@ -56,9 +56,7 @@ impl std::fmt::Display for LockError {
 impl std::error::Error for LockError {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
-            LockError::CreateFailed(_, err) | LockError::DirectoryCreateFailed(_, err) => {
-                Some(err)
-            }
+            LockError::CreateFailed(_, err) | LockError::DirectoryCreateFailed(_, err) => Some(err),
             LockError::AlreadyLocked(_) => None,
         }
     }
@@ -114,8 +112,8 @@ impl DataDirLock {
         }
 
         // Create/open the lock file
-        let file = File::create(&lock_path)
-            .map_err(|e| LockError::CreateFailed(lock_path.clone(), e))?;
+        let file =
+            File::create(&lock_path).map_err(|e| LockError::CreateFailed(lock_path.clone(), e))?;
 
         // Try to acquire an exclusive lock (non-blocking)
         match file.try_lock_exclusive() {
@@ -189,6 +187,7 @@ impl Drop for DataDirLock {
 }
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
 mod tests {
     use super::*;
     use tempfile::TempDir;

crates/raft/src/idempotency.rs

@@ -155,7 +155,13 @@ impl IdempotencyCache {
         }
 
         // Not a duplicate, insert the new result
-        self.insert(namespace_id, vault_id, client_id.to_string(), sequence, result);
+        self.insert(
+            namespace_id,
+            vault_id,
+            client_id.to_string(),
+            sequence,
+            result,
+        );
         None
     }
 
@@ -280,7 +286,10 @@ mod tests {
 
         // Same client, same sequence, but different vault should NOT be duplicate
         let cached = cache.check(1, 2, "client-1", 1);
-        assert!(cached.is_none(), "different vault should not be a duplicate");
+        assert!(
+            cached.is_none(),
+            "different vault should not be a duplicate"
+        );
 
         // Now insert for vault 2
         cache.insert(1, 2, "client-1".to_string(), 1, result);
@@ -302,7 +311,10 @@ mod tests {
 
         // Same client, same vault, same sequence, but different namespace should NOT be duplicate
         let cached = cache.check(2, 1, "client-1", 1);
-        assert!(cached.is_none(), "different namespace should not be a duplicate");
+        assert!(
+            cached.is_none(),
+            "different namespace should not be a duplicate"
+        );
     }
 
     /// DESIGN.md compliance test: sequence numbers must be monotonically increasing.

crates/raft/src/learner_refresh.rs

@@ -28,8 +28,8 @@ use tonic::transport::Channel;
 use tracing::{debug, info, warn};
 
 use crate::log_storage::AppliedStateAccessor;
-use crate::proto::system_discovery_service_client::SystemDiscoveryServiceClient;
 use crate::proto::GetSystemStateRequest;
+use crate::proto::system_discovery_service_client::SystemDiscoveryServiceClient;
 use crate::types::{LedgerNodeId, LedgerTypeConfig};
 
 /// Default refresh interval for learner background task.
@@ -154,11 +154,7 @@ impl LearnerRefreshJob {
 
         membership
             .voter_ids()
-            .filter_map(|id| {
-                membership
-                    .get_node(&id)
-                    .map(|node| (id, node.addr.clone()))
-            })
+            .filter_map(|id| membership.get_node(&id).map(|node| (id, node.addr.clone())))
             .collect()
     }
 

crates/raft/src/lib.rs

@@ -51,14 +51,18 @@ pub mod proto {
 
 pub use auto_recovery::{AutoRecoveryConfig, AutoRecoveryJob, RecoveryResult};
 pub use batching::{BatchConfig, BatchError, BatchWriter, BatchWriterHandle};
-pub use learner_refresh::{CachedSystemState, LearnerRefreshConfig, LearnerRefreshJob};
 pub use block_compaction::BlockCompactor;
 pub use file_lock::{DataDirLock, LockError};
 pub use idempotency::IdempotencyCache;
+pub use learner_refresh::{CachedSystemState, LearnerRefreshConfig, LearnerRefreshJob};
 pub use log_storage::{
     AppliedState, AppliedStateAccessor, NamespaceMeta, RaftLogStore, SequenceCounters,
     VaultHealthStatus, VaultMeta,
 };
+pub use multi_raft::{
+    MultiRaftConfig, MultiRaftError, MultiRaftManager, MultiRaftStats, ShardConfig, ShardGroup,
+};
+pub use multi_shard_server::MultiShardLedgerServer;
 pub use orphan_cleanup::OrphanCleanupJob;
 pub use pagination::{PageToken, PageTokenCodec, PageTokenError};
 pub use peer_maintenance::PeerMaintenance;
@@ -67,13 +71,9 @@ pub use raft_network::{GrpcRaftNetwork, GrpcRaftNetworkFactory};
 pub use rate_limit::{NamespaceRateLimiter, RateLimitExceeded};
 pub use saga_orchestrator::SagaOrchestrator;
 pub use server::LedgerServer;
-pub use multi_shard_server::MultiShardLedgerServer;
 pub use shard_router::{
     RouterConfig, RouterStats, RoutingError, RoutingInfo, ShardConnection, ShardRouter,
 };
-pub use multi_raft::{
-    MultiRaftConfig, MultiRaftError, MultiRaftManager, MultiRaftStats, ShardConfig, ShardGroup,
-};
 // Re-export multi-shard service types
 pub use services::{
     ForwardClient, MultiShardReadService, MultiShardResolver, MultiShardWriteService,