From 7c2361757e8fae1cb631516317a8f02495abd69d Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 26 Aug 2026 21:24:37 +0800 Subject: [PATCH] fix(ecstore): bound copy-source shard read-ahead (#6663) --- crates/ecstore/src/cluster/rpc/remote_disk.rs | 580 +++++++- crates/ecstore/src/disk/error.rs | 112 ++ crates/ecstore/src/erasure/coding/decode.rs | 1258 +++++++++++++++-- crates/ecstore/src/io_support/bitrot.rs | 15 +- .../src/set_disk/core/io_primitives.rs | 94 ++ crates/ecstore/src/set_disk/mod.rs | 95 ++ crates/ecstore/src/set_disk/ops/object.rs | 122 +- crates/ecstore/src/set_disk/read.rs | 80 +- crates/ecstore/src/store/object.rs | 27 + rustfs/src/app/multipart_usecase.rs | 8 +- rustfs/src/app/object_usecase.rs | 11 +- rustfs/src/app/storage_api.rs | 4 +- scripts/run_get_codec_streaming_smoke.sh | 1 + 13 files changed, 2262 insertions(+), 145 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 6410cf6d6..2116f7fb5 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -35,7 +35,11 @@ use crate::disk::{ health_state::{RuntimeDriveHealthState, get_drive_returning_probe_interval, record_drive_runtime_state}, validate_batch_read_version_item_count, }; -use crate::disk::{disk_store::DiskHealthTracker, error::DiskError, local::ScanGuard}; +use crate::disk::{ + disk_store::DiskHealthTracker, + error::{DiskError, is_terminal_read_error, terminal_read_error_to_io}, + local::ScanGuard, +}; use crate::set_disk::DEFAULT_READ_BUFFER_SIZE; use bytes::Bytes; use futures::lock::Mutex; @@ -324,6 +328,39 @@ where } } +/// Mark a terminal fresh-shard recovery failure for adaptive retirement while +/// retaining its typed `DiskError` and original I/O kind. The decoder checks +/// the marker independently of the kind because not-found and transport +/// failures are terminal too, but must not be reported as timeouts. +fn remote_read_error_to_io(error: DiskError) -> io::Error { + terminal_read_error_to_io(error) +} + +/// Retire a remote shard after its stream can no longer be trusted. A body +/// error that arrives after the one permitted resume is terminal: retaining +/// the reader would let the next stripe poll an already misaligned stream. +fn remote_terminal_io_error(error: io::Error) -> io::Error { + if is_terminal_read_error(&error) { + return error; + } + terminal_read_error_to_io(DiskError::from(error)) +} + +fn remote_terminal_message_to_io(message: &'static str) -> io::Error { + terminal_read_error_to_io(DiskError::Io(io::Error::other(message))) +} + +fn remote_terminal_eof_to_io() -> io::Error { + terminal_read_error_to_io(DiskError::Io(io::Error::new( + io::ErrorKind::UnexpectedEof, + "remote read ended before requested length", + ))) +} + +fn remote_terminal_task_error_to_io(error: JoinError) -> io::Error { + terminal_read_error_to_io(DiskError::other(error)) +} + struct AbortOnDropTask(JoinHandle); impl AbortOnDropTask { @@ -418,6 +455,9 @@ impl RetryingRemoteReader { impl AsyncRead for RetryingRemoteReader { fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if buf.remaining() == 0 { + return Poll::Ready(Ok(())); + } loop { // After the absolute cutoff, let initial progress win over a stale fresh-open. let resume_pending = if let Some(resume) = self.resume.as_mut() { @@ -431,14 +471,14 @@ impl AsyncRead for RetryingRemoteReader { Poll::Ready(Ok(Err(error))) => { self.resume = None; if self.reader.is_none() { - return Poll::Ready(Err(io::Error::other(error))); + return Poll::Ready(Err(remote_read_error_to_io(error))); } continue; } Poll::Ready(Err(error)) => { self.resume = None; if self.reader.is_none() { - return Poll::Ready(Err(io::Error::other(error))); + return Poll::Ready(Err(remote_terminal_task_error_to_io(error))); } continue; } @@ -479,6 +519,9 @@ impl AsyncRead for RetryingRemoteReader { } else { self.resume = None; } + } else if produced == 0 && self.request.length != 0 && self.emitted < self.request.length { + self.reader = None; + return Poll::Ready(Err(remote_terminal_eof_to_io())); } return Poll::Ready(Ok(())); } @@ -494,7 +537,10 @@ impl AsyncRead for RetryingRemoteReader { self.reader = None; continue; } - Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), + Poll::Ready(Err(error)) => { + self.reader = None; + return Poll::Ready(Err(remote_terminal_io_error(error))); + } } } } @@ -585,21 +631,23 @@ impl rustfs_rio::ChunkReader for RetryingRemoteChunkReader { Poll::Ready(Ok(Ok(None))) => { self.resume = None; if self.reader.is_none() { - return Poll::Ready(Err(io::Error::other("remote resume transport did not provide a chunk reader"))); + return Poll::Ready(Err(remote_terminal_message_to_io( + "remote resume transport did not provide a chunk reader", + ))); } continue; } Poll::Ready(Ok(Err(error))) => { self.resume = None; if self.reader.is_none() { - return Poll::Ready(Err(io::Error::other(error))); + return Poll::Ready(Err(remote_read_error_to_io(error))); } continue; } Poll::Ready(Err(error)) => { self.resume = None; if self.reader.is_none() { - return Poll::Ready(Err(io::Error::other(error))); + return Poll::Ready(Err(remote_terminal_task_error_to_io(error))); } continue; } @@ -635,10 +683,26 @@ impl rustfs_rio::ChunkReader for RetryingRemoteChunkReader { return Poll::Ready(Ok(Some(chunk))); } Poll::Ready(Ok(None)) if resume_pending => { + // A clean EOF from the original stream wins when the + // request is unbounded (or has already emitted its full + // bounded length). Waiting for a speculative fresh open + // in that case can turn a successful read into a recovery + // timeout, especially on the read_file/unbounded path. + if self.request.length == 0 || self.emitted >= self.request.length { + self.reader = None; + self.resume = None; + return Poll::Ready(Ok(None)); + } self.reader = None; continue; } - Poll::Ready(Ok(None)) => return Poll::Ready(Ok(None)), + Poll::Ready(Ok(None)) => { + if self.request.length != 0 && self.emitted < self.request.length { + self.reader = None; + return Poll::Ready(Err(remote_terminal_eof_to_io())); + } + return Poll::Ready(Ok(None)); + } Poll::Ready(Err(error)) if !self.retried && is_retryable_remote_body_error(&error) => { self.retried = true; self.reader = None; @@ -651,7 +715,10 @@ impl rustfs_rio::ChunkReader for RetryingRemoteChunkReader { self.reader = None; continue; } - Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), + Poll::Ready(Err(error)) => { + self.reader = None; + return Poll::Ready(Err(remote_terminal_io_error(error))); + } } } } @@ -5154,6 +5221,7 @@ mod tests { enum ResumeReadStep { PartialThenReset(Vec), Data(Vec), + Eof, } #[derive(Debug, Default)] @@ -5412,6 +5480,7 @@ mod tests { struct PendingFreshOpenTransport { fresh_read_drops: Arc, fresh_chunk_drops: Arc, + initial_chunk_eof: bool, } #[async_trait::async_trait] @@ -5429,6 +5498,9 @@ mod tests { } async fn open_read_chunks(&self, _request: ReadStreamRequest) -> Result> { + if self.initial_chunk_eof { + return Ok(Some(resume_step_chunk_reader(ResumeReadStep::Eof))); + } Ok(Some(Box::new(ChunkPartialThenErrorReader { data: None, error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")), @@ -5457,6 +5529,70 @@ mod tests { } } + #[derive(Debug)] + struct TerminalFreshOpenTransport { + fresh_read_opens: Arc, + fresh_chunk_opens: Arc, + chunk_returns_none: bool, + } + + impl TerminalFreshOpenTransport { + fn new(chunk_returns_none: bool) -> Self { + Self { + fresh_read_opens: Arc::new(AtomicUsize::new(0)), + fresh_chunk_opens: Arc::new(AtomicUsize::new(0)), + chunk_returns_none, + } + } + } + + #[async_trait::async_trait] + impl InternodeDataTransport for TerminalFreshOpenTransport { + async fn open_read(&self, _request: ReadStreamRequest) -> Result { + Ok(Box::new(PartialThenErrorReader { + cursor: Cursor::new(Vec::new()), + error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")), + })) + } + + async fn open_read_fresh(&self, _request: ReadStreamRequest) -> Result { + self.fresh_read_opens.fetch_add(1, Ordering::Relaxed); + Err(DiskError::FileNotFound) + } + + async fn open_read_chunks(&self, _request: ReadStreamRequest) -> Result> { + Ok(Some(Box::new(ChunkPartialThenErrorReader { + data: Some(Bytes::from_static(b"x")), + error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")), + }))) + } + + async fn open_read_chunks_fresh(&self, _request: ReadStreamRequest) -> Result> { + self.fresh_chunk_opens.fetch_add(1, Ordering::Relaxed); + if self.chunk_returns_none { + Ok(None) + } else { + Err(DiskError::FileNotFound) + } + } + + async fn open_write(&self, _request: WriteStreamRequest) -> Result { + panic!("open_write should not be used in terminal fresh-open tests"); + } + + async fn open_walk_dir(&self, _request: WalkDirStreamRequest) -> Result { + panic!("open_walk_dir should not be used in terminal fresh-open tests"); + } + + fn name(&self) -> &'static str { + "terminal-fresh-open-test" + } + + fn capabilities(&self) -> InternodeDataTransportCapabilities { + InternodeDataTransportCapabilities::tcp_http() + } + } + fn resume_step_reader(step: ResumeReadStep) -> FileReader { match step { ResumeReadStep::PartialThenReset(data) => Box::new(PartialThenErrorReader { @@ -5464,6 +5600,7 @@ mod tests { error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")), }), ResumeReadStep::Data(data) => Box::new(Cursor::new(data)), + ResumeReadStep::Eof => Box::new(Cursor::new(Vec::new())), } } @@ -5477,6 +5614,7 @@ mod tests { data: Some(Bytes::from(data)), error: None, }), + ResumeReadStep::Eof => Box::new(ChunkPartialThenErrorReader { data: None, error: None }), } } @@ -5553,6 +5691,430 @@ mod tests { } } + fn partial_hashed_shard(shard_size: usize) -> (rustfs_utils::HashAlgorithm, Vec, usize) { + let checksum = rustfs_utils::HashAlgorithm::HighwayHash256S; + let data = vec![0x5a; shard_size]; + let hash_bytes = { + let hash = checksum.hash_encode(&data); + hash.as_ref().to_vec() + }; + let hash_len = hash_bytes.len(); + let encoded_length = hash_len + data.len(); + let mut prefix = Vec::with_capacity(hash_len + shard_size / 2); + prefix.extend_from_slice(&hash_bytes); + prefix.extend_from_slice(&data[..shard_size / 2]); + (checksum, prefix, encoded_length) + } + + #[test] + fn remote_read_error_conversion_preserves_recovery_classification() { + for disk_error in [DiskError::Timeout, DiskError::SourceStalled] { + let error = remote_read_error_to_io(disk_error); + assert_eq!(error.kind(), std_io::ErrorKind::TimedOut); + } + + let error = remote_read_error_to_io(DiskError::Timeout); + assert!( + error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .is_some() + ); + assert!(matches!(DiskError::from(error), DiskError::Timeout)); + + let error = remote_read_error_to_io(DiskError::SourceStalled); + assert!(matches!(DiskError::from(error), DiskError::SourceStalled)); + + let error = + remote_read_error_to_io(DiskError::Io(io::Error::new(std_io::ErrorKind::ConnectionReset, "connection reset"))); + assert_eq!(error.kind(), std_io::ErrorKind::ConnectionReset); + assert!(crate::disk::error::is_terminal_read_error(&error)); + assert!(matches!(DiskError::from(error), DiskError::Io(inner) if inner.kind() == std_io::ErrorKind::ConnectionReset)); + } + + #[test] + fn remote_reader_zero_capacity_poll_is_a_noop() { + let transport: Arc = Arc::new(PendingFreshOpenTransport::default()); + let mut reader = RetryingRemoteReader::new_with_timeouts( + Box::new(Cursor::new(b"x".to_vec())), + transport, + resume_request(1), + None, + None, + ); + let mut empty = []; + let mut read_buf = ReadBuf::new(&mut empty); + let mut cx = Context::from_waker(std::task::Waker::noop()); + + assert!(matches!(Pin::new(&mut reader).poll_read(&mut cx, &mut read_buf), Poll::Ready(Ok(())))); + assert!(reader.reader.is_some(), "zero-capacity polls must not retire the remote reader"); + + let mut output = Vec::new(); + futures::executor::block_on(reader.read_to_end(&mut output)).expect("the reader should remain usable"); + assert_eq!(output, b"x"); + } + + #[tokio::test(start_paused = true)] + async fn remote_reader_fresh_open_timeout_preserves_timed_out_kind() { + let transport = Arc::new(PendingFreshOpenTransport::default()); + let transport_for_reader: Arc = transport.clone(); + let mut reader = RetryingRemoteReader::new_with_timeouts( + resume_step_reader(ResumeReadStep::PartialThenReset(Vec::new())), + transport_for_reader, + resume_request(1), + None, + Some(Duration::from_secs(1)), + ); + + let error = reader + .read_to_end(&mut Vec::new()) + .await + .expect_err("a hung fresh open must surface its recovery timeout"); + + assert_eq!(error.kind(), std_io::ErrorKind::TimedOut); + assert!( + error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .is_some() + ); + assert!(matches!(DiskError::from(error), DiskError::Timeout)); + assert_eq!(transport.fresh_read_drops.load(Ordering::Relaxed), 1); + } + + #[tokio::test(start_paused = true)] + async fn remote_chunk_reader_fresh_open_timeout_preserves_timed_out_kind() { + let transport = Arc::new(PendingFreshOpenTransport::default()); + let transport_for_reader: Arc = transport.clone(); + let mut reader = RetryingRemoteChunkReader::new_with_timeouts( + resume_step_chunk_reader(ResumeReadStep::PartialThenReset(b"x".to_vec())), + transport_for_reader, + resume_request(2), + None, + Some(Duration::from_secs(1)), + ); + + let error = reader + .read_to_end(&mut Vec::new()) + .await + .expect_err("a hung fresh chunk open must surface its recovery timeout"); + + assert_eq!(error.kind(), std_io::ErrorKind::TimedOut); + assert!( + error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .is_some() + ); + assert!(matches!(DiskError::from(error), DiskError::Timeout)); + assert_eq!(transport.fresh_chunk_drops.load(Ordering::Relaxed), 1); + } + + #[tokio::test(start_paused = true)] + async fn remote_chunk_reader_unbounded_clean_eof_wins_over_speculative_resume() { + let transport = Arc::new(PendingFreshOpenTransport { + initial_chunk_eof: true, + ..PendingFreshOpenTransport::default() + }); + let transport_for_reader: Arc = transport.clone(); + let mut reader = RetryingRemoteChunkReader::new_with_timeouts( + resume_step_chunk_reader(ResumeReadStep::Eof), + transport_for_reader, + resume_request(0), + Some(Duration::ZERO), + Some(Duration::from_secs(1)), + ); + + let mut output = Vec::new(); + reader + .read_to_end(&mut output) + .await + .expect("clean EOF from an unbounded original stream should finish the read"); + assert!(output.is_empty()); + // The executor may abort the speculative task before it is first + // polled, in which case the pending-open future never constructs its + // drop probe. The dedicated drop-cancellation tests cover the + // already-polled case; this regression only needs to establish that a + // clean unbounded EOF is not converted into a recovery timeout. + } + + #[tokio::test(start_paused = true)] + async fn remote_reader_fresh_open_non_timeout_error_is_retired_from_adaptive_decode() { + let transport = Arc::new(TerminalFreshOpenTransport::new(false)); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteReader::new_with_timeouts( + resume_step_reader(ResumeReadStep::PartialThenReset(Vec::new())), + transport_for_reader, + resume_request(8), + None, + Some(Duration::from_secs(1)), + ); + let shard = BitrotReader::new(ShardReader::Stream(Box::new(retry)), 8, rustfs_utils::HashAlgorithm::None, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, 8), 0, 16); + + let (_, first_errors) = parallel.read().await; + assert!(matches!(first_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + assert_eq!(transport.fresh_read_opens.load(Ordering::Relaxed), 1); + + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + assert_eq!(transport.fresh_read_opens.load(Ordering::Relaxed), 1); + } + + #[tokio::test(start_paused = true)] + async fn remote_chunk_reader_missing_fresh_reader_is_retired_from_adaptive_decode() { + let transport = Arc::new(TerminalFreshOpenTransport::new(true)); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteChunkReader::new_with_timeouts( + resume_step_chunk_reader(ResumeReadStep::PartialThenReset(b"x".to_vec())), + transport_for_reader, + resume_request(8), + None, + Some(Duration::from_secs(1)), + ); + let shard = BitrotReader::new(ShardReader::Chunked(Box::new(retry)), 8, rustfs_utils::HashAlgorithm::None, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, 8), 0, 16); + + let (_, first_errors) = parallel.read().await; + assert!( + matches!(first_errors.first().and_then(Option::as_ref), Some(DiskError::Io(error)) if error.kind() == std_io::ErrorKind::Other), + "unexpected first errors: {first_errors:?}" + ); + assert_eq!(transport.fresh_chunk_opens.load(Ordering::Relaxed), 1); + + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + assert_eq!(transport.fresh_chunk_opens.load(Ordering::Relaxed), 1); + } + + #[tokio::test] + async fn remote_reader_fresh_open_short_eof_is_retired_from_adaptive_decode() { + // A successful fresh open can still end before the bounded request. + // Treat that as terminal immediately so the next stripe does not poll + // an already exhausted reader and defer the failure to BitrotReader. + let transport = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::Eof])); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteReader::new_with_timeouts( + resume_step_reader(ResumeReadStep::PartialThenReset(b"01".to_vec())), + transport_for_reader, + resume_request(7), + None, + None, + ); + let shard = BitrotReader::new(ShardReader::Stream(Box::new(retry)), 7, rustfs_utils::HashAlgorithm::None, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, 7), 0, 14); + + let (_, first_errors) = parallel.read().await; + assert!(matches!( + first_errors.first().and_then(Option::as_ref), + Some(DiskError::Io(error)) if error.kind() == std_io::ErrorKind::UnexpectedEof + )); + assert_eq!( + transport + .fresh_read_requests + .lock() + .expect("fresh read request lock should not be poisoned") + .len(), + 1 + ); + + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + 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_fresh_open_short_eof_is_retired_from_adaptive_decode() { + let transport = Arc::new(ResumeTransport::with_chunk_steps(vec![ResumeReadStep::Eof])); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteChunkReader::new_with_timeouts( + resume_step_chunk_reader(ResumeReadStep::PartialThenReset(b"01".to_vec())), + transport_for_reader, + resume_request(7), + None, + None, + ); + let shard = BitrotReader::new(ShardReader::Chunked(Box::new(retry)), 7, rustfs_utils::HashAlgorithm::None, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, 7), 0, 14); + + let (_, first_errors) = parallel.read().await; + assert!(matches!( + first_errors.first().and_then(Option::as_ref), + Some(DiskError::Io(error)) if error.kind() == std_io::ErrorKind::UnexpectedEof + )); + assert_eq!( + transport + .fresh_chunk_requests + .lock() + .expect("fresh chunk request lock should not be poisoned") + .len(), + 1 + ); + + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + assert_eq!( + transport + .fresh_chunk_requests + .lock() + .expect("fresh chunk request lock should not be poisoned") + .len(), + 1 + ); + } + + #[tokio::test] + async fn remote_reader_second_body_reset_is_retired_from_adaptive_decode() { + // The first connection emits a prefix, the one permitted fresh + // connection emits another prefix, and then resets again. The second + // reset must retire the reader so the next stripe cannot consume a + // misaligned stream. + let transport = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::PartialThenReset(b"23".to_vec())])); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteReader::new_with_timeouts( + resume_step_reader(ResumeReadStep::PartialThenReset(b"01".to_vec())), + transport_for_reader, + resume_request(7), + None, + None, + ); + let shard = BitrotReader::new(ShardReader::Stream(Box::new(retry)), 7, rustfs_utils::HashAlgorithm::None, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, 7), 0, 14); + + let (_, first_errors) = parallel.read().await; + assert!(matches!( + first_errors.first().and_then(Option::as_ref), + Some(DiskError::Io(error)) if error.kind() == std_io::ErrorKind::ConnectionReset + )); + assert_eq!( + transport + .fresh_read_requests + .lock() + .expect("fresh read request lock should not be poisoned") + .len(), + 1 + ); + + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + 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_second_body_reset_is_retired_from_adaptive_decode() { + let transport = Arc::new(ResumeTransport::with_chunk_steps(vec![ResumeReadStep::PartialThenReset(b"23".to_vec())])); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteChunkReader::new_with_timeouts( + resume_step_chunk_reader(ResumeReadStep::PartialThenReset(b"01".to_vec())), + transport_for_reader, + resume_request(7), + None, + None, + ); + let shard = BitrotReader::new(ShardReader::Chunked(Box::new(retry)), 7, rustfs_utils::HashAlgorithm::None, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, 7), 0, 14); + + let (_, first_errors) = parallel.read().await; + assert!(matches!( + first_errors.first().and_then(Option::as_ref), + Some(DiskError::Io(error)) if error.kind() == std_io::ErrorKind::ConnectionReset + )); + assert_eq!( + transport + .fresh_chunk_requests + .lock() + .expect("fresh chunk request lock should not be poisoned") + .len(), + 1 + ); + + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + assert_eq!( + transport + .fresh_chunk_requests + .lock() + .expect("fresh chunk request lock should not be poisoned") + .len(), + 1 + ); + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn remote_reader_hashed_timeout_is_retired_from_adaptive_decode() { + temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_DISK_READ_TIMEOUT, Some("5"))], async { + const SHARD_SIZE: usize = 64; + let (checksum, encoded_prefix, encoded_length) = partial_hashed_shard(SHARD_SIZE); + let transport = Arc::new(PendingFreshOpenTransport::default()); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteReader::new_with_timeouts( + resume_step_reader(ResumeReadStep::PartialThenReset(encoded_prefix)), + transport_for_reader, + resume_request(encoded_length), + None, + Some(Duration::from_secs(1)), + ); + let shard = BitrotReader::new(ShardReader::Stream(Box::new(retry)), SHARD_SIZE, checksum, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, SHARD_SIZE), 0, SHARD_SIZE * 2); + + let (first_buffers, first_errors) = parallel.read().await; + assert!(first_buffers.first().and_then(Option::as_ref).is_none()); + assert!(matches!(first_errors.first().and_then(Option::as_ref), Some(DiskError::Timeout))); + assert_eq!(transport.fresh_read_drops.load(Ordering::Relaxed), 1); + + // A TimedOut error retires the dead slot. The next stripe therefore + // reports the slot as unavailable instead of polling a reader whose + // fresh connection already timed out. + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + }) + .await; + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn remote_chunk_reader_hashed_timeout_is_retired_from_adaptive_decode() { + temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_DISK_READ_TIMEOUT, Some("5"))], async { + const SHARD_SIZE: usize = 64; + let (checksum, encoded_prefix, encoded_length) = partial_hashed_shard(SHARD_SIZE); + let transport = Arc::new(PendingFreshOpenTransport::default()); + let transport_for_reader: Arc = transport.clone(); + let retry = RetryingRemoteChunkReader::new_with_timeouts( + resume_step_chunk_reader(ResumeReadStep::PartialThenReset(encoded_prefix)), + transport_for_reader, + resume_request(encoded_length), + None, + Some(Duration::from_secs(1)), + ); + let shard = BitrotReader::new(ShardReader::Chunked(Box::new(retry)), SHARD_SIZE, checksum, false); + let mut parallel = ParallelReader::new(vec![Some(shard)], Erasure::new(1, 0, SHARD_SIZE), 0, SHARD_SIZE * 2); + + let (first_buffers, first_errors) = parallel.read().await; + assert!(first_buffers.first().and_then(Option::as_ref).is_none()); + assert!(matches!(first_errors.first().and_then(Option::as_ref), Some(DiskError::Timeout))); + assert_eq!(transport.fresh_chunk_drops.load(Ordering::Relaxed), 1); + + let (_, second_errors) = parallel.read().await; + assert!(matches!(second_errors.first().and_then(Option::as_ref), Some(DiskError::FileNotFound))); + }) + .await; + } + #[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())])); diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index 6f771f790..dff44d78b 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -23,6 +23,16 @@ pub type Result = core::result::Result; const METACACHE_OUTPUT_STREAM_CLOSED: &str = "metacache output stream closed"; +/// Marker carried by a shard-read `io::Error` when the underlying reader can +/// no longer be realigned after a fresh remote open failed. The marker is +/// deliberately separate from the `ErrorKind`: a terminal read must retire +/// its reader, while its original typed disk error and I/O kind still need to +/// survive quorum/error mapping. +#[derive(Debug)] +pub(crate) struct TerminalReadError { + source: DiskError, +} + // DiskError == StorageErr #[derive(Debug, thiserror::Error)] pub enum DiskError { @@ -168,6 +178,67 @@ pub enum DiskError { RemoteClientUnavailable(String), } +impl TerminalReadError { + pub(crate) fn new(source: DiskError) -> Self { + Self { source } + } + + fn into_source(self) -> DiskError { + self.source + } +} + +impl std::fmt::Display for TerminalReadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.source.fmt(f) + } +} + +impl StdError for TerminalReadError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(&self.source) + } +} + +fn classify_internode_missing_error(error: &InternodeHttpError) -> Option { + if error.is_remote_file_not_found() { + return Some(DiskError::FileNotFound); + } + if error.is_remote_volume_not_found() { + return Some(DiskError::VolumeNotFound); + } + None +} + +/// Wrap a terminal shard-read failure without changing its typed +/// classification. Timeout-like disk errors retain `TimedOut`; other errors +/// retain their inner I/O kind or use `Other` when no more specific kind exists. +pub(crate) fn terminal_read_error_to_io(error: DiskError) -> io::Error { + let kind = match &error { + DiskError::Io(inner) => inner.kind(), + DiskError::SourceStalled | DiskError::Timeout => io::ErrorKind::TimedOut, + DiskError::DiskNotFound + | DiskError::FileNotFound + | DiskError::FileVersionNotFound + | DiskError::PathNotFound + | DiskError::VolumeNotFound => io::ErrorKind::NotFound, + DiskError::DiskAccessDenied | DiskError::FileAccessDenied | DiskError::VolumeAccessDenied => { + io::ErrorKind::PermissionDenied + } + DiskError::DiskFull => io::ErrorKind::StorageFull, + DiskError::FileCorrupt | DiskError::PartMissingOrCorrupt | DiskError::BitrotHashAlgoInvalid => io::ErrorKind::InvalidData, + _ => io::ErrorKind::Other, + }; + io::Error::new(kind, TerminalReadError::new(error)) +} + +/// Whether an I/O error marks a shard reader as terminal for adaptive decode. +pub(crate) fn is_terminal_read_error(error: &io::Error) -> bool { + error + .get_ref() + .is_some_and(|source| source.downcast_ref::().is_some()) +} + impl From for DiskError { fn from(error: crate::erasure::coding::ErasureConstructionError) -> Self { Self::Io(error.into_io_error()) @@ -344,6 +415,21 @@ impl From for DiskError { return DiskError::VolumeNotFound; } } + let e = match e.downcast::() { + Ok(terminal_error) => { + let source = terminal_error.into_source(); + if let DiskError::Io(io_error) = &source + && let Some(internode_error) = io_error + .get_ref() + .and_then(|source| source.downcast_ref::()) + && let Some(classified) = classify_internode_missing_error(internode_error) + { + return classified; + } + return source; + } + Err(e) => e, + }; match e.downcast::() { Ok(disk_error) => disk_error, // Mirror `From for StorageError`: a StorageError boxed @@ -679,6 +765,32 @@ mod tests { use super::*; use std::collections::HashMap; + #[test] + fn terminal_read_error_preserves_kind_and_disk_classification() { + let timeout = terminal_read_error_to_io(DiskError::Timeout); + assert_eq!(timeout.kind(), io::ErrorKind::TimedOut); + assert!(is_terminal_read_error(&timeout)); + assert!(matches!(DiskError::from(timeout), DiskError::Timeout)); + + let missing = terminal_read_error_to_io(DiskError::FileNotFound); + assert_eq!(missing.kind(), io::ErrorKind::NotFound); + assert!(is_terminal_read_error(&missing)); + assert!(matches!(DiskError::from(missing), DiskError::FileNotFound)); + + let reset = terminal_read_error_to_io(DiskError::Io(io::Error::new(io::ErrorKind::ConnectionReset, "connection reset"))); + assert_eq!(reset.kind(), io::ErrorKind::ConnectionReset); + assert!(is_terminal_read_error(&reset)); + assert!(matches!(DiskError::from(reset), DiskError::Io(error) if error.kind() == io::ErrorKind::ConnectionReset)); + + for (remote_error, expected) in [ + (rustfs_rio::new_test_remote_file_not_found_http_io_error(), DiskError::FileNotFound), + (rustfs_rio::new_test_remote_volume_not_found_http_io_error(), DiskError::VolumeNotFound), + ] { + let wrapped = terminal_read_error_to_io(DiskError::Io(remote_error)); + assert_eq!(DiskError::from(wrapped), expected); + } + } + #[test] fn other_preserves_erasure_construction_source_chain() { use crate::erasure::coding::ErasureConstructionError; diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index 8b616c058..7c8b38b54 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/crates/ecstore/src/erasure/coding/decode.rs @@ -20,7 +20,7 @@ use crate::diagnostics::get::{ record_get_object_pipeline_failure, record_get_stage_duration_if_enabled, }; use crate::disk::disk_store::get_object_disk_read_timeout; -use crate::disk::error::Error; +use crate::disk::error::{Error, is_terminal_read_error}; use crate::disk::error_reduce::reduce_errs; use crate::erasure::codec::workspace::ShardBufferPool; use crate::erasure::coding::{BitrotReader, Erasure}; @@ -36,12 +36,16 @@ use std::future::Future; use std::io; use std::io::ErrorKind; use std::pin::Pin; +use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::io::AsyncWrite; use tokio::io::AsyncWriteExt; use tracing::{debug, error, warn}; type ShardReadFuture<'a> = Pin, Error>, bool)> + Send + 'a>>; +type OwnedShardReadFuture<'a, R> = + Pin, Error>, Option>, bool)> + Send + 'a>>; +pub(crate) type DeferredReaderReopener = Arc Option> + Send + Sync>; type ShardIndexes = SmallVec<[usize; INLINE_SHARD_SLOTS]>; type ActiveReaders = SmallVec<[bool; INLINE_SHARD_SLOTS]>; @@ -56,6 +60,41 @@ const SHARD_LOCALITY_SCHEDULING_OFF: &str = "off"; const SHARD_LOCALITY_SCHEDULING_OBSERVE: &str = "observe"; const SHARD_LOCALITY_SCHEDULING_ON: &str = "on"; +/// Read-ahead contract selected by the caller of the erasure decoder. +/// +/// Ordinary GETs retain the configured overlap and all-shard lockstep +/// behavior. A server-side copy holds its source while the destination can +/// apply backpressure, so it uses the demand-bound variant: no speculative +/// stripe read and data shards only until reconstruction needs parity. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(crate) enum DecodeReadPolicy { + #[default] + Default, + DemandBound, +} + +/// Maximum number of deferred parity reads that a demand-bound stripe may have +/// in flight at once. The window is deliberately small: a 16+16 layout must +/// not turn one slow data shard into 16 simultaneous HTTP/H2 opens. The +/// window is refilled as results arrive, so larger erasure sets still make +/// progress without an unbounded fan-out. +const MAX_DEMAND_BOUND_PARITY_IN_FLIGHT: usize = 4; + +tokio::task_local! { + static DECODE_READ_POLICY: DecodeReadPolicy; +} + +pub(crate) fn decode_read_policy() -> DecodeReadPolicy { + DECODE_READ_POLICY.try_with(|policy| *policy).unwrap_or_default() +} + +pub(crate) async fn with_decode_read_policy(policy: DecodeReadPolicy, future: F) -> F::Output +where + F: Future, +{ + DECODE_READ_POLICY.scope(policy, future).await +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ShardLocalitySchedulingMode { Off, @@ -136,10 +175,11 @@ const DEFAULT_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE: bool = false; /// Whether the data-shards-only lockstep GET read is enabled (backlog#923). pub(crate) fn get_lockstep_data_shards_only_enabled() -> bool { - rustfs_utils::get_env_bool( - ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, - DEFAULT_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, - ) + matches!(decode_read_policy(), DecodeReadPolicy::DemandBound) + || rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, + DEFAULT_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, + ) } /// Get whether bitrot-decode overlap is enabled. @@ -170,7 +210,8 @@ fn is_bitrot_decode_overlap_enabled() -> bool { /// pre-existing strictly-serial read → reconstruct → emit behaviour, byte for /// byte. fn legacy_stripe_prefetch_enabled() -> bool { - get_decode_stripe_prefetch_count() > 1 || is_bitrot_decode_overlap_enabled() + !matches!(decode_read_policy(), DecodeReadPolicy::DemandBound) + && (get_decode_stripe_prefetch_count() > 1 || is_bitrot_decode_overlap_enabled()) } /// Outcome of reconstructing and emitting a single already-read stripe in the @@ -268,6 +309,92 @@ fn shard_role(index: usize, data_shards: usize) -> &'static str { } } +#[allow(clippy::too_many_arguments)] +async fn read_shard_result( + index: usize, + read_cost: ShardReadCost, + reader: &mut BitrotReader, + recycled_buf: Option>, + shard_size: usize, + data_shards: usize, + read_timeout: Duration, + metrics_path: Option<&'static str>, +) -> (Result, Error>, bool) +where + R: crate::erasure::coding::ShardSource, +{ + let role = shard_role(index, data_shards); + // Capacity, not length: `read_appending` writes every byte it returns, so + // the buffer never needs zeroing first (rustfs/backlog#1159). + let mut buf = recycled_buf.unwrap_or_else(|| Vec::with_capacity(shard_size)); + buf.clear(); + let read_start = metrics_path.map(|_| Instant::now()); + let read_result = if read_timeout.is_zero() { + reader.read_appending(&mut buf, shard_size).await + } else { + match tokio::time::timeout(read_timeout, reader.read_appending(&mut buf, shard_size)).await { + Ok(result) => result, + Err(_) => { + let timeout_error = io::Error::new(ErrorKind::TimedOut, "shard read timed out"); + let error_class = classify_io_error(&timeout_error).as_str(); + if let Some(path) = metrics_path { + rustfs_io_metrics::record_get_object_shard_read_observation( + path, + index, + role, + read_cost.as_str(), + GET_SHARD_READ_OUTCOME_ERROR, + error_class, + 0, + read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()), + reader.last_verify_duration().as_secs_f64(), + ); + } + return (Err(Error::from(timeout_error)), true); + } + } + }; + + match read_result { + Ok(n) => { + debug_assert_eq!(buf.len(), n, "read_appending must grow the buffer by exactly n"); + if let Some(path) = metrics_path { + rustfs_io_metrics::record_get_object_shard_read_observation( + path, + index, + role, + read_cost.as_str(), + GET_SHARD_READ_OUTCOME_SUCCESS, + GET_SHARD_READ_ERROR_NONE, + n, + read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()), + reader.last_verify_duration().as_secs_f64(), + ); + } + (Ok(buf), false) + } + Err(e) => { + let verify_duration_secs = reader.last_verify_duration().as_secs_f64(); + let error_class = classify_io_error(&e).as_str(); + let should_retire = e.kind() == ErrorKind::TimedOut || is_terminal_read_error(&e); + if let Some(path) = metrics_path { + rustfs_io_metrics::record_get_object_shard_read_observation( + path, + index, + role, + read_cost.as_str(), + GET_SHARD_READ_OUTCOME_ERROR, + error_class, + 0, + read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()), + verify_duration_secs, + ); + } + (Err(Error::from(e)), should_retire) + } + } +} + #[allow(clippy::too_many_arguments)] fn read_shard<'a, R>( index: usize, @@ -285,75 +412,18 @@ where let role = shard_role(index, data_shards); if let Some(reader) = reader { Box::pin(async move { - // Capacity, not length: `read_appending` writes every byte it returns, so - // the buffer never needs zeroing first (rustfs/backlog#1159). - let mut buf = recycled_buf.unwrap_or_else(|| Vec::with_capacity(shard_size)); - buf.clear(); - let read_start = metrics_path.map(|_| Instant::now()); - let read_result = if read_timeout.is_zero() { - reader.read_appending(&mut buf, shard_size).await - } else { - match tokio::time::timeout(read_timeout, reader.read_appending(&mut buf, shard_size)).await { - Ok(result) => result, - Err(_) => { - let timeout_error = io::Error::new(ErrorKind::TimedOut, "shard read timed out"); - let error_class = classify_io_error(&timeout_error).as_str(); - if let Some(path) = metrics_path { - rustfs_io_metrics::record_get_object_shard_read_observation( - path, - index, - role, - read_cost.as_str(), - GET_SHARD_READ_OUTCOME_ERROR, - error_class, - 0, - read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()), - reader.last_verify_duration().as_secs_f64(), - ); - } - return (index, read_cost, Err(Error::from(timeout_error)), true); - } - } - }; - - match read_result { - Ok(n) => { - debug_assert_eq!(buf.len(), n, "read_appending must grow the buffer by exactly n"); - if let Some(path) = metrics_path { - rustfs_io_metrics::record_get_object_shard_read_observation( - path, - index, - role, - read_cost.as_str(), - GET_SHARD_READ_OUTCOME_SUCCESS, - GET_SHARD_READ_ERROR_NONE, - n, - read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()), - reader.last_verify_duration().as_secs_f64(), - ); - } - (index, read_cost, Ok(buf), false) - } - Err(e) => { - let verify_duration_secs = reader.last_verify_duration().as_secs_f64(); - let error_class = classify_io_error(&e).as_str(); - let should_retire = e.kind() == ErrorKind::TimedOut; - if let Some(path) = metrics_path { - rustfs_io_metrics::record_get_object_shard_read_observation( - path, - index, - role, - read_cost.as_str(), - GET_SHARD_READ_OUTCOME_ERROR, - error_class, - 0, - read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()), - verify_duration_secs, - ); - } - (index, read_cost, Err(Error::from(e)), should_retire) - } - } + let (result, should_retire) = read_shard_result( + index, + read_cost, + reader, + recycled_buf, + shard_size, + data_shards, + read_timeout, + metrics_path, + ) + .await; + (index, read_cost, result, should_retire) }) } else { Box::pin(async move { @@ -375,6 +445,121 @@ where } } +#[allow(clippy::too_many_arguments)] +fn read_shard_owned<'a, R>( + index: usize, + read_cost: ShardReadCost, + reader: Option>, + recycled_buf: Option>, + shard_size: usize, + data_shards: usize, + read_timeout: Duration, + metrics_path: Option<&'static str>, +) -> OwnedShardReadFuture<'a, R> +where + R: crate::erasure::coding::ShardSource + 'a, +{ + let role = shard_role(index, data_shards); + Box::pin(async move { + let Some(mut reader) = reader else { + if let Some(path) = metrics_path { + rustfs_io_metrics::record_get_object_shard_read_observation( + path, + index, + role, + read_cost.as_str(), + GET_SHARD_READ_OUTCOME_MISSING, + GET_SHARD_READ_ERROR_MISSING, + 0, + 0.0, + 0.0, + ); + } + return (index, read_cost, Err(Error::FileNotFound), None, false); + }; + let (result, should_retire) = read_shard_result( + index, + read_cost, + &mut reader, + recycled_buf, + shard_size, + data_shards, + read_timeout, + metrics_path, + ) + .await; + (index, read_cost, result, Some(reader), should_retire) + }) +} + +#[allow(clippy::too_many_arguments)] +fn launch_owned_shard<'a, R>( + sets: &mut FuturesUnordered>, + readers: &mut [Option>], + buffers: &mut ShardBufferPool, + active: &mut [bool], + scheduled: &mut usize, + index: usize, + shard_size: usize, + data_shards: usize, + read_cost: ShardReadCost, + read_timeout: Duration, + metrics_path: Option<&'static str>, +) -> bool +where + R: crate::erasure::coding::ShardSource + 'a, +{ + let Some(reader) = readers.get_mut(index).and_then(Option::take) else { + return false; + }; + launch_owned_reader( + sets, + buffers, + active, + scheduled, + index, + reader, + shard_size, + data_shards, + read_cost, + read_timeout, + metrics_path, + ) +} + +#[allow(clippy::too_many_arguments)] +fn launch_owned_reader<'a, R>( + sets: &mut FuturesUnordered>, + buffers: &mut ShardBufferPool, + active: &mut [bool], + scheduled: &mut usize, + index: usize, + reader: BitrotReader, + shard_size: usize, + data_shards: usize, + read_cost: ShardReadCost, + read_timeout: Duration, + metrics_path: Option<&'static str>, +) -> bool +where + R: crate::erasure::coding::ShardSource + 'a, +{ + let recycled_buf = Some(buffers.take(index, shard_size)); + *scheduled += 1; + active[index] = true; + sets.push(read_shard_owned( + index, + read_cost, + Some(reader), + recycled_buf, + shard_size, + data_shards, + read_timeout, + metrics_path, + )); + true +} + pin_project! { pub(crate) struct ParallelReader { #[pin] @@ -400,6 +585,11 @@ pub(crate) struct ParallelReader { // it to the current stripe when it is engaged mid-object (backlog#923). engaged: SmallVec<[bool; INLINE_SHARD_SLOTS]>, deferred_handles: Vec>, + // Copy-source hedges use a fresh deferred reader so cancelling a hedge + // never consumes the unopened reader reserved for a later stripe. The + // vector is empty for callers that do not provide a reopen factory (tests + // and the ordinary GET path retain the handle-based behavior). + deferred_reopeners: Vec>>, stripe_index: usize, } } @@ -607,6 +797,7 @@ where stripe_state: None, engaged, deferred_handles: Vec::new(), + deferred_reopeners: Vec::new(), stripe_index: 0, } } @@ -622,6 +813,17 @@ where self.deferred_handles = handles; self } + + /// Attach factories for unopened parity readers. A factory must return a + /// reader already aligned to the requested stripe. Keeping the original + /// deferred reader in `self.readers` lets a cancelled hedge be discarded + /// without poisoning the next-stripe reserve. + pub(crate) fn with_deferred_parity_reopeners(mut self, mut reopeners: Vec>>) -> Self { + reopeners.resize_with(self.readers.len(), || None); + reopeners.truncate(self.readers.len()); + self.deferred_reopeners = reopeners; + self + } } #[allow(clippy::too_many_arguments)] @@ -675,6 +877,34 @@ fn shard_read_hedge_delay(read_timeout: Duration) -> Option { } } +/// Return the number of new deferred parity readers that may be admitted for +/// the current demand-bound stripe. The window is based on concrete state, +/// not on setup candidates: at most `missing_data + 1` parity reads are useful +/// for a verification quorum, and the global in-flight cap keeps a wide EC +/// layout from opening every remaining shard at once. As a read completes the +/// caller invokes this again, which refills one slot after a failure or a +/// successful-but-insufficient parity result. +fn demand_bound_parity_admission_limit(shards: &[Option>], active: &[bool], data_shards: usize) -> usize { + let missing_data = shards.iter().take(data_shards).filter(|shard| shard.is_none()).count(); + if missing_data == 0 { + return 0; + } + + let successes = shards.iter().filter(|shard| shard.is_some()).count(); + let needed_for_verification = (data_shards + 1).saturating_sub(successes); + let desired = missing_data + .saturating_add(1) + .min(needed_for_verification) + .min(MAX_DEMAND_BOUND_PARITY_IN_FLIGHT); + let active_parity = active + .iter() + .enumerate() + .skip(data_shards) + .filter(|(_, is_active)| **is_active) + .count(); + desired.saturating_sub(active_parity) +} + fn shard_locality_remote_avoid_potential(remote_scheduled: usize, low_cost_available: usize, data_shards: usize) -> usize { let theoretical_remote_needed = data_shards.saturating_sub(low_cost_available); remote_scheduled.saturating_sub(theoretical_remote_needed) @@ -1045,6 +1275,11 @@ where /// realigned (no pending deferred handle) is likewise retired instead of /// being read out of position. async fn read_lockstep(&mut self, state: &mut StripeReadState) { + if matches!(decode_read_policy(), DecodeReadPolicy::DemandBound) { + self.read_lockstep_demand_bound(state).await; + return; + } + let num_readers = self.readers.len(); state.reset(num_readers, self.data_shards); let shard_size = if self.offset + self.shard_size > self.shard_file_size { @@ -1102,6 +1337,7 @@ where } let data_shards = self.data_shards; + let read_timeout = self.read_timeout; let metrics_path = self.metrics_path; let locality_preference_enabled = self.locality_preference_enabled; @@ -1295,17 +1531,431 @@ where } } - /// Attempt to bring an as-yet-unread parity reader into the lockstep read - /// set at `stripe_index`. + /// Demand-bound lockstep stripe read used by server-side copy sources. /// - /// At stripe 0 every reader is still positioned at the stream start, so - /// engagement is trivially aligned. Past stripe 0 the parity reader must - /// still be an unopened deferred reader: its pending open offset is - /// advanced by `stripe_index` bitrot blocks (the `bitrot_encoded_range` - /// geometry) so its first read returns the current stripe. A parity reader - /// that cannot be realigned is retired for the rest of the object, - /// mirroring the retire-on-error rule: reading it would return an earlier - /// stripe and reintroduce the backlog#832 desync. + /// The ordinary lockstep path can cancel every in-flight reader once it + /// has a quorum because all of its parity readers are already engaged. + /// Copy sources keep parity unopened until a data reader is missing. A + /// hedge therefore has to race the deferred parity reads against the + /// original data reads and may retire the latter only after the parity has + /// produced an actual decode-plus-verification quorum. The futures own + /// their readers so disjoint data/parity slots can be admitted while the + /// other group is still pending; dropping an abandoned future retires its + /// stream without leaving a borrowed slot behind. + async fn read_lockstep_demand_bound(&mut self, state: &mut StripeReadState) { + let num_readers = self.readers.len(); + state.reset(num_readers, self.data_shards); + let shard_size = if self.offset + self.shard_size > self.shard_file_size { + self.shard_file_size - self.offset + } else { + self.shard_size + }; + + let (shards, errs) = state.parts_mut(); + if shard_size == 0 { + return; + } + + self.offset += shard_size; + let stripe_index = self.stripe_index; + self.stripe_index += 1; + self.buffers.ensure_slots(num_readers); + + // A data slot retired on an earlier stripe is already missing. The + // bounded parity launcher below admits enough substitutes before the + // first data future is polled, preserving the lockstep alignment. + let missing_data_readers = self.readers.iter().take(self.data_shards).filter(|r| r.is_none()).count(); + + let data_shards = self.data_shards; + let read_timeout = self.read_timeout; + let metrics_path = self.metrics_path; + let stripe_read_start = metrics_path.map(|_| Instant::now()); + let mut retire_readers = ShardIndexes::new(); + let mut scheduled = 0usize; + let mut success = 0usize; + let mut completed = 0usize; + let mut failed = 0usize; + let mut first_shard_recorded = false; + let mut active = vec![false; num_readers]; + let mut temporary_parity = vec![false; num_readers]; + // A deferred parity slot is attempted at most once per stripe. A + // failed disposable hedge keeps its unopened reserve for the next + // stripe, but must not be relaunched in a tight same-stripe retry + // loop (which would defeat the bounded fan-out and amplify a remote + // outage). + let mut attempted_parity = vec![false; num_readers]; + // Once a data reader has returned an error (or was already missing at + // setup), the loss is permanent for lockstep alignment. Use the + // deferred handle and keep parity engaged across subsequent stripes; + // disposable reopeners are reserved for an as-yet unresolved slow + // data reader. + let mut data_failure_seen = missing_data_readers > 0; + let mut sets: FuturesUnordered> = FuturesUnordered::new(); + // Once a deferred parity reader has been admitted, a concrete + // `data_shards + 1` result is enough to finish a degraded stripe and + // abandon only the still-pending readers. Setup counts never set this + // flag: they are candidates, not successful shards. + let mut fallback_admitted = false; + + // Move engaged readers into owned futures. This leaves the slots free + // so a deferred parity reader can be admitted while these reads wait. + for i in 0..num_readers { + if !self.engaged[i] || self.readers[i].is_none() { + continue; + } + let read_cost = self.read_costs.get(i).copied().unwrap_or(ShardReadCost::Unknown); + let _ = launch_owned_shard( + &mut sets, + &mut self.readers, + &mut self.buffers, + &mut active, + &mut scheduled, + i, + shard_size, + data_shards, + read_cost, + read_timeout, + metrics_path, + ); + } + + if missing_data_readers > 0 { + let want = (missing_data_readers + 1).min(MAX_DEMAND_BOUND_PARITY_IN_FLIGHT); + if self.launch_demand_bound_parity( + stripe_index, + want, + &mut sets, + &mut active, + &mut temporary_parity, + &mut attempted_parity, + false, + &mut scheduled, + shard_size, + data_shards, + read_timeout, + metrics_path, + ) > 0 + { + fallback_admitted = true; + } + } + + let hedge_delay = shard_read_hedge_delay(read_timeout); + let hedge_sleep = hedge_delay.map(tokio::time::sleep); + tokio::pin!(hedge_sleep); + let mut hedged = false; + + loop { + let item = if !hedged { + match hedge_sleep.as_mut().as_pin_mut() { + Some(sleep) => tokio::select! { + biased; + item = sets.next() => item, + _ = sleep => { + hedged = true; + // Do not cancel a pending data read based on setup + // counts. Admit every still-unengaged parity + // reader as a bounded hedge batch; only concrete + // successful results below can satisfy the quorum. + let data_missing = shards.iter().take(data_shards).any(|shard| shard.is_none()); + if data_missing { + let launched = self.launch_demand_bound_parity( + stripe_index, + demand_bound_parity_admission_limit(shards, &active, data_shards), + &mut sets, + &mut active, + &mut temporary_parity, + &mut attempted_parity, + true, + &mut scheduled, + shard_size, + data_shards, + read_timeout, + metrics_path, + ); + if launched > 0 { + fallback_admitted = true; + } + } + continue; + } + }, + None => sets.next().await, + } + } else { + sets.next().await + }; + + let Some((i, _read_cost, result, reader, should_retire)) = item else { + // A fast failure can drain the initial data futures before the + // hedge timer fires (and a zero timeout intentionally has no + // timer). Do not return a false quorum just because the + // FuturesUnordered is momentarily empty: admit the deferred + // parity candidates and race them now. + let data_missing = shards.iter().take(data_shards).any(|shard| shard.is_none()); + data_failure_seen |= errs.iter().take(data_shards).any(Option::is_some); + if data_missing && success <= data_shards { + let launched = self.launch_demand_bound_parity( + stripe_index, + demand_bound_parity_admission_limit(shards, &active, data_shards), + &mut sets, + &mut active, + &mut temporary_parity, + &mut attempted_parity, + !data_failure_seen, + &mut scheduled, + shard_size, + data_shards, + read_timeout, + metrics_path, + ); + if launched > 0 { + fallback_admitted = true; + continue; + } + } + break; + }; + let result_failed = result.is_err(); + active[i] = false; + completed += 1; + if !first_shard_recorded { + if let Some(path) = metrics_path { + record_get_stage_duration_if_enabled(path, GET_STAGE_STRIPE_READ_FIRST_SHARD, stripe_read_start); + } + first_shard_recorded = true; + } + + match result { + Ok(v) => { + shards[i] = Some(v); + success += 1; + // A successful reader consumed exactly one aligned stripe + // and remains usable on the following stripe. + if temporary_parity[i] { + // A reopener hedge is disposable. Keep the unopened + // reserve untouched even when the hedge wins: promoting + // the one-stripe reader would make every later healthy + // stripe read parity and would leave a reset/timeout + // without a way to reopen it at the next stripe. + drop(reader); + self.engaged[i] = false; + } else if !should_retire { + self.readers[i] = reader; + } + } + Err(e) => { + failed += 1; + if i < data_shards { + data_failure_seen = true; + } + if temporary_parity[i] { + // A disposable hedge reader is independent of the + // unopened deferred reserve. Its timeout/reset may be + // transient, so discard only the hedge and keep the + // reserve available for a later stripe. The factory + // already removed a slot when it could not produce an + // aligned reader at all; an error after launch must + // not turn that setup failure policy into permanent + // disk retirement. Do not publish this disposable + // error into `errs`: `emit_decoded_stripe` uses that + // vector for terminal FileNotFound/FileCorrupt + // attribution, and a speculative failure must not + // fail a stripe that later reaches a real quorum. + self.engaged[i] = false; + } else { + errs[i] = Some(e); + // Lockstep cannot safely reuse a reader after any + // error, even when the low-level classifier called it + // nonfatal. + self.readers[i] = None; + retire_readers.push(i); + } + } + } + + // A degraded stripe should keep a small parity window full. Admit + // it immediately after any result (especially a fast data error, + // or a zero-timeout read where no hedge timer exists). Refill one + // slot after a parity failure/success rather than opening every + // candidate at once. Pending data futures stay in `sets` and can + // still win the race if the source recovers. + let data_missing = shards.iter().take(data_shards).any(|shard| shard.is_none()); + if data_missing && success <= data_shards && (result_failed || fallback_admitted) { + let launched = self.launch_demand_bound_parity( + stripe_index, + demand_bound_parity_admission_limit(shards, &active, data_shards), + &mut sets, + &mut active, + &mut temporary_parity, + &mut attempted_parity, + !data_failure_seen, + &mut scheduled, + shard_size, + data_shards, + read_timeout, + metrics_path, + ); + if launched > 0 { + fallback_admitted = true; + } + } + + // Once a real quorum is present, abandon only the still-pending + // futures. If a parity read failed, the pending original data + // reader remains in the race and can still rescue the stripe; no + // optimistic setup count may retire it early. A healthy stripe + // can also finish as soon as every data shard has returned, even + // when the hedge timer has not fired. + let succeeded = shards.iter().filter(|shard| shard.is_some()).count(); + let data_missing = shards.iter().take(data_shards).any(|shard| shard.is_none()); + if !data_missing { + break; + } + if fallback_admitted && succeeded > data_shards { + break; + } + } + + // Dropping `sets` cancels all remaining owned reads. A temporary + // parity hedge has an untouched deferred reserve in `self.readers`, so + // it can be abandoned without poisoning the next stripe. All other + // active readers were consumed in this stripe and must be retired. + for i in 0..num_readers { + if active[i] { + if temporary_parity[i] { + // The factory reader is disposable; leave the original + // deferred reader unengaged and available for a later + // stripe. + self.engaged[i] = false; + continue; + } + if shards[i].is_none() && errs[i].is_none() { + errs[i] = Some(Error::from(io::Error::new(ErrorKind::TimedOut, "shard read hedged after a slow shard"))); + retire_readers.push(i); + } + self.readers[i] = None; + } + } + drop(sets); + + if let Some(path) = metrics_path { + record_get_stage_duration_if_enabled(path, GET_STAGE_STRIPE_READ_QUORUM, stripe_read_start); + rustfs_io_metrics::record_get_object_shard_read_fanout(path, scheduled, completed, success, failed); + } + + for i in retire_readers { + self.readers[i] = None; + } + } + + /// Launch a bounded demand admission for deferred parity. `disposable` + /// selects a speculative hedge (a fresh reopener whose reserve remains + /// unopened) versus a confirmed loss (the deferred handle is engaged and + /// retained across stripes). Tests and legacy callers without a reopener + /// use the handle-based reader as a persistent fallback. + #[allow(clippy::too_many_arguments)] + fn launch_demand_bound_parity<'a>( + &mut self, + stripe_index: usize, + max_new: usize, + sets: &mut FuturesUnordered>, + active: &mut [bool], + temporary_parity: &mut [bool], + attempted_parity: &mut [bool], + disposable: bool, + scheduled: &mut usize, + shard_size: usize, + data_shards: usize, + read_timeout: Duration, + metrics_path: Option<&'static str>, + ) -> usize + where + R: 'a, + { + let mut launched = 0; + for idx in self.data_shards..self.readers.len() { + if launched >= max_new || self.engaged[idx] || self.readers[idx].is_none() || active[idx] || attempted_parity[idx] { + continue; + } + + // Mark before invoking the factory/handle so a setup failure is + // also bounded to one attempt for this stripe. + attempted_parity[idx] = true; + + let reopener = self.deferred_reopeners.get(idx).and_then(Option::clone); + let (reader, temporary) = if disposable { + if let Some(reopener) = reopener { + let Some(reader) = reopener(stripe_index) else { + // A factory failure is terminal for this parity slot. + // Do not leave an apparently available reader that + // cannot be aligned to the current stripe. + self.readers[idx] = None; + continue; + }; + (reader, true) + } else { + // Tests and legacy callers without a factory retain the + // handle-based fallback. It is a persistent admission, + // because consuming that reserve is the only safe way to + // keep the stream aligned for the next stripe. + if !self.try_engage_parity(idx, stripe_index) { + continue; + } + let Some(reader) = self.readers[idx].take() else { + self.engaged[idx] = false; + continue; + }; + (reader, false) + } + } else if self.try_engage_parity(idx, stripe_index) { + let Some(reader) = self.readers[idx].take() else { + self.engaged[idx] = false; + continue; + }; + (reader, false) + } else if let Some(reopener) = reopener { + // A setup without a stripe handle can still make a known + // missing slot persistent by promoting the factory reader. + // This is a compatibility fallback; production CopySource + // setup supplies both a reserve and a handle. + let Some(reader) = reopener(stripe_index) else { + self.readers[idx] = None; + continue; + }; + // A non-disposable reopener is the persistent reserve when a + // setup did not retain a stripe handle. Mark it engaged just + // like the handle path so subsequent stripes reuse the + // aligned reader instead of reopening the remote shard. + self.engaged[idx] = true; + (reader, false) + } else { + continue; + }; + + let read_cost = self.read_costs.get(idx).copied().unwrap_or(ShardReadCost::Unknown); + if launch_owned_reader( + sets, + &mut self.buffers, + active, + scheduled, + idx, + reader, + shard_size, + data_shards, + read_cost, + read_timeout, + metrics_path, + ) { + temporary_parity[idx] = temporary; + launched += 1; + } else if !temporary { + self.engaged[idx] = false; + } + } + launched + } + fn try_engage_parity(&mut self, idx: usize, stripe_index: usize) -> bool { if stripe_index == 0 { self.engaged[idx] = true; @@ -1539,7 +2189,7 @@ impl Erasure { W: AsyncWrite + Send + Sync + Unpin, R: crate::erasure::coding::ShardSource, { - self.decode_inner(writer, readers, offset, length, total_length, None, Vec::new()) + self.decode_inner(writer, readers, offset, length, total_length, None, Vec::new(), Vec::new()) .await } @@ -1557,13 +2207,17 @@ impl Erasure { W: AsyncWrite + Send + Sync + Unpin, R: crate::erasure::coding::ShardSource, { - self.decode_inner(writer, readers, offset, length, total_length, Some(read_costs), Vec::new()) + self.decode_inner(writer, readers, offset, length, total_length, Some(read_costs), Vec::new(), Vec::new()) .await } /// GET decode entry point that also carries the deferred-parity stripe /// handles from bitrot reader setup, so unengaged parity readers can be /// opened aligned to the stripe where a data shard fails (backlog#923). + #[allow( + dead_code, + reason = "kept as the compatibility wrapper for existing decode callers and tests" + )] #[allow(clippy::too_many_arguments)] pub(crate) async fn decode_with_stripe_handles( &self, @@ -1579,8 +2233,49 @@ impl Erasure { W: AsyncWrite + Send + Sync + Unpin, R: crate::erasure::coding::ShardSource, { - self.decode_inner(writer, readers, offset, length, total_length, read_costs, deferred_handles) - .await + self.decode_with_stripe_handles_and_reopeners( + writer, + readers, + offset, + length, + total_length, + read_costs, + deferred_handles, + Vec::new(), + ) + .await + } + + /// Decode entry point with disposable, stripe-aligned parity reopeners. + /// CopySource uses these to hedge a slow data read without consuming the + /// unopened parity reserve when the data stream recovers first. + #[allow(clippy::too_many_arguments)] + pub(crate) async fn decode_with_stripe_handles_and_reopeners( + &self, + writer: &mut W, + readers: Vec>>, + offset: usize, + length: usize, + total_length: usize, + read_costs: Option>, + deferred_handles: Vec>, + deferred_reopeners: Vec>>, + ) -> (usize, Option) + where + W: AsyncWrite + Send + Sync + Unpin, + R: crate::erasure::coding::ShardSource, + { + self.decode_inner( + writer, + readers, + offset, + length, + total_length, + read_costs, + deferred_handles, + deferred_reopeners, + ) + .await } /// Reconstruct and emit one already-read stripe. @@ -1708,6 +2403,7 @@ impl Erasure { total_length: usize, read_costs: Option>, deferred_handles: Vec>, + deferred_reopeners: Vec>>, ) -> (usize, Option) where W: AsyncWrite + Send + Sync + Unpin, @@ -1756,7 +2452,8 @@ impl Erasure { } else { ParallelReader::new_for_decode(readers, self.clone(), offset, total_length, Some(GET_OBJECT_PATH_LEGACY_DUPLEX)) } - .with_deferred_parity_handles(deferred_handles); + .with_deferred_parity_handles(deferred_handles) + .with_deferred_parity_reopeners(deferred_reopeners); let start = offset / self.block_size; let end = end_offset.saturating_sub(1) / self.block_size; @@ -1992,12 +2689,6 @@ mod tests { #[test] fn parallel_reader_keeps_stripe_scratch_out_of_line() { - eprintln!( - "parallel_reader={} stripe_state={} cached_state={}", - std::mem::size_of::>>>(), - std::mem::size_of::(), - std::mem::size_of::>>() - ); assert_eq!( std::mem::size_of::>>(), std::mem::size_of::(), @@ -2156,11 +2847,16 @@ mod tests { sleep: Option>>, }, Pending, + /// Parks without self-waking so a surrounding timer can make a + /// deterministic cancellation decision (unlike `Pending`, which is + /// intentionally a busy-waking fixture for timeout tests). + Parked, PartialThenPending { data: Vec, emitted: bool, }, TimedOut, + TerminalFileNotFound, /// Serves `cursor` (typically the first stripe's bytes) normally, then /// once it is exhausted parks on a long `sleep` instead of returning EOF — /// modelling a shard whose *next*-stripe read never completes (a wedged or @@ -2192,6 +2888,7 @@ mod tests { cx.waker().wake_by_ref(); Poll::Pending } + TestShardReader::Parked => Poll::Pending, TestShardReader::PartialThenPending { data, emitted } => { if *emitted { cx.waker().wake_by_ref(); @@ -2204,6 +2901,9 @@ mod tests { Poll::Ready(Ok(())) } TestShardReader::TimedOut => Poll::Ready(Err(io::Error::new(ErrorKind::TimedOut, "test shard read timed out"))), + TestShardReader::TerminalFileNotFound => { + Poll::Ready(Err(crate::disk::error::terminal_read_error_to_io(Error::FileNotFound))) + } TestShardReader::PrefixThenSlow { cursor, stall, sleep } => { let before = buf.filled().len(); match Pin::new(cursor).poll_read(cx, buf) { @@ -3110,6 +3810,36 @@ mod tests { }); } + /// A copy source must remain demand-bound even when an operator has opted + /// into the ordinary GET overlap switches. The policy also enables the + /// deferred-parity lockstep mode so healthy copies do not consume parity + /// streams until reconstruction needs them. + #[tokio::test] + #[serial_test::serial] + async fn demand_bound_policy_disables_stripe_read_ahead_and_uses_data_only_lockstep() { + temp_env::async_with_vars( + [ + (ENV_RUSTFS_GET_DECODE_STRIPE_PREFETCH_COUNT, Some("8")), + (ENV_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, Some("false")), + ], + async { + assert!(legacy_stripe_prefetch_enabled()); + assert!(!get_lockstep_data_shards_only_enabled()); + + with_decode_read_policy(DecodeReadPolicy::DemandBound, async { + assert!(!legacy_stripe_prefetch_enabled()); + assert!(get_lockstep_data_shards_only_enabled()); + }) + .await; + + assert!(legacy_stripe_prefetch_enabled()); + assert!(!get_lockstep_data_shards_only_enabled()); + }, + ) + .await; + } + /// Cancel-safety (https://github.com/rustfs/backlog/issues/1310): when the /// current stripe's emit fails (client disconnect / broken pipe), the /// speculatively prefetched next-stripe read must be *cancelled*, not waited @@ -4174,6 +4904,342 @@ mod tests { assert_eq!(DATA_SHARDS + 1, bufs.iter().filter(|buf| buf.is_some()).count()); } + /// Demand-bound lockstep regression: a slow data shard must be hedged as + /// soon as deferred parity can provide the decode-plus-verification quorum. + /// Before this guard, the hedge timer only looked at already-completed + /// readers, so a 2+2 stripe with one ready data shard waited out the full + /// read timeout even though both parity readers were available to engage. + #[tokio::test] + async fn test_demand_bound_lockstep_hedges_to_deferred_parity_quorum() { + const NUM_SHARDS: usize = 1; + const BLOCK_SIZE: usize = 64; + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const SHARD_SIZE: usize = BLOCK_SIZE / DATA_SHARDS; + + let hash_algo = HashAlgorithm::None; + let slow_until = TokioInstant::now() + Duration::from_secs(60); + let readers = vec![ + Some(BitrotReader::new( + TestShardReader::ReadyAt { + cursor: Cursor::new(vec![0_u8; SHARD_SIZE * NUM_SHARDS]), + ready_at: slow_until, + sleep: None, + }, + SHARD_SIZE, + hash_algo.clone(), + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![1_u8; SHARD_SIZE * NUM_SHARDS])), + SHARD_SIZE, + hash_algo.clone(), + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![2_u8; SHARD_SIZE * NUM_SHARDS])), + SHARD_SIZE, + hash_algo.clone(), + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![3_u8; SHARD_SIZE * NUM_SHARDS])), + SHARD_SIZE, + hash_algo, + false, + )), + ]; + + let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); + let (bufs, errs, engaged, readers_remaining) = with_decode_read_policy(DecodeReadPolicy::DemandBound, async { + let mut parallel_reader = ParallelReader::new_with_metrics_path_read_costs_timeout_and_reconstruction_verification( + readers, + erasure, + 0, + NUM_SHARDS * BLOCK_SIZE, + None, + vec![ShardReadCost::Unknown; DATA_SHARDS + PARITY_SHARDS], + Duration::from_secs(60), + true, + ); + let (bufs, errs) = tokio::time::timeout(Duration::from_secs(2), parallel_reader.read()) + .await + .expect("deferred parity must cover a hedged data shard without waiting for read_timeout"); + ( + bufs, + errs, + parallel_reader.engaged.clone(), + parallel_reader.readers.iter().map(Option::is_some).collect::>(), + ) + }) + .await; + + assert!(matches!(&errs[0], Some(DiskError::Io(err)) if err.kind() == ErrorKind::TimedOut)); + assert_eq!(bufs.iter().filter(|buf| buf.is_some()).count(), DATA_SHARDS + 1); + assert_eq!(engaged.as_slice(), &[true, true, true, true]); + assert_eq!(readers_remaining, vec![false, true, true, true]); + } + + /// A fast data failure must admit deferred parity immediately. There is + /// intentionally no hedge timer when `read_timeout == 0`, so relying on + /// the timer would drain the initial futures and return a false quorum + /// before the healthy parity readers are ever opened. + #[tokio::test] + async fn test_demand_bound_admits_parity_after_fast_data_failure_without_timer() { + const NUM_SHARDS: usize = 1; + const BLOCK_SIZE: usize = 64; + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const SHARD_SIZE: usize = BLOCK_SIZE / DATA_SHARDS; + + let hash_algo = HashAlgorithm::None; + let readers = vec![ + Some(BitrotReader::new(TestShardReader::TimedOut, SHARD_SIZE, hash_algo.clone(), false)), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![1_u8; SHARD_SIZE * NUM_SHARDS])), + SHARD_SIZE, + hash_algo.clone(), + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![2_u8; SHARD_SIZE * NUM_SHARDS])), + SHARD_SIZE, + hash_algo.clone(), + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![3_u8; SHARD_SIZE * NUM_SHARDS])), + SHARD_SIZE, + hash_algo, + false, + )), + ]; + + let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); + let (bufs, errs, engaged, readers_remaining) = with_decode_read_policy(DecodeReadPolicy::DemandBound, async { + let mut parallel_reader = ParallelReader::new_with_metrics_path_read_timeout_and_reconstruction_verification( + readers, + erasure, + 0, + NUM_SHARDS * BLOCK_SIZE, + None, + Duration::ZERO, + true, + ); + let (bufs, errs) = tokio::time::timeout(Duration::from_millis(500), parallel_reader.read()) + .await + .expect("fast data failure must immediately race healthy parity without a hedge timer"); + ( + bufs, + errs, + parallel_reader.engaged.clone(), + parallel_reader.readers.iter().map(Option::is_some).collect::>(), + ) + }) + .await; + + assert!(matches!(&errs[0], Some(DiskError::Io(err)) if err.kind() == ErrorKind::TimedOut)); + assert_eq!(bufs.iter().filter(|buf| buf.is_some()).count(), DATA_SHARDS + 1); + assert_eq!(engaged.as_slice(), &[true, true, true, true]); + assert_eq!(readers_remaining, vec![false, true, true, true]); + } + + #[tokio::test] + async fn test_demand_bound_canceled_hedge_preserves_deferred_parity_for_next_stripe() { + const BLOCK_SIZE: usize = 64; + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const SHARD_SIZE: usize = BLOCK_SIZE / DATA_SHARDS; + + let hash_algo = HashAlgorithm::None; + let parity_calls = Arc::new(AtomicUsize::new(0)); + let mut reopeners: Vec>> = vec![None; DATA_SHARDS + PARITY_SHARDS]; + for (idx, slot) in reopeners.iter_mut().enumerate().skip(DATA_SHARDS).take(PARITY_SHARDS) { + let parity_calls = Arc::clone(&parity_calls); + *slot = Some(Arc::new(move |_stripe_index| { + let call = parity_calls.fetch_add(1, Ordering::SeqCst); + let reader = if call == 0 { + // One hedge succeeds before the slow data reader returns; + // it must still remain disposable rather than being + // promoted into the next stripe. + TestShardReader::Ready(Cursor::new(vec![idx as u8; SHARD_SIZE])) + } else if call < PARITY_SHARDS { + TestShardReader::Parked + } else { + TestShardReader::Ready(Cursor::new(vec![idx as u8; SHARD_SIZE * 2])) + }; + Some(BitrotReader::new(reader, SHARD_SIZE, HashAlgorithm::None, false)) + })); + } + + let slow_until = TokioInstant::now() + Duration::from_millis(150); + let readers = vec![ + Some(BitrotReader::new( + TestShardReader::ReadyAt { + cursor: Cursor::new(vec![0_u8; SHARD_SIZE * 4]), + ready_at: slow_until, + sleep: None, + }, + SHARD_SIZE, + hash_algo.clone(), + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![1_u8; SHARD_SIZE * 4])), + SHARD_SIZE, + hash_algo.clone(), + false, + )), + Some(BitrotReader::new(TestShardReader::Pending, SHARD_SIZE, hash_algo.clone(), false)), + Some(BitrotReader::new(TestShardReader::Pending, SHARD_SIZE, hash_algo, false)), + ]; + + let (first_parity_reserved, second_result) = with_decode_read_policy(DecodeReadPolicy::DemandBound, async { + let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); + let mut parallel_reader = ParallelReader::new_with_metrics_path_read_timeout_and_reconstruction_verification( + readers, + erasure, + 0, + BLOCK_SIZE * 4, + None, + Duration::from_secs(1), + true, + ) + .with_deferred_parity_reopeners(reopeners); + + let (first_buffers, first_errors) = tokio::time::timeout(Duration::from_secs(1), parallel_reader.read()) + .await + .expect("healthy data recovery must not wait for canceled parity hedges"); + assert_eq!(first_buffers.iter().filter(|buffer| buffer.is_some()).count(), DATA_SHARDS + 1); + assert!(first_errors.iter().take(DATA_SHARDS).all(Option::is_none)); + assert_eq!(parity_calls.load(Ordering::SeqCst), PARITY_SHARDS); + assert!(parallel_reader.readers[2].is_some()); + assert!(parallel_reader.readers[3].is_some()); + assert!(!parallel_reader.engaged[2]); + assert!(!parallel_reader.engaged[3]); + + // A healthy following stripe must not inherit the one-stripe hedge + // reader. If a successful temporary hedge were promoted, this + // read would schedule parity again and violate demand-bound + // read-ahead. + let (healthy_buffers, healthy_errors) = tokio::time::timeout(Duration::from_secs(1), parallel_reader.read()) + .await + .expect("a healthy following stripe must complete without parity fan-out"); + assert!(healthy_errors.iter().take(DATA_SHARDS).all(Option::is_none)); + assert_eq!(healthy_buffers.iter().filter(|buffer| buffer.is_some()).count(), DATA_SHARDS); + assert_eq!(parity_calls.load(Ordering::SeqCst), PARITY_SHARDS); + assert!(!parallel_reader.engaged[2]); + assert!(!parallel_reader.engaged[3]); + + // A later stripe loses data shard 0. The deferred reserves must be + // available again; the second pair of factory calls returns the + // aligned parity bytes for this stripe. + parallel_reader.readers[0] = + Some(BitrotReader::new(TestShardReader::TimedOut, SHARD_SIZE, HashAlgorithm::None, false)); + let (second_buffers, second_errors) = tokio::time::timeout(Duration::from_secs(1), parallel_reader.read()) + .await + .expect("the next degraded stripe must reuse the preserved parity reserve"); + assert!(matches!(&second_errors[0], Some(DiskError::Io(error)) if error.kind() == ErrorKind::TimedOut)); + assert_eq!(second_buffers.iter().filter(|buffer| buffer.is_some()).count(), DATA_SHARDS + 1); + + // Once the parity readers have been persistently engaged, another + // degraded stripe reuses their aligned streams; no new factory + // calls (and therefore no new remote opens) are permitted. + let (third_buffers, third_errors) = tokio::time::timeout(Duration::from_secs(1), parallel_reader.read()) + .await + .expect("persistent parity readers must cover a subsequent degraded stripe"); + assert!(third_errors[0].is_none(), "an already-retired data slot has no new read error"); + assert_eq!(third_buffers.iter().filter(|buffer| buffer.is_some()).count(), DATA_SHARDS + 1); + assert_eq!(parity_calls.load(Ordering::SeqCst), PARITY_SHARDS * 2); + ( + parallel_reader.readers[2].is_some() && parallel_reader.readers[3].is_some(), + (third_buffers, third_errors), + ) + }) + .await; + + assert!(first_parity_reserved); + assert_eq!(parity_calls.load(Ordering::SeqCst), PARITY_SHARDS * 2); + // `second_result` carries the third (persistently degraded) stripe; + // the data slot was already retired by the preceding stripe, so it + // must not emit a fresh timeout or trigger another remote open. + let (buffers, errors) = second_result; + assert!(errors[0].is_none()); + assert_eq!(buffers.iter().filter(|buffer| buffer.is_some()).count(), DATA_SHARDS + 1); + } + + /// A disposable parity hedge is advisory. Its terminal error must not be + /// copied into the stripe error vector, because the emitter treats + /// FileNotFound/FileCorrupt there as an object-level failure even after a + /// healthy data reader has recovered and supplied a complete stripe. + #[tokio::test] + async fn test_demand_bound_disposable_parity_error_does_not_poison_recovered_stripe() { + const BLOCK_SIZE: usize = 64; + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const SHARD_SIZE: usize = BLOCK_SIZE / DATA_SHARDS; + + let parity_calls = Arc::new(AtomicUsize::new(0)); + let mut reopeners: Vec>> = vec![None; DATA_SHARDS + PARITY_SHARDS]; + for (idx, slot) in reopeners.iter_mut().enumerate().skip(DATA_SHARDS).take(PARITY_SHARDS) { + let parity_calls = Arc::clone(&parity_calls); + *slot = Some(Arc::new(move |_stripe_index| { + let call = parity_calls.fetch_add(1, Ordering::SeqCst); + let reader = if call == 0 { + TestShardReader::TerminalFileNotFound + } else { + TestShardReader::Ready(Cursor::new(vec![idx as u8; SHARD_SIZE])) + }; + Some(BitrotReader::new(reader, SHARD_SIZE, HashAlgorithm::None, false)) + })); + } + + let slow_until = TokioInstant::now() + Duration::from_millis(150); + let readers = vec![ + Some(BitrotReader::new( + TestShardReader::ReadyAt { + cursor: Cursor::new(vec![0_u8; SHARD_SIZE]), + ready_at: slow_until, + sleep: None, + }, + SHARD_SIZE, + HashAlgorithm::None, + false, + )), + Some(BitrotReader::new( + TestShardReader::Ready(Cursor::new(vec![1_u8; SHARD_SIZE])), + SHARD_SIZE, + HashAlgorithm::None, + false, + )), + Some(BitrotReader::new(TestShardReader::Pending, SHARD_SIZE, HashAlgorithm::None, false)), + Some(BitrotReader::new(TestShardReader::Pending, SHARD_SIZE, HashAlgorithm::None, false)), + ]; + + let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE); + let mut output = Vec::new(); + let (written, error) = with_decode_read_policy(DecodeReadPolicy::DemandBound, async { + erasure + .decode_with_stripe_handles_and_reopeners( + &mut output, + readers, + 0, + BLOCK_SIZE, + BLOCK_SIZE, + None, + Vec::new(), + reopeners, + ) + .await + }) + .await; + + assert_eq!(parity_calls.load(Ordering::SeqCst), PARITY_SHARDS); + assert_eq!(written, BLOCK_SIZE); + assert_eq!(output.len(), BLOCK_SIZE); + assert!(error.is_none(), "a failed disposable hedge must not fail a recovered stripe: {error:?}"); + } + /// Lockstep verification-quorum regression (backlog#1156). When a data shard is /// missing, the hedge must settle only at `data_shards + 1` (decode quorum plus /// a reconstruction-verification source), never at exactly `data_shards` — that diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index 88a3da910..371a93560 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -346,16 +346,11 @@ impl AsyncRead for DeferredObjectReader { } fn disk_error_to_io_error(err: DiskError) -> io::Error { - let kind = match err { - DiskError::Timeout | DiskError::SourceStalled => io::ErrorKind::TimedOut, - DiskError::DiskNotFound | DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::PathNotFound => { - io::ErrorKind::NotFound - } - DiskError::FileCorrupt | DiskError::PartMissingOrCorrupt | DiskError::BitrotHashAlgoInvalid => io::ErrorKind::InvalidData, - DiskError::Io(io_err) => return io_err, - _ => io::ErrorKind::Other, - }; - io::Error::new(kind, err.to_string()) + // Keep the typed disk error attached to deferred-reader failures. The + // decoder uses the marker to retire a stream that can no longer be + // realigned, while quorum reduction still sees Timeout/NotFound instead + // of an opaque `DiskError::Io` wrapper. + crate::disk::error::terminal_read_error_to_io(err) } async fn open_disk_reader( diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 6d2c6cf6c..a05884f72 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -1517,6 +1517,7 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter( } pub(in crate::set_disk) type ObjectBitrotReader = BitrotReader; +pub(in crate::set_disk) type DeferredReaderReopener = crate::erasure::coding::decode::DeferredReaderReopener; pub(in crate::set_disk) type BitrotReaderTask<'a> = Pin, DiskError>)> + Send + 'a>>; @@ -1533,6 +1534,10 @@ pub(in crate::set_disk) struct BitrotReaderSetup { /// readers. The lockstep GET decode uses them to open a parity shard /// aligned to the stripe where a data shard failed (backlog#923). pub(in crate::set_disk) deferred_stripe_handles: Vec>, + /// Factories for a fresh, stripe-aligned parity reader. CopySource hedges + /// use these disposable readers so an abandoned hedge leaves the original + /// deferred reserve untouched. + pub(in crate::set_disk) deferred_reopeners: Vec>, pub(in crate::set_disk) errors: Vec>, pub(in crate::set_disk) scheduled: Vec, pub(in crate::set_disk) attempted: Vec, @@ -1595,6 +1600,16 @@ pub(in crate::set_disk) fn get_bitrot_reader_setup_strategy( mode: BitrotReaderSetupMode, prefer_data_blocks_first: bool, ) -> BitrotReaderSetupStrategy { + // CopyObject holds the source reader behind a backpressured destination. + // Keep its setup demand-bound even when an operator has retained the + // legacy all-shards environment setting for ordinary GETs. + if matches!( + crate::set_disk::get_object_read_policy(), + crate::set_disk::GetObjectReadPolicy::CopySource + ) { + return BitrotReaderSetupStrategy::DataBlocksFirst; + } + match mode { BitrotReaderSetupMode::ReadQuorum if prefer_data_blocks_first @@ -1620,6 +1635,7 @@ impl BitrotReaderSetup { Self { readers: (0..shards).map(|_| None).collect(), deferred_stripe_handles: (0..shards).map(|_| None).collect(), + deferred_reopeners: (0..shards).map(|_| None).collect(), errors: vec![Some(DiskError::DiskNotFound); shards], scheduled: vec![false; shards], attempted: vec![false; shards], @@ -1814,6 +1830,41 @@ pub(in crate::set_disk) fn next_unscheduled_reader_index( .find(|idx| !setup.scheduled[*idx]) } +/// Build a cloneable opener for an unopened deferred shard. The returned +/// reader is aligned to the requested stripe before its first poll, while the +/// source reader created during setup remains untouched as a reserve. +#[allow(clippy::too_many_arguments)] +fn deferred_reader_reopener( + inline_data: Option, + disk: Option, + bucket: &str, + path: &str, + read_offset: usize, + read_length: usize, + shard_size: usize, + checksum_algo: HashAlgorithm, + skip_verify_bitrot: bool, + use_mmap_read: bool, +) -> DeferredReaderReopener { + let bucket = bucket.to_owned(); + let path = path.to_owned(); + Arc::new(move |stripe_index| { + let (reader, handle) = create_deferred_bitrot_reader_with_stripe_handle( + inline_data.clone(), + disk.clone(), + &bucket, + &path, + read_offset, + read_length, + shard_size, + checksum_algo.clone(), + skip_verify_bitrot, + use_mmap_read, + ); + handle.advance_stripes(stripe_index).then_some(reader) + }) +} + #[allow(clippy::too_many_arguments)] pub(in crate::set_disk) fn fill_deferred_bitrot_readers( setup: &mut BitrotReaderSetup, @@ -1836,6 +1887,15 @@ pub(in crate::set_disk) fn fill_deferred_bitrot_readers( return; } + // Only CopySource uses disposable, stripe-aligned reopeners. Ordinary GET + // readers use the existing deferred handle and should not retain one + // heap-allocated closure (plus cloned path/disk state) for every parity + // slot. + let copy_source_demand_bound = matches!( + crate::set_disk::get_object_read_policy(), + crate::set_disk::GetObjectReadPolicy::CopySource + ); + for idx in 0..disks.len() { if setup.attempted[idx] { continue; @@ -1849,6 +1909,20 @@ pub(in crate::set_disk) fn fill_deferred_bitrot_readers( let disk = disks[idx].clone(); let data_dir = files[idx].data_dir.unwrap_or_default(); let path = format!("{object}/{data_dir}/part.{part_number}"); + let reopener = copy_source_demand_bound.then(|| { + deferred_reader_reopener( + inline_data.clone(), + disk.clone(), + bucket, + &path, + read_offset, + read_length, + shard_size, + checksum_algo.clone(), + skip_verify_bitrot, + use_mmap_read, + ) + }); let (reader, stripe_handle) = create_deferred_bitrot_reader_with_stripe_handle( inline_data, disk, @@ -1862,6 +1936,7 @@ pub(in crate::set_disk) fn fill_deferred_bitrot_readers( use_mmap_read, ); setup.retain_deferred_reader(idx, reader, stripe_handle); + setup.deferred_reopeners[idx] = reopener; } // With the data-shards-only lockstep gate on (backlog#923), the GET decode @@ -1887,6 +1962,20 @@ pub(in crate::set_disk) fn fill_deferred_bitrot_readers( let disk = disks[idx].clone(); let data_dir = files[idx].data_dir.unwrap_or_default(); let path = format!("{object}/{data_dir}/part.{part_number}"); + let reopener = copy_source_demand_bound.then(|| { + deferred_reader_reopener( + inline_data.clone(), + disk.clone(), + bucket, + &path, + read_offset, + read_length, + shard_size, + checksum_algo.clone(), + skip_verify_bitrot, + use_mmap_read, + ) + }); let (reader, stripe_handle) = create_deferred_bitrot_reader_with_stripe_handle( inline_data, disk, @@ -1901,6 +1990,7 @@ pub(in crate::set_disk) fn fill_deferred_bitrot_readers( ); setup.readers[idx] = Some(reader); setup.deferred_stripe_handles[idx] = Some(stripe_handle); + setup.deferred_reopeners[idx] = reopener; } } @@ -2210,6 +2300,10 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_with_prefere let strategy = get_bitrot_reader_setup_strategy(mode, prefer_data_blocks_first); if use_mmap_read + && !matches!( + crate::set_disk::get_object_read_policy(), + crate::set_disk::GetObjectReadPolicy::CopySource + ) && let Some(mut setup) = try_create_bitrot_readers_via_batch_pread( files, disks, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 0d0f1fb57..a63348116 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -792,6 +792,65 @@ const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX: &str = "RUSTFS_GET_M const ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH: &str = "RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH"; const DEFAULT_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH: bool = true; +/// Identifies the caller's read contract for policies that are deliberately +/// narrower than the storage API's ordinary GET contract. +/// +/// Server-side copy consumes a source reader while a destination writer is +/// applying backpressure. Its source read must not speculatively open the +/// next multipart part: those extra shard streams can share an internode H2 +/// connection with the current part and starve the lockstep decoder. Keep +/// this context internal so the public `ObjectOptions` and storage traits do +/// not acquire a copy-only field. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(crate) enum GetObjectReadPolicy { + #[default] + Default, + CopySource, +} + +impl GetObjectReadPolicy { + pub(crate) const fn allows_multipart_setup_prefetch(self) -> bool { + matches!(self, Self::Default) + } +} + +tokio::task_local! { + static GET_OBJECT_READ_POLICY: GetObjectReadPolicy; + static GET_OBJECT_READ_CANCELLATION: tokio_util::sync::CancellationToken; +} + +pub(crate) fn get_object_read_policy() -> GetObjectReadPolicy { + GET_OBJECT_READ_POLICY.try_with(|policy| *policy).unwrap_or_default() +} + +pub(crate) async fn with_get_object_read_policy(policy: GetObjectReadPolicy, future: F) -> F::Output +where + F: std::future::Future, +{ + let decode_policy = match policy { + GetObjectReadPolicy::Default => crate::erasure::coding::decode::DecodeReadPolicy::Default, + GetObjectReadPolicy::CopySource => crate::erasure::coding::decode::DecodeReadPolicy::DemandBound, + }; + crate::erasure::coding::decode::with_decode_read_policy(decode_policy, GET_OBJECT_READ_POLICY.scope(policy, future)).await +} + +/// Return the request-owned cancellation token for a copy source, when one is +/// installed. The token is read before the detached legacy producer is spawned; +/// Tokio task-local values do not cross that spawn boundary on their own. +pub(crate) fn get_object_read_cancellation() -> Option { + GET_OBJECT_READ_CANCELLATION.try_with(|token| token.clone()).ok() +} + +pub(crate) async fn with_get_object_read_cancellation( + cancellation: tokio_util::sync::CancellationToken, + future: F, +) -> F::Output +where + F: std::future::Future, +{ + GET_OBJECT_READ_CANCELLATION.scope(cancellation, future).await +} + static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); mod core; @@ -2296,6 +2355,7 @@ enum GetCodecStreamingFallbackReason { InvalidMinSize, ReadQuorumNotSafe, MultipartPartLimit, + CopySourceDemandBound, } impl GetCodecStreamingFallbackReason { @@ -2317,6 +2377,7 @@ impl GetCodecStreamingFallbackReason { Self::InvalidMinSize => "invalid_min_size", Self::ReadQuorumNotSafe => "read_quorum_not_safe", Self::MultipartPartLimit => "multipart_part_limit", + Self::CopySourceDemandBound => "copy_source_demand_bound", } } } @@ -2623,6 +2684,17 @@ fn get_codec_streaming_reader_gate( prefer_data_blocks_first_reader_setup: false, }; } + if matches!(get_object_read_policy(), GetObjectReadPolicy::CopySource) { + // The codec reader has its own bounded fill worker. It may still + // request an additional stripe for a plain single-part object even + // when multipart setup prefetch is disabled, so copy sources use the + // legacy demand-bound reader for every object class. + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::CopySourceDemandBound), + prefer_data_blocks_first_reader_setup: false, + }; + } if !config.rollout.is_opted_in() { return GetCodecStreamingGate { object_class, @@ -5912,6 +5984,29 @@ mod tests { use tokio::fs; use tokio::io::AsyncReadExt; + #[tokio::test] + async fn copy_source_read_policy_is_scoped_and_demand_bound() { + assert_eq!(get_object_read_policy(), GetObjectReadPolicy::Default); + assert!(GetObjectReadPolicy::Default.allows_multipart_setup_prefetch()); + assert!(!GetObjectReadPolicy::CopySource.allows_multipart_setup_prefetch()); + + with_get_object_read_policy(GetObjectReadPolicy::CopySource, async { + assert_eq!(get_object_read_policy(), GetObjectReadPolicy::CopySource); + assert!(!get_object_read_policy().allows_multipart_setup_prefetch()); + assert_eq!( + crate::erasure::coding::decode::decode_read_policy(), + crate::erasure::coding::decode::DecodeReadPolicy::DemandBound + ); + }) + .await; + + assert_eq!(get_object_read_policy(), GetObjectReadPolicy::Default); + assert_eq!( + crate::erasure::coding::decode::decode_read_policy(), + crate::erasure::coding::decode::DecodeReadPolicy::Default + ); + } + #[test] fn complete_part_error_maps_confirmed_missing_to_invalid_part() { for err in ["file not found", "Specified part could not be found", "part.7 not found"] { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 8f550be82..aa75a27e7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -61,6 +61,8 @@ use http::HeaderValue; use rustfs_utils::path::decode_dir_object; use std::future::Future; use std::sync::OnceLock; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, ReadBuf}; use tokio_util::sync::CancellationToken; const OLD_DATA_CLEANUP_RECEIPT_FILE: &str = ".rustfs-old-data-cleanup-receipt.json"; @@ -1120,6 +1122,37 @@ where Ok((reader, offset, length)) } +/// Cancels a detached legacy GET producer when its consumer is dropped. +/// +/// The producer owns the shard readers and the object read lock, while the +/// consumer owns only the duplex read half. Closing that half eventually +/// unblocks a writer, but can leave a producer stuck in reader setup or remote +/// recovery until a lower-level timeout fires. This small boundary wrapper +/// provides an explicit cancellation signal without changing the public +/// `GetObjectReader` shape. +struct ProducerCancellationReader { + inner: R, + cancellation: CancellationToken, +} + +impl ProducerCancellationReader { + fn new(inner: R, cancellation: CancellationToken) -> Self { + Self { inner, cancellation } + } +} + +impl AsyncRead for ProducerCancellationReader { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl Drop for ProducerCancellationReader { + fn drop(&mut self) { + self.cancellation.cancel(); + } +} + fn data_read_metadata_early_stop_request_shape_allowed(range: &Option, opts: &ObjectOptions) -> bool { range.is_none() && opts.part_number.is_none() @@ -1907,12 +1940,25 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // lookup on the streaming miss path (ODC-16). reader.body_source = body_source; + // The producer is otherwise detached from the returned reader. Tie its + // lifetime to the source stream so a cancelled copy (or an abandoned + // GET) releases in-flight shard opens, response bodies, and the read + // lock immediately instead of waiting for a disk timeout. + let producer_cancellation = crate::set_disk::get_object_read_cancellation(); + if let Some(cancellation) = producer_cancellation.as_ref() { + reader.stream = Box::new(ProducerCancellationReader::new(reader.stream, cancellation.clone())); + } + // let disks = disks.clone(); let bucket = bucket.to_owned(); let object = object.to_owned(); let set_index = self.set_index; let pool_index = self.pool_index; let skip_verify = opts.skip_verify_bitrot; + // The producer runs in a separate Tokio task, so carry the caller's + // read policy across the task boundary explicitly. Tokio task-local + // values are not inherited by spawned tasks. + let read_policy = crate::set_disk::get_object_read_policy(); let erasure_cache = Arc::clone(&self.erasure_cache); let (fi, files, disks) = snapshot.into_owned(); tokio::spawn(async move { @@ -1922,26 +1968,40 @@ 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( - &bucket, - &object, - erasure_cache, - offset, - length, - &mut writer, - fi, - files, - &disks, - set_index, - pool_index, - skip_verify, - false, - GET_OBJECT_PATH_LEGACY_DUPLEX, - object_class.as_str(), - size_bucket, - ) - .await; - if let Err(e) = &producer_result { + let producer_result = tokio::select! { + biased; + result = crate::set_disk::with_get_object_read_policy( + read_policy, + Self::get_object_with_fileinfo( + &bucket, + &object, + erasure_cache, + offset, + length, + &mut writer, + fi, + files, + &disks, + set_index, + pool_index, + skip_verify, + false, + GET_OBJECT_PATH_LEGACY_DUPLEX, + object_class.as_str(), + size_bucket, + ), + ) => result, + _ = async { + if let Some(cancellation) = producer_cancellation.as_ref() { + cancellation.cancelled().await; + } else { + std::future::pending::<()>().await; + } + } => Err(Error::OperationCanceled), + }; + if let Err(e) = &producer_result + && !matches!(e, Error::OperationCanceled) + { let reason = classify_storage_error(e); if reason == GetObjectFailureReason::DownstreamClosed { debug!( @@ -3794,6 +3854,28 @@ mod legacy_duplex_producer_reader_tests { assert_eq!(out, b"complete"); } + #[tokio::test] + async fn producer_cancellation_reader_cancels_pending_producer_on_drop() { + let cancellation = CancellationToken::new(); + let producer_cancellation = cancellation.clone(); + let producer = tokio::spawn(async move { + tokio::select! { + _ = producer_cancellation.cancelled() => true, + _ = std::future::pending::<()>() => false, + } + }); + + let reader = ProducerCancellationReader::new(tokio::io::empty(), cancellation); + drop(reader); + + assert!( + tokio::time::timeout(std::time::Duration::from_secs(1), producer) + .await + .expect("dropping the consumer should cancel the producer promptly") + .expect("producer task should not panic") + ); + } + #[tokio::test] async fn legacy_duplex_reader_ignores_zero_capacity_read_buf() { let (mut writer, reader) = tokio::io::duplex(64); diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 569ba57c0..ce3f47016 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -792,7 +792,7 @@ impl SetDisks { let use_mmap_read = object_mmap_read_enabled(); let files = Arc::new(files); let disks = Arc::new(disks); - let prefetch_enabled = is_multipart_reader_setup_prefetch_enabled(); + let prefetch_enabled = multipart_reader_setup_prefetch_enabled(get_object_read_policy()); let mut prefetched: Option<(usize, PrefetchedReaderSetup)> = None; let mut total_read = 0; @@ -1065,8 +1065,9 @@ impl SetDisks { let unattempted_data_shards = !reader_setup.data_shards_attempted(erasure.data_shards); let readers = reader_setup.readers; let deferred_stripe_handles = reader_setup.deferred_stripe_handles; + let deferred_reopeners = reader_setup.deferred_reopeners; let (written, err) = erasure - .decode_with_stripe_handles( + .decode_with_stripe_handles_and_reopeners( writer, readers, part_offset, @@ -1074,6 +1075,7 @@ impl SetDisks { part_size, read_costs, deferred_stripe_handles, + deferred_reopeners, ) .await; let decode_elapsed = decode_stage_start.elapsed(); @@ -1476,6 +1478,7 @@ impl SetDisks { erasure.clone(), reader_setup.readers, reader_setup.deferred_stripe_handles, + reader_setup.deferred_reopeners, read_costs, part_offset, part_length, @@ -1488,6 +1491,7 @@ impl SetDisks { let readers = reader_setup.readers; let deferred_stripe_handles = reader_setup.deferred_stripe_handles; + let deferred_reopeners = reader_setup.deferred_reopeners; let source = if let Some(read_costs) = read_costs { coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification( readers, @@ -1506,7 +1510,8 @@ impl SetDisks { Some(metrics_path), ) } - .with_deferred_parity_handles(deferred_stripe_handles); + .with_deferred_parity_handles(deferred_stripe_handles) + .with_deferred_parity_reopeners(deferred_reopeners); let engine = build_get_codec_streaming_decode_engine(erasure.clone())?; let reader = coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?; @@ -1535,6 +1540,10 @@ fn multipart_part_checksum_algo(fi: &FileInfo, part_number: usize) -> HashAlgori } } +fn multipart_reader_setup_prefetch_enabled(policy: GetObjectReadPolicy) -> bool { + policy.allows_multipart_setup_prefetch() && is_multipart_reader_setup_prefetch_enabled() +} + /// Run one part's bitrot reader setup and measure its wall-clock duration. /// /// Shared by the synchronous path and the prefetch task in @@ -1762,10 +1771,12 @@ impl Drop for LazyMultipartCodecStreamingReader { /// background task drives the decode into the write half while the returned /// reader drains the read half. No extra file descriptors are opened — the /// readers are moved in from the setup that just ran. +#[allow(clippy::too_many_arguments)] fn build_legacy_per_part_fallback_reader( erasure: coding::Erasure, readers: Vec>, deferred_stripe_handles: Vec>, + deferred_reopeners: Vec>, read_costs: Option>, part_offset: usize, part_length: usize, @@ -1775,7 +1786,7 @@ fn build_legacy_per_part_fallback_reader( let (read_half, mut write_half) = tokio::io::duplex(buffer); let decode = tokio::spawn(async move { let (_written, err) = erasure - .decode_with_stripe_handles( + .decode_with_stripe_handles_and_reopeners( &mut write_half, readers, part_offset, @@ -1783,6 +1794,7 @@ fn build_legacy_per_part_fallback_reader( part_size, read_costs, deferred_stripe_handles, + deferred_reopeners, ) .await; // Dropping `write_half` on return signals EOF to the reader half. @@ -3236,6 +3248,7 @@ mod metadata_cache_tests { mod tests { use super::*; use crate::erasure::coding::BitrotWriter; + use serial_test::serial; use std::io::{Cursor, ErrorKind, IoSlice}; use std::sync::{ Arc, @@ -3246,6 +3259,15 @@ mod tests { const CODEC_STREAMING_TEST_BUCKET: &str = "bucket"; const CODEC_STREAMING_TEST_OBJECT: &str = "object"; + #[test] + #[serial] + fn multipart_reader_setup_prefetch_is_disabled_only_for_copy_sources() { + temp_env::with_var(ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH, Some("true"), || { + assert!(multipart_reader_setup_prefetch_enabled(GetObjectReadPolicy::Default)); + assert!(!multipart_reader_setup_prefetch_enabled(GetObjectReadPolicy::CopySource)); + }); + } + #[tokio::test] async fn downstream_writer_marks_closed_duplex_reader_as_downstream_close() { let (reader, inner) = tokio::io::duplex(64); @@ -5481,6 +5503,7 @@ mod tests { erasure, setup.readers, setup.deferred_stripe_handles, + Vec::new(), None, 0, data.len(), @@ -5531,6 +5554,7 @@ mod tests { erasure, setup.readers, setup.deferred_stripe_handles, + Vec::new(), None, 0, part2_len, @@ -5571,6 +5595,7 @@ mod tests { erasure, setup.readers, setup.deferred_stripe_handles, + Vec::new(), None, 0, data.len(), @@ -6007,6 +6032,49 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn codec_streaming_reader_gate_keeps_copy_source_demand_bound_for_all_classes() { + temp_env::async_with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + ], + async { + let fi = codec_streaming_test_fileinfo(1024, 2); + let object_info = codec_streaming_test_object_info(&fi); + let normal = codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true); + assert_eq!(normal.decision, GetCodecStreamingDecision::Use); + + let plain_fi = codec_streaming_test_fileinfo(1024, 1); + let plain_object_info = codec_streaming_test_object_info(&plain_fi); + let normal_plain = codec_streaming_reader_gate_for_test(&None, &plain_object_info, &plain_fi, true); + assert_eq!(normal_plain.object_class, GetCodecStreamingObjectClass::PlainSinglePart); + assert_eq!(normal_plain.decision, GetCodecStreamingDecision::Use); + + let copy = with_get_object_read_policy(GetObjectReadPolicy::CopySource, async { + let multipart = codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true); + let plain = codec_streaming_reader_gate_for_test(&None, &plain_object_info, &plain_fi, true); + (multipart, plain) + }) + .await; + assert_eq!( + copy.0.decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::CopySourceDemandBound) + ); + assert_eq!( + copy.1.decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::CopySourceDemandBound) + ); + }, + ) + .await; + } + #[test] fn codec_streaming_reader_gate_keeps_multipart_default_off() { temp_env::with_vars( @@ -6108,6 +6176,10 @@ mod tests { assert_eq!(GetCodecStreamingFallbackReason::InvalidMinSize.as_str(), "invalid_min_size"); assert_eq!(GetCodecStreamingFallbackReason::ReadQuorumNotSafe.as_str(), "read_quorum_not_safe"); assert_eq!(GetCodecStreamingFallbackReason::MultipartPartLimit.as_str(), "multipart_part_limit"); + assert_eq!( + GetCodecStreamingFallbackReason::CopySourceDemandBound.as_str(), + "copy_source_demand_bound" + ); assert_eq!(GetCodecStreamingObjectClass::PlainSinglePart.as_str(), "plain_single_part"); assert_eq!(GetCodecStreamingObjectClass::Range.as_str(), "range"); assert_eq!(GetCodecStreamingObjectClass::Encrypted.as_str(), "encrypted"); diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index c5b7c321c..75fdc1aa4 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -2468,6 +2468,33 @@ impl ECStore { Self::resolve_decommission_tiered_object_result(result, bucket, &object) } + /// Open a source reader for a server-side copy. + /// + /// Copy consumers hold the source reader while a destination write can + /// apply backpressure. Keep that read contract explicit at the storage + /// boundary so the lower-level legacy multipart pipeline can suppress its + /// speculative next-part setup without changing the public `ObjectIO` + /// trait or `ObjectOptions` layout. + pub async fn get_object_reader_for_copy( + &self, + bucket: &str, + object: &str, + range: Option, + h: HeaderMap, + opts: &ObjectOptions, + ) -> Result<(GetObjectReader, tokio_util::sync::CancellationToken)> { + let cancellation = tokio_util::sync::CancellationToken::new(); + let reader = crate::set_disk::with_get_object_read_cancellation( + cancellation.clone(), + crate::set_disk::with_get_object_read_policy( + crate::set_disk::GetObjectReadPolicy::CopySource, + self.handle_get_object_reader(bucket, object, range, h, opts), + ), + ) + .await?; + Ok((reader, cancellation)) + } + #[instrument(level = "debug", skip(self, h))] #[hotpath::measure(impl_type = "ECStore")] pub(super) async fn handle_get_object_reader( diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 89d116d14..7bff61075 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -33,7 +33,9 @@ use super::storage_api::multipart_usecase::contract::http::HTTPPreconditions; use super::storage_api::multipart_usecase::contract::multipart::{ CompletePart, MAX_MULTIPART_PART_NUMBER, MultipartOperations as _, MultipartUploadResult, }; -use super::storage_api::multipart_usecase::contract::object::{ObjectIO as _, ObjectOperations as _}; +#[cfg(test)] +use super::storage_api::multipart_usecase::contract::object::ObjectIO as _; +use super::storage_api::multipart_usecase::contract::object::ObjectOperations as _; use super::storage_api::multipart_usecase::contract::range::HTTPRangeSpec; use super::storage_api::multipart_usecase::data_usage::{ quota_object_size, record_bucket_object_version_write_memory, record_bucket_object_write_memory, @@ -1443,8 +1445,8 @@ impl DefaultMultipartUsecase { .into()); } - let src_reader = store - .get_object_reader(&src_bucket, &src_key, rs.clone(), h, &get_opts) + let (src_reader, _source_cancellation) = store + .get_object_reader_for_copy(&src_bucket, &src_key, rs.clone(), h, &get_opts) .await .map_err(map_get_object_reader_error)?; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 022951744..0f5c39bbc 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -7948,11 +7948,18 @@ impl DefaultObjectUsecase { .into()); } - let gr = store - .get_object_reader(&src_bucket, &src_key, None, h, &src_get_opts) + let (gr, source_cancellation) = store + .get_object_reader_for_copy(&src_bucket, &src_key, None, h, &src_get_opts) .await .map_err(map_get_object_reader_error)?; + // The commit owner is intentionally detached so SetDisk can finish + // its rename/cleanup and post-commit publication if the HTTP caller + // goes away. Keep a request-owned guard for the source producer: + // cancellation drops the source read promptly, while the detached + // commit task retains the guards it needs to complete safely. + let _source_cancellation_guard = source_cancellation.clone().drop_guard(); + let mut src_info = gr.object_info.clone(); // A copy reads the source plaintext, so it needs the source key's decrypt permission diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index c9b3438af..40b4c459f 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1185,7 +1185,9 @@ pub(crate) mod multipart_usecase { } pub(crate) mod object { - pub(crate) use super::super::super::storage_contracts::{ObjectIO, ObjectOperations}; + #[cfg(test)] + pub(crate) use super::super::super::storage_contracts::ObjectIO; + pub(crate) use super::super::super::storage_contracts::ObjectOperations; } pub(crate) mod range { diff --git a/scripts/run_get_codec_streaming_smoke.sh b/scripts/run_get_codec_streaming_smoke.sh index b8631634e..bfedc9496 100755 --- a/scripts/run_get_codec_streaming_smoke.sh +++ b/scripts/run_get_codec_streaming_smoke.sh @@ -710,6 +710,7 @@ Static gate reasons: - multipart_part_limit - invalid_min_size - read_quorum_not_safe +- copy_source_demand_bound Relevant focused proof points: - set_disk::read::tests::codec_streaming_reader_gate_defaults_to_disabled