chore: integrate main for recovery verification

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

@@ -15737,7 +15737,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,