rust volume: share the I/O-error tracker between Volume and EcVolume (#11351)

* rust volume: share the I/O-error tracker between Volume and EcVolume

Volume and EcVolume each carried the same three fields - a mutex-held
last error, a consecutive count and a sticky quarantine flag - and the
same four methods over them, identical except for the path qualifier on
is_storage_io_error. The tolerance the count is compared against was a
fourth copy: heartbeat.rs held VOLUME_IO_ERROR_TOLERANCE for volumes,
ec_volume.rs held IO_ERROR_TOLERANCE for EC, and the volume test helper
open-coded the same 3, so the two paths could drift apart silently.

Go keeps this in one place already: weed/storage/io_error.go holds
IoErrorTracker, IoErrorTolerance and isStorageIoError, and Volume embeds
the tracker. Go's EcVolume has to re-implement it only because those
fields are unexported and EC lives in another package.

storage::io_error::IoErrorTracker now owns that state, with record /
state / should_quarantine / mark_quarantined / reset and the single
IO_ERROR_TOLERANCE. is_storage_io_error moves into the same file, so it
sits with the tracker that is now its only caller, the way io_error.go
is laid out. Both volume kinds embed one tracker and keep their existing
method names as delegates, so the ~16 internal call sites and the
readers in heartbeat.rs, store.rs and grpc_server.rs change only where
the two threshold comparisons become should_quarantine().

Volume::last_io_error and EcVolume::reset_io_error_state had no callers
and are gone.

Unchanged: what counts as a storage-media error - is_storage_io_error
changed file, not body, and is still the single predicate both volume
kinds share, where Go's EcVolume tests EIO directly and so misses the
Windows codes. Also unchanged: the tolerance value, the metric increment
on every counted error, and the sticky quarantine - a success clears the
count and the last error but never the flag, which only reset lifts. In
the heartbeat the state read moved inside the quarantine branch, so the
common path no longer takes the tracker's mutex or clones the last-error
string; should_quarantine's two relaxed loads run either way.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* rust volume: hoist absolute_display_path into server

handlers.rs and ui.rs each held a byte-identical copy of the helper that
turns a configured -dir into an absolute path for display. The status
JSON and the status page are meant to show the same directory, so the
two copies had to be edited together to stay that way.

The helper now lives in server/mod.rs as pub(crate) and both callers use
it. No behaviour change: same body, same call sites.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* rust volume: keep EcVolume::reset_io_error_state

Moving both volume types onto the shared IoErrorTracker dropped
EcVolume's public reset while Volume kept its own, so the two sides of
the tracker drifted apart.

mark_quarantined is sticky: a later successful read clears the error
count through record(), but the quarantine flag only comes down through
reset(). Without the delegate an EC volume that hit sustained media
errors could not be returned to service in place once the storage was
repaired. Go exposes the same method as EcVolume.ResetIoErrorState
(weed/storage/erasure_coding/ec_volume.go:114).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* rust volume: name the shared tracker after Go's IoErrorTracker

- check_read_write_error, get_io_error_state, mark_io_quarantined,
  reset_io_error_state match weed/storage/io_error.go one to one
- io_error module is pub(crate) like the io module beside it
- restore EcVolume::reset_io_error_state so both volume kinds expose the
  same recovery surface
- trim comments that restate the code

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/server/handlers.rs

@@ -15,6 +15,7 @@ use axum::http::{HeaderMap, Method, Request, StatusCode, header};
 use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};
 
+use super::absolute_display_path;
 use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint};
 use super::volume_server::{VolumeServerState, normalize_outgoing_http_url, to_http_address};
 use crate::config::ReadMode;
