perf(ecstore): improve erasure write diagnostics and single-block performance (#3280)

* docs(object-capacity): add localized crate docs * fix(ecstore): improve quorum and transport diagnostics * perf(ecstore): add safe single-block write fast path * refactor(ecstore): collapse layered small write paths * chore(docs): keep issue 662 design note tracked * fix(docs): restore issue 662 design note * chore(docs): keep issue 662 design local only * feat(obs): add internode reliability metrics and dashboard * feat(obs): extend internode diagnostics and service logging * fix(docs): use AGENTS guide filename * perf(ecstore): reuse owned buffer in small encode * fix(ecstore): tighten small write diagnostics --------- Co-authored-by: cxymds <Cxymds@qq.com>
2026-08-09 22:59:59 +00:00 · 2026-06-08 17:45:56 +08:00
parent ddd35badad
commit ed73952cb6
16 changed files with 2455 additions and 123 deletions
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use rustfs_rio::{InternodeHttpError, InternodeHttpErrorKind};
 use std::hash::{Hash, Hasher};
 use std::io::{self};
 use std::path::PathBuf;
@@ -181,6 +182,26 @@ impl DiskError {
        matches!(err, &DiskError::FileVersionNotFound)
    }

+    pub fn is_retryable_internode_write_failure(&self) -> bool {
+        match self {
+            DiskError::Io(io_error) => io_error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<InternodeHttpError>())
+                .is_some_and(|err| err.kind().is_retryable()),
+            _ => false,
+        }
+    }
+
+    pub fn internode_http_error_kind(&self) -> Option<InternodeHttpErrorKind> {
+        match self {
+            DiskError::Io(io_error) => io_error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<InternodeHttpError>())
+                .map(InternodeHttpError::kind),
+            _ => None,
+        }
+    }
+
    // /// If all errors are of the same fatal disk error type, returns the corresponding error.
    // /// Otherwise, returns Ok.
    // pub fn check_disk_fatal_errs(errs: &[Option<Error>]) -> Result<()> {
@@ -241,7 +262,10 @@ impl From<rustfs_filemeta::Error> for DiskError {

 impl From<std::io::Error> for DiskError {
    fn from(e: std::io::Error) -> Self {
-        e.downcast::<DiskError>().unwrap_or_else(DiskError::Io)
+        match e.downcast::<DiskError>() {
+            Ok(disk_error) => disk_error,
+            Err(io_error) => DiskError::Io(io_error),
+        }
    }
 }

@@ -14,6 +14,19 @@

 use crate::disk::error::Error;

+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WriteQuorumFailureSummary {
+    pub required: usize,
+    pub achieved: usize,
+    pub failed: usize,
+    pub total: usize,
+    pub offline_disks: usize,
+    pub ignored_failures: usize,
+    pub retryable_failures: usize,
+    pub dominant_error: Option<Error>,
+    pub dominant_error_label: &'static str,
+}
+
 pub static OBJECT_OP_IGNORED_ERRS: &[Error] = &[
    Error::DiskNotFound,
    Error::FaultyDisk,
@@ -77,6 +90,58 @@ pub fn reduce_errs(errors: &[Option<Error>], ignored_errs: &[Error]) -> (usize,
    }
 }

+pub fn build_write_quorum_failure_summary(
+    errors: &[Option<Error>],
+    ignored_errs: &[Error],
+    quorum: usize,
+) -> WriteQuorumFailureSummary {
+    let total = errors.len();
+    let achieved = errors.iter().filter(|err| err.is_none()).count();
+    let failed = total.saturating_sub(achieved);
+    let offline_disks = count_errs(errors, &Error::DiskNotFound);
+    let ignored_failures = errors
+        .iter()
+        .filter_map(|err| err.as_ref())
+        .filter(|err| is_ignored_err(ignored_errs, err))
+        .count();
+    let retryable_failures = count_retryable_failures(errors);
+    let (_, dominant_error) = reduce_errs(errors, ignored_errs);
+    let dominant_error_label = dominant_error_label(errors, ignored_errs, dominant_error.as_ref());
+
+    WriteQuorumFailureSummary {
+        required: quorum,
+        achieved,
+        failed,
+        total,
+        offline_disks,
+        ignored_failures,
+        retryable_failures,
+        dominant_error,
+        dominant_error_label,
+    }
+}
+
+fn dominant_error_label(errors: &[Option<Error>], ignored_errs: &[Error], dominant_error: Option<&Error>) -> &'static str {
+    let Some(dominant_error) = dominant_error else {
+        return "nil_dominated";
+    };
+
+    if dominant_error == &Error::DiskNotFound {
+        return "disk_not_found";
+    }
+    if dominant_error == &Error::ShortWrite {
+        return "short_write";
+    }
+
+    errors
+        .iter()
+        .filter_map(|err| err.as_ref())
+        .find(|err| !is_ignored_err(ignored_errs, err) && *err == dominant_error)
+        .and_then(Error::internode_http_error_kind)
+        .map(|kind| kind.metric_label())
+        .unwrap_or("other_error")
+}
+
 pub fn is_ignored_err(ignored_errs: &[Error], err: &Error) -> bool {
    ignored_errs.iter().any(|e| e == err)
 }
