Merge branch 'main' into cxymds/fix-1853-lock-lease-reaper

crates/ecstore/src/cluster/rpc/internode_data_transport.rs

@@ -233,17 +233,11 @@ pub struct NsScannerCapabilityRequest {
 #[async_trait]
 pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
     async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader>;
-    async fn open_read_fresh(&self, request: ReadStreamRequest) -> Result<FileReader> {
-        self.open_read(request).await
-    }
     /// Opens an owned-chunk stream when this transport can retain receive-buffer
     /// ownership. `None` preserves the established `open_read` fallback.
     async fn open_read_chunks(&self, _request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
         Ok(None)
     }
-    async fn open_read_chunks_fresh(&self, request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
-        self.open_read_chunks(request).await
-    }
     async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter>;
     async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result<FileReader>;
     async fn open_ns_scanner(&self, _request: NsScannerStreamRequest) -> Result<FileReader> {
@@ -275,15 +269,6 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
         ))
     }
 
-    async fn open_read_fresh(&self, request: ReadStreamRequest) -> Result<FileReader> {
-        let url = build_read_file_stream_url(&request);
-        let mut headers = json_headers();
-        build_auth_headers(&url, &Method::GET, &mut headers)?;
-        Ok(Box::new(
-            HttpReader::new_fresh_connection_with_stall_timeout(url, Method::GET, headers, None, request.stall_timeout).await?,
-        ))
-    }
-
     async fn open_read_chunks(&self, request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
         let url = build_read_file_stream_url(&request);
         let mut headers = json_headers();
@@ -293,16 +278,6 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
         )))
     }
 
-    async fn open_read_chunks_fresh(&self, request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
-        let url = build_read_file_stream_url(&request);
-        let mut headers = json_headers();
-        build_auth_headers(&url, &Method::GET, &mut headers)?;
-        Ok(Some(Box::new(
-            HttpChunkReader::new_fresh_connection_with_stall_timeout(url, Method::GET, headers, None, request.stall_timeout)
-                .await?,
-        )))
-    }
-
     async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter> {
         let server_epoch = self.put_file_auth_capability(&request.endpoint).await?;
         let nonce = server_epoch.map(|_| Uuid::new_v4());

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -57,17 +57,15 @@ use serde::{Serialize, de::DeserializeOwned};
 use std::{
     io::Cursor,
     path::PathBuf,
-    pin::Pin,
     sync::{
         Arc,
         atomic::{AtomicU32, Ordering},
     },
-    task::{Context, Poll},
     time::Duration,
 };
 use tokio::time;
 use tokio::{
-    io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
+    io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
     net::TcpStream,
     time::timeout,
 };
@@ -216,231 +214,6 @@ where
     }
 }
 
