diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 36dd9dbf3..d4527fad6 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -128,6 +128,39 @@ pub const ENV_OBJECT_DISK_READ_TIMEOUT: &str = "RUSTFS_OBJECT_DISK_READ_TIMEOUT" /// Default disk read timeout in seconds. pub const DEFAULT_OBJECT_DISK_READ_TIMEOUT: u64 = 10; +/// Environment variable for the per-shard erasure write stall timeout (seconds). +/// +/// A single shard write (or shard-writer shutdown) that makes no forward +/// progress for longer than this budget is failed and its disk is dropped +/// before commit, so a black-hole peer that accepts the connection but never +/// drains the body cannot pin an otherwise-healthy write quorum forever +/// (see `MultiWriter` in `erasure/coding/encode.rs`). The budget is re-armed on +/// every shard write, so it bounds a *stall* rather than the total transfer +/// time of a large object. +/// +/// Unit: seconds (u64). `0` disables the stall deadline (previous behavior: +/// wait indefinitely). Default: 30 seconds. +pub const ENV_OBJECT_DISK_WRITE_STALL_TIMEOUT: &str = "RUSTFS_OBJECT_DISK_WRITE_STALL_TIMEOUT"; + +/// Default per-shard erasure write stall timeout in seconds. +pub const DEFAULT_OBJECT_DISK_WRITE_STALL_TIMEOUT: u64 = 30; + +/// Environment variable for the absolute per-object erasure write cap (seconds). +/// +/// Optional administrator backstop against a "slow-drip" peer that produces +/// just enough forward progress to reset the per-shard stall timeout on every +/// block while never converging. When set, the shard writers for one object are +/// engaged for at most this long in aggregate before a stalled writer is failed +/// and dropped. It is disabled by default because a legitimate large upload +/// over a slow-but-honest link must not be killed on total time alone; the +/// per-shard stall timeout is the primary guarantee. +/// +/// Unit: seconds (u64). `0` (default) disables the absolute cap. +pub const ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP: &str = "RUSTFS_OBJECT_DISK_WRITE_ABSOLUTE_CAP"; + +/// Default absolute per-object erasure write cap in seconds (`0` = disabled). +pub const DEFAULT_OBJECT_DISK_WRITE_ABSOLUTE_CAP: u64 = 0; + /// Environment variable for minimum GetObject timeout in seconds. /// /// When dynamic timeout calculation is enabled, this is the minimum timeout diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index f3f98d4a1..e9d240fac 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -217,6 +217,27 @@ pub fn get_object_disk_read_timeout() -> Duration { ) } +/// Per-shard erasure write stall budget: a shard write (or shutdown) that makes +/// no forward progress for this long is failed and its disk dropped before +/// commit. Re-armed on every shard write, so it bounds a stall rather than the +/// whole transfer. `0` disables the deadline (wait indefinitely). +pub fn get_object_disk_write_stall_timeout() -> Duration { + Duration::from_secs(rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_DISK_WRITE_STALL_TIMEOUT, + rustfs_config::DEFAULT_OBJECT_DISK_WRITE_STALL_TIMEOUT, + )) +} + +/// Optional absolute per-object erasure write cap (administrator slow-drip +/// backstop). `0` (default) disables the cap; the per-shard stall timeout is the +/// primary guarantee. +pub fn get_object_disk_write_absolute_cap() -> Duration { + Duration::from_secs(rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP, + rustfs_config::DEFAULT_OBJECT_DISK_WRITE_ABSOLUTE_CAP, + )) +} + pub fn get_drive_active_check_interval() -> Duration { Duration::from_secs(rustfs_utils::get_env_u64( rustfs_config::ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS, @@ -1653,6 +1674,32 @@ mod tests { }); } + #[test] + fn object_disk_write_stall_timeout_default_and_override() { + temp_env::with_var_unset(rustfs_config::ENV_OBJECT_DISK_WRITE_STALL_TIMEOUT, || { + assert_eq!( + get_object_disk_write_stall_timeout(), + Duration::from_secs(rustfs_config::DEFAULT_OBJECT_DISK_WRITE_STALL_TIMEOUT) + ); + }); + temp_env::with_var(rustfs_config::ENV_OBJECT_DISK_WRITE_STALL_TIMEOUT, Some("9"), || { + assert_eq!(get_object_disk_write_stall_timeout(), Duration::from_secs(9)); + }); + temp_env::with_var(rustfs_config::ENV_OBJECT_DISK_WRITE_STALL_TIMEOUT, Some("0"), || { + assert!(get_object_disk_write_stall_timeout().is_zero(), "0 disables the stall deadline"); + }); + } + + #[test] + fn object_disk_write_absolute_cap_defaults_disabled() { + temp_env::with_var_unset(rustfs_config::ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP, || { + assert!(get_object_disk_write_absolute_cap().is_zero(), "absolute cap is disabled by default"); + }); + temp_env::with_var(rustfs_config::ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP, Some("120"), || { + assert_eq!(get_object_disk_write_absolute_cap(), Duration::from_secs(120)); + }); + } + #[test] fn object_disk_read_timeout_uses_default_when_unset() { temp_env::with_var_unset(rustfs_config::ENV_OBJECT_DISK_READ_TIMEOUT, || { diff --git a/crates/ecstore/src/erasure/coding/encode.rs b/crates/ecstore/src/erasure/coding/encode.rs index 4bc4a9224..8509275b8 100644 --- a/crates/ecstore/src/erasure/coding/encode.rs +++ b/crates/ecstore/src/erasure/coding/encode.rs @@ -171,19 +171,90 @@ fn quorum_dominant_error_metric_label(summary: &WriteQuorumFailureSummary) -> &' dominant_error_summary_label(summary) } +/// Progress deadlines that bound how long shard writers may stall. +/// +/// A single shard write (or writer shutdown) is a unit of forward progress: for +/// a remote shard the underlying `HttpWriter` buffers the bytes and hands them +/// to a background HTTP task, so a peer that accepts the connection but never +/// drains the body eventually wedges the write once the bounded buffers fill. +/// Without a deadline that stalled writer is awaited forever and pins an +/// otherwise-healthy write quorum (rustfs/backlog#1319). +/// +/// * `stall_timeout` is re-armed on every shard write, so it bounds a stall, not +/// the total transfer time of a large object — a slow-but-honest writer that +/// keeps completing shards is never killed. +/// * `absolute_cap` is an optional administrator backstop: the shard writers for +/// one object are engaged for at most this long in aggregate. It defends +/// against a "slow-drip" peer that produces just enough progress to reset the +/// per-shard timeout on every block while never converging. Disabled by +/// default so a legitimate large upload over a slow link is not killed on +/// total time alone. +#[derive(Debug, Clone, Copy)] +pub(crate) struct WriteProgressPolicy { + stall_timeout: Option, + absolute_cap: Option, +} + +impl WriteProgressPolicy { + /// Build a policy, mapping a zero duration to "disabled" for both knobs. + pub(crate) fn new(stall_timeout: std::time::Duration, absolute_cap: std::time::Duration) -> Self { + Self { + stall_timeout: (!stall_timeout.is_zero()).then_some(stall_timeout), + absolute_cap: (!absolute_cap.is_zero()).then_some(absolute_cap), + } + } +} + +impl Default for WriteProgressPolicy { + fn default() -> Self { + Self::new( + crate::disk::disk_store::get_object_disk_write_stall_timeout(), + crate::disk::disk_store::get_object_disk_write_absolute_cap(), + ) + } +} + pub(crate) struct MultiWriter<'a> { writers: &'a mut [Option], write_quorum: usize, errs: Vec>, + policy: WriteProgressPolicy, + /// Absolute deadline for the whole object, armed lazily on the first + /// progress-bounded operation when `policy.absolute_cap` is set. + absolute_deadline: Option, } impl<'a> MultiWriter<'a> { pub fn new(writers: &'a mut [Option], write_quorum: usize) -> Self { + Self::with_policy(writers, write_quorum, WriteProgressPolicy::default()) + } + + pub fn with_policy(writers: &'a mut [Option], write_quorum: usize, policy: WriteProgressPolicy) -> Self { let length = writers.len(); MultiWriter { writers, write_quorum, errs: vec![None; length], + policy, + absolute_deadline: None, + } + } + + /// Effective budget for one shard operation: the smaller of the per-shard + /// stall timeout and the time remaining until the object's absolute cap. + /// Returns `None` when neither deadline is configured (wait indefinitely). + fn next_progress_budget(&mut self) -> Option { + if let Some(cap) = self.policy.absolute_cap { + let deadline = *self + .absolute_deadline + .get_or_insert_with(|| tokio::time::Instant::now() + cap); + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + match self.policy.stall_timeout { + Some(stall) => Some(stall.min(remaining)), + None => Some(remaining), + } + } else { + self.policy.stall_timeout } } @@ -214,13 +285,30 @@ impl<'a> MultiWriter<'a> { pub async fn write(&mut self, data: Vec) -> std::io::Result<()> { assert_eq!(data.len(), self.writers.len()); + let budget = self.next_progress_budget(); { let mut futures = FuturesUnordered::new(); for ((writer_opt, err), shard) in self.writers.iter_mut().zip(self.errs.iter_mut()).zip(data.iter()) { if err.is_some() { continue; // Skip if we already have an error for this writer } - futures.push(Self::write_shard(writer_opt, err, shard)); + // A shard write that makes no forward progress within `budget` is + // failed and its disk dropped, so a stalled peer cannot pin an + // otherwise-healthy write quorum (rustfs/backlog#1319). `budget` + // is recomputed per block, so it bounds a stall — not the total + // transfer time — and a slow-but-honest writer is never killed. + futures.push(async move { + match budget { + Some(budget) => match tokio::time::timeout(budget, Self::write_shard(writer_opt, err, shard)).await { + Ok(()) => {} + Err(_elapsed) => { + *err = Some(Error::Timeout); + *writer_opt = None; + } + }, + None => Self::write_shard(writer_opt, err, shard).await, + } + }); } while let Some(()) = futures.next().await {} } @@ -269,13 +357,30 @@ impl<'a> MultiWriter<'a> { pub async fn shutdown(&mut self) -> std::io::Result<()> { crate::hp_guard!("MultiWriter::shutdown"); + let budget = self.next_progress_budget(); { let mut futures = FuturesUnordered::new(); for (writer_opt, err) in self.writers.iter_mut().zip(self.errs.iter_mut()) { if err.is_some() { continue; } - futures.push(Self::shutdown_writer(writer_opt, err)); + // Shutdown flushes the tail and waits for the remote to finish the + // HTTP request; a black-hole peer that never responds would hang + // here forever for a small object whose bytes were fully buffered + // (so `write` never blocked). Bound it with the same progress + // budget and drop the stalled writer before the quorum check. + futures.push(async move { + match budget { + Some(budget) => match tokio::time::timeout(budget, Self::shutdown_writer(writer_opt, err)).await { + Ok(()) => {} + Err(_elapsed) => { + *err = Some(Error::Timeout); + *writer_opt = None; + } + }, + None => Self::shutdown_writer(writer_opt, err).await, + } + }); } while let Some(()) = futures.next().await {} } @@ -728,10 +833,12 @@ mod tests { use crate::erasure::coding::{BitrotWriterWrapper, CustomWriter}; use rustfs_rio::HardLimitReader; use rustfs_utils::HashAlgorithm; + use std::future::Future; use std::io::Cursor; use std::pin::Pin; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; + use std::time::Duration; use tokio::io::{AsyncWrite, AsyncWriteExt}; #[derive(Clone, Default)] @@ -821,6 +928,101 @@ mod tests { } } + /// A writer that models a "black-hole" peer: it accepts up to `stall_after` + /// bytes and then parks every subsequent `poll_write` forever (never + /// registering a waker), the way a remote `HttpWriter` wedges once its + /// bounded buffers fill and the peer never drains the body. With + /// `stall_after = 0` it stalls on the very first byte. `poll_shutdown` + /// completes only when `stall_shutdown` is false. + #[derive(Clone)] + struct StallingWriter { + accepted: Arc, + stall_after: usize, + stall_shutdown: bool, + } + + impl StallingWriter { + fn stalls_on_write(stall_after: usize) -> Self { + Self { + accepted: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + stall_after, + stall_shutdown: false, + } + } + + fn stalls_on_shutdown() -> Self { + Self { + accepted: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + stall_after: usize::MAX, + stall_shutdown: true, + } + } + } + + impl AsyncWrite for StallingWriter { + fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + use std::sync::atomic::Ordering; + let already = self.accepted.load(Ordering::SeqCst); + if already >= self.stall_after { + // Black hole: no forward progress, no waker registered. + return Poll::Pending; + } + let take = buf.len().min(self.stall_after - already); + self.accepted.fetch_add(take, Ordering::SeqCst); + Poll::Ready(Ok(take)) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + if self.stall_shutdown { + Poll::Pending + } else { + Poll::Ready(Ok(())) + } + } + } + + /// A writer that always makes forward progress but each `poll_write` first + /// waits `delay` (on the runtime timer). Under a paused virtual clock this + /// deterministically models a slow-but-honest peer: it must never be killed + /// by the per-shard stall deadline as long as `delay < stall_timeout`, yet an + /// absolute cap can still bound its aggregate engagement. + struct SlowWriter { + delay: std::time::Duration, + timer: Option>>, + } + + impl SlowWriter { + fn new(delay: std::time::Duration) -> Self { + Self { delay, timer: None } + } + } + + impl AsyncWrite for SlowWriter { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let delay = self.delay; + let timer = self.timer.get_or_insert_with(|| Box::pin(tokio::time::sleep(delay))); + match timer.as_mut().poll(cx) { + Poll::Ready(()) => { + self.timer = None; + Poll::Ready(Ok(buf.len())) + } + Poll::Pending => Poll::Pending, + } + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + fn bitrot_writer(writer: W, shard_size: usize) -> BitrotWriterWrapper where W: AsyncWrite + Send + Sync + Unpin + 'static, @@ -828,6 +1030,16 @@ mod tests { BitrotWriterWrapper::new(CustomWriter::new_tokio_writer(writer), shard_size, HashAlgorithm::HighwayHash256S) } + /// Like `bitrot_writer` but with no interleaved hash, so one shard write + /// issues a single `poll_write` to the underlying fake. Used by the paused + /// virtual-clock stall tests so each block maps to exactly one modeled delay. + fn bitrot_writer_plain(writer: W, shard_size: usize) -> BitrotWriterWrapper + where + W: AsyncWrite + Send + Sync + Unpin + 'static, + { + BitrotWriterWrapper::new(CustomWriter::new_tokio_writer(writer), shard_size, HashAlgorithm::None) + } + #[tokio::test] async fn helper_writers_cover_flush_and_shutdown_paths() { let mut failing_write = FailingWriteWriter; @@ -922,6 +1134,194 @@ mod tests { assert!(shutdown_err.contains("erasure write quorum")); } + // The production wiring (`MultiWriter::new`) must arm a real deadline by + // default, and honor `0` as "disabled". This guards against the fix silently + // regressing to wait-forever behavior on the default PUT path. + #[test] + fn default_write_progress_policy_is_armed_and_configurable() { + temp_env::with_var_unset(rustfs_config::ENV_OBJECT_DISK_WRITE_STALL_TIMEOUT, || { + temp_env::with_var_unset(rustfs_config::ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP, || { + let policy = WriteProgressPolicy::default(); + assert_eq!( + policy.stall_timeout, + Some(Duration::from_secs(rustfs_config::DEFAULT_OBJECT_DISK_WRITE_STALL_TIMEOUT)), + "the default PUT path must be protected by a stall deadline" + ); + assert_eq!(policy.absolute_cap, None, "the absolute cap is disabled by default"); + }); + }); + + temp_env::with_var(rustfs_config::ENV_OBJECT_DISK_WRITE_STALL_TIMEOUT, Some("0"), || { + let policy = WriteProgressPolicy::default(); + assert_eq!(policy.stall_timeout, None, "0 disables the stall deadline (wait indefinitely)"); + }); + } + + fn four_shards() -> Vec { + vec![ + Bytes::from_static(b"shard-0"), + Bytes::from_static(b"shard-1"), + Bytes::from_static(b"shard-2"), + Bytes::from_static(b"shard-3"), + ] + } + + /// Four full-`shard_size` shards. A shard shorter than `shard_size` marks the + /// bitrot writer `finished`, so multi-block tests must send full shards to + /// keep writing across iterations. + fn four_full_shards(shard_size: usize) -> Vec { + (0..4).map(|i| Bytes::from(vec![0x40 + i as u8; shard_size])).collect() + } + + // rustfs/backlog#1319: a single black-hole writer that never accepts a byte + // must be failed within the stall budget so the remaining 3/4 writers still + // meet a write quorum of 3 — the write must not wait forever. + #[tokio::test(start_paused = true)] + async fn multi_writer_write_stall_fails_black_hole_but_meets_quorum() { + let committed: Vec>>> = (0..3).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + let mut writers = vec![ + Some(bitrot_writer_plain(StallingWriter::stalls_on_write(0), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[0].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[1].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[2].clone()), 64)), + ]; + + { + let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO); + let mut mw = MultiWriter::with_policy(&mut writers, 3, policy); + mw.write(four_shards()) + .await + .expect("one stalled writer must still satisfy a 3/4 write quorum without hanging"); + } + + assert!(writers[0].is_none(), "the black-hole writer must be failed and dropped before commit"); + assert!(writers[1].is_some() && writers[2].is_some() && writers[3].is_some()); + } + + // Two black-hole writers drop the healthy count to 2/4, below the quorum of + // 3, so the write must fail cleanly (not hang). + #[tokio::test(start_paused = true)] + async fn multi_writer_write_stall_two_black_holes_fail_quorum() { + let committed: Vec>>> = (0..2).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + let mut writers = vec![ + Some(bitrot_writer_plain(StallingWriter::stalls_on_write(0), 64)), + Some(bitrot_writer_plain(StallingWriter::stalls_on_write(0), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[0].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[1].clone()), 64)), + ]; + + let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO); + let mut mw = MultiWriter::with_policy(&mut writers, 3, policy); + let err = mw + .write(four_shards()) + .await + .expect_err("two stalled writers must fail the write quorum instead of hanging"); + assert!(err.to_string().contains("Failed to write data")); + } + + // A small object whose bytes were fully buffered leaves `write` succeeding + // for a black-hole peer; the stall must then be caught during shutdown so it + // cannot pin the quorum there either. + #[tokio::test(start_paused = true)] + async fn multi_writer_shutdown_stall_fails_black_hole_but_meets_quorum() { + let committed: Vec>>> = (0..3).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + let mut writers = vec![ + Some(bitrot_writer_plain(StallingWriter::stalls_on_shutdown(), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[0].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[1].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[2].clone()), 64)), + ]; + + { + let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO); + let mut mw = MultiWriter::with_policy(&mut writers, 3, policy); + mw.write(four_shards()).await.expect("all writers accept the buffered shard"); + mw.shutdown() + .await + .expect("a single shutdown stall must still satisfy a 3/4 quorum without hanging"); + } + + assert!(writers[0].is_none(), "the writer that stalled shutdown must be failed and dropped"); + } + + #[tokio::test(start_paused = true)] + async fn multi_writer_shutdown_stall_two_black_holes_fail_quorum() { + let committed: Vec>>> = (0..2).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + let mut writers = vec![ + Some(bitrot_writer_plain(StallingWriter::stalls_on_shutdown(), 64)), + Some(bitrot_writer_plain(StallingWriter::stalls_on_shutdown(), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[0].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[1].clone()), 64)), + ]; + + let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO); + let mut mw = MultiWriter::with_policy(&mut writers, 3, policy); + mw.write(four_shards()).await.expect("all writers accept the buffered shard"); + let err = mw + .shutdown() + .await + .expect_err("two shutdown stalls must fail the shutdown quorum instead of hanging"); + assert!(err.to_string().contains("Failed to shutdown writers")); + } + + // A slow-but-honest writer that keeps completing shards (delay < stall + // budget) must never be killed, even across many blocks, when no absolute cap + // is configured. + #[tokio::test(start_paused = true)] + async fn multi_writer_slow_but_progressing_writer_is_not_killed() { + let mut writers = vec![ + Some(bitrot_writer_plain(SlowWriter::new(Duration::from_secs(1)), 64)), + Some(bitrot_writer_plain(SlowWriter::new(Duration::from_secs(1)), 64)), + Some(bitrot_writer_plain(SlowWriter::new(Duration::from_secs(1)), 64)), + Some(bitrot_writer_plain(SlowWriter::new(Duration::from_secs(1)), 64)), + ]; + + { + let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO); + let mut mw = MultiWriter::with_policy(&mut writers, 3, policy); + for _ in 0..8 { + mw.write(four_full_shards(64)) + .await + .expect("a writer that keeps making progress within the stall budget must not be failed"); + } + } + + assert!(writers.iter().all(Option::is_some), "no slow-but-honest writer should be dropped"); + } + + // Slow-drip defense: the per-shard stall timer resets on every block, so a + // peer that dribbles just enough progress each block is not caught by the + // stall timeout alone. The optional absolute cap bounds its aggregate + // engagement and fails it within a finite budget, while the healthy writers + // still meet quorum. + #[tokio::test(start_paused = true)] + async fn multi_writer_absolute_cap_bounds_slow_drip_writer() { + let committed: Vec>>> = (0..3).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + let mut writers = vec![ + Some(bitrot_writer_plain(SlowWriter::new(Duration::from_secs(8)), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[0].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[1].clone()), 64)), + Some(bitrot_writer_plain(DeferredCommitWriter::new(committed[2].clone()), 64)), + ]; + + { + // stall (10s) never fires for an 8s-per-block writer, but the 15s + // absolute cap does once its aggregate engagement crosses the cap. + let policy = WriteProgressPolicy::new(Duration::from_secs(10), Duration::from_secs(15)); + let mut mw = MultiWriter::with_policy(&mut writers, 3, policy); + for _ in 0..4 { + mw.write(four_full_shards(64)) + .await + .expect("healthy writers keep the 3/4 quorum while the drip writer is bounded"); + } + } + + assert!( + writers[0].is_none(), + "the slow-drip writer must be failed within the absolute cap, not held forever" + ); + } + #[tokio::test] async fn drain_queued_inflight_bytes_consumes_pending_blocks() { let (tx, mut rx) = mpsc::channel(2); diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index 6cc6a6fb8..33bdce62d 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -976,21 +976,21 @@ impl Stream for ReceiverStream { } } -pin_project! { - pub struct HttpWriter { - url:String, - method: Method, - headers: HeaderMap, - err_rx: tokio::sync::oneshot::Receiver, - start_tx: Option>, - sender: PollSender>, - handle: tokio::task::JoinHandle>, - pending_chunk: BytesMut, - finish:bool, - track_internode_metrics: bool, - internode_operation: Option<&'static str>, - - } +// Not pin-projected: every field is `Unpin` and the `AsyncWrite` impl accesses +// them through `get_mut()`, so a plain struct lets us add a manual `Drop` +// (pin-project forbids one) to abort the background HTTP task — see below. +pub struct HttpWriter { + url: String, + method: Method, + headers: HeaderMap, + err_rx: tokio::sync::oneshot::Receiver, + start_tx: Option>, + sender: PollSender>, + handle: tokio::task::JoinHandle>, + pending_chunk: BytesMut, + finish: bool, + track_internode_metrics: bool, + internode_operation: Option<&'static str>, } const HTTP_WRITER_CHANNEL_CAPACITY: usize = 8; @@ -1476,6 +1476,20 @@ impl AsyncWrite for HttpWriter { } } +impl Drop for HttpWriter { + /// Abort the background HTTP request when the writer is dropped without a + /// clean shutdown. On a stall timeout the caller drops this writer to fail + /// its shard (rustfs/backlog#1319); a black-hole peer would otherwise leave + /// the spawned task holding the connection and its buffered body alive. A + /// cleanly shut-down writer has already joined the task, so aborting a + /// finished handle is a no-op. This does not — and cannot — unsend bytes + /// already handed to the transport: those land only in the upload's unique + /// tmp path and are reclaimed by tmp GC, never touching a committed object. + fn drop(&mut self) { + self.handle.abort(); + } +} + #[cfg(test)] mod tests { use super::*; @@ -1877,18 +1891,10 @@ mod tests { }; let writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap(); - let HttpWriter { - handle, - sender, - start_tx, - .. - } = writer; - drop(start_tx); - drop(sender); - handle - .await - .expect("HttpWriter background task should not panic") - .expect("an unstarted HttpWriter should stop cleanly"); + // Dropping an unstarted writer aborts its parked background task + // (rustfs/backlog#1319) before it can ever send a PUT. + drop(writer); + tokio::time::sleep(Duration::from_millis(50)).await; assert_eq!(state.put_count.load(Ordering::SeqCst), 0); assert!(state.put_bodies.lock().await.is_empty()); @@ -1896,6 +1902,70 @@ mod tests { server_handle.abort(); } + /// A PUT handler that registers the request, then parks forever without ever + /// sending a response — modeling a black-hole peer that accepts the + /// connection but never completes the request, so the writer's background + /// task stays parked at `request.send().await` until it is aborted. + async fn hanging_put(State(state): State, _body: Body) -> impl IntoResponse { + state.put_count.fetch_add(1, Ordering::SeqCst); + std::future::pending::<()>().await; + StatusCode::OK + } + + // rustfs/backlog#1319: when a stalled remote writer is dropped (the encode + // path drops it to fail its shard), the background HTTP task must be aborted + // so it stops holding the connection and buffered body — it must not linger. + #[tokio::test] + async fn http_writer_drop_aborts_background_request_to_hanging_peer() { + let state = TestState::default(); + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let addr = listener.local_addr().expect("listener local address should be available"); + let app = Router::new() + .route("/hang", axum::routing::put(hanging_put)) + .with_state(state.clone()); + let server_handle = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let url = format!("http://{addr}/hang"); + let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap(); + // A >1MiB write starts the request and hands the body to the background + // task, which then parks in the handler that never responds. + writer.write_all(&vec![0xa5u8; HTTP_WRITER_BUFFER_SIZE + 1]).await.unwrap(); + // Let the server register the request, so the background task is parked + // (alive) at send() before we drop the writer. + for _ in 0..100 { + if state.put_count.load(Ordering::SeqCst) >= 1 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(state.put_count.load(Ordering::SeqCst), 1, "server should have accepted the request"); + + // Observe the background task via its abort handle; it must still be + // running before the drop, then be aborted (finished) after it. + let task = writer.handle.abort_handle(); + assert!(!task.is_finished(), "background task should still be running before drop"); + + drop(writer); + + let mut aborted = false; + for _ in 0..500 { + if task.is_finished() { + aborted = true; + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(aborted, "dropping the writer must abort the background HTTP task"); + + server_handle.abort(); + } + #[tokio::test] async fn http_writer_shutdown_without_write_sends_empty_put() { let state = TestState::default();