@@ -85,6 +150,14 @@ pub fn count_errs(errors: &[Option<Error>], err: &Error) -> usize {
    errors.iter().filter(|&e| e.as_ref() == Some(err)).count()
 }

+pub fn count_retryable_failures(errors: &[Option<Error>]) -> usize {
+    errors
+        .iter()
+        .filter_map(|err| err.as_ref())
+        .filter(|err| err.is_retryable_internode_write_failure())
+        .count()
+}
+
 pub fn is_all_buckets_not_found(errs: &[Option<Error>]) -> bool {
    for err in errs.iter() {
        if let Some(err) = err {
@@ -163,6 +236,51 @@ mod tests {
        assert!(!is_ignored_err(&ignored, &e2));
    }

+    #[test]
+    fn test_build_write_quorum_failure_summary() {
+        let retryable = Error::from(rustfs_rio::new_test_internode_http_io_error(
+            rustfs_rio::InternodeHttpErrorKind::ConnectionReset,
+        ));
+        let non_retryable = err_io("other");
+        let errors = vec![
+            None,
+            None,
+            None,
+            None,
+            None,
+            Some(retryable),
+            Some(non_retryable),
+            Some(Error::DiskNotFound),
+        ];
+
+        let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 6);
+        assert_eq!(summary.required, 6);
+        assert_eq!(summary.achieved, 5);
+        assert_eq!(summary.failed, 3);
+        assert_eq!(summary.total, 8);
+        assert_eq!(summary.offline_disks, 1);
+        assert_eq!(summary.ignored_failures, 1);
+        assert_eq!(summary.retryable_failures, 1);
+        assert_eq!(summary.dominant_error, None);
+        assert_eq!(summary.dominant_error_label, "nil_dominated");
+    }
+
+    #[test]
+    fn test_build_write_quorum_failure_summary_preserves_internode_label() {
+        let retryable_a = Error::from(rustfs_rio::new_test_internode_http_io_error(
+            rustfs_rio::InternodeHttpErrorKind::ConnectionReset,
+        ));
+        let retryable_b = Error::from(rustfs_rio::new_test_internode_http_io_error(
+            rustfs_rio::InternodeHttpErrorKind::ConnectionReset,
+        ));
+        let errors = vec![Some(retryable_a), Some(retryable_b)];
+
+        let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 2);
+
+        assert_eq!(summary.retryable_failures, 2);
+        assert_eq!(summary.dominant_error_label, "connection_reset");
+    }
+
    #[test]
    fn test_reduce_errs_nil_tiebreak() {
        // Error::Nil and another error have the same count, should prefer Nil
@@ -13,8 +13,9 @@
 // limitations under the License.

 use crate::disk::error::Error;
-use crate::disk::error_reduce::count_errs;
-use crate::disk::error_reduce::{OBJECT_OP_IGNORED_ERRS, reduce_write_quorum_errs};
+use crate::disk::error_reduce::{
+    OBJECT_OP_IGNORED_ERRS, WriteQuorumFailureSummary, build_write_quorum_failure_summary, reduce_write_quorum_errs,
+};
 use crate::erasure_coding::BitrotWriterWrapper;
 use crate::erasure_coding::Erasure;
 use bytes::Bytes;
@@ -50,6 +51,28 @@ async fn drain_queued_inflight_bytes(rx: &mut mpsc::Receiver<Vec<Bytes>>) {
    }
 }

+fn dominant_error_summary_label(summary: &WriteQuorumFailureSummary) -> &'static str {
+    summary.dominant_error_label
+}
+
+fn format_write_quorum_failure(summary: &WriteQuorumFailureSummary) -> String {
+    format!(
+        "erasure write quorum (required={}, achieved={}, failed={}, total={}, offline-disks={}/{}, retryable-failures={}, dominant-error={})",
+        summary.required,
+        summary.achieved,
+        summary.failed,
+        summary.total,
+        summary.offline_disks,
+        summary.total,
+        summary.retryable_failures,
+        dominant_error_summary_label(summary)
+    )
+}
+
+fn quorum_dominant_error_metric_label(summary: &WriteQuorumFailureSummary) -> &'static str {
+    dominant_error_summary_label(summary)
+}
+
 pub(crate) struct MultiWriter<'a> {
    writers: &'a mut [Option<BitrotWriterWrapper>],
    write_quorum: usize,