-fn is_retryable_remote_body_error(error: &io::Error) -> bool {
-    if error
-        .get_ref()
-        .and_then(|source| source.downcast_ref::<rustfs_rio::BodyStalled>())
-        .is_some()
-    {
-        return true;
-    }
-
-    matches!(
-        error.kind(),
-        io::ErrorKind::ConnectionReset
-            | io::ErrorKind::BrokenPipe
-            | io::ErrorKind::ConnectionAborted
-            | io::ErrorKind::UnexpectedEof
-    )
-}
-
-fn resumed_read_request(request: &ReadStreamRequest, emitted: usize) -> io::Result<ReadStreamRequest> {
-    let offset = request
-        .offset
-        .checked_add(emitted)
-        .ok_or_else(|| io::Error::other("remote read resume offset overflow"))?;
-    let length = if request.length == 0 {
-        0
-    } else {
-        request
-            .length
-            .checked_sub(emitted)
-            .ok_or_else(|| io::Error::other("remote read resume offset exceeds requested length"))?
-    };
-    Ok(ReadStreamRequest {
-        offset,
-        length,
-        ..request.clone()
-    })
-}
-
-type ReadResumeFuture = tokio::task::JoinHandle<Result<FileReader>>;
-
-struct RetryingRemoteReader {
-    reader: Option<FileReader>,
-    transport: Arc<dyn InternodeDataTransport>,
-    request: ReadStreamRequest,
-    emitted: usize,
-    retried: bool,
-    resume: Option<ReadResumeFuture>,
-}
-
-impl RetryingRemoteReader {
-    fn new(reader: FileReader, transport: Arc<dyn InternodeDataTransport>, request: ReadStreamRequest) -> Self {
-        Self {
-            reader: Some(reader),
-            transport,
-            request,
-            emitted: 0,
-            retried: false,
-            resume: None,
-        }
-    }
-
-    fn start_resume(&mut self) -> io::Result<()> {
-        if self.request.length != 0 && self.emitted >= self.request.length {
-            self.reader = None;
-            return Ok(());
-        }
-        let request = resumed_read_request(&self.request, self.emitted)?;
-        let transport = Arc::clone(&self.transport);
-        self.resume = Some(tokio::spawn(async move { transport.open_read_fresh(request).await }));
-        Ok(())
-    }
-}
-
-impl AsyncRead for RetryingRemoteReader {
-    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
-        loop {
-            if let Some(resume) = self.resume.as_mut() {
-                match Pin::new(resume).poll(cx) {
-                    Poll::Pending => return Poll::Pending,
-                    Poll::Ready(Ok(Ok(reader))) => {
-                        self.resume = None;
-                        self.reader = Some(reader);
-                    }
-                    Poll::Ready(Ok(Err(error))) => {
-                        self.resume = None;
-                        return Poll::Ready(Err(io::Error::other(error)));
-                    }
-                    Poll::Ready(Err(error)) => {
-                        self.resume = None;
-                        return Poll::Ready(Err(io::Error::other(error)));
-                    }
-                }
-            }
-
-            let Some(reader) = self.reader.as_mut() else {
-                return Poll::Ready(Ok(()));
-            };
-            let before = buf.filled().len();
-            match Pin::new(reader).poll_read(cx, buf) {
-                Poll::Pending => return Poll::Pending,
-                Poll::Ready(Ok(())) => {
-                    let produced = buf.filled().len() - before;
-                    self.emitted = match self.emitted.checked_add(produced) {
-                        Some(emitted) => emitted,
-                        None => return Poll::Ready(Err(io::Error::other("remote read emitted byte count overflow"))),
-                    };
-                    return Poll::Ready(Ok(()));
-                }
-                Poll::Ready(Err(error)) if !self.retried && is_retryable_remote_body_error(&error) => {
-                    self.retried = true;
-                    if let Err(resume_error) = self.start_resume() {
-                        return Poll::Ready(Err(resume_error));
-                    }
-                    continue;
-                }
-                Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
-            }
-        }
-    }
-}
-
-type ChunkResumeFuture = tokio::task::JoinHandle<Result<Option<rustfs_rio::ChunkReaderBox>>>;
-
-struct RetryingRemoteChunkReader {
-    reader: Option<rustfs_rio::ChunkReaderBox>,
-    transport: Arc<dyn InternodeDataTransport>,
-    request: ReadStreamRequest,
-    emitted: usize,
-    retried: bool,
-    resume: Option<ChunkResumeFuture>,
-}
-
-impl RetryingRemoteChunkReader {
-    fn new(reader: rustfs_rio::ChunkReaderBox, transport: Arc<dyn InternodeDataTransport>, request: ReadStreamRequest) -> Self {
-        Self {
-            reader: Some(reader),
-            transport,
-            request,
-            emitted: 0,
-            retried: false,
-            resume: None,
-        }
-    }
-
-    fn start_resume(&mut self) -> io::Result<()> {
-        if self.request.length != 0 && self.emitted >= self.request.length {
-            self.reader = None;
-            return Ok(());
-        }
-        let request = resumed_read_request(&self.request, self.emitted)?;
-        let transport = Arc::clone(&self.transport);
-        self.resume = Some(tokio::spawn(async move { transport.open_read_chunks_fresh(request).await }));
-        Ok(())
-    }
-}
-
-impl AsyncRead for RetryingRemoteChunkReader {
-    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
-        if buf.remaining() == 0 {
-            return Poll::Ready(Ok(()));
-        }
-        match rustfs_rio::ChunkReader::poll_read_chunk(self.as_mut(), cx, buf.remaining()) {
-            Poll::Ready(Ok(Some(chunk))) => {
-                buf.put_slice(&chunk);
-                Poll::Ready(Ok(()))
-            }
-            Poll::Ready(Ok(None)) => Poll::Ready(Ok(())),
-            Poll::Ready(Err(error)) => Poll::Ready(Err(error)),
-            Poll::Pending => Poll::Pending,
-        }
-    }
-}
-
-impl rustfs_rio::ChunkReader for RetryingRemoteChunkReader {
-    fn poll_read_chunk(mut self: Pin<&mut Self>, cx: &mut Context<'_>, max: usize) -> Poll<io::Result<Option<Bytes>>> {
-        loop {
-            if let Some(resume) = self.resume.as_mut() {
-                match Pin::new(resume).poll(cx) {
-                    Poll::Pending => return Poll::Pending,
-                    Poll::Ready(Ok(Ok(Some(reader)))) => {
-                        self.resume = None;
-                        self.reader = Some(reader);
-                    }
-                    Poll::Ready(Ok(Ok(None))) => {
-                        self.resume = None;
-                        self.reader = None;
-                        return Poll::Ready(Err(io::Error::other("remote resume transport did not provide a chunk reader")));
-                    }
-                    Poll::Ready(Ok(Err(error))) => {
-                        self.resume = None;
-                        return Poll::Ready(Err(io::Error::other(error)));
-                    }
-                    Poll::Ready(Err(error)) => {
-                        self.resume = None;
-                        return Poll::Ready(Err(io::Error::other(error)));
-                    }
-                }
-            }
-
-            let Some(reader) = self.reader.as_mut() else {
-                return Poll::Ready(Ok(None));
-            };
-            match rustfs_rio::ChunkReader::poll_read_chunk(Pin::new(reader.as_mut()), cx, max) {
-                Poll::Pending => return Poll::Pending,
-                Poll::Ready(Ok(Some(chunk))) => {
-                    self.emitted = match self.emitted.checked_add(chunk.len()) {
-                        Some(emitted) => emitted,
-                        None => return Poll::Ready(Err(io::Error::other("remote read emitted byte count overflow"))),
-                    };
-                    return Poll::Ready(Ok(Some(chunk)));
-                }
-                Poll::Ready(Ok(None)) => return Poll::Ready(Ok(None)),
-                Poll::Ready(Err(error)) if !self.retried && is_retryable_remote_body_error(&error) => {
-                    self.retried = true;
-                    if let Err(resume_error) = self.start_resume() {
-                        return Poll::Ready(Err(resume_error));
-                    }
-                    continue;
-                }
-                Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
-            }
-        }
-    }
-}
-
 #[derive(Debug)]
 pub struct RemoteDisk {
     pub id: Mutex<Option<Uuid>>,
@@ -2696,7 +2469,7 @@ impl DiskAPI for RemoteDisk {
         }
         let disk = self.disk_ref().await;
         let stall_timeout = get_object_disk_read_timeout();
-        let request = ReadStreamRequest {
+        self.open_read_with_retry(ReadStreamRequest {
             endpoint: self.endpoint.grid_host(),
             disk,
             volume: volume.to_string(),
@@ -2704,9 +2477,8 @@ impl DiskAPI for RemoteDisk {
             offset,
             length,
             stall_timeout: (!stall_timeout.is_zero()).then_some(stall_timeout),
-        };
-        let reader = self.open_read_with_retry(request.clone()).await?;
-        Ok(Box::new(RetryingRemoteReader::new(reader, Arc::clone(&self.data_transport), request)))
+        })
+        .await
     }
 
     async fn read_file_stream_chunks(
@@ -2721,7 +2493,7 @@ impl DiskAPI for RemoteDisk {
         }
         let disk = self.disk_ref().await;
         let stall_timeout = get_object_disk_read_timeout();
-        let request = ReadStreamRequest {
+        self.open_read_chunks_with_retry(ReadStreamRequest {
             endpoint: self.endpoint.grid_host(),
             disk,
             volume: volume.to_string(),
@@ -2729,12 +2501,8 @@ impl DiskAPI for RemoteDisk {
             offset,
             length,
             stall_timeout: (!stall_timeout.is_zero()).then_some(stall_timeout),
-        };
-        let reader = self.open_read_chunks_with_retry(request.clone()).await?;
-        Ok(reader.map(|reader| {
-            Box::new(RetryingRemoteChunkReader::new(reader, Arc::clone(&self.data_transport), request))
-                as rustfs_rio::ChunkReaderBox
-        }))
+        })
+        .await
     }
 
     /// Buffered read for remote disks.
@@ -4355,278 +4123,6 @@ mod tests {
         }
     }
 
-    #[derive(Debug, Clone)]
-    enum ResumeReadStep {
-        PartialThenReset(Vec<u8>),
-        Data(Vec<u8>),
-    }
-
-    #[derive(Debug, Default)]
-    struct ResumeTransport {
-        read_steps: Mutex<Vec<ResumeReadStep>>,
-        chunk_steps: Mutex<Vec<ResumeReadStep>>,
-        read_requests: Mutex<Vec<ReadStreamRequest>>,
-        chunk_requests: Mutex<Vec<ReadStreamRequest>>,
-        fresh_read_requests: Mutex<Vec<ReadStreamRequest>>,
-        fresh_chunk_requests: Mutex<Vec<ReadStreamRequest>>,
-    }
-
-    impl ResumeTransport {
-        fn with_read_steps(read_steps: Vec<ResumeReadStep>) -> Self {
-            Self {
-                read_steps: Mutex::new(read_steps),
-                ..Self::default()
-            }
-        }
-
-        fn with_chunk_steps(chunk_steps: Vec<ResumeReadStep>) -> Self {
-            Self {
-                chunk_steps: Mutex::new(chunk_steps),
-                ..Self::default()
-            }
-        }
-    }
-
-    #[derive(Debug)]
-    struct ChunkPartialThenErrorReader {
-        data: Option<Bytes>,
-        error: Option<io::Error>,
-    }
-
-    impl rustfs_rio::ChunkReader for ChunkPartialThenErrorReader {
-        fn poll_read_chunk(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, max: usize) -> Poll<io::Result<Option<Bytes>>> {
-            if let Some(mut data) = self.data.take() {
-                let take = data.len().min(max);
-                let chunk = data.split_to(take);
-                if !data.is_empty() {
-                    self.data = Some(data);
-                }
-                return Poll::Ready(Ok(Some(chunk)));
-            }
-            if let Some(error) = self.error.take() {
-                return Poll::Ready(Err(error));
-            }
-            Poll::Ready(Ok(None))
-        }
-    }
-
-    impl AsyncRead for ChunkPartialThenErrorReader {
-        fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
-            Poll::Ready(Err(io::Error::other("chunk reader must use chunk handoff")))
-        }
-    }
-
-    fn resume_step_reader(step: ResumeReadStep) -> FileReader {
-        match step {
-            ResumeReadStep::PartialThenReset(data) => Box::new(PartialThenErrorReader {
-                cursor: Cursor::new(data),
-                error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")),
-            }),
-            ResumeReadStep::Data(data) => Box::new(Cursor::new(data)),
-        }
-    }
-
-    fn resume_step_chunk_reader(step: ResumeReadStep) -> rustfs_rio::ChunkReaderBox {
-        match step {
-            ResumeReadStep::PartialThenReset(data) => Box::new(ChunkPartialThenErrorReader {
-                data: Some(Bytes::from(data)),
-                error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")),
-            }),
-            ResumeReadStep::Data(data) => Box::new(ChunkPartialThenErrorReader {
-                data: Some(Bytes::from(data)),
-                error: None,
-            }),
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl InternodeDataTransport for ResumeTransport {
-        async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader> {
-            self.read_requests
-                .lock()
-                .expect("read request lock should not be poisoned")
-                .push(request);
-            let step = self
-                .read_steps
-                .lock()
-                .expect("read steps lock should not be poisoned")
-                .remove(0);
-            Ok(resume_step_reader(step))
-        }
-
-        async fn open_read_fresh(&self, request: ReadStreamRequest) -> Result<FileReader> {
-            self.fresh_read_requests
-                .lock()
-                .expect("fresh read request lock should not be poisoned")
-                .push(request.clone());
-            self.open_read(request).await
-        }
-
-        async fn open_read_chunks(&self, request: ReadStreamRequest) -> Result<Option<rustfs_rio::ChunkReaderBox>> {
-            self.chunk_requests
-                .lock()
-                .expect("chunk request lock should not be poisoned")
-                .push(request);
-            let step = self
-                .chunk_steps
-                .lock()
-                .expect("chunk steps lock should not be poisoned")
-                .remove(0);
-            Ok(Some(resume_step_chunk_reader(step)))
-        }
-
-        async fn open_read_chunks_fresh(&self, request: ReadStreamRequest) -> Result<Option<rustfs_rio::ChunkReaderBox>> {
-            self.fresh_chunk_requests
-                .lock()
-                .expect("fresh chunk request lock should not be poisoned")
-                .push(request.clone());
-            self.open_read_chunks(request).await
-        }
-
-        async fn open_write(&self, _request: WriteStreamRequest) -> Result<FileWriter> {
-            panic!("open_write should not be used in remote read resume tests");
-        }
-
-        async fn open_walk_dir(&self, _request: WalkDirStreamRequest) -> Result<FileReader> {
-            panic!("open_walk_dir should not be used in remote read resume tests");
-        }
-
-        fn name(&self) -> &'static str {
-            "resume-test"
-        }
-
-        fn capabilities(&self) -> InternodeDataTransportCapabilities {
-            InternodeDataTransportCapabilities::tcp_http()
-        }
-    }
-
-    fn resume_request(length: usize) -> ReadStreamRequest {
-        ReadStreamRequest {
-            endpoint: "http://remote".to_string(),
-            disk: "disk".to_string(),
-            volume: "volume".to_string(),
-            path: "path".to_string(),
-            offset: 7,
-            length,
-            stall_timeout: None,
-        }
-    }
-
-    #[tokio::test]
-    async fn remote_reader_resumes_from_emitted_bytes_without_duplicates() {
-        let transport = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::Data(b"456789".to_vec())]));
-        let request = resume_request(10);
-        let reader = resume_step_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec()));
-        let mut reader = RetryingRemoteReader::new(reader, transport.clone(), request);
-        let mut output = Vec::new();
-        reader
-            .read_to_end(&mut output)
-            .await
-            .expect("one body reset should be resumed");
-
-        assert_eq!(output, b"0123456789");
-        let requests = transport
-            .read_requests
-            .lock()
-            .expect("read request lock should not be poisoned");
-        assert_eq!(requests.len(), 1);
-        assert_eq!(requests[0].offset, 11);
-        assert_eq!(requests[0].length, 6);
-        assert_eq!(
-            transport
-                .fresh_read_requests
-                .lock()
-                .expect("fresh read request lock should not be poisoned")
-                .len(),
-            1
-        );
-    }
-
-    #[tokio::test]
-    async fn remote_chunk_reader_resumes_from_emitted_bytes_without_duplicates() {
-        let transport = Arc::new(ResumeTransport::with_chunk_steps(vec![ResumeReadStep::Data(b"456789".to_vec())]));
-        let request = resume_request(10);
-        let reader = resume_step_chunk_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec()));
-        let mut reader = RetryingRemoteChunkReader::new(reader, transport.clone(), request);
-        let mut output = Vec::new();
-        reader
-            .read_to_end(&mut output)
-            .await
-            .expect("chunk body reset should be resumed");
-
-        assert_eq!(output, b"0123456789");
-        let requests = transport
-            .chunk_requests
-            .lock()
-            .expect("chunk request lock should not be poisoned");
-        assert_eq!(requests.len(), 1);
-        assert_eq!(requests[0].offset, 11);
-        assert_eq!(requests[0].length, 6);
-        assert_eq!(
-            transport
-                .fresh_chunk_requests
-                .lock()
-                .expect("fresh chunk request lock should not be poisoned")
-                .len(),
-            1
-        );
-    }
-
-    #[tokio::test]
-    async fn remote_reader_retries_at_most_once_and_preserves_non_retryable_errors() {
-        let transport = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::PartialThenReset(b"456".to_vec())]));
-        let mut reader = RetryingRemoteReader::new(
-            resume_step_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec())),
-            transport.clone(),
-            resume_request(7),
-        );
-        let error = reader
-            .read_to_end(&mut Vec::new())
-            .await
-            .expect_err("second reset must not retry");
-        assert_eq!(error.kind(), std_io::ErrorKind::ConnectionReset);
-        assert_eq!(
-            transport
-                .read_requests
-                .lock()
-                .expect("read request lock should not be poisoned")
-                .len(),
-            1
-        );
-
-        let transport = Arc::new(ResumeTransport::default());
-        let reader = PartialThenErrorReader {
-            cursor: Cursor::new(b"data".to_vec()),
-            error: Some(io::Error::new(std_io::ErrorKind::PermissionDenied, "permission denied")),
-        };
-        let mut reader = RetryingRemoteReader::new(Box::new(reader), transport.clone(), resume_request(4));
-        let error = reader
-            .read_to_end(&mut Vec::new())
-            .await
-            .expect_err("non-retryable errors must not retry");
-        assert_eq!(error.kind(), std_io::ErrorKind::PermissionDenied);
-        assert!(
-            transport
-                .read_requests
-                .lock()
-                .expect("read request lock should not be poisoned")
-                .is_empty()
-        );
-    }
-
-    #[test]
-    fn resumed_read_request_checks_large_offsets() {
-        let request = ReadStreamRequest {
-            offset: usize::MAX - 1,
-            length: 0,
-            ..resume_request(0)
-        };
-        assert!(resumed_read_request(&request, 2).is_err());
-
-        let request = resume_request(4);
-        assert!(resumed_read_request(&request, 5).is_err());
-    }
-
     fn init_tracing(filter_level: Level) {
         INIT.call_once(|| {
             let _ = tracing_subscriber::fmt()

crates/ecstore/src/set_disk/ops/object.rs

@@ -1093,8 +1093,6 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let (rd, wd) = tokio::io::duplex(duplex_buffer_size);
         debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer");
 
-        let (producer_terminal_tx, producer_terminal_rx) = tokio::sync::oneshot::channel();
-        let rd = LegacyDuplexProducerReader::new(rd, producer_terminal_rx);
         let (mut reader, offset, length) =
             get_object_reader_with_context(&self.ctx, Box::new(rd), range, &object_info, opts, &h).await?;
         // Carry the hook probe result so the app layer skips its now-redundant
@@ -1116,7 +1114,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             // `get_object_with_fileinfo` also waits on `writer`, so an outer timeout
             // would incorrectly treat downstream backpressure as disk-read latency.
             // Disk read timeouts must be enforced at the actual disk I/O operations.
-            let producer_result = Self::get_object_with_fileinfo(
+            if let Err(e) = Self::get_object_with_fileinfo(
                 &bucket,
                 &object,
                 erasure_cache,
@@ -1134,9 +1132,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 object_class.as_str(),
                 size_bucket,
             )
-            .await;
-            if let Err(e) = &producer_result {
-                let reason = classify_storage_error(e);
+            .await
+            {
+                let reason = classify_storage_error(&e);
                 if reason == GetObjectFailureReason::DownstreamClosed {
                     debug!(
                         event = EVENT_SET_DISK_WRITE,
@@ -1175,7 +1173,6 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                     );
                 }
             };
-            let _ = producer_terminal_tx.send(producer_result.map(|_| ()));
         });
 
         Ok(reader)
@@ -2321,159 +2318,6 @@ impl<R: AsyncRead + Unpin> AsyncRead for TransitionUploadReader<R> {
     }
 }
 
-struct LegacyDuplexProducerReader<R> {
-    inner: R,
-    terminal: Option<tokio::sync::oneshot::Receiver<Result<()>>>,
-    inner_eof: bool,
-}
-
-impl<R> LegacyDuplexProducerReader<R> {
-    fn new(inner: R, terminal: tokio::sync::oneshot::Receiver<Result<()>>) -> Self {
-        Self {
-            inner,
-            terminal: Some(terminal),
-            inner_eof: false,
-        }
-    }
-}
-
-impl<R: AsyncRead + Unpin> AsyncRead for LegacyDuplexProducerReader<R> {
-    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
-        if !self.inner_eof {
-            let before = buf.filled().len();
-            match Pin::new(&mut self.inner).poll_read(cx, buf) {
-                Poll::Pending => return Poll::Pending,
-                Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
-                Poll::Ready(Ok(())) if buf.filled().len() > before => return Poll::Ready(Ok(())),
-                Poll::Ready(Ok(())) => {
-                    self.inner_eof = true;
-                }
-            }
-        }
-
-        let Some(terminal) = self.terminal.as_mut() else {
-            return Poll::Ready(Ok(()));
-        };
-        match Pin::new(terminal).poll(cx) {
-            Poll::Pending => Poll::Pending,
-            Poll::Ready(Ok(Ok(()))) => {
-                self.terminal = None;
-                Poll::Ready(Ok(()))
-            }
-            Poll::Ready(Ok(Err(err))) => {
-                self.terminal = None;
-                Poll::Ready(Err(std::io::Error::other(err)))
-            }
-            Poll::Ready(Err(_)) => {
-                self.terminal = None;
-                Poll::Ready(Err(std::io::Error::other(StorageError::Unexpected)))
-            }
-        }
-    }
-}
-
-#[cfg(test)]
-mod legacy_duplex_producer_reader_tests {
-    use super::*;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-
-    fn storage_error_source(error: &std::io::Error) -> &StorageError {
-        error
-            .get_ref()
-            .and_then(|source| source.downcast_ref::<StorageError>())
-            .expect("legacy duplex terminal error should retain StorageError source")
-    }
-
-    #[tokio::test]
-    async fn legacy_duplex_reader_allows_clean_completion() {
-        let (mut writer, reader) = tokio::io::duplex(64);
-        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
-        writer
-            .write_all(b"complete")
-            .await
-            .expect("duplex write should fit in buffer");
-        drop(writer);
-        terminal_tx.send(Ok(())).expect("terminal receiver should remain installed");
-
-        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
-        let mut out = Vec::new();
-        reader
-            .read_to_end(&mut out)
-            .await
-            .expect("clean producer completion should surface clean EOF");
-
-        assert_eq!(out, b"complete");
-    }
-
-    #[tokio::test]
-    async fn legacy_duplex_reader_surfaces_terminal_error_after_partial_data() {
-        let (mut writer, reader) = tokio::io::duplex(64);
-        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
-        writer.write_all(b"partial").await.expect("duplex write should fit in buffer");
-        drop(writer);
-        terminal_tx
-            .send(Err(StorageError::FileCorrupt))
-            .expect("terminal receiver should remain installed");
-
-        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
-        let mut out = Vec::new();
-        let err = reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("terminal producer error must not become clean EOF");
-
-        assert_eq!(out, b"partial");
-        assert!(matches!(storage_error_source(&err), StorageError::FileCorrupt));
-    }
-
-    #[tokio::test]
-    async fn legacy_duplex_reader_surfaces_terminal_error_after_declared_length() {
-        let (mut writer, reader) = tokio::io::duplex(64);
-        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
-        writer.write_all(b"exact").await.expect("duplex write should fit in buffer");
-        drop(writer);
-        terminal_tx
-            .send(Err(StorageError::Io(std::io::Error::new(
-                std::io::ErrorKind::ConnectionReset,
-                "remote body reset after final byte",
-            ))))
-            .expect("terminal receiver should remain installed");
-
-        let reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
-        let mut reader =
-            HashReader::from_stream(reader, 5, 5, None, None, false).expect("hash reader should accept exact declared length");
-        let mut out = Vec::new();
-        let err = reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("producer terminal error after the declared length must still fail");
-
-        assert_eq!(out, b"exact");
-        assert!(
-            matches!(storage_error_source(&err), StorageError::Io(io_error) if io_error.kind() == std::io::ErrorKind::ConnectionReset)
-        );
-    }
-
-    #[tokio::test]
-    async fn legacy_duplex_reader_fails_closed_when_terminal_channel_closes() {
-        let (mut writer, reader) = tokio::io::duplex(64);
-        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel::<Result<()>>();
-        writer.write_all(b"body").await.expect("duplex write should fit in buffer");
-        drop(writer);
-        drop(terminal_tx);
-
-        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
-        let mut out = Vec::new();
-        let err = reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("producer disappearance must fail closed");
-
-        assert_eq!(out, b"body");
-        assert!(matches!(storage_error_source(&err), StorageError::Unexpected));
-    }
-}
-
 struct TransitionUploadWriter<W> {
     inner: W,
     produced: u64,

crates/lock/src/client/local.rs

@@ -15,8 +15,10 @@
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, SystemTime};
 use tokio::sync::RwLock;
+use tokio::time::Instant;
 
 use crate::{
     FastLockGuard, GlobalLockManager, LockClient, LockId, LockInfo, LockManager, LockMetadata, LockPriority, LockRequest,
@@ -26,43 +28,51 @@ use crate::{
 /// Default shard count for guard storage (must be power of 2)
 const DEFAULT_GUARD_SHARD_COUNT: usize = 64;
 
+type GuardShard = Arc<RwLock<HashMap<LockId, LocalGuardEntry>>>;
+type GuardStorage = Arc<Vec<GuardShard>>;
+
 /// Local lock client using FastLock with sharded guard storage for better concurrency
 #[derive(Debug)]
 pub struct LocalClient {
     /// Sharded guard storage to reduce lock contention
-    guard_storage: Vec<Arc<RwLock<HashMap<LockId, LocalGuardEntry>>>>,
+    guard_storage: GuardStorage,
     /// Mask for fast shard index calculation (shard_count - 1)
     shard_mask: usize,
     /// Optional lock manager (if None, uses global singleton)
     manager: Option<Arc<GlobalLockManager>>,
+    reaper_started: AtomicBool,
+    reaper_interval: Duration,
 }
 
 #[derive(Debug)]
 struct LocalGuardEntry {
     guard: FastLockGuard,
     expires_at: SystemTime,
+    deadline: Instant,
     ttl: Duration,
-    /// Owner recorded at acquire time; used only for reclaim diagnostics (#899).
-    owner: String,
 }
 
 impl LocalGuardEntry {
-    fn new(guard: FastLockGuard, ttl: Duration, owner: String) -> Self {
+    fn new(guard: FastLockGuard, ttl: Duration) -> Self {
         let now = SystemTime::now();
+        let monotonic_now = Instant::now();
         Self {
             guard,
-            expires_at: now + ttl,
+            expires_at: now.checked_add(ttl).unwrap_or(now),
+            deadline: monotonic_now.checked_add(ttl).unwrap_or(monotonic_now),
             ttl,
-            owner,
         }
     }
 
     fn is_expired(&self) -> bool {
-        self.expires_at <= SystemTime::now()
+        self.deadline <= Instant::now()
     }
 
     fn refresh(&mut self) {
-        self.expires_at = SystemTime::now() + self.ttl;
+        let now = SystemTime::now();
+        let monotonic_now = Instant::now();
+        self.expires_at = now.checked_add(self.ttl).unwrap_or(now);
+        self.deadline = monotonic_now.checked_add(self.ttl).unwrap_or(monotonic_now);
     }
 }
 
@@ -77,26 +87,38 @@ impl LocalClient {
     pub fn with_shard_count(shard_count: usize) -> Self {
         assert!(shard_count.is_power_of_two(), "Shard count must be power of 2");
 
-        let guard_storage: Vec<Arc<RwLock<HashMap<LockId, LocalGuardEntry>>>> =
-            (0..shard_count).map(|_| Arc::new(RwLock::new(HashMap::new()))).collect();
+        let guard_storage: Vec<GuardShard> = (0..shard_count).map(|_| Arc::new(RwLock::new(HashMap::new()))).collect();
 
+        Self::with_storage(Arc::new(guard_storage), None, crate::fast_lock::CLEANUP_INTERVAL)
+    }
+
+    fn with_storage(guard_storage: GuardStorage, manager: Option<Arc<GlobalLockManager>>, reaper_interval: Duration) -> Self {
+        let shard_count = guard_storage.len();
+        debug_assert!(shard_count.is_power_of_two());
         Self {
             guard_storage,
             shard_mask: shard_count - 1,
-            manager: None,
+            manager,
+            reaper_started: AtomicBool::new(false),
+            reaper_interval,
         }
     }
 
     /// Create new local client with a specific lock manager
     /// This allows simulating multi-node environments where each node has its own lock backend
     pub fn with_manager(manager: Arc<GlobalLockManager>) -> Self {
-        Self {
-            guard_storage: (0..DEFAULT_GUARD_SHARD_COUNT)
-                .map(|_| Arc::new(RwLock::new(HashMap::new())))
-                .collect(),
-            shard_mask: DEFAULT_GUARD_SHARD_COUNT - 1,
-            manager: Some(manager),
-        }
+        let guard_storage = (0..DEFAULT_GUARD_SHARD_COUNT)
+            .map(|_| Arc::new(RwLock::new(HashMap::new())))
+            .collect();
+        Self::with_storage(Arc::new(guard_storage), Some(manager), crate::fast_lock::CLEANUP_INTERVAL)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn with_manager_and_reaper_interval(manager: Arc<GlobalLockManager>, reaper_interval: Duration) -> Self {
+        let guard_storage = (0..DEFAULT_GUARD_SHARD_COUNT)
+            .map(|_| Arc::new(RwLock::new(HashMap::new())))
+            .collect();
+        Self::with_storage(Arc::new(guard_storage), Some(manager), reaper_interval)
     }
 
     /// Get the lock manager (injected manager if available, otherwise global singleton)
@@ -118,52 +140,61 @@ impl LocalClient {
     }
 
     async fn reclaim_expired_guards_for_resource(&self, resource: &crate::ObjectKey) -> usize {
-        let mut reclaimed = 0usize;
+        let expired_entries = Self::extract_expired_guards(&self.guard_storage, Some(resource)).await;
+        Self::release_reclaimed_guards(expired_entries, Some(resource))
+    }
 
-        for shard in &self.guard_storage {
-            let expired_entries = {
-                let mut guards = shard.write().await;
-                let mut retained = HashMap::with_capacity(guards.len());
-                let mut expired_entries = Vec::new();
+    async fn extract_expired_guards(storage: &GuardStorage, resource: Option<&crate::ObjectKey>) -> Vec<LocalGuardEntry> {
+        let mut expired_entries = Vec::new();
+        for shard in storage.iter() {
+            let mut guards = shard.write().await;
+            expired_entries.extend(
+                guards
+                    .extract_if(|lock_id, entry| {
+                        resource.is_none_or(|resource| &lock_id.resource == resource) && entry.is_expired()
+                    })
+                    .map(|(_, entry)| entry),
+            );
+        }
+        expired_entries
+    }
 
-                for (lock_id, entry) in std::mem::take(&mut *guards) {
-                    if &lock_id.resource == resource && entry.is_expired() {
-                        expired_entries.push(entry);
-                    } else {
-                        retained.insert(lock_id, entry);
-                    }
-                }
-
-                *guards = retained;
-                expired_entries
-            };
-
-            for mut entry in expired_entries {
-                // An expired entry whose owner never refreshed it (a dead coordinator, #698) is
-                // reclaimed so a live contender can re-form quorum. With guard heartbeats in place
-                // (#899) a live owner keeps its entry from expiring, so reaching here means the
-                // lease genuinely lapsed. Surface it for observability; the reclaim decision itself
-                // is unchanged.
-                let since_last_refresh = entry
-                    .expires_at
-                    .checked_sub(entry.ttl)
-                    .and_then(|last_refresh| SystemTime::now().duration_since(last_refresh).ok())
-                    .unwrap_or(entry.ttl);
-                tracing::warn!(
-                    owner = %entry.owner,
-                    resource = %resource,
-                    ttl_ms = entry.ttl.as_millis() as u64,
-                    since_last_refresh_ms = since_last_refresh.as_millis() as u64,
-                    "reclaiming expired lock guard whose lease was not refreshed"
-                );
+    fn release_reclaimed_guards(mut entries: Vec<LocalGuardEntry>, resource: Option<&crate::ObjectKey>) -> usize {
+        let reclaimed = entries.len();
+        for entry in &mut entries {
+            let _ = entry.guard.release();
+        }
+        if reclaimed > 0 {
+            for _ in 0..reclaimed {
                 rustfs_io_metrics::record_lock_reclaimed();
-                let _ = entry.guard.release();
-                reclaimed = reclaimed.saturating_add(1);
+            }
+            if let Some(resource) = resource {
+                tracing::debug!(event = "lock_guard_reclaimed", resource = %resource, count = reclaimed, "expired lock guards reclaimed");
+            } else {
+                tracing::debug!(event = "lock_guard_reaper_sweep", count = reclaimed, "expired lock guards reclaimed");
             }
         }
-
         reclaimed
     }
+
+    fn ensure_reaper(&self) {
+        if self.reaper_started.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        let storage = Arc::downgrade(&self.guard_storage);
+        let interval = self.reaper_interval;
+        tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(interval);
+            loop {
+                ticker.tick().await;
+                let Some(storage) = storage.upgrade() else {
+                    break;
+                };
+                let expired_entries = Self::extract_expired_guards(&storage, None).await;
+                Self::release_reclaimed_guards(expired_entries, None);
+            }
+        });
+    }
 }
 
 impl Default for LocalClient {
@@ -175,28 +206,36 @@ impl Default for LocalClient {
 #[async_trait::async_trait]
 impl LockClient for LocalClient {
     async fn acquire_lock(&self, request: &LockRequest) -> Result<LockResponse> {
+        self.ensure_reaper();
         let lock_manager = self.get_lock_manager();
         let reclaimed_before_acquire = self.reclaim_expired_guards_for_resource(&request.resource).await;
+        let acquire_deadline = Instant::now()
+            .checked_add(request.acquire_timeout)
+            .unwrap_or_else(Instant::now);
 
-        let build_lock_request = || match request.lock_type {
+        let build_lock_request = |acquire_timeout| match request.lock_type {
             LockType::Exclusive => crate::ObjectLockRequest::new_write(request.resource.clone(), request.owner.clone())
-                .with_acquire_timeout(request.acquire_timeout),
+                .with_acquire_timeout(acquire_timeout),
             LockType::Shared => crate::ObjectLockRequest::new_read(request.resource.clone(), request.owner.clone())
-                .with_acquire_timeout(request.acquire_timeout),
+                .with_acquire_timeout(acquire_timeout),
         };
 
         let mut retried_after_reclaim = reclaimed_before_acquire > 0;
         loop {
-            match lock_manager.acquire_lock(build_lock_request()).await {
+            let remaining = acquire_deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                return Ok(LockResponse::failure("Lock acquisition timeout", request.acquire_timeout));
+            }
+            match lock_manager.acquire_lock(build_lock_request(remaining)).await {
                 Ok(guard) => {
                     let lock_id = request.lock_id.clone();
                     let acquired_at = SystemTime::now();
-                    let expires_at = acquired_at + request.ttl;
+                    let expires_at = acquired_at.checked_add(request.ttl).unwrap_or(acquired_at);
 
                     {
                         let shard = self.get_shard(&lock_id);
                         let mut guards = shard.write().await;
-                        guards.insert(lock_id.clone(), LocalGuardEntry::new(guard, request.ttl, request.owner.clone()));
+                        guards.insert(lock_id.clone(), LocalGuardEntry::new(guard, request.ttl));
                     }
 
                     let lock_info = LockInfo {
@@ -317,3 +356,111 @@ impl LockClient for LocalClient {
         true
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{GlobalLockManager, LockClient, LockRequest, LockType};
+
+    fn request(resource: crate::ObjectKey, owner: &str, ttl: Duration) -> LockRequest {
+        LockRequest::new(resource, LockType::Exclusive, owner)
+            .with_ttl(ttl)
+            .with_acquire_timeout(Duration::from_millis(80))
+    }
+
+    async fn wait_until_reaped(client: &LocalClient, lock_id: &LockId) {
+        for _ in 0..80 {
+            if client.check_status(lock_id).await.unwrap().is_none() {
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+        panic!("lock guard was not reaped before test deadline");
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn expired_guard_is_reaped_without_resource_reacquire() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5));
+        let request = request(crate::ObjectKey::new("bucket", "unique-chunk"), "owner-a", Duration::from_millis(10));
+        let lock_id = request.lock_id.clone();
+
+        assert!(client.acquire_lock(&request).await.unwrap().success);
+        assert!(client.check_status(&lock_id).await.unwrap().is_some());
+        tokio::time::sleep(Duration::from_millis(15)).await;
+        wait_until_reaped(&client, &lock_id).await;
+
+        let direct = manager
+            .acquire_lock(crate::ObjectLockRequest::new_write(request.resource.clone(), "owner-b"))
+            .await;
+        assert!(direct.is_ok());
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn sibling_client_cannot_reclaim_but_owner_reaper_releases_shared_lock() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let owner = LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5));
+        let contender = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_millis(5));
+        let request_a = request(crate::ObjectKey::new("bucket", "shared-resource"), "owner-a", Duration::from_millis(10));
+        assert!(owner.acquire_lock(&request_a).await.unwrap().success);
+
+        let request_b = request(request_a.resource.clone(), "owner-b", Duration::from_millis(20))
+            .with_acquire_timeout(Duration::from_millis(5));
+        assert!(!contender.acquire_lock(&request_b).await.unwrap().success);
+
+        tokio::time::sleep(Duration::from_millis(25)).await;
+        assert!(owner.check_status(&request_a.lock_id).await.unwrap().is_none());
+        assert!(contender.acquire_lock(&request_b).await.unwrap().success);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn refresh_wins_before_deadline_and_reaper_wins_after_deadline() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_millis(5));
+        let request = request(crate::ObjectKey::new("bucket", "refresh-race"), "owner-a", Duration::from_millis(25));
+        let lock_id = request.lock_id.clone();
+        assert!(client.acquire_lock(&request).await.unwrap().success);
+
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        assert!(client.refresh(&lock_id).await.unwrap());
+        tokio::time::sleep(Duration::from_millis(15)).await;
+        assert!(client.check_status(&lock_id).await.unwrap().is_some());
+        wait_until_reaped(&client, &lock_id).await;
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn zero_ttl_is_reaped_and_oversized_ttl_does_not_panic() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_millis(5));
+
+        let zero = request(crate::ObjectKey::new("bucket", "zero-ttl"), "owner-zero", Duration::ZERO);
+        let zero_id = zero.lock_id.clone();
+        assert!(client.acquire_lock(&zero).await.unwrap().success);
+        wait_until_reaped(&client, &zero_id).await;
+
+        let huge = request(crate::ObjectKey::new("bucket", "huge-ttl"), "owner-huge", Duration::MAX);
+        let huge_id = huge.lock_id.clone();
+        assert!(client.acquire_lock(&huge).await.unwrap().success);
+        wait_until_reaped(&client, &huge_id).await;
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn acquire_retry_preserves_total_deadline() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_secs(60));
+        let first = request(crate::ObjectKey::new("bucket", "deadline-budget"), "owner-a", Duration::from_millis(10));
+        assert!(client.acquire_lock(&first).await.unwrap().success);
+
+        let second =
+            request(first.resource.clone(), "owner-b", Duration::from_millis(30)).with_acquire_timeout(Duration::from_millis(60));
+        let started = Instant::now();
+        let response = client.acquire_lock(&second).await.unwrap();
+        assert!(!response.success, "the first attempt consumed the caller's acquire budget");
+        assert!(
+            started.elapsed() < Duration::from_millis(100),
+            "reclaim retry must not double the acquire budget"
+        );
+        let recovered = client.acquire_lock(&second).await.unwrap();
+        assert!(recovered.success, "the reclaimed guard must be available to the next request");
+    }
+}

crates/lock/src/namespace/tests.rs

@@ -840,6 +840,117 @@ async fn test_namespace_lock_distributed_reclaims_expired_same_resource_after_fa
     );
 }
 
+#[tokio::test]
+async fn four_node_failed_release_converges_without_replica_repair() {
+    let managers = (0..4).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let flaky_clients = managers
+        .iter()
+        .map(|manager| {
+            Arc::new(FlakyReleaseClient {
+                inner: LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5)),
+                failed_releases_remaining: AtomicUsize::new(usize::MAX),
+                release_attempts: AtomicUsize::new(0),
+            })
+        })
+        .collect::<Vec<_>>();
+    let clients = flaky_clients
+        .iter()
+        .map(|client| client.clone() as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+    let lock = NamespaceLock::Distributed(DistributedLock::new("four-node-expired-lease".to_string(), clients, 3));
+    let resource = create_test_object_key("bucket", "object-four-node-expired");
+    let request = LockRequest::new(resource.clone(), LockType::Exclusive, "owner-a")
+        .with_acquire_timeout(Duration::from_millis(300))
+        .with_ttl(Duration::from_millis(40));
+
+    let mut guard = lock
+        .acquire_guard(&request)
+        .await
+        .expect("initial acquire should not error")
+        .expect("initial acquire should reach quorum");
+    assert!(guard.release(), "release should be acknowledged while RPC cleanup is pending");
+
+    for _ in 0..40 {
+        if flaky_clients.iter().all(|client| client.release_attempts() >= 3) {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+    loop {
+        let all_reaped =
+            futures::future::join_all(flaky_clients.iter().map(|client| client.inner.check_status(&request.lock_id)))
+                .await
+                .into_iter()
+                .all(|status| status.expect("status should not error").is_none());
+        if all_reaped {
+            break;
+        }
+        assert!(tokio::time::Instant::now() < deadline, "all four local lease entries must converge");
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+
+    for suffix in ["chunk-0", "chunk-1", ".rustfs.sys/multipart/upload-0"] {
+        for client in &flaky_clients {
+            let orphan = LockRequest::new(create_test_object_key("bucket", suffix), LockType::Exclusive, "orphan")
+                .with_ttl(Duration::from_millis(25));
+            assert!(client.inner.acquire_lock(&orphan).await.expect("orphan acquire").success);
+        }
+    }
+    tokio::time::sleep(Duration::from_millis(80)).await;
+
+    let recovered = lock
+        .acquire_guard(
+            &LockRequest::new(resource, LockType::Exclusive, "owner-b")
+                .with_acquire_timeout(Duration::from_millis(300))
+                .with_ttl(Duration::from_millis(40)),
+        )
+        .await
+        .expect("recovery acquire should not error")
+        .expect("four-node quorum should recover after local reapers run");
+    drop(recovered);
+}
+
+#[tokio::test]
+async fn four_node_stale_quorum_contention_respects_acquire_deadline() {
+    let managers = (0..4).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let node_clients = managers
+        .iter()
+        .map(|manager| Arc::new(LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5))))
+        .collect::<Vec<_>>();
+    let resource = create_test_object_key("bucket", "stale-quorum");
+    let stale = LockRequest::new(resource.clone(), LockType::Exclusive, "stale-owner").with_ttl(Duration::from_millis(180));
+    for client in &node_clients {
+        assert!(client.acquire_lock(&stale).await.expect("stale acquire").success);
+    }
+
+    let clients = node_clients
+        .iter()
+        .map(|client| client.clone() as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+    let lock = NamespaceLock::Distributed(DistributedLock::new("stale-quorum-deadline".to_string(), clients, 3));
+    let contender = LockRequest::new(resource.clone(), LockType::Exclusive, "new-owner")
+        .with_acquire_timeout(Duration::from_millis(150))
+        .with_ttl(Duration::from_millis(100));
+    let started = tokio::time::Instant::now();
+    let response = lock.acquire_guard(&contender).await.expect("contention should not error");
+    assert!(response.is_none(), "unexpired leases must not be force-reclaimed");
+    assert!(started.elapsed() < Duration::from_millis(350), "acquire must respect its deadline");
+
+    tokio::time::sleep(Duration::from_millis(80)).await;
+    let recovered = lock
+        .acquire_guard(
+            &LockRequest::new(resource, LockType::Exclusive, "new-owner")
+                .with_acquire_timeout(Duration::from_millis(300))
+                .with_ttl(Duration::from_millis(100)),
+        )
+        .await
+        .expect("post-expiry acquire should not error")
+        .expect("quorum should recover after local reapers clear stale leases");
+    drop(recovered);
+}
+
 #[tokio::test]
 async fn test_namespace_lock_distributed_retries_transient_acquire_timeout() {
     let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();

crates/rio/src/http_reader.rs

@@ -138,12 +138,6 @@ impl std::fmt::Display for InternodeHttpErrorKind {
     }
 }
 
-#[derive(thiserror::Error, Debug, Clone, Copy, Eq, PartialEq)]
-#[error("internode body stalled for {timeout:?}")]
-pub struct BodyStalled {
-    pub timeout: Duration,
-}
-
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct InternodeHttpRequestContext {
     method: String,
@@ -277,10 +271,6 @@ pub fn internode_http_timeout_error(method: &Method, url: &str) -> io::Error {
     internode_kind_error(method, url, internode_rpc_operation(url), InternodeHttpErrorKind::ConnectTimeout)
 }
 
-fn body_stalled_error(stall_timeout: Duration) -> io::Error {
-    Error::new(io::ErrorKind::TimedOut, BodyStalled { timeout: stall_timeout })
-}
-
 /// Clone an internode HTTP I/O error while retaining its structured classification.
 ///
 /// The underlying transport source is intentionally omitted because it is not
@@ -715,13 +705,6 @@ async fn get_http_client(url: &str) -> io::Result<Client> {
     Ok(cached.client_for(disable_proxy))
 }
 
-async fn get_fresh_http_client(url: &str) -> io::Result<Client> {
-    let tuning = internode_http_client_tuning();
-    let disable_proxy = should_disable_proxy_for_url(url, tuning);
-    let outbound_tls = crate::http_runtime_sources::outbound_tls_state().await;
-    build_http_client(disable_proxy, tuning, &outbound_tls).await
-}
-
 fn internode_request_context(method: &Method, url: &str, operation: Option<&'static str>) -> InternodeHttpRequestContext {
     let target = reqwest::Url::parse(url)
         .ok()
@@ -969,28 +952,6 @@ impl HttpReader {
         Self::with_capacity_and_stall_timeout(url, method, headers, body, 0, stall_timeout).await
     }
 
-    pub async fn new_fresh_connection_with_stall_timeout(
-        url: String,
-        method: Method,
-        headers: HeaderMap,
-        body: Option<Vec<u8>>,
-        stall_timeout: Option<Duration>,
-    ) -> io::Result<Self> {
-        let init = Self::open(&url, &method, &headers, body, stall_timeout, true).await?;
-        Ok(Self {
-            inner: StreamReader::new(init.stream),
-            url,
-            method,
-            headers,
-            track_internode_metrics: init.track_internode_metrics,
-            internode_operation: init.internode_operation,
-            stall_timer: None,
-            stall_timeout: init.stall_timeout,
-            request_started: init.request_started,
-            duration_recorded: false,
-        })
-    }
-
     /// Create a new HttpReader from a URL. The request is performed immediately.
     pub async fn with_capacity(
         url: String,
@@ -1010,7 +971,7 @@ impl HttpReader {
         _read_buf_size: usize,
         stall_timeout: Option<Duration>,
     ) -> io::Result<Self> {
-        let init = Self::open(&url, &method, &headers, body, stall_timeout, false).await?;
+        let init = Self::open(&url, &method, &headers, body, stall_timeout).await?;
         Ok(Self {
             inner: StreamReader::new(init.stream),
             url,
@@ -1031,16 +992,10 @@ impl HttpReader {
         headers: &HeaderMap,
         body: Option<Vec<u8>>,
         stall_timeout: Option<Duration>,
-        force_fresh_connection: bool,
     ) -> io::Result<HttpReaderInit> {
         let track_internode_metrics = is_internode_rpc_url(url);
         let internode_operation = internode_rpc_operation(url);
-        let client = if force_fresh_connection {
-            get_fresh_http_client(url).await
-        } else {
-            get_http_client(url).await
-        }
-        .inspect_err(|_| {
+        let client = get_http_client(url).await.inspect_err(|_| {
             record_internode_error(track_internode_metrics, internode_operation);
         })?;
         let mut request: RequestBuilder = client.request(method.clone(), url).headers(headers.clone());
@@ -1130,7 +1085,10 @@ impl AsyncRead for HttpReader {
                     );
                     record_internode_stall_timeout(*this.track_internode_metrics, *this.internode_operation);
                     record_internode_error(*this.track_internode_metrics, *this.internode_operation);
-                    Poll::Ready(Err(body_stalled_error(stall_timeout)))
+                    Poll::Ready(Err(Error::new(
+                        io::ErrorKind::TimedOut,
+                        "HttpReader stall timeout: no data received before deadline",
+                    )))
                 } else {
                     Poll::Pending
                 }
@@ -1156,28 +1114,7 @@ impl HttpChunkReader {
         body: Option<Vec<u8>>,
         stall_timeout: Option<Duration>,
     ) -> io::Result<Self> {
-        let init = HttpReader::open(&url, &method, &headers, body, stall_timeout, false).await?;
-        Ok(Self {
-            inner: init.stream,
-            current: None,
-            track_internode_metrics: init.track_internode_metrics,
-            internode_operation: init.internode_operation,
-            stall_timer: None,
-            stall_timeout: init.stall_timeout,
-            request_started: init.request_started,
-            duration_recorded: false,
-            consecutive_empty_chunks: 0,
-        })
-    }
-
-    pub async fn new_fresh_connection_with_stall_timeout(
-        url: String,
-        method: Method,
-        headers: HeaderMap,
-        body: Option<Vec<u8>>,
-        stall_timeout: Option<Duration>,
-    ) -> io::Result<Self> {
-        let init = HttpReader::open(&url, &method, &headers, body, stall_timeout, true).await?;
+        let init = HttpReader::open(&url, &method, &headers, body, stall_timeout).await?;
         Ok(Self {
             inner: init.stream,
             current: None,
@@ -1280,7 +1217,10 @@ impl ChunkReader for HttpChunkReader {
                         );
                         record_internode_stall_timeout(*this.track_internode_metrics, *this.internode_operation);
                         record_internode_error(*this.track_internode_metrics, *this.internode_operation);
-                        return Poll::Ready(Err(body_stalled_error(stall_timeout)));
+                        return Poll::Ready(Err(Error::new(
+                            io::ErrorKind::TimedOut,
+                            "HttpReader stall timeout: no data received before deadline",
+                        )));
                     }
                     return Poll::Pending;
                 }
@@ -2439,46 +2379,6 @@ mod tests {
             Err(err) => err,
         };
         assert_eq!(err.kind(), io::ErrorKind::TimedOut);
-        let stalled = err
-            .get_ref()
-            .and_then(|source| source.downcast_ref::<BodyStalled>())
-            .expect("stall timeout should retain typed body-stalled source");
-        assert_eq!(stalled.timeout, Duration::from_millis(20));
-
-        handle.abort();
-    }
-
-    #[tokio::test]
-    async fn http_chunk_reader_stall_timeout_retains_typed_source() {
-        let state = TestState::default();
-        let Some((base_url, handle)) = start_test_server(state).await else {
-            return;
-        };
-        let url = base_url.replace("/stream", "/stall");
-        let mut reader =
-            HttpChunkReader::new_with_stall_timeout(url, Method::GET, HeaderMap::new(), None, Some(Duration::from_millis(20)))
-                .await
-                .expect("chunk reader should open");
-
-        let first = std::future::poll_fn(|cx| Pin::new(&mut reader).poll_read_chunk(cx, 64))
-            .await
-            .expect("initial body chunk should arrive")
-            .expect("initial body chunk should not be EOF");
-        assert_eq!(first, b"hello"[..]);
-
-        let err = tokio::time::timeout(
-            Duration::from_secs(1),
-            std::future::poll_fn(|cx| Pin::new(&mut reader).poll_read_chunk(cx, 64)),
-        )
-        .await
-        .expect("stall timeout should wake chunk reader")
-        .expect_err("chunk reader should return a timeout error");
-        assert_eq!(err.kind(), io::ErrorKind::TimedOut);
-        let stalled = err
-            .get_ref()
-            .and_then(|source| source.downcast_ref::<BodyStalled>())
-            .expect("chunk stall timeout should retain typed body-stalled source");
-        assert_eq!(stalled.timeout, Duration::from_millis(20));
 
         handle.abort();
     }

rustfs/src/app/object_usecase.rs

@@ -2173,11 +2173,10 @@ fn get_object_resume_control(ctx: GetObjectResumeContext) -> GetObjectResumeCont
 /// disks" failures keep the existing fail-loud behavior.
 fn is_object_relocation_error(err: &std::io::Error) -> bool {
     let Some(inner) = err.get_ref() else { return false };
-    match inner.downcast_ref::<StorageError>() {
-        Some(StorageError::FileNotFound | StorageError::ObjectNotFound(..) | StorageError::InsufficientReadQuorum(..)) => true,
-        Some(StorageError::Io(source)) => source.kind() == std::io::ErrorKind::NotFound,
-        _ => false,
-    }
+    matches!(
+        inner.downcast_ref::<StorageError>(),
+        Some(StorageError::FileNotFound | StorageError::ObjectNotFound(..) | StorageError::InsufficientReadQuorum(..))
+    )
 }
 
 /// Resolve the S3 request-body inter-chunk read timeout from the environment.
@@ -13007,7 +13006,6 @@ mod tests {
             StorageError::FileNotFound,
             StorageError::ObjectNotFound("test-bucket".to_string(), "relocated-object".to_string()),
             StorageError::InsufficientReadQuorum("test-bucket".to_string(), "relocated-object".to_string()),
-            StorageError::Io(std::io::Error::new(std::io::ErrorKind::NotFound, "relocated shard disappeared")),
         ] {
             let reopen_count = Arc::new(AtomicUsize::new(0));
             let control = counting_resume_control(Arc::clone(&reopen_count), |emitted| {

rustfs/src/error.rs

@@ -669,23 +669,6 @@ mod tests {
         assert!(api_error.source.is_some());
     }
 
-    #[test]
-    fn test_api_error_from_storage_io_copy_object_terminal_error_stays_internal() {
-        let io_error = IoError::other(StorageError::FileCorrupt);
-        let storage_error: StorageError = io_error.into();
-        assert!(matches!(storage_error, StorageError::FileCorrupt));
-
-        let api_error: ApiError = storage_error.into();
-
-        assert_eq!(api_error.code, S3ErrorCode::InternalError);
-        let source = api_error
-            .source
-            .as_deref()
-            .and_then(|source| source.downcast_ref::<StorageError>())
-            .expect("API error should retain the storage error source");
-        assert!(matches!(source, StorageError::FileCorrupt));
-    }
-
     #[test]
     fn test_api_error_from_iam_error() {
         let iam_error = rustfs_iam::error::Error::other("IAM test error");