fix(ecstore): preserve multipart shards within write budgets

Co-authored-by: cursoragent <cursoragent@users.noreply.github.com>

.github/workflows/cla.yml

@@ -95,7 +95,7 @@ jobs:
 
       - name: Run CLA Bot
         if: github.event_name != 'merge_group'
-        uses: overtrue/cla-bot@f054da99d5777e47e1af9933737b9c0c5584be8d # v0.1.0
+        uses: overtrue/cla-bot@4bd1387c19691168c42fcf5edfa17b15819b2ab3 # v0.1.1
         with:
           github-token: ${{ github.token }}
           registry-token: ${{ steps.registry-token.outputs.token }}

crates/e2e_test/src/protocols/README.md

@@ -77,9 +77,7 @@ core protocol surface plus cross-protocol consistency:
   existing path returns an error, bad-password authentication is rejected
 - Cross-protocol via aws-sdk-s3: SFTP write then S3 read with SHA256 match,
   S3 write then SFTP read with SHA256 match
-- Cross-API directory visibility: SFTP-created sub-directory visible via S3
-  ListObjectsV2, S3-created `__XLDIR__` marker visible via SFTP readdir as a
-  directory entry
+- Cross-API directory visibility: SFTP-created sub-directory visible via S3 ListObjectsV2, and an S3 trailing-slash directory object visible via SFTP readdir as a directory entry
 
 #### sftp_compliance (`test_sftp_compliance_suite`)
 

crates/e2e_test/src/protocols/sftp_core.rs

