ci: bump overtrue/cla-bot to v0.1.1 (#9189)

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

.docker/observability/otel-collector-config.yaml

@@ -28,15 +28,6 @@ processors:
     check_interval: 1s
     limit_mib: 1024
     spike_limit_mib: 256
-  transform/metrics:
-    error_mode: ignore
-    metric_statements:
-      # RustFS process metrics already expose process_pid as a metric label.
-      # Remove only the duplicate datapoint label and keep the shared resource
-      # process.pid available to sibling metrics and lock histograms.
-      - context: datapoint
-        statements:
-          - delete_key(attributes, "process_pid") where IsMatch(metric.name, "^rustfs_system_process_")
   transform/logs:
     error_mode: ignore
     log_statements:
@@ -121,7 +112,7 @@ service:
       exporters: [ otlp/tempo, otlp/jaeger ]
     metrics:
       receivers: [ otlp ]
-      processors: [ transform/metrics, memory_limiter, batch ]
+      processors: [ batch ]
       exporters: [ prometheus ]
     logs:
       receivers: [ otlp ]

.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/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/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;
             }