@@ -109,25 +132,18 @@ impl<'a> MultiWriter<'a> {
        }

        if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) {
-            error!(
-                "reduce_write_quorum_errs: {:?}, offline-disks={}/{}, errs={:?}",
-                write_err,
-                count_errs(&self.errs, &Error::DiskNotFound),
-                self.writers.len(),
-                self.errs
-            );
-            return Err(std::io::Error::other(format!(
-                "Failed to write data: {} (offline-disks={}/{})",
-                write_err,
-                count_errs(&self.errs, &Error::DiskNotFound),
-                self.writers.len()
-            )));
+            let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum);
+            let summary_text = format_write_quorum_failure(&summary);
+            rustfs_io_metrics::internode_metrics::global_internode_metrics()
+                .record_erasure_write_quorum_failure("write", quorum_dominant_error_metric_label(&summary));
+            error!("reduce_write_quorum_errs: {:?}, {}, errs={:?}", write_err, summary_text, self.errs);
+            return Err(std::io::Error::other(format!("Failed to write data: {summary_text}")));
        }

+        let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum);
        Err(std::io::Error::other(format!(
-            "Failed to write data:  (offline-disks={}/{}): {}",
-            count_errs(&self.errs, &Error::DiskNotFound),
-            self.writers.len(),
+            "Failed to write data: {}: {}",
+            format_write_quorum_failure(&summary),
            self.errs
                .iter()
                .map(|e| e.as_ref().map_or_else(|| "<nil>".to_string(), |e| e.to_string()))
@@ -171,25 +187,21 @@ impl<'a> MultiWriter<'a> {
        }

        if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) {
+            let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum);
+            let summary_text = format_write_quorum_failure(&summary);
+            rustfs_io_metrics::internode_metrics::global_internode_metrics()
+                .record_erasure_write_quorum_failure("shutdown", quorum_dominant_error_metric_label(&summary));
            error!(
-                "reduce_write_quorum_errs during shutdown: {:?}, offline-disks={}/{}, errs={:?}",
-                write_err,
-                count_errs(&self.errs, &Error::DiskNotFound),
-                self.writers.len(),
-                self.errs
+                "reduce_write_quorum_errs during shutdown: {:?}, {}, errs={:?}",
+                write_err, summary_text, self.errs
            );
-            return Err(std::io::Error::other(format!(
-                "Failed to shutdown writers: {} (offline-disks={}/{})",
-                write_err,
-                count_errs(&self.errs, &Error::DiskNotFound),
-                self.writers.len()
-            )));
+            return Err(std::io::Error::other(format!("Failed to shutdown writers: {summary_text}")));
        }

+        let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum);
        Err(std::io::Error::other(format!(
-            "Failed to shutdown writers: (offline-disks={}/{}): {}",
-            count_errs(&self.errs, &Error::DiskNotFound),
-            self.writers.len(),
+            "Failed to shutdown writers: {}: {}",
+            format_write_quorum_failure(&summary),
            self.errs
                .iter()
                .map(|e| e.as_ref().map_or_else(|| "<nil>".to_string(), |e| e.to_string()))
@@ -200,6 +212,49 @@ impl<'a> MultiWriter<'a> {
 }

 impl Erasure {
+    async fn encode_small_direct<R>(
+        self: Arc<Self>,
+        mut reader: R,
+        writers: &mut [Option<BitrotWriterWrapper>],
+        quorum: usize,
+        require_single_block: bool,
+    ) -> std::io::Result<(R, usize)>
+    where
+        R: AsyncRead + Send + Sync + Unpin,
+    {
+        use tokio::io::AsyncReadExt;
+
+        let mut buf = Vec::with_capacity(self.block_size);
+        let total = if require_single_block {
+            let read_limit = self
+                .block_size
+                .checked_add(1)
+                .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "erasure block_size is too large"))?;
+            let read_limit = u64::try_from(read_limit)
+                .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "erasure block_size exceeds u64"))?;
+            (&mut reader).take(read_limit).read_to_end(&mut buf).await?
+        } else {
+            reader.read_to_end(&mut buf).await?
+        };
+
+        if total == 0 {
+            return Ok((reader, 0));
+        }
+
+        if require_single_block && total > self.block_size {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "single-block non-inline fast path expects total <= block_size",
+            ));
+        }
+
+        let shards = self.encode_data_owned(buf)?;
+        let mut mw = MultiWriter::new(writers, quorum);
+        mw.write(shards).await?;
+        mw.shutdown().await?;
+        Ok((reader, total))
+    }
+
    pub async fn encode<R>(
        self: Arc<Self>,
        mut reader: R,
@@ -307,27 +362,28 @@ impl Erasure {
    /// Reads all data, encodes directly, writes shards sequentially.
    pub async fn encode_inline_small<R>(
        self: Arc<Self>,
-        mut reader: R,
+        reader: R,
        writers: &mut [Option<BitrotWriterWrapper>],
        quorum: usize,
    ) -> std::io::Result<(R, usize)>
    where
        R: AsyncRead + Send + Sync + Unpin,
    {
-        use tokio::io::AsyncReadExt;
+        self.encode_small_direct(reader, writers, quorum, false).await
+    }

-        let mut buf = Vec::with_capacity(self.block_size);
-        let total = reader.read_to_end(&mut buf).await?;
-
-        if total == 0 {
-            return Ok((reader, 0));
-        }
-
-        let shards = self.encode_data(&buf)?;
-        let mut mw = MultiWriter::new(writers, quorum);
-        mw.write(shards).await?;
-        mw.shutdown().await?;
-        Ok((reader, total))
+    /// Fast path for single-block non-inline objects: avoids the producer/consumer
+    /// pipeline in `encode()` while keeping the same writer/quorum/shutdown semantics.
+    pub async fn encode_single_block_non_inline<R>(
+        self: Arc<Self>,
+        reader: R,
+        writers: &mut [Option<BitrotWriterWrapper>],
+        quorum: usize,
+    ) -> std::io::Result<(R, usize)>
+    where
+        R: AsyncRead + Send + Sync + Unpin,
+    {
+        self.encode_small_direct(reader, writers, quorum, true).await
    }
 }

@@ -491,6 +547,75 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn encode_single_block_non_inline_payload_writes_all_shards() {
+        const DATA_SHARDS: usize = 2;
+        const PARITY_SHARDS: usize = 2;
+        const TOTAL_SHARDS: usize = DATA_SHARDS + PARITY_SHARDS;
+        const BLOCK_SIZE: usize = 64;
+
+        let committed: Vec<Arc<Mutex<Vec<u8>>>> = (0..TOTAL_SHARDS).map(|_| Arc::new(Mutex::new(Vec::new()))).collect();
+
+        let mut writers: Vec<Option<BitrotWriterWrapper>> = committed
+            .iter()
+            .map(|c| {
+                Some(BitrotWriterWrapper::new(
+                    CustomWriter::new_tokio_writer(DeferredCommitWriter::new(c.clone())),
+                    BLOCK_SIZE / DATA_SHARDS,
+                    HashAlgorithm::HighwayHash256S,
+                ))
+            })
+            .collect();
+
+        let payload = b"hello single block";
+        let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
+        let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload.to_vec()));
+        let (_reader, total) = erasure
+            .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS)
+            .await
+            .unwrap();
+
+        assert_eq!(total, payload.len());
+        for (i, c) in committed.iter().enumerate() {
+            assert!(!c.lock().unwrap().is_empty(), "shard {i} should have received data");
+        }
+    }
+
+    #[tokio::test]
+    async fn encode_single_block_non_inline_rejects_multi_block_payload() {
+        const DATA_SHARDS: usize = 2;
+        const PARITY_SHARDS: usize = 2;
+        const TOTAL_SHARDS: usize = DATA_SHARDS + PARITY_SHARDS;
+        const BLOCK_SIZE: usize = 64;
+
+        let committed: Vec<Arc<Mutex<Vec<u8>>>> = (0..TOTAL_SHARDS).map(|_| Arc::new(Mutex::new(Vec::new()))).collect();
+
+        let mut writers: Vec<Option<BitrotWriterWrapper>> = committed
+            .iter()
+            .map(|c| {
+                Some(BitrotWriterWrapper::new(
+                    CustomWriter::new_tokio_writer(DeferredCommitWriter::new(c.clone())),
+                    BLOCK_SIZE / DATA_SHARDS,
+                    HashAlgorithm::HighwayHash256S,
+                ))
+            })
+            .collect();
+
+        let payload = vec![1u8; BLOCK_SIZE + 1];
+        let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
+        let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload));
+        let err = erasure
+            .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS)
+            .await
+            .expect_err("single-block fast path must reject oversized readers");
+
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+        assert!(err.to_string().contains("single-block non-inline fast path"));
+        for c in committed {
+            assert!(c.lock().unwrap().is_empty());
+        }
+    }
+
    #[test]
    fn encode_channel_capacity_never_returns_zero() {
        assert_eq!(encode_channel_capacity(0, 1024), 1);
@@ -498,6 +623,29 @@ mod tests {
        assert_eq!(encode_channel_capacity(4096, 1024), 1);
    }

+    #[test]
+    fn write_quorum_failure_summary_uses_stable_dominant_error_label() {
+        let err = Error::from(rustfs_rio::new_test_internode_http_io_error(
+            rustfs_rio::InternodeHttpErrorKind::ConnectionReset,
+        ));
+        let summary = WriteQuorumFailureSummary {
+            required: 2,
+            achieved: 0,
+            failed: 2,
+            total: 2,
+            offline_disks: 0,
+            ignored_failures: 0,
+            retryable_failures: 2,
+            dominant_error: Some(err),
+            dominant_error_label: "connection_reset",
+        };
+        let text = format_write_quorum_failure(&summary);
+
+        assert!(text.contains("dominant-error=connection_reset"));
+        assert!(!text.contains("/rustfs/rpc/put_file_stream"));
+        assert!(!text.contains("PUT "));
+    }
+
    #[test]
    fn encode_channel_capacity_respects_budget_and_hard_cap() {
        assert_eq!(encode_channel_capacity(4 * 1024 * 1024, 32 * 1024 * 1024), 8);
@@ -36,7 +36,7 @@ use metrics::counter;
 use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo};
 use rustfs_io_metrics::internode_metrics::{
    INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC,
-    global_internode_metrics,
+    INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics,
 };
 use rustfs_protos::evict_failed_connection;
 use rustfs_protos::proto_gen::node_service::RenamePartRequest;
@@ -74,6 +74,9 @@ enum FailureHealthAction {
    IgnoreFailure,
 }

+const REMOTE_DISK_OPEN_WRITE_MAX_ATTEMPTS: usize = 2;
+const REMOTE_DISK_OPEN_WRITE_RETRY_BACKOFF: Duration = Duration::from_millis(20);
+
 async fn copy_stream_with_buffer<R, W>(reader: &mut R, writer: &mut W, buffer_size: usize) -> io::Result<u64>
 where
    R: AsyncRead + Unpin,
@@ -119,6 +122,10 @@ impl RemoteDisk {
        err_text.contains("httpreader stream error") || err_text.contains("error decoding response body")
    }

+    fn is_retryable_open_write_error(err: &DiskError) -> bool {
+        err.is_retryable_internode_write_failure()
+    }
+
    pub(crate) async fn new(ep: &Endpoint, opt: &DiskOption, data_transport: Arc<dyn InternodeDataTransport>) -> Result<Self> {
        let addr = if let Some(port) = ep.url.port() {
            format!("{}://{}:{}", ep.url.scheme(), ep.url.host_str().unwrap(), port)
@@ -156,6 +163,50 @@ impl RemoteDisk {
        self.health.last_capacity_snapshot()
    }

+    async fn open_write_with_retry(&self, request: WriteStreamRequest) -> Result<FileWriter> {
+        let mut attempt = 1;
+        let mut last_retry_classification = None;
+        loop {
+            match self.data_transport.open_write(request.clone()).await {
+                Ok(writer) => {
+                    if attempt > 1
+                        && let Some(classification) = last_retry_classification
+                    {
+                        global_internode_metrics().record_retry_success_for_operation_and_backend(
+                            rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM,
+                            INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
+                            classification,
+                        );
+                    }
+                    return Ok(writer);
+                }
+                Err(err) if attempt < REMOTE_DISK_OPEN_WRITE_MAX_ATTEMPTS && Self::is_retryable_open_write_error(&err) => {
+                    if let Some(classification) = err.internode_http_error_kind() {
+                        let classification = classification.metric_label();
+                        global_internode_metrics().record_retry_for_operation_and_backend(
+                            rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM,
+                            INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
+                            classification,
+                        );
+                        last_retry_classification = Some(classification);
+                    }
+                    debug!(
+                        endpoint = %request.endpoint,
+                        volume = %request.volume,
+                        path = %request.path,
+                        append = request.append,
+                        size = request.size,
+                        attempt,
+                        "retrying remote open_write after retryable transport error"
+                    );
+                    tokio::time::sleep(REMOTE_DISK_OPEN_WRITE_RETRY_BACKOFF).await;
+                    attempt += 1;
+                }
+                Err(err) => return Err(err),
+            }
+        }
+    }
+
    pub fn record_capacity_probe(&self, total: u64, used: u64, free: u64) {
        self.health.record_capacity_probe(total, used, free);
    }
@@ -1342,16 +1393,15 @@ impl DiskAPI for RemoteDisk {
            return Err(DiskError::FaultyDisk);
        }
        let disk = self.disk_ref().await;
-        self.data_transport
-            .open_write(WriteStreamRequest {
-                endpoint: self.endpoint.grid_host(),
-                disk,
-                volume: volume.to_string(),
-                path: path.to_string(),
-                append: true,
-                size: 0,
-            })
-            .await
+        self.open_write_with_retry(WriteStreamRequest {
+            endpoint: self.endpoint.grid_host(),
+            disk,
+            volume: volume.to_string(),
+            path: path.to_string(),
+            append: true,
+            size: 0,
+        })
+        .await
    }

    #[tracing::instrument(level = "debug", skip(self))]
@@ -1368,16 +1418,15 @@ impl DiskAPI for RemoteDisk {
            return Err(DiskError::FaultyDisk);
        }
        let disk = self.disk_ref().await;
-        self.data_transport
-            .open_write(WriteStreamRequest {
-                endpoint: self.endpoint.grid_host(),
-                disk,
-                volume: volume.to_string(),
-                path: path.to_string(),
-                append: false,
-                size: file_size,
-            })
-            .await
+        self.open_write_with_retry(WriteStreamRequest {
+            endpoint: self.endpoint.grid_host(),
+            disk,
+            volume: volume.to_string(),
+            path: path.to_string(),
+            append: false,
+            size: file_size,
+        })
+        .await
    }

    #[tracing::instrument(level = "debug", skip(self))]
@@ -1832,6 +1881,35 @@ mod tests {
        }
    }