@@ -3621,16 +3622,6 @@ async fn read_remote_chunk_needle(
 // Helpers
 // ============================================================================
 
-fn absolute_display_path(path: &str) -> String {
-    let p = std::path::Path::new(path);
-    if p.is_absolute() {
-        return path.to_string();
-    }
-    std::env::current_dir()
-        .map(|cwd| cwd.join(p).to_string_lossy().to_string())
-        .unwrap_or_else(|_| path.to_string())
-}
-
 fn build_disk_statuses(store: &crate::storage::store::Store) -> Vec<serde_json::Value> {
     let mut disk_statuses = Vec::new();
     for loc in &store.locations {

seaweed-volume/src/server/heartbeat.rs

@@ -24,7 +24,6 @@ use crate::storage::volume_report::VolumeReportKey;
 use crate::storage::volume_report_hash::report_hash;
 
 const DUPLICATE_UUID_RETRY_MESSAGE: &str = "duplicate UUIDs detected, retrying connection";
-const VOLUME_IO_ERROR_TOLERANCE: i32 = 3;
 const MAX_DUPLICATE_UUID_RETRIES: u32 = 3;
 
 /// Configuration for the heartbeat client.
@@ -952,8 +951,8 @@ fn build_heartbeat_with_ec_status(
             let volume_size = vol.dat_file_size().unwrap_or(0);
             let mut should_delete_volume = false;
 
-            let (_, io_count, io_quarantined) = vol.get_io_error_state();
-            if io_quarantined || io_count >= VOLUME_IO_ERROR_TOLERANCE {
+            if vol.should_quarantine() {
+                let (_, io_count, io_quarantined) = vol.get_io_error_state();
                 if !io_quarantined {
                     vol.mark_io_quarantined();
                     warn!(

seaweed-volume/src/server/mod.rs

@@ -38,6 +38,19 @@ pub fn status_with_context(context: &str, err: VolumeError) -> Status {
     Status::new(status.code(), format!("{context}: {}", status.message()))
 }
 
+/// Render a configured disk directory as an absolute path for display, so the
+/// status JSON and the UI show the same thing for a relative `-dir`. Falls
+/// back to the configured spelling when the current directory cannot be read.
+pub(crate) fn absolute_display_path(path: &str) -> String {
+    let p = std::path::Path::new(path);
+    if p.is_absolute() {
+        return path.to_string();
+    }
+    std::env::current_dir()
+        .map(|cwd| cwd.join(p).to_string_lossy().to_string())
+        .unwrap_or_else(|_| path.to_string())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

seaweed-volume/src/server/ui.rs

@@ -1,5 +1,6 @@
 use std::fmt::Write as _;
 
+use crate::server::absolute_display_path;
 use crate::server::server_stats;
 use crate::server::volume_server::VolumeServerState;
 use crate::storage::store::Store;
@@ -450,16 +451,6 @@ fn collect_ui_data(
     (disk_rows, volumes, remote_volumes, ec_volumes)
 }
 
-fn absolute_display_path(path: &str) -> String {
-    let p = std::path::Path::new(path);
-    if p.is_absolute() {
-        return path.to_string();
-    }
-    std::env::current_dir()
-        .map(|cwd| cwd.join(p).to_string_lossy().to_string())
-        .unwrap_or_else(|_| path.to_string())
-}
-
 fn join_i64(values: &[i64]) -> String {
     values
         .iter()

seaweed-volume/src/storage/erasure_coding/ec_volume.rs

@@ -13,13 +13,11 @@ use crate::pb::master_pb;
 use crate::storage::erasure_coding::ec_locate;
 use crate::storage::erasure_coding::ec_shard::*;
 use crate::storage::io::read_exact_at;
+use crate::storage::io_error::IoErrorTracker;
 use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size};
 use crate::storage::types::*;
 use crate::storage::volume_open::open_volume_file;
 
-/// An erasure-coded volume managing its local shards and index.
-pub const IO_ERROR_TOLERANCE: i32 = 3;
-
 /// The shard-location cache: where each shard lives, when that was last learned
 /// from the master, and whether a read has since disproved it. One struct behind
 /// one lock, because the freshness heuristic judges all three together (the map
@@ -46,6 +44,7 @@ pub(crate) struct ShardLocationCache {
     stale: bool,
 }
 
+/// An erasure-coded volume managing its local shards and index.
 pub struct EcVolume {
     pub volume_id: VolumeId,
     pub collection: String,
@@ -104,9 +103,9 @@ pub struct EcVolume {
     /// sidecar other than the one those two fields actually hold.
     pub(crate) bitrot_source_dir: String,
 
-    io_error_count: std::sync::atomic::AtomicI32,
-    io_error_quarantined: std::sync::atomic::AtomicBool,
-    last_io_error: std::sync::Mutex<Option<String>>,
+    /// Consecutive storage-media errors and the quarantine they lead to,
+    /// for EC volume health monitoring.
+    io_errors: IoErrorTracker,
 }
 
 /// Locate the `.vif` for a (collection, vid) by preferring the data dir
@@ -459,9 +458,7 @@ impl EcVolume {
             bitrot: None,
             bitrot_status: crate::storage::erasure_coding::ec_bitrot::BitrotStatus::Off,
             bitrot_source_dir: String::new(),
-            io_error_count: std::sync::atomic::AtomicI32::new(0),
-            io_error_quarantined: std::sync::atomic::AtomicBool::new(false),
-            last_io_error: std::sync::Mutex::new(None),
+            io_errors: IoErrorTracker::default(),
         };
 
         // Open .ecx file (sorted index) in read/write mode for in-place deletion marking.
@@ -1006,48 +1003,26 @@ impl EcVolume {
         refresh
     }
 
-    // ---- I/O error tracking (mirrors Go's EcVolume IoErrorTracker) ----
+    // ---- I/O error tracking ----
 
     pub fn check_read_write_error(&self, err: Option<&io::Error>) {
-        use std::sync::atomic::Ordering;
-        if let Some(e) = err
-            && crate::storage::volume::is_storage_io_error(e)
-        {
-            self.io_error_count.fetch_add(1, Ordering::Relaxed);
-            if let Ok(mut guard) = self.last_io_error.lock() {
-                *guard = Some(e.to_string());
-            }
-            crate::metrics::STORAGE_IO_ERROR_COUNTER.inc();
-            return;
-        }
-        self.io_error_count.store(0, Ordering::Relaxed);
-        if let Ok(mut guard) = self.last_io_error.lock()
-            && guard.is_some()
-        {
-            *guard = None;
-        }
+        self.io_errors.check_read_write_error(err);
     }
 
     pub fn get_io_error_state(&self) -> (Option<String>, i32, bool) {
-        use std::sync::atomic::Ordering;
-        let err = self.last_io_error.lock().ok().and_then(|g| g.clone());
-        let count = self.io_error_count.load(Ordering::Relaxed);
-        let quarantined = self.io_error_quarantined.load(Ordering::Relaxed);
-        (err, count, quarantined)
+        self.io_errors.get_io_error_state()
+    }
+
+    pub fn should_quarantine(&self) -> bool {
+        self.io_errors.should_quarantine()
     }
 
     pub fn mark_io_quarantined(&self) {
-        self.io_error_quarantined
-            .store(true, std::sync::atomic::Ordering::Relaxed);
+        self.io_errors.mark_io_quarantined();
     }
 
     pub fn reset_io_error_state(&self) {
-        use std::sync::atomic::Ordering;
-        self.io_error_count.store(0, Ordering::Relaxed);
-        self.io_error_quarantined.store(false, Ordering::Relaxed);
-        if let Ok(mut guard) = self.last_io_error.lock() {
-            *guard = None;
-        }
+        self.io_errors.reset_io_error_state();
     }
 
     // ---- Index operations ----

seaweed-volume/src/storage/io_error.rs

@@ -0,0 +1,202 @@
+//! Consecutive storage-media error tracking shared by `Volume` and
+//! `EcVolume`. Mirrors Go's `weed/storage/io_error.go`.
+
+use std::io;
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicBool, AtomicI32, Ordering};
+
+/// Consecutive storage-media errors allowed before the volume is quarantined.
+pub(crate) const IO_ERROR_TOLERANCE: i32 = 3;
+
+/// Returns true for I/O errors that indicate faulty storage media, not
+/// transient/network failures. On Unix this is EIO; on Windows it covers
+/// ERROR_CRC and ERROR_IO_DEVICE, which the kernel returns for failing disks.
+pub(crate) fn is_storage_io_error(e: &io::Error) -> bool {
+    #[cfg(unix)]
+    {
+        e.raw_os_error() == Some(libc::EIO)
+    }
+    #[cfg(windows)]
+    {
+        const ERROR_CRC: i32 = 23;
+        const ERROR_IO_DEVICE: i32 = 1117;
+        return e.raw_os_error() == Some(ERROR_CRC) || e.raw_os_error() == Some(ERROR_IO_DEVICE);
+    }
+    #[cfg(not(any(unix, windows)))]
+    {
+        false
+    }
+}
+
+/// Consecutive storage-media error state for one volume. `quarantined` is
+/// sticky: once set it survives later successful I/O and is lifted only by
+/// `reset_io_error_state`.
+#[derive(Default)]
+pub(crate) struct IoErrorTracker {
+    last: Mutex<Option<String>>,
+    count: AtomicI32,
+    quarantined: AtomicBool,
+}
+
+impl IoErrorTracker {
+    /// `Some(e)` records a failure, `None` a success. Only storage-media
+    /// failures count; every other outcome clears the count and last error.
+    pub(crate) fn check_read_write_error(&self, err: Option<&io::Error>) {
+        if let Some(e) = err
+            && is_storage_io_error(e)
+        {
+            self.count.fetch_add(1, Ordering::Relaxed);
+            if let Ok(mut guard) = self.last.lock() {
+                *guard = Some(e.to_string());
+            }
+            crate::metrics::STORAGE_IO_ERROR_COUNTER.inc();
+            return;
+        }
+        self.count.store(0, Ordering::Relaxed);
+        if let Ok(mut guard) = self.last.lock()
+            && guard.is_some()
+        {
+            *guard = None;
+        }
+    }
+
+    /// The last recorded error, the consecutive count, and the quarantine flag.
+    pub(crate) fn get_io_error_state(&self) -> (Option<String>, i32, bool) {
+        let err = self.last.lock().ok().and_then(|g| g.clone());
+        let count = self.count.load(Ordering::Relaxed);
+        let quarantined = self.quarantined.load(Ordering::Relaxed);
+        (err, count, quarantined)
+    }
+
+    pub(crate) fn should_quarantine(&self) -> bool {
+        self.quarantined.load(Ordering::Relaxed)
+            || self.count.load(Ordering::Relaxed) >= IO_ERROR_TOLERANCE
+    }
+
+    pub(crate) fn mark_io_quarantined(&self) {
+        self.quarantined.store(true, Ordering::Relaxed);
+    }
+
+    pub(crate) fn reset_io_error_state(&self) {
+        self.count.store(0, Ordering::Relaxed);
+        self.quarantined.store(false, Ordering::Relaxed);
+        if let Ok(mut guard) = self.last.lock() {
+            *guard = None;
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn set_last_io_error_for_test(&self, err: Option<&str>) {
+        if let Ok(mut guard) = self.last.lock() {
+            *guard = err.map(|value| value.to_string());
+        }
+        if err.is_some() {
+            self.count.store(IO_ERROR_TOLERANCE, Ordering::Relaxed);
+        } else {
+            self.count.store(0, Ordering::Relaxed);
+        }
+    }
+}
+
+// The tracker only reacts to errors `is_storage_io_error` recognises, which is
+// nothing at all on a platform that is neither Unix nor Windows.
+#[cfg(all(test, any(unix, windows)))]
+mod tests {
+    use super::*;
+
+    /// An OS error the platform reports for failing storage media.
+    #[cfg(unix)]
+    fn media_error() -> io::Error {
+        io::Error::from_raw_os_error(libc::EIO)
+    }
+
+    /// An OS error the platform reports for failing storage media.
+    #[cfg(windows)]
+    fn media_error() -> io::Error {
+        const ERROR_IO_DEVICE: i32 = 1117;
+        io::Error::from_raw_os_error(ERROR_IO_DEVICE)
+    }
+
+    #[test]
+    fn check_read_write_error_counts_consecutive_media_errors() {
+        let tracker = IoErrorTracker::default();
+        tracker.check_read_write_error(Some(&media_error()));
+        tracker.check_read_write_error(Some(&media_error()));
+
+        let (last, count, quarantined) = tracker.get_io_error_state();
+        assert_eq!(last, Some(media_error().to_string()));
+        assert_eq!(count, 2);
+        assert!(!quarantined);
+    }
+
+    #[test]
+    fn success_clears_the_count_and_the_last_error() {
+        let tracker = IoErrorTracker::default();
+        tracker.check_read_write_error(Some(&media_error()));
+        tracker.check_read_write_error(None);
+
+        assert_eq!(tracker.get_io_error_state(), (None, 0, false));
+    }
+
+    #[test]
+    fn non_media_error_clears_the_count() {
+        let tracker = IoErrorTracker::default();
+        tracker.check_read_write_error(Some(&media_error()));
+        tracker.check_read_write_error(Some(&io::Error::new(
+            io::ErrorKind::NotFound,
+            "no such file",
+        )));
+
+        assert_eq!(tracker.get_io_error_state(), (None, 0, false));
+    }
+
+    #[test]
+    fn should_quarantine_only_once_the_tolerance_is_reached() {
+        let tracker = IoErrorTracker::default();
+        for _ in 1..IO_ERROR_TOLERANCE {
+            tracker.check_read_write_error(Some(&media_error()));
+            assert!(!tracker.should_quarantine());
+        }
+        tracker.check_read_write_error(Some(&media_error()));
+        assert!(tracker.should_quarantine());
+    }
+
+    #[test]
+    fn quarantine_survives_later_successful_io() {
+        let tracker = IoErrorTracker::default();
+        tracker.mark_io_quarantined();
+        tracker.check_read_write_error(None);
+
+        assert_eq!(tracker.get_io_error_state(), (None, 0, true));
+        assert!(tracker.should_quarantine());
+    }
+
+    #[test]
+    fn reset_io_error_state_lifts_the_quarantine() {
+        let tracker = IoErrorTracker::default();
+        tracker.check_read_write_error(Some(&media_error()));
+        tracker.mark_io_quarantined();
+        tracker.reset_io_error_state();
+
+        assert_eq!(tracker.get_io_error_state(), (None, 0, false));
+        assert!(!tracker.should_quarantine());
+    }
+
+    #[test]
+    fn test_helper_arms_a_sustained_error() {
+        let tracker = IoErrorTracker::default();
+        tracker.set_last_io_error_for_test(Some("input/output error"));
+        assert!(tracker.should_quarantine());
+        assert_eq!(
+            tracker.get_io_error_state(),
+            (
+                Some("input/output error".to_string()),
+                IO_ERROR_TOLERANCE,
+                false
+            )
+        );
+
+        tracker.set_last_io_error_for_test(None);
+        assert_eq!(tracker.get_io_error_state(), (None, 0, false));
+    }
+}

seaweed-volume/src/storage/mod.rs

@@ -2,6 +2,7 @@ pub mod disk_location;
 pub mod erasure_coding;
 pub mod idx;
 pub(crate) mod io;
+pub(crate) mod io_error;
 pub mod needle;
 pub mod needle_map;
 pub mod store;

seaweed-volume/src/storage/store.rs

@@ -1144,16 +1144,11 @@ impl Store {
             for (vid, ec_vol) in loc.ec_volumes() {
                 if ec_vol.is_time_to_destroy() {
                     expired_vids.push(*vid);
+                } else if ec_vol.should_quarantine() {
+                    io_quarantined_vids.push(*vid);
                 } else {
-                    let (_, io_count, quarantined) = ec_vol.get_io_error_state();
-                    if quarantined
-                        || io_count >= crate::storage::erasure_coding::ec_volume::IO_ERROR_TOLERANCE
-                    {
-                        io_quarantined_vids.push(*vid);
-                    } else {
-                        ec_shards
-                            .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32));
-                    }
+                    ec_shards
+                        .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32));
                 }
             }
 

seaweed-volume/src/storage/volume.rs

@@ -22,6 +22,7 @@ use tracing::{debug, error, info, warn};
 
 use crate::storage::idx;
 use crate::storage::io::read_exact_at;
+use crate::storage::io_error::IoErrorTracker;
 use crate::storage::needle::needle::{self, Needle, NeedleError, get_actual_size};
 use crate::storage::needle_map::sorted_file::SortedFileNeedleMap;
 use crate::storage::needle_map::{CompactNeedleMap, NeedleMap, NeedleMapKind, RedbNeedleMap};
@@ -112,26 +113,6 @@ fn exceeds_expected_compacted_size(expected_live_bytes: u64, dst_dat_size: u64)
     expected_live_bytes > dst_dat_size
 }
 
-/// Returns true for I/O errors that indicate faulty storage media, not
-/// transient/network failures. On Unix this is EIO; on Windows it covers
-/// ERROR_CRC and ERROR_IO_DEVICE, which the kernel returns for failing disks.
-pub fn is_storage_io_error(e: &io::Error) -> bool {
-    #[cfg(unix)]
-    {
-        e.raw_os_error() == Some(libc::EIO)
-    }
-    #[cfg(windows)]
-    {
-        const ERROR_CRC: i32 = 23;
-        const ERROR_IO_DEVICE: i32 = 1117;
-        return e.raw_os_error() == Some(ERROR_CRC) || e.raw_os_error() == Some(ERROR_IO_DEVICE);
-    }
-    #[cfg(not(any(unix, windows)))]
-    {
-        false
-    }
-}
-
 // ============================================================================
 // VolumeInfo (.vif persistence)
 // ============================================================================
@@ -710,14 +691,9 @@ pub struct Volume {
     /// Compaction speed limit in bytes per second (0 = unlimited).
     pub compaction_byte_per_second: i64,
 
-    /// Tracks the last I/O error (EIO) for volume health monitoring.
-    /// Uses Mutex for interior mutability so reads (&self) can clear/set it.
-    last_io_error: Mutex<Option<String>>,
-    /// Consecutive EIO count; reset on success or non-EIO errors.
-    io_error_count: std::sync::atomic::AtomicI32,
-    /// Sticky quarantine flag set after sustained EIO; cleared only by
-    /// explicit recovery (mirrors Go's markIoQuarantined).
-    io_error_quarantined: std::sync::atomic::AtomicBool,
+    /// Consecutive storage-media errors and the quarantine they lead to,
+    /// for volume health monitoring.
+    io_errors: IoErrorTracker,
 
     /// Protobuf VolumeInfo for tiered storage (.vif file).
     ///
@@ -799,9 +775,7 @@ impl Volume {
             last_compact_revision: 0,
             is_compacting: false,
             compaction_byte_per_second: 0,
-            last_io_error: Mutex::new(None),
-            io_error_count: std::sync::atomic::AtomicI32::new(0),
-            io_error_quarantined: std::sync::atomic::AtomicBool::new(false),
+            io_errors: IoErrorTracker::default(),
             volume_info: PbVolumeInfo::default(),
         };
 
@@ -838,9 +812,7 @@ impl Volume {
             last_compact_revision: 0,
             is_compacting: false,
             compaction_byte_per_second: 0,
-            last_io_error: Mutex::new(None),
-            io_error_count: std::sync::atomic::AtomicI32::new(0),
-            io_error_quarantined: std::sync::atomic::AtomicBool::new(false),
+            io_errors: IoErrorTracker::default(),
             volume_info: PbVolumeInfo::default(),
         }
     }
@@ -4400,69 +4372,30 @@ impl Volume {
         self.dir != self.dir_idx && has_ecx(&volume_file_name(&self.dir, &self.collection, self.id))
     }
 
-    /// Check if an I/O error is a storage-media failure and record it for
-    /// health monitoring. On success (None), clears any previously recorded
-    /// EIO error. Matches Go's `checkReadWriteError` in volume_write.go.
+    /// Matches Go's `checkReadWriteError` in `weed/storage/io_error.go`.
     fn check_read_write_error(&self, err: Option<&io::Error>) {
-        use std::sync::atomic::Ordering;
-        if let Some(e) = err
-            && is_storage_io_error(e)
-        {
-            self.io_error_count.fetch_add(1, Ordering::Relaxed);
-            if let Ok(mut guard) = self.last_io_error.lock() {
-                *guard = Some(e.to_string());
-            }
-            crate::metrics::STORAGE_IO_ERROR_COUNTER.inc();
-            return;
-        }
-        self.io_error_count.store(0, Ordering::Relaxed);
-        if let Ok(mut guard) = self.last_io_error.lock()
-            && guard.is_some()
-        {
-            *guard = None;
-        }
-    }
-
-    /// Returns the last recorded I/O error string, if any.
-    pub fn last_io_error(&self) -> Option<String> {
-        self.last_io_error.lock().ok()?.clone()
+        self.io_errors.check_read_write_error(err);
     }
 
     pub fn get_io_error_state(&self) -> (Option<String>, i32, bool) {
-        use std::sync::atomic::Ordering;
-        let err = self.last_io_error.lock().ok().and_then(|g| g.clone());
-        let count = self.io_error_count.load(Ordering::Relaxed);
-        let quarantined = self.io_error_quarantined.load(Ordering::Relaxed);
-        (err, count, quarantined)
+        self.io_errors.get_io_error_state()
+    }
+
+    pub fn should_quarantine(&self) -> bool {
+        self.io_errors.should_quarantine()
     }
 
     pub fn mark_io_quarantined(&self) {
-        self.io_error_quarantined
-            .store(true, std::sync::atomic::Ordering::Relaxed);
+        self.io_errors.mark_io_quarantined();
     }
 
     pub fn reset_io_error_state(&self) {
-        use std::sync::atomic::Ordering;
-        self.io_error_count.store(0, Ordering::Relaxed);
-        self.io_error_quarantined.store(false, Ordering::Relaxed);
-        if let Ok(mut guard) = self.last_io_error.lock() {
-            *guard = None;
-        }
+        self.io_errors.reset_io_error_state();
     }
 
     #[cfg(test)]
     pub(crate) fn set_last_io_error_for_test(&self, err: Option<&str>) {
-        use std::sync::atomic::Ordering;
-        if let Ok(mut guard) = self.last_io_error.lock() {
-            *guard = err.map(|value| value.to_string());
-        }
-        // Set count at/above the heartbeat tolerance (3) so the test
-        // helper reflects a sustained error, not a single transient one.
-        if err.is_some() {
-            self.io_error_count.store(3, Ordering::Relaxed);
-        } else {
-            self.io_error_count.store(0, Ordering::Relaxed);
-        }
+        self.io_errors.set_last_io_error_for_test(err);
     }
 
     #[cfg(test)]