@@ -394,9 +394,9 @@ pub async fn test_sftp_core_operations() -> Result<()> {
         info!("PASS: S3-written object matches via S3 GetObject and SFTP READ");
 
         // --- Cross-API directory visibility: SFTP mkdir, S3 ListObjectsV2 ---
-        // SFTP mkdir writes a __XLDIR__ marker. The rustfs S3 listing path
-        // decodes that marker back to a trailing-slash key, so the asserted
-        // pattern is "subdir_sftp/".
+        // SFTP mkdir writes a zero-byte object at the trailing-slash key.
+        // Storage encodes that key to the __XLDIR__ marker, and the S3
+        // listing path decodes the marker back to a trailing-slash key.
         info!("Testing SFTP: SFTP-created sub-directory visible via S3 ListObjectsV2");
         let sftp_subdir_name = "subdir_sftp";
         let sftp_subdir_path = format!("/{bucket}/{sftp_subdir_name}");
@@ -422,20 +422,21 @@ pub async fn test_sftp_core_operations() -> Result<()> {
         );
         info!("PASS: SFTP mkdir visible to S3 ListObjectsV2");
 
-        // --- Cross-API directory visibility: S3 marker, SFTP readdir ---
-        // aws-sdk-s3 PutObject writes a zero-byte marker keyed with the
-        // __XLDIR__ suffix. SFTP readdir must decode the marker back to a
-        // bare directory entry whose file_type reports as a directory.
-        info!("Testing SFTP: S3-created __XLDIR__ marker visible via SFTP readdir");
+        // --- Cross-API directory visibility: S3 directory object, SFTP readdir ---
+        // aws-sdk-s3 PutObject writes a zero-byte object at the trailing-slash
+        // key. The __XLDIR__ suffix is reserved on the client-key boundary, so
+        // a direct put of that suffix is rejected. Storage encodes the slash
+        // key, and SFTP readdir must show a directory entry.
+        info!("Testing SFTP: S3-created directory object visible via SFTP readdir");
         let s3_subdir_name = "subdir_s3";
-        let s3_subdir_marker_key = format!("{s3_subdir_name}{XLDIR_SUFFIX}");
+        let s3_subdir_key = format!("{s3_subdir_name}/");
         s3.put_object()
             .bucket(bucket)
-            .key(&s3_subdir_marker_key)
+            .key(&s3_subdir_key)
             .body(ByteStream::from_static(b""))
             .send()
             .await
-            .map_err(|e| anyhow!("S3 PutObject {}/{} failed: {:?}", bucket, s3_subdir_marker_key, e))?;
+            .map_err(|e| anyhow!("S3 PutObject {}/{} failed: {:?}", bucket, s3_subdir_key, e))?;
         let bucket_entries: Vec<(String, bool)> = sftp
             .read_dir(&bucket_path)
             .await?
@@ -444,9 +445,9 @@ pub async fn test_sftp_core_operations() -> Result<()> {
         let visible_via_sftp = bucket_entries.iter().any(|(name, is_dir)| name == s3_subdir_name && *is_dir);
         assert!(
             visible_via_sftp,
-            "S3-created marker must appear as a directory in SFTP readdir: entries were {bucket_entries:?}"
+            "S3-created directory object must appear as a directory in SFTP readdir: entries were {bucket_entries:?}"
         );
-        info!("PASS: S3 marker visible to SFTP readdir as a directory");
+        info!("PASS: S3 directory object visible to SFTP readdir as a directory");
 
         // --- Pre-cleanup of cross-protocol fixtures ---
         // Removes the new files and sub-directories so the existing rmdir

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

@@ -996,21 +996,32 @@ fn internode_idempotent_read_retries() -> usize {
 static PREWARMED_PEERS: std::sync::LazyLock<std::sync::Mutex<std::collections::HashSet<String>>> =
     std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new()));
 
+fn claim_control_channel_prewarm(addr: &str) -> bool {
+    let Ok(mut prewarmed) = PREWARMED_PEERS.lock() else {
+        return false;
+    };
+    prewarmed.insert(addr.to_owned())
+}
+
+fn release_control_channel_prewarm(addr: &str) {
+    if let Ok(mut prewarmed) = PREWARMED_PEERS.lock() {
+        prewarmed.remove(addr);
+    }
+}
+
 /// Best-effort background prewarm of a peer's control channel (grpc-optimization P3-1). Deduped per
-/// peer; a dial failure just falls through to the existing lazy connect and recovery monitor.
+/// peer; a dial failure releases the marker so a later disk construction can retry the prewarm.
 fn spawn_control_channel_prewarm(addr: String) {
-    {
-        let Ok(mut prewarmed) = PREWARMED_PEERS.lock() else {
-            return;
-        };
-        if !prewarmed.insert(addr.clone()) {
-            return;
-        }
+    if !claim_control_channel_prewarm(&addr) {
+        return;
     }
     tokio::spawn(async move {
         match node_service_time_out_client_no_auth(&addr).await {
             Ok(_) => debug!(addr = %addr, "internode control channel prewarmed"),
-            Err(err) => debug!(addr = %addr, error = %err, "internode control channel prewarm failed (best-effort)"),
+            Err(err) => {
+                release_control_channel_prewarm(&addr);
+                debug!(addr = %addr, error = %err, "internode control channel prewarm failed (best-effort)");
+            }
         }
     });
 }
@@ -3680,7 +3691,13 @@ impl DiskAPI for RemoteDisk {
                         );
                     }
                     let rpc_started = read_version_stage_timer(batch_read_version_attribution_enabled);
-                    let response = match client.batch_read_version(request).await {
+                    let response_result = client.batch_read_version(request).await;
+                    if let Some(rpc_started) = rpc_started {
+                        crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_duration(
+                            rpc_started.elapsed(),
+                        );
+                    }
+                    let response = match response_result {
                         Ok(response) => {
                             record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RPC_ROUNDTRIP, rpc_started);
                             response.into_inner()
@@ -4695,7 +4712,10 @@ impl DiskAPI for RemoteDisk {
                     read_multiple_req_bin: read_multiple_req_bin.into(),
                 });
 
-                let response = client.read_multiple(request).await?.into_inner();
+                let rpc_started = Instant::now();
+                let response = client.read_multiple(request).await;
+                crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_multiple_duration(rpc_started.elapsed());
+                let response = response?.into_inner();
 
                 if !response.success {
                     return Err(response.error.unwrap_or_default().into());
@@ -4746,6 +4766,7 @@ impl DiskAPI for RemoteDisk {
                 attach_mutation_body_digest(&mut request, canonical_body, "write_all")?;
 
                 crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_request();
+                let rpc_started = Instant::now();
                 let response = match client.write_all(request).await {
                     Ok(response) => response.into_inner(),
                     Err(err) => {
@@ -4754,6 +4775,7 @@ impl DiskAPI for RemoteDisk {
                     }
                 };
 
+                crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_duration(rpc_started.elapsed());
                 crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_sent_bytes(data_len);
 
                 if !response.success {
@@ -4851,6 +4873,7 @@ impl DiskAPI for RemoteDisk {
                 });
 
                 crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_request();
+                let rpc_started = Instant::now();
                 let response = match client.read_all(request).await {
                     Ok(response) => response.into_inner(),
                     Err(err) => {
@@ -4859,6 +4882,8 @@ impl DiskAPI for RemoteDisk {
                     }
                 };
 
+                crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_duration(rpc_started.elapsed());
+
                 if !response.success {
                     crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_error();
                     return Err(response.error.unwrap_or_default().into());
@@ -11975,4 +12000,15 @@ mod tests {
 
         accept_task.abort();
     }
+
+    #[test]
+    fn prewarm_claim_is_deduplicated_and_released_after_failure() {
+        let addr = "http://prewarm-retry-test.invalid:9000";
+        release_control_channel_prewarm(addr);
+        assert!(claim_control_channel_prewarm(addr));
+        assert!(!claim_control_channel_prewarm(addr));
+        release_control_channel_prewarm(addr);
+        assert!(claim_control_channel_prewarm(addr));
+        release_control_channel_prewarm(addr);
+    }
 }

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

@@ -481,6 +481,14 @@ pub(crate) fn record_remote_disk_grpc_write_all_request() {
         .record_outgoing_request_for_operation_and_backend(INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
 }
 
+pub(crate) fn record_remote_disk_grpc_write_all_duration(duration: Duration) {
+    global_internode_metrics().record_duration_for_operation_and_backend(
+        INTERNODE_OPERATION_GRPC_WRITE_ALL,
+        INTERNODE_TRANSPORT_BACKEND_GRPC,
+        duration,
+    );
+}
+
 pub(crate) fn record_remote_disk_grpc_write_all_sent_bytes(bytes: usize) {
     global_internode_metrics().record_sent_bytes_for_operation_and_backend(
         INTERNODE_OPERATION_GRPC_WRITE_ALL,
@@ -521,6 +529,14 @@ pub(crate) fn record_remote_disk_grpc_read_all_request() {
         .record_outgoing_request_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
 }
 
+pub(crate) fn record_remote_disk_grpc_read_all_duration(duration: Duration) {
+    global_internode_metrics().record_duration_for_operation_and_backend(
+        INTERNODE_OPERATION_GRPC_READ_ALL,
+        INTERNODE_TRANSPORT_BACKEND_GRPC,
+        duration,
+    );
+}
+
 #[cfg(feature = "diagnostic-read-version-caller")]
 fn read_version_request_mode(opts: &ReadOptions) -> ReadVersionRequestMode {
     match (opts.read_data, opts.healing, opts.incl_free_versions) {
@@ -598,6 +614,17 @@ pub(crate) fn record_remote_disk_grpc_batch_read_version_stage(stage: &'static s
     );
 }
 
+pub(crate) fn record_remote_disk_grpc_batch_read_version_duration(duration: Duration) {
+    if !rustfs_io_metrics::get_stage_metrics_enabled() {
+        return;
+    }
+    global_internode_metrics().record_duration_for_operation_and_backend(
+        INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION,
+        INTERNODE_TRANSPORT_BACKEND_GRPC,
+        duration,
+    );
+}
+
 pub(crate) fn record_remote_disk_grpc_batch_read_version_error() {
     if !rustfs_io_metrics::get_stage_metrics_enabled() {
         return;
@@ -690,6 +717,14 @@ pub(crate) fn record_remote_disk_grpc_read_multiple_recv_bytes(bytes: usize) {
     record_grpc_payload_size(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, bytes);
 }
 
+pub(crate) fn record_remote_disk_grpc_read_multiple_duration(duration: Duration) {
+    global_internode_metrics().record_duration_for_operation_and_backend(
+        INTERNODE_OPERATION_GRPC_READ_MULTIPLE,
+        INTERNODE_TRANSPORT_BACKEND_GRPC,
+        duration,
+    );
+}
+
 /// Payload-size threshold (bytes) above which a unary internode gRPC response is counted as a
 /// "large payload" for alerting. Env-overridable via `RUSTFS_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES`.
 fn internode_rpc_large_payload_warn_bytes() -> usize {

crates/ecstore/src/core/pool_meta_recovery_codec.rs

@@ -315,7 +315,7 @@ pub(super) async fn encode_inline_pool_meta_replicas(
         .encode_with_shard_integrity(
             Cursor::new(Bytes::copy_from_slice(canonical)),
             &mut writers,
-            raw.len(),
+            crate::erasure::coding::encode::ShardWritePolicy::object(raw.len()),
             1,
             IntegrityEncodeMode::Inline(canonical.len()),
             protected,

crates/ecstore/src/erasure/coding/encode.rs

@@ -326,6 +326,40 @@ impl Default for WriteProgressPolicy {
     }
 }
 
+/// Caller-specific completion behavior for an erasure write.
+///
+/// A multipart upload must not discard a different healthy shard from each
+/// part merely because another quorum finished first: the final object needs
+/// a quorum of disks carrying every part. Pending writers still obey the
+/// existing progress and absolute deadlines and never count as completed.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct ShardWritePolicy {
+    quorum: usize,
+    post_quorum: PostQuorumPolicy,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum PostQuorumPolicy {
+    Grace,
+    ExistingBudget,
+}
+
+impl ShardWritePolicy {
+    pub(crate) fn object(quorum: usize) -> Self {
+        Self {
+            quorum,
+            post_quorum: PostQuorumPolicy::Grace,
+        }
+    }
+
+    pub(crate) fn multipart(quorum: usize) -> Self {
+        Self {
+            quorum,
+            post_quorum: PostQuorumPolicy::ExistingBudget,
+        }
+    }
+}
+
 /// After write quorum is already in hand, how long a slower shard may still
 /// finish before it is dropped. The per-shard stall budget (30s by default)
 /// is what a black-hole peer is allowed to consume *before* quorum exists.
@@ -338,13 +372,15 @@ fn post_quorum_straggler_grace(stall_timeout: Option<std::time::Duration>) -> Op
     stall_timeout.map(|stall| stall.min(WRITE_QUORUM_STRAGGLER_GRACE))
 }
 
-/// Drive shard ops until write quorum has succeeded and stragglers have had
-/// [`post_quorum_straggler_grace`] to finish. Returns `true` for each slot
-/// whose future completed; `false` means the caller must fail that slot.
+/// Drive shard operations within their existing deadlines. Ordinary object
+/// writes additionally stop after the post-quorum grace; multipart writes keep
+/// polling all candidates within the same operation budget. A `false` slot
+/// never completed and must be excluded from commit.
 async fn await_write_quorum<F>(
     write_quorum: usize,
     stall_timeout: Option<std::time::Duration>,
     slot_count: usize,
+    post_quorum: PostQuorumPolicy,
     mut futures: FuturesUnordered<F>,
 ) -> Vec<bool>
 where
@@ -376,6 +412,7 @@ where
         if succeeded {
             successes = successes.saturating_add(1);
             if successes >= write_quorum
+                && post_quorum == PostQuorumPolicy::Grace
                 && grace.is_none()
                 && let Some(delay) = post_quorum_straggler_grace(stall_timeout)
             {
@@ -404,6 +441,7 @@ pub(crate) struct MultiWriter<'a> {
     writers: &'a mut [Option<BitrotWriterWrapper>],
     integrity: Option<&'a mut IntegrityBuilder>,
     write_quorum: usize,
+    post_quorum: PostQuorumPolicy,
     errs: Vec<Option<Error>>,
     policy: WriteProgressPolicy,
     /// Absolute deadline for the whole object, armed lazily on the first
@@ -412,16 +450,21 @@ pub(crate) struct MultiWriter<'a> {
 }
 
 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 new(writers: &'a mut [Option<BitrotWriterWrapper>], write_policy: ShardWritePolicy) -> Self {
+        Self::with_policy(writers, write_policy, WriteProgressPolicy::default())
     }
 
-    pub fn with_policy(writers: &'a mut [Option<BitrotWriterWrapper>], write_quorum: usize, policy: WriteProgressPolicy) -> Self {
+    pub fn with_policy(
+        writers: &'a mut [Option<BitrotWriterWrapper>],
+        write_policy: ShardWritePolicy,
+        policy: WriteProgressPolicy,
+    ) -> Self {
         let length = writers.len();
         MultiWriter {
             writers,
             integrity: None,
-            write_quorum,
+            write_quorum: write_policy.quorum,
+            post_quorum: write_policy.post_quorum,
             errs: vec![None; length],
             policy,
             absolute_deadline: None,
@@ -506,9 +549,8 @@ impl<'a> MultiWriter<'a> {
                 // 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
-                // while quorum is still open. Once quorum has succeeded, a
-                // straggler only gets `WRITE_QUORUM_STRAGGLER_GRACE` more time.
+                // transfer time. Only ordinary object writes add the shorter
+                // post-quorum grace; multipart preserves this operation budget.
                 futures.push(async move {
                     match budget {
                         Some(budget) => match tokio::time::timeout(budget, Self::write_shard(writer_opt, err, shard)).await {
@@ -524,7 +566,7 @@ impl<'a> MultiWriter<'a> {
                     (index, succeeded)
                 });
             }
-            await_write_quorum(write_quorum, stall_timeout, slot_count, futures).await
+            await_write_quorum(write_quorum, stall_timeout, slot_count, self.post_quorum, futures).await
         };
         abandon_unfinished_writers(self.writers, &mut self.errs, &saw);
 
@@ -587,8 +629,8 @@ impl<'a> MultiWriter<'a> {
                 // 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.
-                // After quorum has shut down, do not keep waiting out the rest of
-                // that budget for a peer that has stopped answering.
+                // Ordinary object writes also apply their post-quorum grace;
+                // multipart waits for all candidates within this same budget.
                 futures.push(async move {
                     match budget {
                         Some(budget) => match tokio::time::timeout(budget, Self::shutdown_writer(writer_opt, err)).await {
@@ -604,7 +646,7 @@ impl<'a> MultiWriter<'a> {
                     (index, succeeded)
                 });
             }
-            await_write_quorum(write_quorum, stall_timeout, slot_count, futures).await
+            await_write_quorum(write_quorum, stall_timeout, slot_count, self.post_quorum, futures).await
         };
         abandon_unfinished_writers(self.writers, &mut self.errs, &saw);
 
@@ -639,7 +681,7 @@ impl Erasure {
         self: Arc<Self>,
         reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         part_number: usize,
         mode: IntegrityEncodeMode,
         protected: bool,
@@ -668,19 +710,19 @@ impl Erasure {
             }
             IntegrityEncodeMode::SingleBlock(hint) => {
                 let (reader, size) = self
-                    .encode_small_direct(reader, writers, quorum, true, hint, integrity.as_mut())
+                    .encode_small_direct(reader, writers, write_policy, true, hint, integrity.as_mut())
                     .await?;
                 (reader, size, None)
             }
             IntegrityEncodeMode::Streaming => {
                 let (reader, size) = self
-                    .encode_with_ingest_mode(reader, writers, quorum, use_bytesmut_ingest(), integrity.as_mut())
+                    .encode_with_ingest_mode(reader, writers, write_policy, use_bytesmut_ingest(), integrity.as_mut())
                     .await?;
                 (reader, size, None)
             }
             IntegrityEncodeMode::Batched => {
                 let (reader, size) = self
-                    .encode_batched_with_integrity(reader, writers, quorum, integrity.as_mut())
+                    .encode_batched_with_integrity(reader, writers, write_policy, integrity.as_mut())
                     .await?;
                 (reader, size, None)
             }
@@ -745,7 +787,7 @@ impl Erasure {
         self: Arc<Self>,
         mut reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         require_single_block: bool,
         size_hint: usize,
         integrity: Option<&mut IntegrityBuilder>,
@@ -780,7 +822,7 @@ impl Erasure {
         }
 
         let block = self.encode_data_owned_block(buf)?;
-        let mut mw = MultiWriter::new(writers, quorum);
+        let mut mw = MultiWriter::new(writers, write_policy);
         mw.integrity = integrity;
         mw.write_block(&block).await?;
         mw.shutdown().await?;
@@ -847,7 +889,7 @@ impl Erasure {
         R: AsyncRead + Send + Sync + Unpin + 'static,
     {
         let use_bytesmut_ingest = use_bytesmut_ingest();
-        self.encode_with_ingest_mode(reader, writers, quorum, use_bytesmut_ingest, None)
+        self.encode_with_ingest_mode(reader, writers, ShardWritePolicy::object(quorum), use_bytesmut_ingest, None)
             .await
     }
 
@@ -859,7 +901,7 @@ impl Erasure {
         self: Arc<Self>,
         mut reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         use_bytesmut_ingest: bool,
         integrity: Option<&mut IntegrityBuilder>,
     ) -> std::io::Result<(R, usize)>
@@ -961,7 +1003,7 @@ impl Erasure {
             Ok((reader, total))
         }));
 
-        let mut writers = MultiWriter::new(writers, quorum);
+        let mut writers = MultiWriter::new(writers, write_policy);
         writers.integrity = integrity;
 
         let mut write_err = None;
@@ -1017,14 +1059,15 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin + 'static,
     {
-        self.encode_batched_with_integrity(reader, writers, quorum, None).await
+        self.encode_batched_with_integrity(reader, writers, ShardWritePolicy::object(quorum), None)
+            .await
     }
 
     async fn encode_batched_with_integrity<R>(
         self: Arc<Self>,
         mut reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         integrity: Option<&mut IntegrityBuilder>,
     ) -> std::io::Result<(R, usize)>
     where
@@ -1105,7 +1148,7 @@ impl Erasure {
             Ok((reader, total))
         }));
 
-        let mut writers = MultiWriter::new(writers, quorum);
+        let mut writers = MultiWriter::new(writers, write_policy);
         writers.integrity = integrity;
         let mut write_err = None;
 
@@ -1165,7 +1208,7 @@ impl Erasure {
         R: AsyncRead + Send + Sync + Unpin,
     {
         let size_hint = self.block_size;
-        self.encode_small_direct(reader, writers, quorum, false, size_hint, None)
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), false, size_hint, None)
             .await
     }
 
@@ -1182,7 +1225,7 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin,
     {
-        self.encode_small_direct(reader, writers, quorum, false, size_hint, None)
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), false, size_hint, None)
             .await
     }
 
@@ -1199,7 +1242,8 @@ impl Erasure {
         R: AsyncRead + Send + Sync + Unpin,
     {
         let size_hint = self.block_size;
-        self.encode_small_direct(reader, writers, quorum, true, size_hint, None).await
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), true, size_hint, None)
+            .await
     }
 
     /// Size-aware single-block fast path. `size_hint` only controls the bounded
@@ -1215,7 +1259,8 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin,
     {
-        self.encode_small_direct(reader, writers, quorum, true, size_hint, None).await
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), true, size_hint, None)
+            .await
     }
 }
 
@@ -1632,6 +1677,219 @@ mod tests {
         }
     }
 
+    #[derive(Clone, Copy, Debug)]
+    enum DelayedPhase {
+        Write,
+        Flush,
+        Shutdown,
+    }
+
+    struct PhaseDelayedWriter {
+        inner: DeferredCommitWriter,
+        phase: DelayedPhase,
+        delay: Duration,
+        timer: Option<Pin<Box<tokio::time::Sleep>>>,
+        ready: bool,
+    }
+
+    impl PhaseDelayedWriter {
+        fn poll_delay(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+            if self.ready {
+                return Poll::Ready(());
+            }
+            let timer = self.timer.get_or_insert_with(|| Box::pin(tokio::time::sleep(self.delay)));
+            match timer.as_mut().poll(cx) {
+                Poll::Pending => Poll::Pending,
+                Poll::Ready(()) => {
+                    self.ready = true;
+                    Poll::Ready(())
+                }
+            }
+        }
+    }
+
+    impl AsyncWrite for PhaseDelayedWriter {
+        fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bytes: &[u8]) -> Poll<std::io::Result<usize>> {
+            if matches!(self.phase, DelayedPhase::Write) && self.poll_delay(cx).is_pending() {
+                return Poll::Pending;
+            }
+            Pin::new(&mut self.inner).poll_write(cx, bytes)
+        }
+
+        fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            if matches!(self.phase, DelayedPhase::Flush) && self.poll_delay(cx).is_pending() {
+                return Poll::Pending;
+            }
+            Pin::new(&mut self.inner).poll_flush(cx)
+        }
+
+        fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            if matches!(self.phase, DelayedPhase::Shutdown) && self.poll_delay(cx).is_pending() {
+                return Poll::Pending;
+            }
+            Pin::new(&mut self.inner).poll_shutdown(cx)
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn multipart_budget_keeps_slow_writers_in_every_encode_mode() {
+        const BLOCK: usize = 64;
+        for make_mode in [
+            || IntegrityEncodeMode::SingleBlock(BLOCK),
+            || IntegrityEncodeMode::Streaming,
+            || IntegrityEncodeMode::Batched,
+        ] {
+            for phase in [DelayedPhase::Write, DelayedPhase::Flush, DelayedPhase::Shutdown] {
+                for delay in [Duration::from_secs(2), Duration::from_secs(5)] {
+                    let committed: Vec<_> = (0..4).map(|_| Arc::new(Mutex::new(Vec::new()))).collect();
+                    let mut writers = committed
+                        .iter()
+                        .enumerate()
+                        .map(|(slot, data)| {
+                            if slot == 0 {
+                                Some(bitrot_writer(
+                                    PhaseDelayedWriter {
+                                        inner: DeferredCommitWriter::new(data.clone()),
+                                        phase,
+                                        delay,
+                                        timer: None,
+                                        ready: false,
+                                    },
+                                    BLOCK / 2,
+                                ))
+                            } else {
+                                Some(bitrot_writer(DeferredCommitWriter::new(data.clone()), BLOCK / 2))
+                            }
+                        })
+                        .collect::<Vec<_>>();
+                    let payload = vec![0x5a; BLOCK];
+                    let erasure = Arc::new(Erasure::new(2, 2, BLOCK));
+                    let started = tokio::time::Instant::now();
+                    erasure
+                        .encode_with_shard_integrity(
+                            Cursor::new(payload.clone()),
+                            &mut writers,
+                            ShardWritePolicy::multipart(3),
+                            1,
+                            make_mode(),
+                            false,
+                        )
+                        .await
+                        .expect("multipart must keep finite slow writers within the existing budget");
+                    assert!(started.elapsed() >= delay, "must await the slow phase: {phase:?}");
+                    assert!(writers.iter().all(Option::is_some), "all four completed shards must survive");
+                    for data in &committed {
+                        let bytes = data.lock().expect("committed bytes").clone();
+                        crate::erasure::coding::bitrot_verify(
+                            Cursor::new(bytes.clone()),
+                            bytes.len(),
+                            BLOCK / 2,
+                            HashAlgorithm::HighwayHash256S,
+                            BLOCK / 2,
+                        )
+                        .await
+                        .expect("each completed shard must retain valid bitrot framing");
+                    }
+                }
+            }
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn multipart_budget_expires_blackholes_without_restarting_deadlines() {
+        for shutdown in [false, true] {
+            for blackholes in [1, 2] {
+                for absolute in [Duration::ZERO, Duration::from_secs(3)] {
+                    let mut writers = (0..4)
+                        .map(|slot| {
+                            if slot < blackholes {
+                                Some(bitrot_writer_plain(
+                                    if shutdown {
+                                        StallingWriter::stalls_on_shutdown()
+                                    } else {
+                                        StallingWriter::stalls_on_write(0)
+                                    },
+                                    64,
+                                ))
+                            } else {
+                                Some(bitrot_writer_plain(DeferredCommitWriter::new(Arc::new(Mutex::new(Vec::new()))), 64))
+                            }
+                        })
+                        .collect::<Vec<_>>();
+                    let started = tokio::time::Instant::now();
+                    let result = {
+                        let mut writer = MultiWriter::with_policy(
+                            &mut writers,
+                            ShardWritePolicy::multipart(3),
+                            WriteProgressPolicy::new(Duration::from_secs(7), absolute),
+                        );
+                        if shutdown {
+                            writer.write(four_shards()).await.expect("buffered writes should complete");
+                            tokio::time::sleep(Duration::from_secs(1)).await;
+                            writer.shutdown().await
+                        } else {
+                            writer.write(four_shards()).await
+                        }
+                    };
+                    assert_eq!(result.is_ok(), blackholes == 1, "only a confirmed quorum may succeed");
+                    let expected = if absolute.is_zero() {
+                        Duration::from_secs(7 + u64::from(shutdown))
+                    } else {
+                        absolute
+                    };
+                    assert_eq!(started.elapsed(), expected, "budget must not restart at quorum or shutdown");
+                    assert!(writers[..blackholes].iter().all(Option::is_none));
+                    assert!(writers[blackholes..].iter().all(Option::is_some));
+                }
+            }
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn multipart_budget_cancellation_drops_pending_writers() {
+        struct DropSignal(Option<oneshot::Sender<()>>);
+        impl Drop for DropSignal {
+            fn drop(&mut self) {
+                if let Some(tx) = self.0.take() {
+                    let _ = tx.send(());
+                }
+            }
+        }
+        impl AsyncWrite for DropSignal {
+            fn poll_write(self: Pin<&mut Self>, _: &mut Context<'_>, bytes: &[u8]) -> Poll<std::io::Result<usize>> {
+                Poll::Ready(Ok(bytes.len()))
+            }
+            fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+            fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Pending
+            }
+        }
+        let (tx, rx) = oneshot::channel();
+        let (entered_tx, entered) = oneshot::channel();
+        let task = tokio::spawn(async move {
+            let mut writers = vec![Some(bitrot_writer_plain(DropSignal(Some(tx)), 64))];
+            writers.extend(
+                (0..3).map(|_| Some(bitrot_writer_plain(DeferredCommitWriter::new(Arc::new(Mutex::new(Vec::new()))), 64))),
+            );
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::multipart(3));
+            writer.write(four_shards()).await.expect("buffered writes should succeed");
+            let _ = entered_tx.send(());
+            writer.shutdown().await
+        });
+        entered.await.expect("write must reach shutdown");
+        tokio::task::yield_now().await;
+        assert!(!task.is_finished());
+        task.abort();
+        assert!(
+            task.await
+                .expect_err("caller cancellation should cancel the task")
+                .is_cancelled()
+        );
+        rx.await.expect("pending writer must be dropped with its owning task");
+    }
+
     /// 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
@@ -1704,12 +1962,16 @@ mod tests {
         let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
 
         let encode = match pipeline {
-            EncodePipeline::Vec => {
-                tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, false, None).await })
-            }
-            EncodePipeline::BytesMut => {
-                tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await })
-            }
+            EncodePipeline::Vec => tokio::spawn(async move {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
+                    .await
+            }),
+            EncodePipeline::BytesMut => tokio::spawn(async move {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+                    .await
+            }),
             EncodePipeline::Batched => tokio::spawn(async move { erasure.encode_batched(reader, &mut writers, 1).await }),
         };
 
@@ -1767,8 +2029,16 @@ mod tests {
         ))];
 
         let result = match pipeline {
-            EncodePipeline::Vec => erasure.encode_with_ingest_mode(reader, &mut writers, 1, false, None).await,
-            EncodePipeline::BytesMut => erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await,
+            EncodePipeline::Vec => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
+                    .await
+            }
+            EncodePipeline::BytesMut => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+                    .await
+            }
             EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, 1).await,
         };
 
@@ -1841,8 +2111,16 @@ mod tests {
         let mut writers = vec![Some(bitrot_writer(FailAfterReaderErrorWriter { reader_failed }, BLOCK_SIZE))];
 
         let result = match pipeline {
-            EncodePipeline::Vec => erasure.encode_with_ingest_mode(reader, &mut writers, 1, false, None).await,
-            EncodePipeline::BytesMut => erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await,
+            EncodePipeline::Vec => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
+                    .await
+            }
+            EncodePipeline::BytesMut => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+                    .await
+            }
             EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, 1).await,
         };
 
@@ -1879,7 +2157,7 @@ mod tests {
         let erasure_for_task = erasure.clone();
         let encode = tokio::spawn(async move {
             erasure_for_task
-                .encode_with_ingest_mode(reader, &mut writers, 1, false, None)
+                .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
                 .await
         });
 
@@ -2048,8 +2326,16 @@ mod tests {
             let erasure = Arc::new(Erasure::new(2, 2, 64));
             let result = match path {
                 "direct" => erasure.encode_single_block_non_inline(reader, &mut writers, 2).await,
-                "vec" => erasure.encode_with_ingest_mode(reader, &mut writers, 2, false, None).await,
-                "bytesmut" => erasure.encode_with_ingest_mode(reader, &mut writers, 2, true, None).await,
+                "vec" => {
+                    erasure
+                        .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(2), false, None)
+                        .await
+                }
+                "bytesmut" => {
+                    erasure
+                        .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(2), true, None)
+                        .await
+                }
                 "batched" => erasure.encode_batched(reader, &mut writers, 2).await,
                 _ => unreachable!(),
             };
@@ -2094,7 +2380,7 @@ mod tests {
     async fn multi_writer_short_write_fails_before_shutdown() {
         let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 32))];
         let err = {
-            let mut writer = MultiWriter::new(&mut writers, 1);
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(1));
             writer
                 .write(vec![Bytes::from_static(b"short-write payload")])
                 .await
@@ -2111,7 +2397,7 @@ mod tests {
     async fn multi_writer_reports_fallback_summary_when_only_offline_writers_remain() {
         let mut writers = vec![None, None];
         let (err, summary) = {
-            let mut writer = MultiWriter::new(&mut writers, 1);
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(1));
             let err = writer
                 .write(vec![Bytes::from_static(b"offline-a"), Bytes::from_static(b"offline-b")])
                 .await
@@ -2131,7 +2417,7 @@ mod tests {
         assert!(summary.contains("required=1"));
 
         let (shutdown_err, summary) = {
-            let mut writer = MultiWriter::new(&mut writers, 1);
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(1));
             let err = writer
                 .shutdown()
                 .await
@@ -2155,7 +2441,7 @@ mod tests {
     async fn multi_writer_reports_quorum_failure_when_quorum_exceeds_writer_count() {
         let committed = Arc::new(Mutex::new(Vec::new()));
         let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed), 32))];
-        let mut writer = MultiWriter::new(&mut writers, 2);
+        let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(2));
 
         let err = writer
             .write(vec![Bytes::from_static(b"quorum impossible")])
@@ -2244,7 +2530,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards())
                 .await
                 .expect("one stalled writer must still satisfy a 3/4 write quorum without hanging");
