chore: integrate current main for field regression

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=7c705942d3a9e583ebaa85845daa21c4f81fdbf36def24db931aaef38636c53e
-sha256-linux=5c33dc395233671cb77452f8fb12900ecdbeb368632222b96835d310bce1aab8
+sha256-darwin=16b3cd69887a73b468c84012bf5c0cb7e24360fcaf6ab3c13967020e37ef86d2
+sha256-linux=c7c7f0c7ce2610bc0a86a518bfa67a3ed56d862aa44f05b4a92200fabf51d6c7

.config/e2e-nightly-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=3814cdde8cacf4187e18188d342ed70931cc32e268e42e5bb438d565bb4d0f81
-sha256-linux=663025a80273579f95ae3f49af02c0afcbfce94df09701c360a13e607b7637db
+sha256-darwin=e84e04208f19dc0596b10d4d1b799ce43d6cc431415902f9c5458aa8232bc2d8
+sha256-linux=1f9c7add72c44d1c12ef4ac050bb7917431be657c0c90b29f70c3df56f8bbae3

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -33,7 +33,7 @@ mod tests {
     use sha2::{Digest, Sha256};
     use std::collections::HashSet;
     use std::error::Error;
-    use std::io::{Read, Write};
+    use std::io::{Read, Seek, SeekFrom, Write};
     use std::net::SocketAddr;
     use std::path::{Path, PathBuf};
     use std::process::Command;
@@ -1654,6 +1654,89 @@ mod tests {
         TargetEndpointBlackhole,
     }
 
+    async fn release_heal_commit_barrier_after_shutdown(
+        log: &Path,
+        log_offset: u64,
+        barrier: &Path,
+    ) -> Result<serde_json::Value, Box<dyn Error + Send + Sync>> {
+        // Observe within the existing graceful process-stop budget.
+        timeout(Duration::from_secs(45), async {
+            loop {
+                let mut file = std::fs::File::open(log)?;
+                file.seek(SeekFrom::Start(log_offset))?;
+                let mut tail = String::new();
+                file.read_to_string(&mut tail)?;
+                if let Some(event) = tail
+                    .lines()
+                    .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
+                    .find(|event| {
+                        event["event"] == "server_shutdown_state"
+                            && event["component"] == "main"
+                            && event["subsystem"] == "startup"
+                            && event["state"] == "stopping"
+                    })
+                {
+                    // Local heal has stopped and HTTP shutdown has been signalled.
+                    // Let admitted peer RPCs finish; keeping this synthetic barrier
+                    // until process exit would make the fixture block its own drain.
+                    std::fs::remove_file(barrier)?;
+                    return Ok(event);
+                }
+                sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await?
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn heal_commit_barrier_waits_for_new_shutdown_event() -> Result<(), Box<dyn Error + Send + Sync>> {
+        let directory = tempfile::tempdir()?;
+        let log = directory.path().join("server.log");
+        let barrier = directory.path().join("barrier");
+        let event = serde_json::json!({
+            "event": "server_shutdown_state", "component": "main", "subsystem": "startup", "state": "stopping"
+        });
+        std::fs::write(&log, format!("{event}\n"))?;
+        std::fs::write(&barrier, "bucket/prefix/")?;
+        let log_offset = log.metadata()?.len();
+        let mut writer = std::fs::OpenOptions::new().append(true).open(&log)?;
+        writeln!(writer, "{}", serde_json::json!({"event": "shutdown_signal_received"}))?;
+        writeln!(writer, "{}", serde_json::json!({"event": "server_shutdown_state", "state": "stopping"}))?;
+
+        let release = release_heal_commit_barrier_after_shutdown(&log, log_offset, &barrier);
+        tokio::pin!(release);
+        assert!(
+            futures::poll!(release.as_mut()).is_pending(),
+            "old or unrelated events must not release the barrier"
+        );
+        assert!(barrier.is_file(), "the partial-rebuild barrier must remain armed before shutdown");
+        tokio::time::advance(Duration::from_secs(6)).await;
+        assert!(
+            futures::poll!(release.as_mut()).is_pending(),
+            "background shutdown may take longer than five seconds within the process-stop budget"
+        );
+        writeln!(writer, "{event}")?;
+        tokio::time::advance(Duration::from_millis(10)).await;
+        assert_eq!(release.await?, event, "retain the actual new shutdown witness");
+        assert!(!barrier.exists(), "shutdown must release the synthetic hold");
+        Ok(())
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn heal_commit_barrier_missing_shutdown_fails_without_releasing() -> Result<(), Box<dyn Error + Send + Sync>> {
+        let directory = tempfile::tempdir()?;
+        let log = directory.path().join("server.log");
+        let barrier = directory.path().join("barrier");
+        std::fs::write(&log, "")?;
+        std::fs::write(&barrier, "bucket/prefix/")?;
+        let error = release_heal_commit_barrier_after_shutdown(&log, 0, &barrier)
+            .await
+            .expect_err("missing shutdown evidence must fail the fixture");
+        assert!(error.is::<tokio::time::error::Elapsed>());
+        assert!(barrier.is_file(), "a missing shutdown witness must not release repairs");
+        Ok(())
+    }
+
     async fn run_cluster_root_heal_interruption(scenario: InterruptionScenario) -> Result<(), Box<dyn Error + Send + Sync>> {
         let server_binary = rustfs_binary_path();
         let evidence_case = match scenario {
@@ -1791,6 +1874,7 @@ mod tests {
         let mut created_online_objects = Vec::with_capacity(online_object_count);
         let mut expected_manifests = Vec::with_capacity(online_object_count);
         let mut unclean_shutdown_marker_observed = None;
+        let mut shutdown_barrier_release = None;
         let mut attempt_count = 0usize;
         let max_online_attempts = online_object_count.saturating_mul(topology.total_drives().max(1));
         while expected_manifests.len() < online_object_count {
@@ -2373,7 +2457,21 @@ mod tests {
                     | InterruptionScenario::BackgroundTargetRestartEc84MultiSet
                     | InterruptionScenario::BackgroundCoordinatorRestart
             ) {
-                cluster.stop_node_gracefully(interruption_node).await?;
+                if interruption_node == 1
+                    && let Some(barrier) = &commit_barrier
+                {
+                    let target_log = PathBuf::from(format!("{log_dir}/node1.log"));
+                    let log_offset = target_log.metadata()?.len();
+                    let (stopped, released) = tokio::join!(
+                        cluster.stop_node_gracefully(interruption_node),
+                        release_heal_commit_barrier_after_shutdown(&target_log, log_offset, barrier),
+                    );
+                    // Await both branches even on error so the child is reaped.
+                    stopped?;
+                    shutdown_barrier_release = Some(released?);
+                } else {
+                    cluster.stop_node_gracefully(interruption_node).await?;
+                }
             } else {
                 cluster.stop_node(interruption_node)?;
             }
@@ -2415,7 +2513,9 @@ mod tests {
                     }
                 }
             }
-            if let Some(barrier) = &commit_barrier {
+            if shutdown_barrier_release.is_none()
+                && let Some(barrier) = &commit_barrier
+            {
                 std::fs::remove_file(barrier)?;
             }
             cluster.start_node_from_binary(interruption_node, &server_binary).await?;
@@ -2746,6 +2846,7 @@ mod tests {
                 "all_peers_bound_to_generation_window": true,
                 "pid_before": target_pid, "pid_after": restarted_pid,
                 "unclean_shutdown_marker": unclean_shutdown_marker_observed.unwrap_or(false),
+                "graceful_barrier_release": shutdown_barrier_release,
                 "objects": evidence_objects, "node_listings": node_listings,
             });
             if let Some(outage_write) = outage_write_diagnostic {

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

@@ -5745,6 +5745,7 @@ mod tests {
         object_read_all_disks: Arc<StdMutex<Vec<String>>>,
         format_data: Bytes,
         read_all_data: Bytes,
+        fail_format_reads: Arc<AtomicBool>,
         next_format_read_gate: Arc<StdMutex<Option<Arc<TestFormatReadGate>>>>,
         unsupported_methods: Arc<StdMutex<Vec<String>>>,
     }
@@ -5764,6 +5765,7 @@ mod tests {
                 object_read_all_disks: Arc::default(),
                 format_data,
                 read_all_data,
+                fail_format_reads: Arc::default(),
                 next_format_read_gate: Arc::default(),
                 unsupported_methods: Arc::default(),
             }
@@ -5887,6 +5889,9 @@ mod tests {
                                         gate.entered.notify_one();
                                         gate.release.cancelled().await;
                                     }
+                                    if peer.fail_format_reads.load(Ordering::Acquire) {
+                                        return Err(Status::unavailable("format read temporarily unavailable"));
+                                    }
                                     peer.format_data.clone()
                                 } else {
                                     peer.object_read_all_disks
@@ -8852,6 +8857,124 @@ mod tests {
         peer.stop().await;
     }
 
+    #[tokio::test]
+    #[serial(remote_disk_recovery_probe)]
+    async fn endpoint_reconnect_failure_preserves_registered_remote_recovery() {
+        temp_env::async_with_vars(
+            [
+                (crate::disk::disk_store::ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, Some("true")),
+                (rustfs_config::ENV_DRIVE_RETURNING_SUCCESS_THRESHOLD, Some("1")),
+            ],
+            async {
+                runtime_sources::ensure_test_rpc_secret();
+                let mut reference = crate::layout::format::FormatV3::new(1, 1);
+                reference.erasure.this = reference.erasure.sets[0][0];
+                let peer = TestGrpcPeer::spawn(
+                    Bytes::from(reference.to_json().expect("format should serialize")),
+                    Bytes::from_static(b"recovered-peer-data"),
+                )
+                .await
+                .expect("authenticated regression peer is required");
+                let set = missing_read_set(&peer, reference).await;
+                let disk = crate::disk::new_disk(
+                    &set.set_endpoints[0],
+                    &DiskOption {
+                        cleanup: false,
+                        health_check: true,
+                    },
+                )
+                .await
+                .expect("registered remote disk should construct");
+                let crate::disk::Disk::Remote(remote) = disk.as_ref() else {
+                    panic!("fixture must create a remote disk");
+                };
+                assert!(remote.health_check);
+                remote.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+                set.disks.write().await[0] = Some(Arc::clone(&disk));
+
+                peer.peer.fail_format_reads.store(true, Ordering::Release);
+                set.connect_disks().await;
+                peer.peer.fail_format_reads.store(false, Ordering::Release);
+                let cancelled_after_failure = remote.cancel_token.is_cancelled();
+                assert!(peer.peer.read_all_calls() > 0, "reconnect must attempt the failing format RPC");
+                assert!(
+                    Arc::ptr_eq(&disk, set.disks.read().await[0].as_ref().expect("existing slot must remain")),
+                    "a failed reconnect must not publish its unvalidated candidate"
+                );
+                remote.spawn_recovery_monitor_if_needed();
+                let recovery = time::timeout(Duration::from_secs(5), async {
+                    while !disk.is_online().await {
+                        tokio::task::yield_now().await;
+                    }
+                    disk.read_all("bucket", "object").await
+                })
+                .await;
+                disk.close().await.expect("stop the registered disk's monitoring");
+                peer.stop().await;
+                assert!(!cancelled_after_failure, "failed replacement must retain the registered monitor");
+                assert_eq!(
+                    recovery
+                        .expect("failed reconnect must leave the registered disk able to recover without another sweep")
+                        .expect("the recovered registered disk must serve real I/O"),
+                    Bytes::from_static(b"recovered-peer-data")
+                );
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial(remote_disk_recovery_probe)]
+    async fn endpoint_reconnect_stops_old_remote_monitor_only_after_publication() {
+        runtime_sources::ensure_test_rpc_secret();
+        let mut reference = crate::layout::format::FormatV3::new(1, 1);
+        reference.erasure.this = reference.erasure.sets[0][0];
+        let peer = TestGrpcPeer::spawn(
+            Bytes::from(reference.to_json().expect("format should serialize")),
+            Bytes::from_static(b"replacement-peer-data"),
+        )
+        .await
+        .expect("authenticated regression peer is required");
+        let set = missing_read_set(&peer, reference).await;
+        let disk = crate::disk::new_disk(
+            &set.set_endpoints[0],
+            &DiskOption {
+                cleanup: false,
+                health_check: true,
+            },
+        )
+        .await
+        .expect("registered remote disk should construct");
+        let crate::disk::Disk::Remote(remote) = disk.as_ref() else {
+            panic!("fixture must create a remote disk");
+        };
+        remote.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+        set.disks.write().await[0] = Some(Arc::clone(&disk));
+        let gate = peer.peer.pause_next_format_read();
+        let pending_set = Arc::clone(&set);
+        let pending = tokio::spawn(async move { pending_set.connect_disks().await });
+        time::timeout(Duration::from_secs(5), gate.entered.notified())
+            .await
+            .expect("replacement must reach the format validation barrier");
+        let cancelled_before_publication = remote.cancel_token.is_cancelled();
+        gate.release.cancel();
+        pending.await.expect("validated replacement must finish publication");
+        let replacement = set.disks.read().await[0].clone().expect("replacement must occupy the slot");
+        let readback = replacement.read_all("bucket", "object").await;
+        replacement.close().await.expect("stop replacement monitoring");
+        peer.stop().await;
+        assert!(
+            !cancelled_before_publication,
+            "the registered monitor must survive pending format validation"
+        );
+        assert!(!Arc::ptr_eq(&disk, &replacement), "validated replacement must be published");
+        assert!(remote.cancel_token.is_cancelled(), "the displaced monitor must stop after publication");
+        assert_eq!(
+            readback.expect("replacement must serve real I/O"),
+            Bytes::from_static(b"replacement-peer-data")
+        );
+    }
+
     #[tokio::test]
     #[serial(remote_disk_recovery_probe)]
     async fn endpoint_reconnect_admits_ready_drive_while_another_format_probe_is_blocked() {

crates/ecstore/src/core/pools.rs

@@ -8380,8 +8380,9 @@ where
         })
     });
     let result = save_config_with_opts_and_metadata(pool, object, data, &opts).await;
-    let may_have_mutated = !matches!(&result, Err(Error::PreconditionFailed));
-    if !may_have_mutated {
+    let write_not_started = matches!(&result, Err(Error::Io(inner)) if inner.get_ref().is_some_and(|source| source.is::<crate::set_disk::SystemMetadataWriteNotStarted>()));
+    let may_have_mutated = !write_not_started && !matches!(&result, Err(Error::PreconditionFailed));
+    if matches!(&result, Err(Error::PreconditionFailed)) {
         record_pool_meta_stale_write_rejection(phase);
     }
     let mut source = None;
@@ -8392,6 +8393,8 @@ where
             source = Some(Arc::clone(&original));
             if matches!(original.as_ref(), Error::PreconditionFailed) {
                 Err(Error::PreconditionFailed)
+            } else if write_not_started {
+                Err(Error::other(original))
             } else {
                 Err(Error::other(pool_metadata_error(
                     crate::error::PoolMetadataFailure::TransactionUnknown,
@@ -30826,6 +30829,154 @@ mod pools_tests {
         );
     }
 
+    #[tokio::test]
+    async fn pool_meta_prewrite_quorum_rejection_allows_retry_after_disks_return() {
+        let (dirs, disks, pool) = crate::set_disk::hermetic_set_disks_isolated(4).await;
+        pool.disks.write().await[2..].fill(None);
+        let state = super::PoolMetaWriteState::default();
+        let data =
+            super::initialized_pool_meta_identity_for_test(uuid::Uuid::new_v4(), 1).expect("encode the deployment identity");
+        let mut transaction = state.arm_transaction();
+        save_pool_meta_object_cas(
+            pool.clone(),
+            POOL_META_IDENTITY_NAME,
+            data.clone(),
+            &PoolMetaCasToken::Missing,
+            &PoolMetaPersistenceFence::OfflineRecovery,
+            "identity_cas",
+            &mut transaction,
+        )
+        .await
+        .expect_err("two of four disks cannot satisfy the identity write quorum");
+        drop(transaction);
+        for dir in &dirs {
+            assert!(
+                !dir.path()
+                    .join(crate::disk::RUSTFS_META_BUCKET)
+                    .join(POOL_META_IDENTITY_NAME)
+                    .exists(),
+                "the rejected write must not create an identity replica"
+            );
+        }
+        state
+            .ensure_write_safe("retry after disks return")
+            .expect("a rejected inline write must not poison the startup transaction");
+
+        let ordinary_error = crate::config::com::save_config_with_opts(
+            pool.clone(),
+            "ordinary-config",
+            data.clone(),
+            &ObjectOptions {
+                max_parity: true,
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect_err("ordinary writes must still fail without quorum");
+        assert!(
+            matches!(ordinary_error, Error::InsufficientWriteQuorum(_, _)),
+            "ordinary callers must retain their existing quorum error"
+        );
+
+        *pool.disks.write().await = disks.into_iter().map(Some).collect();
+        let mut retry = state.arm_transaction();
+        save_pool_meta_object_cas(
+            pool.clone(),
+            POOL_META_IDENTITY_NAME,
+            data.clone(),
+            &PoolMetaCasToken::Missing,
+            &PoolMetaPersistenceFence::OfflineRecovery,
+            "identity_cas",
+            &mut retry,
+        )
+        .await
+        .expect("the same transaction must succeed when the disks return");
+        retry.disarm();
+        drop(retry);
+        assert_eq!(
+            crate::config::com::read_config(pool, POOL_META_IDENTITY_NAME)
+                .await
+                .expect("read back the committed identity"),
+            data
+        );
+        state
+            .ensure_write_safe("after successful retry")
+            .expect("retry must leave writes enabled");
+    }
+
+    #[tokio::test]
+    async fn pool_meta_prewrite_quorum_rejection_keeps_an_earlier_write_armed() {
+        let (_dirs, disks, pool) = crate::set_disk::hermetic_set_disks_isolated(4).await;
+        let state = super::PoolMetaWriteState::default();
+        let data =
+            super::initialized_pool_meta_identity_for_test(uuid::Uuid::new_v4(), 1).expect("encode the deployment identity");
+        let mut transaction = state.arm_transaction();
+        let written = save_pool_meta_object_cas(
+            pool.clone(),
+            POOL_META_IDENTITY_NAME,
+            data.clone(),
+            &PoolMetaCasToken::Missing,
+            &PoolMetaPersistenceFence::OfflineRecovery,
+            "prepare_cas",
+            &mut transaction,
+        )
+        .await
+        .expect("the earlier phase must really commit");
+        pool.disks.write().await[2..].fill(None);
+        save_pool_meta_object_cas(
+            pool.clone(),
+            POOL_META_IDENTITY_NAME,
+            data.clone(),
+            &PoolMetaCasToken::Existing(written.etag.expect("committed identity has an ETag")),
+            &PoolMetaPersistenceFence::OfflineRecovery,
+            "identity_cas",
+            &mut transaction,
+        )
+        .await
+        .expect_err("the next phase must fail without a write quorum");
+        drop(transaction);
+        *pool.disks.write().await = disks.into_iter().map(Some).collect();
+        assert_eq!(
+            crate::config::com::read_config(pool, POOL_META_IDENTITY_NAME)
+                .await
+                .expect("the earlier commit must survive the rejected write"),
+            data
+        );
+        let error = state
+            .ensure_write_safe("after a committed phase")
+            .expect_err("a later rejection cannot discharge the earlier write");
+        assert_eq!(
+            error.pool_metadata_failure().expect("retain typed uncertainty").kind,
+            crate::error::PoolMetadataFailure::TransactionUnknown
+        );
+        assert_eq!(error.pool_metadata_failure().expect("retain the earlier mutation").phase, "prepare_cas");
+    }
+
+    #[tokio::test]
+    async fn pool_meta_prewrite_noninline_failure_keeps_transaction_armed() {
+        let (_dirs, _disks, pool) = crate::set_disk::hermetic_set_disks_isolated(4).await;
+        pool.disks.write().await[2..].fill(None);
+        let state = super::PoolMetaWriteState::default();
+        let mut transaction = state.arm_transaction();
+        save_pool_meta_object_cas(
+            pool,
+            POOL_META_IDENTITY_NAME,
+            vec![0x51; 2 * 1024 * 1024],
+            &PoolMetaCasToken::Missing,
+            &PoolMetaPersistenceFence::OfflineRecovery,
+            "identity_cas",
+            &mut transaction,
+        )
+        .await
+        .expect_err("the noninline write must fail without quorum");
+        drop(transaction);
+        assert!(
+            state.ensure_write_safe("after writer setup").is_err(),
+            "a failure after disk writers were created must remain fail-closed"
+        );
+    }
+
     #[derive(Clone, Copy, Debug, PartialEq, Eq)]
     enum IdentityRepairFault {
         Healthy,

crates/ecstore/src/disk/local.rs

@@ -16260,7 +16260,7 @@ mod test {
                 assert!(slots[0][0][1].is_some());
                 drop(slots);
                 drop(entries);
-                let (disks, errors) = crate::store::init_format::init_disks(&endpoints, &options).await;
+                let (disks, errors) = crate::store::init_format::init_disks(&ctx, &endpoints, &options).await;
                 assert!(disks[0].is_none());
                 assert!(disks[1].is_some());
                 assert_eq!(errors[0], Some(DiskError::FileAccessDenied));

crates/ecstore/src/set_disk/mod.rs

@@ -5019,6 +5019,11 @@ impl DiskHealthEntry {
 #[error("committed system metadata write requires repair by its owning writer")]
 pub(crate) struct SystemMetadataRepairRequired;
 
+/// An inline metadata write was rejected before staging or commit began.
+#[derive(Debug, thiserror::Error)]
+#[error("system metadata write rejected before staging: {0}")]
+pub(crate) struct SystemMetadataWriteNotStarted(#[source] pub(crate) Error);
+
 impl SetDisks {
     pub(in crate::set_disk) async fn persist_delete_marker_purge(
         &self,

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

@@ -375,7 +375,10 @@ impl SetDisks {
                         return None;
                     }
 
-                    let _ = disk.close().await;
+                    // A failed remote reconnect must retain its existing recovery monitor.
+                    if disk.is_local() {
+                        let _ = disk.close().await;
+                    }
                 }
 
                 let endpoint = self.set_endpoints.get(i)?;
@@ -497,19 +500,26 @@ impl SetDisks {
         let _ = new_disk.set_disk_id(Some(fm.erasure.this)).await;
 
         if !new_disk.is_local() {
-            let mut disks = self.disks.write().await;
-            let slot = &mut disks[disk_idx];
-            let unchanged = match (slot.as_ref(), previous_disk.as_ref()) {
-                (None, None) => true,
-                (Some(current), Some(previous)) => Arc::ptr_eq(current, previous),
-                _ => false,
+            let displaced = {
+                let mut disks = self.disks.write().await;
+                let slot = &mut disks[disk_idx];
+                let unchanged = match (slot.as_ref(), previous_disk.as_ref()) {
+                    (None, None) => true,
+                    (Some(current), Some(previous)) => Arc::ptr_eq(current, previous),
+                    _ => false,
+                };
+                // A GET probe or another reconnect may have filled this slot while
+                // format I/O was pending. Only the successful publisher owns the
+                // displaced monitor; a failed or superseded probe must retain it.
+                if unchanged {
+                    new_disk.enable_health_check();
+                    slot.replace(new_disk)
+                } else {
+                    None
+                }
             };
-            // A GET probe or another reconnect may have filled this slot while
-            // format I/O was pending. Do not replace its handle or start a
-            // second health monitor for an unpublished handle.
-            if unchanged {
-                new_disk.enable_health_check();
-                *slot = Some(new_disk);
+            if let Some(disk) = displaced {
+                let _ = disk.close().await;
             }
             return false;
         }

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

@@ -4526,7 +4526,16 @@ impl SetDisks {
                     "Set disk write quorum unavailable"
                 );
                 if let Some(write_err) = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, write_quorum) {
-                    return Err(to_object_err(write_err.into(), vec![bucket, object]));
+                    let error = to_object_err(write_err.into(), vec![bucket, object]);
+                    if direct_inline_commit
+                        && is_meta_bucketname(bucket)
+                        && opts.write_completion == crate::object_api::WriteCompletion::TailDrainedWithOwnerRepair
+                    {
+                        // Inline writer setup only checks disk availability;
+                        // no shard or metadata write has been dispatched yet.
+                        return Err(Error::other(crate::set_disk::SystemMetadataWriteNotStarted(error)));
+                    }
+                    return Err(error);
                 }
 
                 return Err(Error::other(format!("not enough disks to write: {errors:?}")));

crates/ecstore/src/store/init.rs

@@ -540,6 +540,7 @@ impl ECStore {
             // failures can still spawn recovery probes for peers that come up
             // after this node.
             let (mut disks, errs) = init_format::init_disks(
+                &instance_ctx,
                 &pool_eps.endpoints,
                 &DiskOption {
                     cleanup: true,

crates/ecstore/src/store/init_format.rs

@@ -44,12 +44,32 @@ const LEGACY_FORMAT_READ_CONCURRENCY: usize = 16;
 const LEGACY_FORMAT_BASE_MAX_BYTES: usize = 16 * 1024;
 const LEGACY_FORMAT_MAX_BYTES_PER_DISK: usize = 64;
 
-pub async fn init_disks(eps: &Endpoints, opt: &DiskOption) -> (Vec<Option<DiskStore>>, Vec<Option<DiskError>>) {
+pub async fn init_disks(
+    instance_ctx: &InstanceContext,
+    eps: &Endpoints,
+    opt: &DiskOption,
+) -> (Vec<Option<DiskStore>>, Vec<Option<DiskError>>) {
     let mut futures = Vec::with_capacity(eps.as_ref().len());
+    let local_disk_map = instance_ctx.local_disk_map();
+    let local_disks = local_disk_map.read().await;
 
     for ep in eps.as_ref().iter() {
-        futures.push(new_disk(ep, opt));
+        // Foundation disks may already serve RPC writes. Opening them again
+        // with startup cleanup can move live staging out from under a writer.
+        let registered = if ep.is_local {
+            local_disks.get(&ep.to_string()).cloned().flatten()
+        } else {
+            None
+        };
+        futures.push(async move {
+            match registered {
+                Some(disk) if disk.endpoint() == *ep => Ok(disk),
+                Some(_) => Err(DiskError::InconsistentDisk),
+                None => new_disk(ep, opt).await,
+            }
+        });
     }
+    drop(local_disks);
 
     let mut res = Vec::with_capacity(eps.as_ref().len());
     let mut errors = Vec::with_capacity(eps.as_ref().len());
@@ -1030,6 +1050,7 @@ mod tests {
         }
 
         let (disks, errors) = init_disks(
+            &InstanceContext::new(),
             &Endpoints::from(endpoints),
             &DiskOption {
                 cleanup: false,
@@ -1049,6 +1070,91 @@ mod tests {
         (temp_dir, disks)
     }
 
+    #[tokio::test]
+    async fn init_disks_preserves_live_staging_on_registered_local_disks() {
+        let (temp_dir, disks) = local_disks(1).await;
+        let registered = disks[0].as_ref().expect("registered disk should exist");
+        let instance_ctx = Arc::new(InstanceContext::new());
+        record_local_disks(&instance_ctx, vec![registered.clone()]).await;
+        let staging = temp_dir.path().join("disk-0/.rustfs.sys/tmp/in-flight/xl.meta");
+        tokio::fs::create_dir_all(staging.parent().expect("staging has a parent"))
+            .await
+            .expect("create staging directory");
+        tokio::fs::write(&staging, b"in-flight pool metadata")
+            .await
+            .expect("stage metadata through the registered disk");
+
+        let (initialized, errors) = init_disks(
+            &instance_ctx,
+            &Endpoints::from(vec![registered.endpoint()]),
+            &DiskOption {
+                cleanup: true,
+                health_check: false,
+            },
+        )
+        .await;
+
+        assert!(errors.iter().all(Option::is_none), "disk initialization failed: {errors:?}");
+        assert_eq!(
+            tokio::fs::read(&staging)
+                .await
+                .expect("startup must preserve live RPC staging"),
+            b"in-flight pool metadata"
+        );
+        assert!(Arc::ptr_eq(registered, initialized[0].as_ref().expect("initialized disk should exist")));
+    }
+
+    #[tokio::test]
+    async fn init_disks_cleans_stale_staging_without_a_registered_disk() {
+        let temp_dir = tempfile::tempdir().expect("create disk root");
+        let endpoint =
+            Endpoint::try_from(temp_dir.path().to_str().expect("disk root should be UTF-8")).expect("endpoint should parse");
+        let staging = temp_dir.path().join(".rustfs.sys/tmp/stale/xl.meta");
+        tokio::fs::create_dir_all(staging.parent().expect("staging has a parent"))
+            .await
+            .expect("create stale staging directory");
+        tokio::fs::write(&staging, b"stale pool metadata")
+            .await
+            .expect("write stale staging");
+
+        let (disks, errors) = init_disks(
+            &InstanceContext::new(),
+            &Endpoints::from(vec![endpoint]),
+            &DiskOption {
+                cleanup: true,
+                health_check: false,
+            },
+        )
+        .await;
+
+        assert!(errors.iter().all(Option::is_none), "disk initialization failed: {errors:?}");
+        assert!(disks[0].is_some());
+        assert!(!staging.exists(), "a fresh disk open must still clear stale staging");
+    }
+
+    #[tokio::test]
+    async fn init_disks_rejects_registered_disk_topology_mismatch() {
+        let (_temp_dir, disks) = local_disks(1).await;
+        let registered = disks[0].as_ref().expect("registered disk should exist");
+        let instance_ctx = Arc::new(InstanceContext::new());
+        record_local_disks(&instance_ctx, vec![registered.clone()]).await;
+        let mut endpoint = registered.endpoint();
+        endpoint.set_pool_index(1);
+
+        let (initialized, errors) = init_disks(
+            &instance_ctx,
+            &Endpoints::from(vec![endpoint]),
+            &DiskOption {
+                cleanup: true,
+                health_check: false,
+            },
+        )
+        .await;
+
+        assert!(initialized[0].is_none(), "a registered disk must not be reopened in a different slot");
+        assert_eq!(errors[0], Some(DiskError::InconsistentDisk));
+    }
+
     async fn format_bytes(disks: &[Option<DiskStore>]) -> Vec<Option<Vec<u8>>> {
         let mut snapshots = Vec::with_capacity(disks.len());
         for disk in disks {
@@ -1758,6 +1864,7 @@ mod tests {
             disk.set_disk_id(None).await.expect("test disk ID should be reset");
         }
         let (prewarm_disks, prewarm_errors) = init_disks(
+            &InstanceContext::new(),
             &prewarm_endpoints,
             &DiskOption {
                 cleanup: false,

crates/ecstore/src/store/mod.rs

@@ -3295,6 +3295,7 @@ mod tests {
         init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
 
         let (mut disks, errs) = init_disks(
+            &crate::runtime::global::current_ctx(),
             &endpoint_pools.as_ref().first().expect("pool endpoints").endpoints,
             &DiskOption {
                 cleanup: true,