+    #[derive(Debug, Clone)]
+    enum OpenWriteTestStep {
+        Error(DiskError),
+        Success,
+    }
+
+    #[derive(Debug, Clone, Default)]
+    struct RetryingOpenWriteInternodeDataTransport {
+        calls: Arc<StdMutex<Vec<RecordedTransportCall>>>,
+        steps: Arc<StdMutex<Vec<OpenWriteTestStep>>>,
+    }
+
+    impl RetryingOpenWriteInternodeDataTransport {
+        fn with_steps(steps: Vec<OpenWriteTestStep>) -> Self {
+            Self {
+                calls: Arc::new(StdMutex::new(Vec::new())),
+                steps: Arc::new(StdMutex::new(steps)),
+            }
+        }
+
+        fn calls(&self) -> Vec<RecordedTransportCall> {
+            self.calls.lock().expect("recorded transport calls lock poisoned").clone()
+        }
+
+        fn record(&self, call: RecordedTransportCall) {
+            self.calls.lock().expect("recorded transport calls lock poisoned").push(call);
+        }
+    }
+
    #[derive(Debug, Default)]
    struct EmptyTestReader;

@@ -1916,6 +1994,34 @@ mod tests {
        }
    }

+    #[async_trait::async_trait]
+    impl InternodeDataTransport for RetryingOpenWriteInternodeDataTransport {
+        async fn open_read(&self, _request: ReadStreamRequest) -> Result<FileReader> {
+            panic!("open_read should not be used in open_write retry test");
+        }
+
+        async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter> {
+            self.record(RecordedTransportCall::Write(request));
+            let step = self.steps.lock().expect("open_write retry steps lock poisoned").remove(0);
+            match step {
+                OpenWriteTestStep::Error(err) => Err(err),
+                OpenWriteTestStep::Success => Ok(Box::new(SinkTestWriter)),
+            }
+        }
+
+        async fn open_walk_dir(&self, _request: WalkDirStreamRequest) -> Result<FileReader> {
+            panic!("open_walk_dir should not be used in open_write retry test");
+        }
+
+        fn name(&self) -> &'static str {
+            "retrying-open-write"
+        }
+
+        fn capabilities(&self) -> InternodeDataTransportCapabilities {
+            InternodeDataTransportCapabilities::tcp_http()
+        }
+    }
+
    async fn new_remote_disk_with_transport(data_transport: Arc<dyn InternodeDataTransport>) -> RemoteDisk {
        let endpoint = Endpoint {
            url: url::Url::parse("http://remote-node:9000/data/rustfs0").unwrap(),
@@ -2283,6 +2389,47 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn test_remote_disk_create_file_retries_once_on_retryable_open_write_error() {
+        let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![
+            OpenWriteTestStep::Error(DiskError::from(rustfs_rio::new_test_internode_http_io_error(
+                rustfs_rio::InternodeHttpErrorKind::ConnectionReset,
+            ))),
+            OpenWriteTestStep::Success,
+        ]);
+        let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+        rustfs_io_metrics::internode_metrics::global_internode_metrics().reset_for_test();
+
+        let _created = remote_disk
+            .create_file("orig-bucket", "bucket", "object/part.1", 4096)
+            .await
+            .expect("retryable open_write error should recover");
+
+        let calls = transport.calls();
+        assert_eq!(calls.len(), 2, "create_file should retry exactly once");
+        let snapshot = rustfs_io_metrics::internode_metrics::global_internode_metrics().snapshot();
+        assert_eq!(snapshot.outgoing_requests_total, 0);
+    }
+
+    #[tokio::test]
+    async fn test_remote_disk_append_file_does_not_retry_non_retryable_open_write_error() {
+        let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![OpenWriteTestStep::Error(DiskError::from(
+            rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::DnsResolutionFailed),
+        ))]);
+        let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+
+        let err = match remote_disk.append_file("bucket", "object/part.2").await {
+            Ok(_) => panic!("non-retryable open_write error should be returned directly"),
+            Err(err) => err,
+        };
+
+        assert_eq!(
+            err.internode_http_error_kind(),
+            Some(rustfs_rio::InternodeHttpErrorKind::DnsResolutionFailed)
+        );
+        assert_eq!(transport.calls().len(), 1, "append_file should not retry non-retryable errors");
+    }
+
    #[tokio::test]
    async fn test_remote_disk_walk_dir_uses_configured_data_transport() {
        let transport = RecordingInternodeDataTransport::default();
@@ -773,6 +773,37 @@ fn delete_file_info_version_id(version_id: Option<Uuid>) -> Option<Uuid> {
    }
 }