@@ -2271,7 +2557,7 @@ mod tests {
         let started = tokio::time::Instant::now();
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards())
                 .await
                 .expect("quorum must succeed without waiting out the black-hole stall");
@@ -2304,7 +2590,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards())
                 .await
                 .expect("a straggler inside the grace window must still satisfy the write");
@@ -2329,7 +2615,7 @@ mod tests {
         ];
 
         let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-        let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+        let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
         let err = mw
             .write(four_shards())
             .await
@@ -2358,7 +2644,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards()).await.expect("all writers accept the buffered shard");
             mw.shutdown()
                 .await
@@ -2379,7 +2665,7 @@ mod tests {
         ];
 
         let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-        let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+        let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
         mw.write(four_shards()).await.expect("all writers accept the buffered shard");
         let err = mw
             .shutdown()
@@ -2408,7 +2694,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             for _ in 0..8 {
                 mw.write(four_full_shards(64))
                     .await
@@ -2438,7 +2724,7 @@ mod tests {
             // 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);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             for _ in 0..4 {
                 mw.write(four_full_shards(64))
                     .await
@@ -2583,7 +2869,7 @@ mod tests {
         let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
         let reader = tokio::io::BufReader::new(Cursor::new(payload.clone()));
         let (_reader, written) = erasure
-            .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, true, None)
+            .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(DATA_SHARDS), true, None)
             .await
             .expect("BytesMut ingest path should encode the streaming payload");
 
@@ -2612,7 +2898,9 @@ mod tests {
         let reader = tokio::io::BufReader::new(Cursor::new(vec![0x5a; block_size * 2]));
 
         rustfs_io_metrics::set_put_stage_metrics_enabled(true);
-        let result = erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await;
+        let result = erasure
+            .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+            .await;
         rustfs_io_metrics::set_put_stage_metrics_enabled(false);
 
         let (_reader, written) = result.expect("bytesmut streaming encode should complete");
@@ -3230,12 +3518,12 @@ mod tests {
         let (_reader, total) = match pipeline {
             EncodePipeline::Vec => {
                 erasure
-                    .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, false, None)
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(DATA_SHARDS), false, None)
                     .await
             }
             EncodePipeline::BytesMut => {
                 erasure
-                    .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, true, None)
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(DATA_SHARDS), true, None)
                     .await
             }
             EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, DATA_SHARDS).await,

crates/ecstore/src/erasure/coding/heal.rs

