fix(ecstore): bound remote shard writers with a progress deadline so one black-hole peer cannot pin write quorum (#4925)

A PUT that fans out erasure shards to remote peers awaited every shard writer to completion on both the per-block write and the final shutdown, and the remote HttpWriter had no progress deadline. A peer that accepts the TCP connection but never drains the request body (or never sends a response) therefore wedges the writer forever once the bounded buffers fill, pinning an otherwise-healthy write quorum indefinitely — a cluster-level write-availability hazard triggered by a single bad peer (rustfs/backlog#1319, https://github.com/rustfs/backlog/issues/1319).

MultiWriter now wraps each shard write and each shard-writer shutdown in a forward-progress deadline. The budget is re-armed on every block, so it bounds a stall rather than the total transfer time of a large object: a slow-but-honest writer that keeps completing shards is never killed, while a writer that makes no progress within the budget is failed and its disk dropped before commit. An optional absolute per-object cap (disabled by default) backstops a slow-drip peer that dribbles just enough progress to reset the per-block timer without ever converging; it is off by default so a legitimate large upload over a slow link is not killed on total time alone. Both knobs come from RUSTFS_OBJECT_DISK_WRITE_STALL_TIMEOUT (default 30s) and RUSTFS_OBJECT_DISK_WRITE_ABSOLUTE_CAP (default 0 = disabled); setting the stall timeout to 0 restores the previous wait-forever behavior for a conservative rollback.

The deadline enforcement lives in MultiWriter (writer-agnostic), so it covers local and remote writers alike and keeps the existing control-flow shape: a timed-out shard is marked failed (Error::Timeout, which is not an ignored error) and excluded from the write quorum exactly like any other shard write failure, and the unchanged nil_count/quorum check then continues on quorum or fails cleanly. This deliberately stays out of the MultiWriter lifecycle / commit-coordinator territory owned by rustfs/backlog#1312.

When a stalled writer is dropped to fail its shard, the remote HttpWriter must stop holding the connection and its buffered body. HttpWriter previously left its spawned request task running on drop; it now aborts that background task in Drop (it is no longer pin-projected, since every field is Unpin and the AsyncWrite impl already used get_mut). Bytes already handed to the transport cannot be unsent, but they land only in this upload's unique tmp path and are reclaimed by tmp GC — they never touch a committed object.

Tests, all on a paused virtual clock so they are deterministic and non-flaky:
- one black-hole writer still meets a 3/4 write quorum without hanging; two black holes fail the quorum cleanly (both for the per-block write and the shutdown paths).
- a slow-but-honest writer that keeps making progress within the stall budget is never failed across many blocks.
- the absolute cap bounds a slow-drip writer within a finite budget while the healthy writers keep quorum.
- the default policy is armed by default and honors 0 as disabled.
- HttpWriter aborts its background request task on drop against a hanging peer.

The toxiproxy/black-hole 4x4 end-to-end acceptance depends on black-box test facilities from rustfs/backlog#1325, which are not built yet; that acceptance is deferred to #1325 and intentionally not faked here.

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

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, || {

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<std::time::Duration>,
+    absolute_cap: Option<std::time::Duration>,
+}
+
+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<BitrotWriterWrapper>],
     write_quorum: usize,
     errs: Vec<Option<Error>>,
+    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<tokio::time::Instant>,
 }
 
 impl<'a> MultiWriter<'a> {
     pub fn new(writers: &'a mut [Option<BitrotWriterWrapper>], write_quorum: usize) -> Self {
+        Self::with_policy(writers, write_quorum, WriteProgressPolicy::default())
+    }
+
+    pub fn with_policy(writers: &'a mut [Option<BitrotWriterWrapper>], 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<std::time::Duration> {
+        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<Bytes>) -> 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<std::sync::atomic::AtomicUsize>,
+        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<std::io::Result<usize>> {
+            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<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            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<Pin<Box<tokio::time::Sleep>>>,
+    }
+
+    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<std::io::Result<usize>> {
+            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<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
     fn bitrot_writer<W>(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<W>(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<Bytes> {
+        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<Bytes> {
+        (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<Arc<Mutex<Vec<u8>>>> = (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<Arc<Mutex<Vec<u8>>>> = (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<Arc<Mutex<Vec<u8>>>> = (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<Arc<Mutex<Vec<u8>>>> = (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<Arc<Mutex<Vec<u8>>>> = (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);

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<std::io::Error>,
-        start_tx: Option<tokio::sync::oneshot::Sender<()>>,
-        sender: PollSender<Option<Bytes>>,
-        handle: tokio::task::JoinHandle<std::io::Result<()>>,
-        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<std::io::Error>,
+    start_tx: Option<tokio::sync::oneshot::Sender<()>>,
+    sender: PollSender<Option<Bytes>>,
+    handle: tokio::task::JoinHandle<std::io::Result<()>>,
+    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<TestState>, _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();