+fn object_fits_single_block(object_size: i64, block_size: usize) -> bool {
+    match usize::try_from(object_size) {
+        Ok(size) => size > 0 && size <= block_size,
+        Err(_) => false,
+    }
+}
+
+fn should_use_inline_small_fast_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> bool {
+    is_inline_buffer && object_fits_single_block(object_size, block_size)
+}
+
+fn should_use_single_block_non_inline_fast_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> bool {
+    !is_inline_buffer && object_fits_single_block(object_size, block_size)
+}
+
+enum SmallWritePath {
+    Inline,
+    SingleBlockNonInline,
+    Pipeline,
+}
+
+fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
+    if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
+        SmallWritePath::Inline
+    } else if should_use_single_block_non_inline_fast_path(is_inline_buffer, object_size, block_size) {
+        SmallWritePath::SingleBlockNonInline
+    } else {
+        SmallWritePath::Pipeline
+    }
+}
+
 #[async_trait::async_trait]
 impl ObjectIO for SetDisks {
    #[tracing::instrument(level = "debug", skip(self))]
@@ -1065,10 +1096,10 @@ impl ObjectIO for SetDisks {
                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
            );

-            let use_fast_path = is_inline_buffer && data.size() <= fi.erasure.block_size as i64;
+            let write_path = classify_small_write_path(is_inline_buffer, data.size(), fi.erasure.block_size);

-            let (reader, w_size) = if use_fast_path {
-                match Arc::new(erasure)
+            let (reader, w_size) = match write_path {
+                SmallWritePath::Inline => match Arc::new(erasure)
                    .encode_inline_small(stream, &mut writers, write_quorum)
                    .await
                {
@@ -1077,15 +1108,24 @@ impl ObjectIO for SetDisks {
                        error!("encode_inline_small err {:?}", e);
                        return Err(e.into());
                    }
-                }
-            } else {
-                match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await {
+                },
+                SmallWritePath::SingleBlockNonInline => match Arc::new(erasure)
+                    .encode_single_block_non_inline(stream, &mut writers, write_quorum)
+                    .await
+                {
+                    Ok((r, w)) => (r, w),
+                    Err(e) => {
+                        error!("encode_single_block_non_inline err {:?}", e);
+                        return Err(e.into());
+                    }
+                },
+                SmallWritePath::Pipeline => match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await {
                    Ok((r, w)) => (r, w),
                    Err(e) => {
                        error!("encode err {:?}", e);
                        return Err(e.into());
                    }
-                }
+                },
            };

            let _ = mem::replace(&mut data.stream, reader);
@@ -2971,7 +3011,18 @@ impl MultipartOperations for SetDisks {
            HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
        );

-        let (reader, w_size) = Arc::new(erasure).encode(stream, &mut writers, write_quorum).await?; // TODO: delete temporary directory on error
+        let write_path = classify_small_write_path(false, data.size(), fi.erasure.block_size);
+
+        let (reader, w_size) = match write_path {
+            SmallWritePath::SingleBlockNonInline => {
+                Arc::new(erasure)
+                    .encode_single_block_non_inline(stream, &mut writers, write_quorum)
+                    .await?
+            }
+            SmallWritePath::Inline | SmallWritePath::Pipeline => {
+                Arc::new(erasure).encode(stream, &mut writers, write_quorum).await?
+            }
+        }; // TODO: delete temporary directory on error

        let _ = mem::replace(&mut data.stream, reader);

@@ -6716,6 +6767,47 @@ mod tests {
        assert_eq!(delete_file_info_version_id(None), None);
    }

+    #[test]
+    fn put_object_fast_path_selection_prefers_inline_only_when_inline_buffer_and_single_block() {
+        assert!(should_use_inline_small_fast_path(true, 1024, 4096));
+        assert!(!should_use_single_block_non_inline_fast_path(true, 1024, 4096));
+        assert!(matches!(classify_small_write_path(true, 1024, 4096), SmallWritePath::Inline));
+
+        assert!(!should_use_inline_small_fast_path(false, 1024, 4096));
+        assert!(should_use_single_block_non_inline_fast_path(false, 1024, 4096));
+        assert!(matches!(
+            classify_small_write_path(false, 1024, 4096),
+            SmallWritePath::SingleBlockNonInline
+        ));
+    }
+
+    #[test]
+    fn put_object_fast_path_selection_rejects_zero_and_multi_block_payloads() {
+        assert!(!should_use_inline_small_fast_path(true, 0, 4096));
+        assert!(!should_use_single_block_non_inline_fast_path(false, 0, 4096));
+        assert!(matches!(classify_small_write_path(true, 0, 4096), SmallWritePath::Pipeline));
+
+        assert!(!should_use_inline_small_fast_path(true, -1, 4096));
+        assert!(!should_use_single_block_non_inline_fast_path(false, -1, 4096));
+        assert!(matches!(classify_small_write_path(false, -1, 4096), SmallWritePath::Pipeline));
+
+        assert!(!should_use_inline_small_fast_path(true, 8192, 4096));
+        assert!(!should_use_single_block_non_inline_fast_path(false, 8192, 4096));
+        assert!(matches!(classify_small_write_path(false, 8192, 4096), SmallWritePath::Pipeline));
+    }
+
+    #[test]
+    fn put_object_part_fast_path_selection_matches_single_block_non_inline_rules() {
+        assert!(should_use_single_block_non_inline_fast_path(false, 4096, 4096));
+        assert!(should_use_single_block_non_inline_fast_path(false, 2048, 4096));
+        assert!(!should_use_single_block_non_inline_fast_path(false, 4097, 4096));
+        assert!(!should_use_single_block_non_inline_fast_path(false, 0, 4096));
+        assert!(matches!(
+            classify_small_write_path(false, 4096, 4096),
+            SmallWritePath::SingleBlockNonInline
+        ));
+    }
+
    #[test]
    fn test_is_cold_storage_class() {
        // Test cold storage classes