@@ -205,7 +205,7 @@ impl super::Erasure {
         // the survivors. Reconstruction correctness is still guaranteed by the
         // read-side quorum check and parity cross-checks below, which are unaffected.
         let write_quorum = 1;
-        let mut writers = MultiWriter::new(writers, write_quorum);
+        let mut writers = MultiWriter::new(writers, super::encode::ShardWritePolicy::object(write_quorum));
         let read_timeout = get_object_disk_read_timeout();
         let shard_file_size = self.shard_file_size(total_length as i64) as usize;
 

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

@@ -112,6 +112,10 @@ use std::time::Duration;
 use tokio::io::AsyncReadExt;
 use tokio::task::JoinSet;
 
+#[cfg(test)]
+#[path = "multipart_quorum_tests.rs"]
+mod multipart_writer_test_hook;
+
 // The erasure-set width bounds fan-out. Await every opener so errors retain
 // their disk slots and every successful writer remains owned until quorum is checked.
 async fn create_part_writers(
@@ -124,6 +128,13 @@ async fn create_part_writers(
         let Some(disk) = disk else {
             return (None, Some(DiskError::DiskNotFound));
         };
+        #[cfg(test)]
+        if let Some(result) = multipart_writer_test_hook::create_writer(disk, path, length, shard_size).await {
+            return match result {
+                Ok(writer) => (Some(writer), None),
+                Err(error) => (None, Some(error)),
+            };
+        }
         match create_bitrot_writer_with_orig_volume(
             false,
             Some(disk),
@@ -2644,8 +2655,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 SmallWritePath::PipelineBatchedLarge => IntegrityEncodeMode::Batched,
                 SmallWritePath::Inline | SmallWritePath::Pipeline => IntegrityEncodeMode::Streaming,
             };
+            let shard_write_policy = crate::erasure::coding::encode::ShardWritePolicy::multipart(write_quorum);
+            #[cfg(test)]
+            let shard_write_policy = multipart_writer_test_hook::write_policy(&shuffle_disks, write_quorum)
+                .unwrap_or(shard_write_policy);
             let (reader, w_size, _, integrity) = Arc::clone(&erasure)
-                .encode_with_shard_integrity(stream, &mut writers, write_quorum, part_id, mode, protected_upload)
+                .encode_with_shard_integrity(
+                    stream,
+                    &mut writers,
+                    shard_write_policy,
+                    part_id,
+                    mode,
+                    protected_upload,
+                )
                 .await?;
 
             if let Some(stage_start) = encode_stage_start {

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

@@ -0,0 +1,303 @@
+//! Exercise participant preservation through real multipart staging and commit.
+
+use super::*;
+use crate::disk::FileWriter;
+use crate::erasure::coding::CustomWriter;
+use crate::set_disk::ops::object::hermetic_set_disks_support::hermetic_set_disks_isolated;
+use crate::storage_api_contracts::object::ObjectIO as _;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll, Waker};
+use tokio::io::AsyncWrite;
+
+#[derive(Clone, Copy)]
+enum Fault {
+    DelayedShutdown,
+    WriteFailure,
+}
+
+struct Round {
+    fault: Fault,
+    legacy_policy: bool,
+    failed_slot: usize,
+    completed: AtomicUsize,
+    delayed_observed: AtomicBool,
+    delayed_waker: Mutex<Option<Waker>>,
+    delayed_dropped: AtomicBool,
+}
+
+type RegisteredWriter = (Arc<Round>, usize);
+static WRITER_ROUNDS: OnceLock<Mutex<HashMap<String, RegisteredWriter>>> = OnceLock::new();
+
+struct WriterRound {
+    endpoints: Vec<String>,
+    round: Arc<Round>,
+}
+
+impl WriterRound {
+    fn install(disks: &[DiskStore], failed_slot: usize, fault: Fault, legacy_policy: bool) -> Self {
+        let round = Arc::new(Round {
+            fault,
+            legacy_policy,
+            failed_slot,
+            completed: AtomicUsize::new(0),
+            delayed_observed: AtomicBool::new(false),
+            delayed_waker: Mutex::new(None),
+            delayed_dropped: AtomicBool::new(false),
+        });
+        let endpoints = disks.iter().map(|disk| disk.endpoint().to_string()).collect::<Vec<_>>();
+        let mut registry = WRITER_ROUNDS
+            .get_or_init(Mutex::default)
+            .lock()
+            .expect("writer hook registry");
+        for (slot, endpoint) in endpoints.iter().enumerate() {
+            assert!(registry.insert(endpoint.clone(), (round.clone(), slot)).is_none());
+        }
+        Self { endpoints, round }
+    }
+}
+
+pub(super) fn write_policy(
+    disks: &[Option<DiskStore>],
+    quorum: usize,
+) -> Option<crate::erasure::coding::encode::ShardWritePolicy> {
+    let registry = WRITER_ROUNDS.get()?.lock().expect("writer hook registry");
+    disks.iter().flatten().find_map(|disk| {
+        registry.get(&disk.endpoint().to_string()).and_then(|(round, _)| {
+            round
+                .legacy_policy
+                .then(|| crate::erasure::coding::encode::ShardWritePolicy::object(quorum))
+        })
+    })
+}
+
+impl Drop for WriterRound {
+    fn drop(&mut self) {
+        let mut registry = WRITER_ROUNDS
+            .get()
+            .expect("installed writer registry")
+            .lock()
+            .expect("writer hook registry");
+        for endpoint in &self.endpoints {
+            registry.remove(endpoint);
+        }
+    }
+}
+
+pub(super) async fn create_writer(
+    disk: &DiskStore,
+    path: &str,
+    length: i64,
+    shard_size: usize,
+) -> Option<disk::error::Result<BitrotWriterWrapper>> {
+    let registered = WRITER_ROUNDS
+        .get()?
+        .lock()
+        .expect("writer hook registry")
+        .get(&disk.endpoint().to_string())
+        .cloned();
+    let (round, slot) = registered?;
+    let length = usize::try_from(length).expect("the fixture writes known-size parts");
+    let encoded_length = coding::bitrot_shard_file_size(length, shard_size, HashAlgorithm::HighwayHash256S);
+    let inner = match disk
+        .create_file(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, path, encoded_length as i64)
+        .await
+    {
+        Ok(inner) => inner,
+        Err(error) => return Some(Err(error)),
+    };
+    let writer = ControlledWriter {
+        inner,
+        round,
+        slot,
+        delay: None,
+        completed: false,
+    };
+    Some(Ok(BitrotWriterWrapper::new(
+        CustomWriter::new_tokio_writer(Box::new(writer)),
+        shard_size,
+        HashAlgorithm::HighwayHash256S,
+    )))
+}
+
+struct ControlledWriter {
+    inner: FileWriter,
+    round: Arc<Round>,
+    slot: usize,
+    delay: Option<Pin<Box<tokio::time::Sleep>>>,
+    completed: bool,
+}
+
+impl Drop for ControlledWriter {
+    fn drop(&mut self) {
+        if self.slot == self.round.failed_slot && !self.completed {
+            self.round.delayed_dropped.store(true, Ordering::Release);
+        }
+    }
+}
+
+impl AsyncWrite for ControlledWriter {
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bytes: &[u8]) -> Poll<io::Result<usize>> {
+        if self.slot == self.round.failed_slot && matches!(self.round.fault, Fault::WriteFailure) {
+            return Poll::Ready(Err(io::Error::new(io::ErrorKind::BrokenPipe, "injected shard write failure")));
+        }
+        Pin::new(&mut self.inner).poll_write(cx, bytes)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        if self.slot == self.round.failed_slot && matches!(self.round.fault, Fault::DelayedShutdown) {
+            self.round.delayed_observed.store(true, Ordering::Release);
+            // Delay only after three real writers finish. This ensures the old
+            // post-quorum grace expires regardless of filesystem scheduling.
+            {
+                let mut waker = self.round.delayed_waker.lock().expect("delayed writer waker");
+                *waker = Some(cx.waker().clone());
+                if self.round.completed.load(Ordering::Acquire) < 3 {
+                    return Poll::Pending;
+                }
+            }
+            let delay = self
+                .delay
+                .get_or_insert_with(|| Box::pin(tokio::time::sleep(Duration::from_secs(2))));
+            if delay.as_mut().poll(cx).is_pending() {
+                return Poll::Pending;
+            }
+        }
+        let result = Pin::new(&mut self.inner).poll_shutdown(cx);
+        if matches!(result, Poll::Ready(Ok(()))) && !self.completed {
+            self.completed = true;
+            self.round.completed.fetch_add(1, Ordering::AcqRel);
+            if let Some(waker) = self.round.delayed_waker.lock().expect("delayed writer waker").take() {
+                waker.wake();
+            }
+        }
+        result
+    }
+}
+
+async fn upload_two_parts(
+    set: &Arc<SetDisks>,
+    disks: &[DiskStore],
+    bucket: &str,
+    object: &str,
+    fault: Fault,
+    legacy_policy: bool,
+) -> (String, Vec<CompletePart>, Vec<u8>, FileInfo) {
+    for disk in disks {
+        disk.make_volume(bucket).await.expect("bucket volume");
+    }
+    let opts = ObjectOptions::default();
+    let upload = set
+        .new_multipart_upload(bucket, object, &opts)
+        .await
+        .expect("new multipart upload");
+    let mut parts = Vec::new();
+    let mut expected = Vec::new();
+    for (number, size, slot) in [(1, 5 * 1024 * 1024, 2), (2, 4096, 3)] {
+        let payload = vec![number as u8; size];
+        expected.extend_from_slice(&payload);
+        let hook = WriterRound::install(disks, slot, fault, legacy_policy);
+        let mut reader = PutObjReader::from_vec(payload);
+        let part = set
+            .put_object_part(bucket, object, &upload.upload_id, number, &mut reader, &opts)
+            .await
+            .expect("three completed shards must admit each part");
+        match fault {
+            Fault::DelayedShutdown if !legacy_policy => {
+                assert!(hook.round.delayed_observed.load(Ordering::Acquire));
+                assert!(!hook.round.delayed_dropped.load(Ordering::Acquire), "finite slow shard must finish");
+                assert_eq!(hook.round.completed.load(Ordering::Acquire), 4, "all real shard writers finished");
+            }
+            Fault::DelayedShutdown | Fault::WriteFailure => {
+                assert!(hook.round.delayed_dropped.load(Ordering::Acquire), "failed shard must be dropped");
+                assert_eq!(hook.round.completed.load(Ordering::Acquire), 3);
+            }
+        }
+        parts.push(CompletePart {
+            part_num: number,
+            etag: part.etag,
+            ..Default::default()
+        });
+    }
+    let (metadata, _) = set
+        .check_upload_id_exists(bucket, object, &upload.upload_id, true)
+        .await
+        .expect("upload metadata");
+    (upload.upload_id, parts, expected, metadata)
+}
+
+#[tokio::test]
+async fn multipart_budget_keeps_different_finite_shutdown_stragglers_and_completes() {
+    let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+    let bucket = "multipart-finite-shutdown";
+    let object = "object";
+    let (upload, parts, expected, metadata) = upload_two_parts(&set, &disks, bucket, object, Fault::DelayedShutdown, false).await;
+    let data_dir = metadata.data_dir.expect("parts-backed data directory");
+    let erasure =
+        coding::Erasure::try_new(metadata.erasure.data_blocks, metadata.erasure.parity_blocks, metadata.erasure.block_size)
+            .expect("upload erasure layout");
+    let stage = SetDisks::get_upload_id_dir(bucket, object, &upload);
+    for disk in &disks {
+        for (number, size) in [(1, 5 * 1024 * 1024), (2, 4096)] {
+            let path = format!("{stage}/{data_dir}/part.{number}");
+            let bytes = disk
+                .read_all(RUSTFS_META_MULTIPART_BUCKET, &path)
+                .await
+                .expect("every disk has every part");
+            let shard_len = erasure.shard_file_size(size);
+            let encoded_len =
+                coding::bitrot_shard_file_size(shard_len as usize, erasure.shard_size(), HashAlgorithm::HighwayHash256S);
+            assert_eq!(bytes.len(), encoded_len, "complete physical shard: {path}");
+        }
+    }
+    set.clone()
+        .complete_multipart_upload(bucket, object, &upload, parts, &ObjectOptions::default())
+        .await
+        .expect("all parts have a common completed quorum");
+    let mut reader = set
+        .get_object_reader(bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default())
+        .await
+        .expect("read committed object");
+    let mut actual = Vec::new();
+    reader.stream.read_to_end(&mut actual).await.expect("whole object readback");
+    assert_eq!(actual, expected);
+}
+
+#[tokio::test]
+async fn multipart_budget_distinct_failed_shards_do_not_bypass_complete_quorum() {
+    let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+    let bucket = "multipart-distinct-failures";
+    let object = "object";
+    let (upload, parts, _, _) = upload_two_parts(&set, &disks, bucket, object, Fault::WriteFailure, false).await;
+    let error = set
+        .clone()
+        .complete_multipart_upload(bucket, object, &upload, parts, &ObjectOptions::default())
+        .await
+        .expect_err("two common complete disks cannot meet the write quorum of three");
+    assert!(error.is_quorum_error(), "expected complete quorum refusal, got {error:?}");
+    set.check_upload_id_exists(bucket, object, &upload, false)
+        .await
+        .expect("failed completion retains retryable upload state");
+    assert!(
+        set.get_object_info(bucket, object, &ObjectOptions::default()).await.is_err(),
+        "failed completion must not publish an object"
+    );
+}
+
+#[tokio::test]
+async fn multipart_budget_old_grace_loses_common_part_quorum() {
+    let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+    let bucket = "multipart-old-grace";
+    let object = "object";
+    let (upload, parts, _, _) = upload_two_parts(&set, &disks, bucket, object, Fault::DelayedShutdown, true).await;
+    let error = set
+        .clone()
+        .complete_multipart_upload(bucket, object, &upload, parts, &ObjectOptions::default())
+        .await
+        .expect_err("the old per-part grace loses the common completion quorum");
+    assert!(error.is_quorum_error(), "expected complete quorum refusal, got {error:?}");
+}

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

@@ -4562,7 +4562,14 @@ impl SetDisks {
                 SmallWritePath::Pipeline => IntegrityEncodeMode::Streaming,
             };
             let (reader, w_size, inline_shards, integrity) = Arc::clone(&erasure)
-                .encode_with_shard_integrity(stream, &mut writers, write_quorum, 1, mode, protect_write)
+                .encode_with_shard_integrity(
+                    stream,
+                    &mut writers,
+                    coding::encode::ShardWritePolicy::object(write_quorum),
+                    1,
+                    mode,
+                    protect_write,
+                )
                 .await?;
             let encode_elapsed = encode_stage_start.map(|stage_start| stage_start.elapsed());
             let encode_ms = encode_elapsed.map(|elapsed| elapsed.as_millis() as u64).unwrap_or_default();

crates/protocols/src/sftp/dir.rs

@@ -56,6 +56,17 @@ pub(super) fn dot_entries() -> Vec<File> {
     ]
 }
 
+/// Client-visible S3 key for an SFTP sub-directory.
+///
+/// `object_key` is the parsed prefix without a trailing slash. The
+/// returned key ends in `/` so the storage layer can encode it to the
+/// reserved directory marker. Callers must not encode the key first:
+/// `validate_object_key` rejects any client key that already contains
+/// the marker suffix.
+fn directory_client_key(object_key: &str) -> String {
+    format!("{object_key}/")
+}
+
 fn directory_listing_input(
     bucket: &str,
     prefix: &str,
@@ -300,19 +311,19 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         Ok(())
     }
 
-    /// MKDIR for a sub-directory path: write a zero-byte object at
-    /// encode_dir_object(prefix + "/"). The encoding maps "foo/" to
-    /// "foo__XLDIR__", which matches the RustFS marker convention used
-    /// by the S3, Swift, and WebDAV backends.
+    /// MKDIR for a sub-directory path: write a zero-byte object at the
+    /// client-visible directory key (`foo/`). Storage encodes that key to
+    /// the on-disk marker `foo__XLDIR__` after the S3 client-key boundary.
+    /// The encoded suffix is reserved, so PutObject must not be given it.
     pub(super) async fn mkdir_subdir_marker(&self, bucket: &str, object_key: &str) -> Result<(), SftpError> {
-        let marker_key = path::encode_dir_object(&format!("{object_key}/"));
-        self.authorize(&S3Action::PutObject, bucket, Some(&marker_key)).await?;
+        let dir_key = directory_client_key(object_key);
+        self.authorize(&S3Action::PutObject, bucket, Some(&dir_key)).await?;
 
         let body = stream::once(async { Ok::<Bytes, std::io::Error>(Bytes::new()) });
         let streaming = StreamingBlob::wrap(body);
         let input = PutObjectInput::builder()
             .bucket(bucket.to_string())
-            .key(marker_key.clone())
+            .key(dir_key)
             .content_length(Some(0))
             .body(Some(streaming))
             .build()
@@ -333,15 +344,14 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
     }
 
     /// RMDIR for a sub-directory path: validate no objects under the
-    /// prefix, then authorise and delete the __XLDIR__ marker that
-    /// represents the directory.
+    /// prefix, then authorise and delete the trailing-slash directory key.
+    /// Storage encodes that key to the on-disk marker.
     pub(super) async fn rmdir_subdir_marker(&self, bucket: &str, object_key: &str) -> Result<(), SftpError> {
-        let prefix = format!("{object_key}/");
-        self.validate_directory_empty(bucket, &prefix).await?;
+        let dir_key = directory_client_key(object_key);
+        self.validate_directory_empty(bucket, &dir_key).await?;
 
-        let marker_key = path::encode_dir_object(&prefix);
-        self.authorize(&S3Action::DeleteObject, bucket, Some(&marker_key)).await?;
-        self.run_backend("delete_object", self.storage.delete_object(bucket, &marker_key, self.credentials()))
+        self.authorize(&S3Action::DeleteObject, bucket, Some(&dir_key)).await?;
+        self.run_backend("delete_object", self.storage.delete_object(bucket, &dir_key, self.credentials()))
             .await?;
         Ok(())
     }
@@ -837,4 +847,95 @@ mod tests {
             "error-level event must carry the SFTP READDIR failed message, captured: {captured}"
         );
     }
+
+    #[tokio::test]
+    async fn mkdir_subdir_puts_trailing_slash_client_key() {
+        let backend = Arc::new(DummyBackend::new());
+        let mut driver = build_driver(Arc::clone(&backend), TEST_PART_SIZE);
+        let authorized = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let record = Arc::clone(&authorized);
+
+        let status = with_test_auth_override(
+            move |action, bucket, key| {
+                record
+                    .lock()
+                    .expect("auth log")
+                    .push((action.as_str(), bucket.to_string(), key.map(str::to_string)));
+                true
+            },
+            driver.mkdir(1, "/bucket/photos/2024".to_string(), russh_sftp::protocol::FileAttributes::default()),
+        )
+        .await
+        .expect("mkdir of an ordinary directory");
+
+        assert!(matches!(status.status_code, StatusCode::Ok));
+        let calls = backend.put_object_calls();
+        assert_eq!(calls.len(), 1, "mkdir must issue one PutObject");
+        assert_eq!(calls[0].bucket, "bucket");
+        assert_eq!(calls[0].key, "photos/2024/");
+        assert!(
+            !calls[0].key.contains(rustfs_utils::path::GLOBAL_DIR_SUFFIX),
+            "the client key must not carry the reserved directory-marker suffix"
+        );
+        let auth = authorized.lock().expect("auth log");
+        assert_eq!(
+            auth.as_slice(),
+            [("s3:PutObject", "bucket".to_string(), Some("photos/2024/".to_string()))]
+        );
+    }
+
+    #[tokio::test]
+    async fn rmdir_subdir_deletes_trailing_slash_client_key() {
+        let backend = Arc::new(DummyBackend::new());
+        backend.queue_delete_object_ok();
+        let mut driver = build_driver(Arc::clone(&backend), TEST_PART_SIZE);
+        let authorized = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let record = Arc::clone(&authorized);
+
+        let status = with_test_auth_override(
+            move |action, bucket, key| {
+                record
+                    .lock()
+                    .expect("auth log")
+                    .push((action.as_str(), bucket.to_string(), key.map(str::to_string)));
+                true
+            },
+            driver.rmdir(2, "/bucket/photos/2024".to_string()),
+        )
+        .await
+        .expect("rmdir of an empty directory");
+
+        assert!(matches!(status.status_code, StatusCode::Ok));
+        let calls = backend.delete_object_calls();
+        assert_eq!(calls.len(), 1, "rmdir must issue one DeleteObject");
+        assert_eq!(calls[0].bucket, "bucket");
+        assert_eq!(calls[0].key, "photos/2024/");
+        assert!(
+            !calls[0].key.contains(rustfs_utils::path::GLOBAL_DIR_SUFFIX),
+            "the client key must not carry the reserved directory-marker suffix"
+        );
+        let auth = authorized.lock().expect("auth log");
+        assert_eq!(
+            auth.as_slice(),
+            [
+                ("s3:ListBucket", "bucket".to_string(), Some("photos/2024/".to_string())),
+                ("s3:DeleteObject", "bucket".to_string(), Some("photos/2024/".to_string())),
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn mkdir_rejects_reserved_directory_marker_before_put() {
+        let backend = Arc::new(DummyBackend::new());
+        let mut driver = build_driver(Arc::clone(&backend), TEST_PART_SIZE);
+        let path = format!("/bucket/sub{}", rustfs_utils::path::GLOBAL_DIR_SUFFIX);
+
+        let err = driver
+            .mkdir(3, path, russh_sftp::protocol::FileAttributes::default())
+            .await
+            .expect_err("a client path containing the reserved marker must be rejected");
+
+        assert_eq!(StatusCode::from(err), StatusCode::BadMessage);
+        assert!(backend.put_object_calls().is_empty(), "rejection must happen before PutObject");
+    }
 }

crates/protocols/src/sftp/driver.rs

@@ -831,14 +831,10 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
 
     /// SSH_FXP_MKDIR, SFTP Internet Draft section 6.6. Bucket-level path
     /// (only the bucket component is set) issues CreateBucket. Sub-bucket
-    /// path issues PutObject of a zero-byte object at the encoded
-    /// directory marker key. MKDIR at the SFTP root returns Failure
-    /// because there is no parent into which a new top-level entity could
-    /// be added.
-    ///
-    /// The directory-marker key is built with rustfs_utils::path::
-    /// encode_dir_object so the key format matches the convention used
-    /// by the rest of RustFS (S3, Swift, WebDAV).
+    /// path issues PutObject of a zero-byte object at the logical prefix
+    /// plus a trailing slash. Storage encodes that client key into the
+    /// directory marker. MKDIR at the SFTP root returns Failure because
+    /// there is no parent into which a new top-level entity could be added.
     #[tracing::instrument(level = "info", skip(self, _attrs), fields(id, path = %sanitise_control_bytes(&path)), err(Debug))]
     async fn mkdir(&mut self, id: u32, path: String, _attrs: FileAttributes) -> Result<Status, Self::Error> {
         self.enforce_server_readonly()?;
@@ -858,8 +854,9 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
     /// SSH_FXP_RMDIR, SFTP Internet Draft section 6.6. Empty check then
     /// delete. Bucket-level path lists the bucket with max_keys=1 and,
     /// on an empty result, calls DeleteBucket. Sub-bucket path lists the
-    /// prefix and, on an empty result, calls DeleteObject on the encoded
-    /// directory marker.
+    /// prefix and, on an empty result, calls DeleteObject on the
+    /// trailing-slash directory key. Storage encodes that key to the
+    /// on-disk marker.
     ///
     /// validate_directory_empty propagates the list error rather than
     /// swallowing it. Without that, a transient backend error during

docs/operations/drive-timeout-tuning.md

@@ -24,6 +24,12 @@ may therefore remain undetected until real I/O reaches it. Missing slots,
 invalidated identities, unhealthy drives, and remote endpoints still get probed
 for recovery. Scanner, heal, and other background services have separate controls.
 
+## Multipart shard completion
+
+UploadPart waits for every still-pending shard operation to finish or reach its existing `RUSTFS_OBJECT_DISK_WRITE_STALL_TIMEOUT` budget (default 30 seconds), bounded further by the remaining `RUSTFS_OBJECT_DISK_WRITE_ABSOLUTE_CAP` when configured. Reaching write quorum does not start the one-second straggler cutoff used by ordinary PUT. This keeps finite slow shards from being discarded on different disks for different parts, which could leave too few disks carrying the entire object at CompleteMultipartUpload.
+
+This does not require every disk to succeed, count an unfinished writer as committed, or reduce the final object's write quorum. Failed and expired writers remain excluded. The tradeoff is that a black-hole peer can delay each part up to the existing operation budget rather than one second after quorum. The stall budget resets per shard-write operation; the optional absolute cap belongs to one part's encoding instance, not the entire multipart upload. Setting both budgets to zero retains the existing unbounded-wait behavior. Actual failures can still leave different parts missing on different disks, in which case completion must fail if a common complete write quorum is unavailable.
+
 ## Background: what these timeouts bound
 
 Every foreground drive operation carries a liveness budget so a hung disk fails

docs/operations/rolling-restart.md

@@ -22,6 +22,16 @@ Each row is a feature whose activation makes older binaries unable to read what
 - Rolling restart (no downtime): restart one node at a time, wait for the restarted node to report `200` on `/health/ready`, and then wait until the other nodes no longer list its drives under `details.storage.unavailableDrives` before touching the next one. The remaining nodes keep serving traffic.
 - Sequential cold start (several nodes down): nodes started before the cluster has quorum come up in degraded mode. The process stays alive, answers `503` with the blocking reason, and recovers automatically as soon as enough peers are online. Do not restart-loop them; keep starting the remaining nodes.
 
+## Returning a recovered node to a load balancer
+
+Probe each backend directly on its S3 listener, using its configured TLS verification when HTTPS is enabled. Require HTTP `200` from `/health/ready` (or `/minio/health/ready`) before routing object requests to that backend. For a workload that also requires write availability, additionally require HTTP `200` from `/minio/health/cluster` on that backend. A read-only workload can use node readiness alone; `/minio/health/cluster/read` has a more conservative lock check and adds no startup-publication guarantee. See the [probe semantics](../architecture/readiness-matrix.md#probe-semantics).
+
+Cluster write/read probes can return `200` while a restarted node still returns `503` for object requests: quorum observations can succeed before startup publishes `FullReady`, including while startup waits for peer re-admission. The node readiness aliases remain `503` during that interval, with `startup_finalization_pending` in `degradedReasons`. Liveness and cluster probes alone are therefore insufficient for backend admission. This also applies when every node's cluster probe succeeds.
+
+Readiness does not certify that healing has completed for every object. The startup peer re-admission wait is bounded and fails open; before stopping another node, also follow the survivor checks in the [rolling restart procedure](#rolling-restart-procedure). No health endpoint is an atomic guarantee that every subsequent object read or write will succeed. The [storage detail contract](../architecture/readiness-matrix.md#storage-detail-contract) describes these sampling limits.
+
+Clients should use bounded retries with backoff and jitter for transient `503` responses. Honor `Retry-After` when present, within the request's retry/deadline budget; it is a suggested delay, not a recovery deadline. The startup admission gate sends `Retry-After: 5` and `x-rustfs-readiness-pending` naming the pending dependency. Other `503` paths can use different headers, so inspect the actual response. Retrying a GET is safe; retrying a write with an ambiguous outcome requires accounting for versioning, concurrent writers, and request preconditions. For recovery validation, retain the first failed response and use distinct payloads or version IDs for overwrites so a successful retry cannot hide a failed assertion or identify the wrong version as durable.
+
 ## Why a single node cannot serve alone
 
 Erasure coding shards every object, including internal metadata such as IAM users, groups, and policies under `.rustfs.sys`, across the drives of a set. Reading an object back needs a read quorum of shards online. With the drives of one set spread over several nodes, one node alone can never satisfy the read quorum; this is a property of erasure coding, not a bug. The cluster becomes readable once enough nodes are up (for internal configuration objects, written with maximum parity, typically about half the nodes of a set).
@@ -59,6 +69,10 @@ On Kubernetes the dial itself is the slow part. A replaced pod gets a new IP, bu
 
 After a distributed node reaches local readiness, it asks the peers in every erasure set it serves how they currently see its drives. It uses the `PeerDriveAdmission` internode RPC, and each peer answers with the same predicate its own `/health/ready` uses for `unavailableDrives`. Startup readiness is published once a strict majority of each set's peers count all of this node's drives in that set as online. In a three-node set that means both peers. While the node waits, `/health/ready` stays `503` with `startup_finalization_pending` and the log shows `event="startup_peer_readmission" state="waiting"`.
 
+Both local dependency checks and peer re-admission rounds are bounded by the startup budget, including in-flight observations and polling delays. An exhausted peer-readmission budget starts no new probe round. Within a round, startup proceeds as soon as every shared set has its required peer majority; pending surplus read-only RPCs are cancelled. Votes are never carried across rounds. The unreachable-peer grace also bounds in-flight rounds: enough reachable replies clear an earlier shortfall before unrelated peers finish, and only a completed round can establish a new unreachable shortfall.
+
+After deferred IAM initialization succeeds, recovery retains that IAM owner while preparing the topology and claiming the application context once. A later local-readiness timeout retries publication with the prepared context; it does not reinitialize IAM or reinstall the context.
+
 The wait always ends in publication:
 
 - `result="admitted"`: the peers re-admitted the node.
@@ -93,7 +107,7 @@ When the whole cluster (or several nodes) went down and nodes are brought back o
 
 | Variable | Default | Effect |
 | --- | --- | --- |
-| `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS` | `120` (`DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS`) | How long startup waits for full readiness before continuing in degraded mode with background recovery. Raising it delays the listener during genuinely slow starts; lowering it surfaces degraded mode sooner. Recovery retries continue regardless of this limit. |
+| `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS` | `120` (`DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS`) | Budget per readiness-publication attempt for local readiness and subsequent peer re-admission, including in-flight checks and polling delays. Local timeout returns an error to the startup/recovery caller; peer re-admission timeout publishes readiness with a warning. Raising it can delay S3 admission; the HTTP listeners are already bound. |
 | `RUSTFS_STARTUP_PEER_READMISSION_ENABLE` | `true` (`DEFAULT_STARTUP_PEER_READMISSION_ENABLE`) | Hold startup readiness until a strict majority of each erasure set's peers count this node's drives online. The wait is bounded by `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS` and fails open. Set `false` to publish on this node's own view only. |
 | `RUSTFS_DRIVE_RETURNING_PROBE_INTERVAL_SECS` | `2` (`DEFAULT_DRIVE_RETURNING_PROBE_INTERVAL_SECS`) | Interval between recovery probes to a remote drive that is `offline` or `returning`. |
 | `RUSTFS_DRIVE_RETURNING_SUCCESS_THRESHOLD` | `3` (`DEFAULT_DRIVE_RETURNING_SUCCESS_THRESHOLD`) | Consecutive successful probes before an `offline` remote drive is `online` again. Together with the interval this bounds how long survivors keep a restarted peer out after they can dial it. |

rustfs/src/server/layer.rs

@@ -4574,66 +4574,112 @@ mod tests {
             async {
                 let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO));
                 let readiness = Arc::new(GlobalReadiness::new());
+                readiness.mark_stage(rustfs_common::SystemStage::IamReady);
                 let inner = CountingHybridService::default();
                 let calls = inner.calls();
-                let mut service = public_health_layer_with_tracker_and_readiness(object_traffic_health, Arc::clone(&readiness))
-                    .await
-                    .layer(inner);
+                let public_health =
+                    public_health_layer_with_tracker_and_readiness(object_traffic_health, Arc::clone(&readiness)).await;
+                let mut service =
+                    crate::server::ReadinessGateLayer::new(Arc::clone(&readiness)).layer(public_health.layer(inner));
 
-                let response = service
-                    .call(
-                        Request::builder()
-                            .method(Method::GET)
-                            .uri(HEALTH_READY_PATH)
-                            .body(Full::<Bytes>::from(Bytes::new()))
-                            .expect("readiness request before admission publication"),
-                    )
-                    .await
-                    .expect("readiness response before admission publication");
-                assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
-                let body = BodyExt::collect(response.into_body())
-                    .await
-                    .expect("readiness body before admission publication")
-                    .to_bytes();
-                let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON");
-                assert_eq!(payload["ready"], false);
-                assert_eq!(payload["details"]["storage"]["ready"], true);
-                assert_eq!(payload["details"]["iam"]["ready"], true);
-                assert_eq!(payload["details"]["lock"]["ready"], true);
-                assert_eq!(payload["degradedReasons"], serde_json::json!(["startup_finalization_pending"]));
+                // Cluster quorum can recover before this node publishes S3 admission.
+                for method in [Method::GET, Method::HEAD] {
+                    for path in [
+                        MINIO_HEALTH_CLUSTER_PATH,
+                        MINIO_HEALTH_CLUSTER_READ_PATH,
+                        HEALTH_COMPAT_LIVE_PATH,
+                    ] {
+                        let response = service
+                            .call(
+                                Request::builder()
+                                    .method(method.clone())
+                                    .uri(path)
+                                    .body(Full::<Bytes>::from(Bytes::new()))
+                                    .expect("probe request before admission publication"),
+                            )
+                            .await
+                            .expect("probe response before admission publication");
+                        assert_eq!(response.status(), StatusCode::OK, "{method} {path}");
+                        let body = BodyExt::collect(response.into_body()).await.expect("probe body").to_bytes();
+                        if method == Method::HEAD {
+                            assert!(body.is_empty(), "HEAD {path}");
+                        } else {
+                            let payload: serde_json::Value = serde_json::from_slice(&body).expect("probe JSON");
+                            if path == HEALTH_COMPAT_LIVE_PATH {
+                                assert_eq!(payload["status"], "ok");
+                                assert!(payload.get("ready").is_none());
+                            } else {
+                                assert_eq!(payload["ready"], true, "{path}");
+                            }
+                        }
+                    }
+                }
 
-                let response = service
-                    .call(
-                        Request::builder()
-                            .method(Method::GET)
-                            .uri(HEALTH_COMPAT_LIVE_PATH)
-                            .body(Full::<Bytes>::from(Bytes::new()))
-                            .expect("liveness request before admission publication"),
-                    )
-                    .await
-                    .expect("liveness response before admission publication");
-                assert_eq!(response.status(), StatusCode::OK);
-                let body = BodyExt::collect(response.into_body())
-                    .await
-                    .expect("liveness body before admission publication")
-                    .to_bytes();
-                let payload: serde_json::Value = serde_json::from_slice(&body).expect("liveness JSON");
-                assert_eq!(payload["status"], "ok");
-                assert!(payload.get("ready").is_none());
-
-                readiness.mark_stage(rustfs_common::SystemStage::FullReady);
-                let response = service
-                    .call(
-                        Request::builder()
-                            .method(Method::HEAD)
-                            .uri(MINIO_HEALTH_READY_PATH)
-                            .body(Full::<Bytes>::from(Bytes::new()))
-                            .expect("readiness request after admission publication"),
-                    )
-                    .await
-                    .expect("readiness response after admission publication");
-                assert_eq!(response.status(), StatusCode::OK);
-                assert_eq!(calls.load(Ordering::SeqCst), 0);
+                for published in [false, true] {
+                    if published {
+                        readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+                    }
+                    for method in [Method::GET, Method::HEAD] {
+                        for path in [HEALTH_READY_PATH, MINIO_HEALTH_READY_PATH] {
+                            let response = service
+                                .call(
+                                    Request::builder()
+                                        .method(method.clone())
+                                        .uri(path)
+                                        .body(Full::<Bytes>::from(Bytes::new()))
+                                        .expect("node readiness request"),
+                                )
+                                .await
+                                .expect("node readiness response");
+                            let expected = if published {
+                                StatusCode::OK
+                            } else {
+                                StatusCode::SERVICE_UNAVAILABLE
+                            };
+                            assert_eq!(response.status(), expected, "{method} {path}, published={published}");
+                            let body = BodyExt::collect(response.into_body())
+                                .await
+                                .expect("readiness body")
+                                .to_bytes();
+                            if method == Method::HEAD {
+                                assert!(body.is_empty(), "HEAD {path}");
+                            } else {
+                                let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON");
+                                assert_eq!(payload["ready"], published);
+                                for component in ["storage", "iam", "lock"] {
+                                    assert_eq!(payload["details"][component]["ready"], true, "{component}");
+                                }
+                                let reasons = if published {
+                                    serde_json::json!([])
+                                } else {
+                                    serde_json::json!(["startup_finalization_pending"])
+                                };
+                                assert_eq!(payload["degradedReasons"], reasons);
+                            }
+                        }
+                    }
+                    assert_eq!(calls.load(Ordering::SeqCst), 0, "health probes bypass the inner service");
+                    for method in [Method::GET, Method::PUT] {
+                        let response = service
+                            .call(
+                                Request::builder()
+                                    .method(method.clone())
+                                    .uri("/bucket/object")
+                                    .body(Full::<Bytes>::from(Bytes::new()))
+                                    .expect("object request"),
+                            )
+                            .await
+                            .expect("object admission response");
+                        if published {
+                            assert_eq!(response.status(), StatusCode::IM_A_TEAPOT, "{method} reaches the inner service");
+                        } else {
+                            assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE, "{method} stays gated");
+                            assert_eq!(response.headers()[http::header::RETRY_AFTER], "5");
+                            assert_eq!(response.headers()["x-rustfs-readiness-pending"], "startup_finalization");
+                        }
+                    }
+                    assert_eq!(calls.load(Ordering::SeqCst), if published { 2 } else { 0 });
+                }
             },
         )
         .await;

rustfs/src/server/mod.rs

@@ -92,6 +92,8 @@ pub(crate) use readiness::collect_node_readiness_report;
 pub(crate) use readiness::observe_peer_drive_admission;
 pub use readiness::publish_ready_when_runtime_ready;
 pub(crate) use readiness::snapshot_dependency_readiness_report;
+#[cfg(test)]
+pub(crate) use readiness::wait_for_runtime_readiness_with;
 pub(crate) use readiness::{collect_cluster_read_health_report, collect_cluster_write_health_report};
 pub(crate) use ssec_transport::NativeTlsTransport;
 

rustfs/src/server/readiness.rs

@@ -75,6 +75,9 @@ fn startup_runtime_readiness_max_wait() -> Duration {
         Duration::from_secs(secs)
     }
 }
+const EVENT_STARTUP_RUNTIME_READINESS: &str = "startup_runtime_readiness";
+const LOG_COMPONENT_SERVER: &str = "server";
+const LOG_SUBSYSTEM_READINESS: &str = "readiness";
 const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready";
 const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total";
 const METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL: &str = "rustfs_pool_metadata_check_timeouts_total";
@@ -1273,6 +1276,15 @@ async fn collect_lock_quorum_observation_uncached() -> LockQuorumObservation {
     }
 }
 
+/// Wait for local runtime dependencies without exceeding the observation budget.
+///
+/// The budget covers both observations and polling sleeps. An immediately ready
+/// observation can succeed with a zero budget. Cancellation drops the in-flight
+/// observation; only success invokes `on_ready`.
+///
+/// # Errors
+/// Returns an error on expiry, retaining the last completed observation when
+/// available. The readiness callback is never invoked after a timeout.
 pub async fn wait_for_runtime_readiness_with<F, Fut, ReadyFn>(
     max_wait: Duration,
     poll_interval: Duration,
@@ -1285,35 +1297,60 @@ where
     ReadyFn: FnMut(DependencyReadiness),
 {
     let startup_deadline = tokio::time::Instant::now() + max_wait;
-
-    loop {
-        let readiness = load_readiness().await;
-        if readiness.storage_ready && readiness.iam_ready && readiness.lock_quorum_ready && readiness.peer_health_ready {
-            on_ready(readiness);
-            return Ok(());
-        }
-
-        if tokio::time::Instant::now() >= startup_deadline {
-            let reason = format!(
-                "startup readiness timed out after {}s: storage_ready={}, iam_ready={}, lock_quorum_ready={}, peer_health_ready={}",
-                max_wait.as_secs(),
-                readiness.storage_ready,
-                readiness.iam_ready,
-                readiness.lock_quorum_ready,
-                readiness.peer_health_ready
+    let mut last_readiness = None;
+    let result = tokio::time::timeout_at(startup_deadline, async {
+        loop {
+            // Tokio timeouts poll the observation first. Keep the initial
+            // zero-budget observation, but start no later one at expiry.
+            if last_readiness.is_some() && tokio::time::Instant::now() >= startup_deadline {
+                return None;
+            }
+            let readiness = load_readiness().await;
+            last_readiness = Some(readiness);
+            if readiness.storage_ready
+                && readiness.iam_ready
+                && readiness.lock_quorum_ready
+                && readiness.peer_health_ready
+                && (max_wait.is_zero() || tokio::time::Instant::now() < startup_deadline)
+            {
+                return Some(readiness);
+            }
+            if tokio::time::Instant::now() >= startup_deadline {
+                return None;
+            }
+            info!(
+                event = EVENT_STARTUP_RUNTIME_READINESS,
+                component = LOG_COMPONENT_SERVER,
+                subsystem = LOG_SUBSYSTEM_READINESS,
+                state = "waiting",
+                storage_ready = readiness.storage_ready,
+                iam_ready = readiness.iam_ready,
+                lock_quorum_ready = readiness.lock_quorum_ready,
+                peer_health_ready = readiness.peer_health_ready,
+                "Runtime node readiness has not been reached yet; delaying ready state publication"
             );
-            return Err(std::io::Error::other(reason));
+            tokio::time::sleep(poll_interval).await;
+        }
+    })
+    .await;
+    match result {
+        Ok(Some(readiness)) => {
+            on_ready(readiness);
+            Ok(())
+        }
+        Ok(None) | Err(_) => {
+            let detail = match last_readiness {
+                Some(readiness) => format!(
+                    "storage_ready={}, iam_ready={}, lock_quorum_ready={}, peer_health_ready={}",
+                    readiness.storage_ready, readiness.iam_ready, readiness.lock_quorum_ready, readiness.peer_health_ready
+                ),
+                None => "no readiness observation completed".to_string(),
+            };
+            Err(std::io::Error::other(format!(
+                "startup readiness timed out after {}s: {detail}",
+                max_wait.as_secs()
+            )))
         }
-
-        info!(
-            target: "rustfs::server::readiness",
-            storage_ready = readiness.storage_ready,
-            iam_ready = readiness.iam_ready,
-            lock_quorum_ready = readiness.lock_quorum_ready,
-            peer_health_ready = readiness.peer_health_ready,
-            "Runtime node readiness has not been reached yet; delaying ready state publication"
-        );
-        tokio::time::sleep(poll_interval).await;
     }
 }
 
@@ -1618,6 +1655,7 @@ mod tests {
     /// predicate counts node-2's drive online, then publish.
     #[tokio::test(start_paused = true)]
     async fn restarted_node_stays_unready_until_survivor_counts_its_drive_online() {
+        use futures::StreamExt;
         use peer_readmission::{
             PeerDriveView, PeerProbe, ReadmissionOutcome, UNREACHABLE_PEER_GRACE, host_drive_observations, readmission_plan,
             wait_for_peer_readmission_with,
@@ -1654,8 +1692,8 @@ mod tests {
         let deadline = tokio::time::Instant::now() + Duration::from_secs(120);
 
         let gate = async {
-            let outcome =
-                wait_for_peer_readmission_with(&plan, deadline, Duration::from_secs(1), UNREACHABLE_PEER_GRACE, || async {
+            let outcome = wait_for_peer_readmission_with(&plan, deadline, Duration::from_secs(1), UNREACHABLE_PEER_GRACE, || {
+                futures::stream::once(async {
                     let mut online_hosts = HashSet::from(["node-0:9000".to_string(), "node-1:9000".to_string()]);
                     if survivor_reaches_restarted.load(Ordering::SeqCst) {
                         online_hosts.insert("node-2:9000".to_string());
@@ -1677,7 +1715,9 @@ mod tests {
                     );
                     HashMap::from([("node-0:9000".to_string(), view.clone()), ("node-1:9000".to_string(), view)])
                 })
-                .await;
+                .flat_map(futures::stream::iter)
+            })
+            .await;
             readiness.mark_stage(rustfs_common::SystemStage::FullReady);
             outcome
         };
@@ -1979,6 +2019,150 @@ mod tests {
         assert_eq!(state_manager.current_state(), ServiceState::Ready);
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn wait_for_runtime_readiness_cancels_a_suspended_observation() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        struct ObservationDrop(Arc<AtomicUsize>);
+        impl Drop for ObservationDrop {
+            fn drop(&mut self) {
+                self.0.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+        let dropped = Arc::new(AtomicUsize::new(0));
+        let published = AtomicUsize::new(0);
+        let started = tokio::time::Instant::now();
+        let wait = wait_for_runtime_readiness_with(
+            Duration::from_secs(2),
+            Duration::from_secs(1),
+            || {
+                let guard = ObservationDrop(Arc::clone(&dropped));
+                async move {
+                    let _guard = guard;
+                    std::future::pending::<DependencyReadiness>().await
+                }
+            },
+            |_| {
+                published.fetch_add(1, Ordering::SeqCst);
+            },
+        );
+        let err = tokio::time::timeout(Duration::from_secs(3), wait)
+            .await
+            .expect("startup budget bounds a suspended dependency observation")
+            .expect_err("no dependency observation completed");
+        assert!(err.to_string().contains("no readiness observation completed"));
+        assert_eq!(started.elapsed(), Duration::from_secs(2));
+        assert_eq!(dropped.load(Ordering::SeqCst), 1);
+        assert_eq!(published.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn wait_for_runtime_readiness_bounds_the_poll_sleep() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let polls = AtomicUsize::new(0);
+        let started = tokio::time::Instant::now();
+        let wait = wait_for_runtime_readiness_with(
+            Duration::from_secs(2),
+            Duration::from_secs(30),
+            || {
+                polls.fetch_add(1, Ordering::SeqCst);
+                future::ready(DependencyReadiness::default())
+            },
+            |_| panic!("an unready observation must never publish"),
+        );
+        let err = tokio::time::timeout(Duration::from_secs(3), wait)
+            .await
+            .expect("startup budget bounds the poll sleep")
+            .expect_err("unready observation");
+        assert_eq!(started.elapsed(), Duration::from_secs(2));
+        assert_eq!(polls.load(Ordering::SeqCst), 1, "expiry must not start another observation");
+        assert!(err.to_string().contains("lock_quorum_ready=false"));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn wait_for_runtime_readiness_publishes_once_before_the_deadline() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let polls = AtomicUsize::new(0);
+        let published = AtomicUsize::new(0);
+        let started = tokio::time::Instant::now();
+        wait_for_runtime_readiness_with(
+            Duration::from_secs(2),
+            Duration::from_secs(1),
+            || {
+                let ready = polls.fetch_add(1, Ordering::SeqCst) > 0;
+                future::ready(DependencyReadiness {
+                    storage_ready: ready,
+                    iam_ready: ready,
+                    lock_quorum_ready: ready,
+                    peer_health_ready: ready,
+                })
+            },
+            |_| {
+                published.fetch_add(1, Ordering::SeqCst);
+            },
+        )
+        .await
+        .expect("dependencies recovered before expiry");
+        assert_eq!(started.elapsed(), Duration::from_secs(1));
+        assert_eq!(published.load(Ordering::SeqCst), 1);
+        assert_eq!(polls.load(Ordering::SeqCst), 2);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn wait_for_runtime_readiness_starts_no_observation_at_expiry() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let polls = AtomicUsize::new(0);
+        let published = AtomicUsize::new(0);
+        let started = tokio::time::Instant::now();
+        wait_for_runtime_readiness_with(
+            Duration::from_secs(2),
+            Duration::from_secs(1),
+            || {
+                let ready = polls.fetch_add(1, Ordering::SeqCst) >= 2;
+                future::ready(DependencyReadiness {
+                    storage_ready: ready,
+                    iam_ready: ready,
+                    lock_quorum_ready: ready,
+                    peer_health_ready: ready,
+                })
+            },
+            |_| {
+                published.fetch_add(1, Ordering::SeqCst);
+            },
+        )
+        .await
+        .expect_err("expiry must prevent a third observation even when immediately ready");
+        assert_eq!(started.elapsed(), Duration::from_secs(2));
+        assert_eq!(polls.load(Ordering::SeqCst), 2);
+        assert_eq!(published.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn wait_for_runtime_readiness_does_not_publish_an_observation_at_expiry() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let published = AtomicUsize::new(0);
+        let started = tokio::time::Instant::now();
+        wait_for_runtime_readiness_with(
+            Duration::from_secs(2),
+            Duration::from_secs(1),
+            || async {
+                tokio::time::sleep(Duration::from_secs(2)).await;
+                DependencyReadiness {
+                    storage_ready: true,
+                    iam_ready: true,
+                    lock_quorum_ready: true,
+                    peer_health_ready: true,
+                }
+            },
+            |_| {
+                published.fetch_add(1, Ordering::SeqCst);
+            },
+        )
+        .await
+        .expect_err("an observation completing at the deadline must not publish");
+        assert_eq!(started.elapsed(), Duration::from_secs(2));
+        assert_eq!(published.load(Ordering::SeqCst), 0);
+    }
+
     #[test]
     fn probe_path_checks_admin_boundaries() {
         assert!(is_probe_path("/minio/admin/v3/info"));

rustfs/src/startup_iam.rs

@@ -85,32 +85,35 @@ where
     publish_ready().await
 }
 
-async fn finalize_iam_recovery(
+async fn prepare_iam_startup(
     iam: Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>,
     store: Arc<ECStore>,
     kms_interface: Arc<KmsServiceManager>,
     readiness: Arc<GlobalReadiness>,
-    state_manager: Option<Arc<ServiceStateManager>>,
     server_ctx: Arc<ServerContextSlot>,
-) -> Result<()> {
+) -> Result<HealTopologyReady> {
     let endpoint_pools = heal_control_endpoint_pools(store.as_ref())?;
-    AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?;
-    let topology_ready =
-        publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools).await?;
-    publish_runtime_ready_after_heal_topology(topology_ready, || async {
-        publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await
+    publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools, || {
+        AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)
     })
     .await
 }
 
-async fn publish_iam_ready_after_heal_topology(
+async fn publish_iam_ready_after_heal_topology<InstallFn>(
     readiness: &GlobalReadiness,
     cache: Arc<tokio::sync::OnceCell<String>>,
     endpoint_pools: crate::storage_api::cluster::EndpointServerPools,
-) -> Result<HealTopologyReady> {
+    install_context: InstallFn,
+) -> Result<HealTopologyReady>
+where
+    InstallFn: FnOnce() -> Result<()>,
+{
     crate::storage_api::startup::heal_control::initialize_heal_topology_fingerprint(cache, endpoint_pools)
         .await
         .map_err(std::io::Error::other)?;
+    // Finish fallible preparation before claiming the one-time context slot.
+    // No fallible/async step may follow a successful claim before this token.
+    install_context()?;
     readiness.mark_stage(SystemStage::IamReady);
     Ok(HealTopologyReady(()))
 }
@@ -145,8 +148,7 @@ fn spawn_iam_recovery_task(
     let finalize_readiness = readiness;
     let finalize_state_manager = state_manager;
     let finalize_server_ctx = server_ctx;
-    let recovered_iam = Arc::new(std::sync::Mutex::new(None));
-    let init_recovered_iam = recovered_iam.clone();
+    let prepare_readiness = finalize_readiness.clone();
     tokio::spawn(async move {
         run_iam_recovery_loop(
             initial_interval,
@@ -154,25 +156,23 @@ fn spawn_iam_recovery_task(
             shutdown_token,
             move || {
                 let store = init_store.clone();
-                let recovered_iam = init_recovered_iam.clone();
-                Box::pin(async move {
-                    let iam = attempt_init_iam_sys(store).await?;
-                    *recovered_iam
-                        .lock()
-                        .map_err(|_| std::io::Error::other("IAM recovery state poisoned"))? = Some(iam);
-                    Ok(())
-                })
+                Box::pin(async move { attempt_init_iam_sys(store).await })
             },
-            move || {
+            move |iam| {
                 let store = finalize_store.clone();
                 let kms_interface = finalize_kms_interface.clone();
+                let readiness = prepare_readiness.clone();
+                let server_ctx = finalize_server_ctx.clone();
+                Box::pin(prepare_iam_startup(iam, store, kms_interface, readiness, server_ctx))
+            },
+            move |topology_ready| {
                 let readiness = finalize_readiness.clone();
                 let state_manager = finalize_state_manager.clone();
-                let server_ctx = finalize_server_ctx.clone();
-                let iam = recovered_iam.lock().ok().and_then(|mut recovered| recovered.take());
                 Box::pin(async move {
-                    let iam = iam.ok_or_else(|| std::io::Error::other("IAM recovered but unavailable"))?;
-                    finalize_iam_recovery(iam, store, kms_interface, readiness, state_manager, server_ctx).await
+                    publish_runtime_ready_after_heal_topology(topology_ready, || async {
+                        publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await
+                    })
+                    .await
                 })
             },
         )
@@ -180,23 +180,27 @@ fn spawn_iam_recovery_task(
     });
 }
 
-type RecoveryFuture = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
+type RecoveryFuture<T = ()> = Pin<Box<dyn Future<Output = Result<T>> + Send>>;
 
-async fn run_iam_recovery_loop<InitFn, FinalizeFn>(
+async fn run_iam_recovery_loop<T, Prepared, InitFn, PrepareFn, FinalizeFn>(
     initial_interval: Duration,
     max_interval: Duration,
     shutdown_token: Option<tokio_util::sync::CancellationToken>,
     mut init_fn: InitFn,
+    mut prepare_fn: PrepareFn,
     mut finalize_fn: FinalizeFn,
 ) where
-    InitFn: FnMut() -> RecoveryFuture,
-    FinalizeFn: FnMut() -> RecoveryFuture,
+    T: Clone,
+    Prepared: Clone,
+    InitFn: FnMut() -> RecoveryFuture<T>,
+    PrepareFn: FnMut(T) -> RecoveryFuture<Prepared>,
+    FinalizeFn: FnMut(Prepared) -> RecoveryFuture,
 {
     let mut attempts: u64 = 0;
     let degraded_since = std::time::Instant::now();
 
-    // Phase 1: retry init until it succeeds
-    loop {
+    // Keep the initialized IAM owner across preparation/publication retries.
+    let recovered = loop {
         attempts += 1;
         let sleep_duration = compute_backoff_interval(attempts, initial_interval, max_interval);
         if let Some(token) = shutdown_token.as_ref() {
@@ -209,7 +213,7 @@ async fn run_iam_recovery_loop<InitFn, FinalizeFn>(
         }
 
         match init_fn().await {
-            Ok(()) => {
+            Ok(recovered) => {
                 let degraded_secs = degraded_since.elapsed().as_secs();
                 info!(
                     event = EVENT_IAM_BOOTSTRAP_RECOVERED,
@@ -219,7 +223,7 @@ async fn run_iam_recovery_loop<InitFn, FinalizeFn>(
                     degraded_duration_secs = degraded_secs,
                     "IAM bootstrap recovered after startup; publishing IAM readiness"
                 );
-                break;
+                break recovered;
             }
             Err(err) => {
                 let next_interval = compute_backoff_interval(attempts + 1, initial_interval, max_interval);
@@ -250,13 +254,24 @@ async fn run_iam_recovery_loop<InitFn, FinalizeFn>(
                 }
             }
         }
-    }
+    };
 
-    // Phase 2: retry finalize until readiness is published
+    // Preparation claims the context once. Only publication retries after it.
+    let mut prepared: Option<Prepared> = None;
     let mut finalize_attempts: u64 = 0;
     loop {
+        let result = match prepared.as_ref() {
+            Some(prepared) => finalize_fn(prepared.clone()).await,
+            None => match prepare_fn(recovered.clone()).await {
+                Ok(value) => {
+                    prepared = Some(value);
+                    continue;
+                }
+                Err(error) => Err(error),
+            },
+        };
         finalize_attempts += 1;
-        match finalize_fn().await {
+        match result {
             Ok(()) => {
                 info!(
                     event = EVENT_IAM_READINESS_PUBLISHED,
@@ -384,11 +399,7 @@ pub(crate) async fn bootstrap_or_defer_iam_init(
 ) -> Result<IamBootstrapDisposition> {
     match attempt_init_iam_sys(store.clone()).await {
         Ok(iam) => {
-            let endpoint_pools = heal_control_endpoint_pools(store.as_ref())?;
-            AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?;
-            let topology_ready =
-                publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools)
-                    .await?;
+            let topology_ready = prepare_iam_startup(iam, store, kms_interface, readiness, server_ctx).await?;
             return Ok(IamBootstrapDisposition::ReadyInline(topology_ready));
         }
         Err(err) => {
@@ -586,7 +597,7 @@ mod tests {
         let expected = crate::storage_api::startup::heal_control::heal_topology_fingerprint(&endpoint_pools)
             .expect("valid topology should hash");
 
-        publish_iam_ready_after_heal_topology(&readiness, Arc::clone(&cache), endpoint_pools)
+        publish_iam_ready_after_heal_topology(&readiness, Arc::clone(&cache), endpoint_pools, || Ok(()))
             .await
             .expect("topology initialization should publish IAM readiness");
 
@@ -599,6 +610,7 @@ mod tests {
             &failed_readiness,
             Arc::clone(&invalid_cache),
             EndpointServerPools::from(Vec::<PoolEndpoints>::new()),
+            || panic!("invalid topology must not claim the context"),
         )
         .await
         .expect_err("invalid topology must block IAM readiness");
@@ -628,7 +640,8 @@ mod tests {
                     Ok(())
                 })
             },
-            move || {
+            |()| Box::pin(async { Ok(()) }),
+            move |_| {
                 let finalize_calls = finalize_calls.clone();
                 Box::pin(async move {
                     let call = finalize_calls.fetch_add(1, Ordering::SeqCst) + 1;
@@ -671,7 +684,8 @@ mod tests {
                     }
                 })
             },
-            move || {
+            |()| Box::pin(async { Ok(()) }),
+            move |_| {
                 let readiness = readiness_for_finalize.clone();
                 let observed_iam_ready = observed_iam_ready_for_finalize.clone();
                 Box::pin(async move {
@@ -691,6 +705,197 @@ mod tests {
         assert!(readiness.is_ready());
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn recovery_publication_timeout_reuses_the_installed_context() {
+        let context = crate::app::gating_test_env::shared_gating_ambient().await;
+        let server_ctx = ServerContextSlot::new();
+        // The legacy fixture exposes its real disks without instance endpoints.
+        let (paths, store) = crate::app::gating_test_env::shared_gating_ecstore_and_disk_paths().await;
+        assert!(Arc::ptr_eq(&store, &context.object_store()));
+        let endpoints = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: paths.len(),
+            endpoints: Endpoints::from(
+                paths
+                    .iter()
+                    .enumerate()
+                    .map(|(index, path)| {
+                        let mut endpoint =
+                            Endpoint::try_from(path.to_str().expect("fixture disk path")).expect("fixture endpoint");
+                        endpoint.set_pool_index(0);
+                        endpoint.set_set_index(0);
+                        endpoint.set_disk_index(index);
+                        endpoint
+                    })
+                    .collect::<Vec<_>>(),
+            ),
+            cmd_line: "recovery-publication-test".to_string(),
+            platform: String::new(),
+        }]);
+        crate::storage_api::startup::heal_control::initialize_heal_topology_fingerprint(
+            server_ctx.heal_topology_fingerprint(),
+            endpoints.clone(),
+        )
+        .await
+        .expect("prepare fixture fingerprint before pausing time");
+        // A fresh paused runtime keeps real fixture I/O and its timer-wheel
+        // rounding out of the exact retry/deadline observations below.
+        tokio::task::spawn_blocking(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .start_paused(true)
+                .build()
+                .expect("paused recovery runtime");
+            runtime.block_on(async move {
+                let readiness = Arc::new(GlobalReadiness::new());
+                let init_calls = Arc::new(AtomicUsize::new(0));
+                let prepare_calls = Arc::new(AtomicUsize::new(0));
+                let publish_calls = Arc::new(AtomicUsize::new(0));
+                let started = tokio::time::Instant::now();
+                let recovery = run_iam_recovery_loop(
+                    Duration::from_secs(5),
+                    Duration::from_secs(30),
+                    None,
+                    {
+                        let context = Arc::clone(&context);
+                        let calls = Arc::clone(&init_calls);
+                        move || {
+                            calls.fetch_add(1, Ordering::SeqCst);
+                            let context = Arc::clone(&context);
+                            Box::pin(async move { Ok(context) })
+                        }
+                    },
+                    {
+                        let server_ctx = Arc::clone(&server_ctx);
+                        let readiness = Arc::clone(&readiness);
+                        let calls = Arc::clone(&prepare_calls);
+                        move |context| {
+                            calls.fetch_add(1, Ordering::SeqCst);
+                            let server_ctx = Arc::clone(&server_ctx);
+                            let readiness = Arc::clone(&readiness);
+                            let endpoints = endpoints.clone();
+                            Box::pin(async move {
+                                publish_iam_ready_after_heal_topology(
+                                    readiness.as_ref(),
+                                    server_ctx.heal_topology_fingerprint(),
+                                    endpoints,
+                                    || server_ctx.try_install(context),
+                                )
+                                .await
+                            })
+                        }
+                    },
+                    {
+                        let readiness = Arc::clone(&readiness);
+                        let calls = Arc::clone(&publish_calls);
+                        move |prepared| {
+                            let attempt = calls.fetch_add(1, Ordering::SeqCst);
+                            let readiness = Arc::clone(&readiness);
+                            Box::pin(async move {
+                                let result = publish_runtime_ready_after_heal_topology(prepared, || async {
+                                    crate::server::wait_for_runtime_readiness_with(
+                                        Duration::from_secs(2),
+                                        Duration::from_secs(1),
+                                        || async move {
+                                            if attempt == 0 {
+                                                std::future::pending::<crate::shared_types::DependencyReadiness>().await
+                                            } else {
+                                                crate::shared_types::DependencyReadiness {
+                                                    storage_ready: true,
+                                                    iam_ready: true,
+                                                    lock_quorum_ready: true,
+                                                    peer_health_ready: true,
+                                                }
+                                            }
+                                        },
+                                        |_| readiness.mark_stage(SystemStage::FullReady),
+                                    )
+                                    .await
+                                })
+                                .await;
+                                if attempt == 0 {
+                                    assert!(result.is_err(), "the first publication must time out");
+                                    assert!(!readiness.is_ready(), "timeout cannot publish FullReady");
+                                }
+                                result
+                            })
+                        }
+                    },
+                );
+                tokio::time::timeout(Duration::from_secs(20), recovery)
+                    .await
+                    .expect("publication retry must succeed without attempting another context installation");
+                assert_eq!(init_calls.load(Ordering::SeqCst), 1);
+                assert_eq!(prepare_calls.load(Ordering::SeqCst), 1);
+                assert_eq!(publish_calls.load(Ordering::SeqCst), 2);
+                assert_eq!(started.elapsed(), Duration::from_secs(12));
+                assert!(readiness.is_ready());
+                assert!(Arc::ptr_eq(&server_ctx.installed_app_context().expect("installed context"), &context));
+                assert_eq!(
+                    server_ctx
+                        .try_install(context)
+                        .expect_err("duplicate installs remain conflicts")
+                        .kind(),
+                    std::io::ErrorKind::AlreadyExists
+                );
+            });
+        })
+        .await
+        .expect("recovery model completes without panic");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn recovery_preparation_retries_reuse_the_initialized_owner() {
+        let owner = Arc::new(AtomicUsize::new(0));
+        let init_calls = Arc::new(AtomicUsize::new(0));
+        let publish_calls = Arc::new(AtomicUsize::new(0));
+        run_iam_recovery_loop(
+            Duration::from_secs(1),
+            Duration::from_secs(2),
+            None,
+            {
+                let owner = Arc::clone(&owner);
+                let calls = Arc::clone(&init_calls);
+                move || {
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    let owner = Arc::clone(&owner);
+                    Box::pin(async move { Ok(owner) })
+                }
+            },
+            {
+                let expected = Arc::clone(&owner);
+                move |owner| {
+                    assert!(Arc::ptr_eq(&expected, &owner), "preparation must keep the same initialized IAM owner");
+                    Box::pin(async move {
+                        if owner.fetch_add(1, Ordering::SeqCst) == 0 {
+                            Err(Error::other("topology not ready"))
+                        } else {
+                            Ok(owner)
+                        }
+                    })
+                }
+            },
+            {
+                let calls = Arc::clone(&publish_calls);
+                move |owner| {
+                    let calls = Arc::clone(&calls);
+                    Box::pin(async move {
+                        assert_eq!(owner.load(Ordering::SeqCst), 2, "publication follows successful preparation");
+                        calls.fetch_add(1, Ordering::SeqCst);
+                        Ok(())
+                    })
+                }
+            },
+        )
+        .await;
+        assert_eq!(init_calls.load(Ordering::SeqCst), 1);
+        assert_eq!(owner.load(Ordering::SeqCst), 2);
+        assert_eq!(publish_calls.load(Ordering::SeqCst), 1);
+        assert_eq!(Arc::strong_count(&owner), 1, "loop releases its initialized and prepared owners");
+    }
+
     #[tokio::test(start_paused = true)]
     async fn recovery_loop_stops_after_shutdown_cancellation() {
         let init_calls = Arc::new(AtomicUsize::new(0));
@@ -711,7 +916,8 @@ mod tests {
                             Err(Error::other("keep retrying"))
                         })
                     },
-                    move || Box::pin(async { Ok(()) }),
+                    |()| Box::pin(async { Ok(()) }),
+                    |()| Box::pin(async { Ok(()) }),
                 )
                 .await;
             }